#!/usr/bin/env python3 """Enrich a list of UK company names or Companies House numbers from the official register (CSV in, CSV out). Reads companies_to_check.csv (one company per row in a column called "lookup": a company number such as 00445790 / SC123456, or a company name), runs the Apify Actor nexgensignal/uk-company-status-records for the rows not yet enriched (one run for numbers, one for names, side by side), and upserts one result row per lookup into enriched_companies.csv: status, type, incorporation date, registered office, SIC codes, accounts and confirmation-statement dates, and a link to Companies House. A state file remembers started runs until their results are saved, so an interrupted save is finished from those runs instead of paying for new ones. Needs Python 3.8+ and nothing else. Set your own token first: export APIFY_TOKEN=your_apify_token (Windows: set APIFY_TOKEN=...) python uk_company_enrich_csv.py Options: --in companies_to_check.csv --out enriched_companies.csv --per-name 5 --refresh --cap 2.00 Recovery: --resume-numbers RUN_ID / --resume-names RUN_ID, --start-new (see the tutorial). Company numbers stay text: open the CSVs with the lookup column set to Text if your spreadsheet strips leading zeros. """ import argparse, csv, datetime as dt, http.client, json, os, re, sys, time, urllib.error, urllib.parse, urllib.request from concurrent.futures import ThreadPoolExecutor ACTOR = "nexgensignal~uk-company-status-records" COLUMNS = ["lookup", "lookup_type", "result", "company_number", "company_name", "company_status", "company_category", "incorporation_date", "registered_address", "reg_post_town", "reg_postcode", "sic_codes", "accounts_next_due", "accounts_last_made_up", "conf_stmt_next_due", "previous_names", "other_matches", "companies_house_url", "snapshot_date", "checked_at"] NUMBER = re.compile(r"^(?:[A-Z]{2}\d{6}|\d{1,8})$") # --- apify run core (shared by the three example scripts; edit _runcore.py and run build_scripts.py) --- import threading _STATE_LOCK = threading.Lock() API = os.environ.get("APIFY_API_BASE", "https://api.apify.com/v2") CONSOLE_RUNS = "https://console.apify.com/actors/runs" class StartUnknown(Exception): """The request that starts a paid run was sent, but no answer came back.""" def api(method, path, token, body=None, query=None): """Reads (GET) are retried on network errors and busy/5xx answers. A run start (POST) is sent exactly once: the Apify API has no idempotency key for starting runs, so repeating it could start a second paid run.""" url = f"{API}{path}" + ("?" + urllib.parse.urlencode(query) if query else "") data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(url, data=data, method=method, headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/json"}) tries = 5 if method == "GET" else 1 for attempt in range(tries): try: with urllib.request.urlopen(req, timeout=120) as r: return json.loads(r.read() or b"null") except urllib.error.HTTPError as e: if e.code == 404 and "/records/" in path: return None # no such record (for example a run that failed on bad input) if method == "GET" and e.code in (429, 500, 502, 503, 504) and attempt < tries - 1: time.sleep(2 ** attempt); continue if method != "GET" and e.code >= 500: raise StartUnknown(f"HTTP {e.code}") # the server may have created the run before failing raise SystemExit(f"Apify API refused {method} {path}: HTTP {e.code} {e.read()[:300]!r}") except (urllib.error.URLError, OSError, http.client.HTTPException) as e: if method == "GET" and attempt < tries - 1: time.sleep(2 ** attempt); continue if method != "GET": raise StartUnknown(f"{type(e).__name__}: {e}") raise def _unknown_message(job, at, extra=""): return (f"START OUTCOME UNKNOWN ({job}): a request to start a paid run was sent at {at} but no answer came " f"back, so it is not known whether Apify started that run. Nothing was started again. {extra}" f"Next: run this script again (it looks for that run and reuses it if found), or check {CONSOLE_RUNS} " f"and rerun with --resume-run RUN_ID to collect that run's results without paying again, or with " f"--start-new if you are sure no run was started.") def find_runs_since(token, actor, since_iso, run_input): """Runs of this Actor (yours) started since the uncertain request, whose saved input equals ours.""" since = dt.datetime.fromisoformat(since_iso) - dt.timedelta(seconds=120) # allow for clock differences runs = (api("GET", f"/acts/{actor}/runs", token, query={"desc": "true", "limit": 25}) or {}).get("data", {}) found = [] for r in runs.get("items", []): started = dt.datetime.fromisoformat(r["startedAt"].replace("Z", "+00:00")) if started < since: continue saved = api("GET", f"/key-value-stores/{r['defaultKeyValueStoreId']}/records/INPUT", token) if saved == run_input: found.append(r["id"]) return found def start_or_resume(token, actor, run_input, cap_usd, job_state, save, resume_id=None, start_new=False, job="run"): """Return the finished run for this job. Reuses a run already started (saved run ID, --resume-run, or a run found after an uncertain start) instead of paying for a new one. job_state is saved to disk by save().""" now = dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds") run = None if resume_id: run = api("GET", f"/actor-runs/{resume_id}", token)["data"] if run["actId"] != api("GET", f"/acts/{actor}", token)["data"]["id"]: raise SystemExit(f"Run {resume_id} belongs to a different Actor; nothing was done.") elif job_state.get("pending_run"): run = api("GET", f"/actor-runs/{job_state['pending_run']}", token)["data"] print(f"Reusing run {run['id']} started earlier (its results were not saved yet); no new run started.") elif job_state.get("start_attempt") and not start_new: att = job_state["start_attempt"] found = find_runs_since(token, actor, att["at"], att["input"]) if len(found) == 1: run = api("GET", f"/actor-runs/{found[0]}", token)["data"] print(f"Found run {run['id']} started by the earlier uncertain request; reusing it, no new run started.") else: extra = (f"{len(found)} matching runs were found: {', '.join(found)}. " if found else "No matching run was found yet. ") raise SystemExit(_unknown_message(job, att["at"], extra)) if run is None: job_state.pop("start_attempt", None) job_state["start_attempt"] = {"at": now, "input": run_input} save() try: run = api("POST", f"/acts/{actor}/runs", token, run_input, {"waitForFinish": 60, "maxTotalChargeUsd": cap_usd})["data"] except StartUnknown as e: raise SystemExit(_unknown_message(job, now, f"({e}) ")) job_state.pop("start_attempt", None) job_state["pending_run"] = run["id"] # kept until this run's results are saved save() while run["status"] in ("READY", "RUNNING", "TIMING-OUT", "ABORTING"): run = api("GET", f"/actor-runs/{run['id']}", token, query={"waitForFinish": 60})["data"] return run def run_results(token, run): items = api("GET", f"/datasets/{run['defaultDatasetId']}/items", token, query={"clean": "true", "format": "json"}) receipt = api("GET", f"/key-value-stores/{run['defaultKeyValueStoreId']}/records/RUN_RECEIPT", token) or {} return items or [], receipt def load_state(path): return json.load(open(path)) if os.path.exists(path) else {} def save_state(path, state): with _STATE_LOCK: # two runs (UK numbers + names) may save at the same moment tmp = path + ".tmp" with open(tmp, "w") as f: json.dump(state, f, indent=1) os.replace(tmp, path) def outcome(run, n_items): """complete | no_match | partial | source_unavailable | failed — from the run status and its status message.""" msg = run.get("statusMessage") or "" if run["status"] != "SUCCEEDED": return "failed" if msg.startswith("SOURCE UNAVAILABLE"): return "source_unavailable" if (msg.startswith(("PARTIAL", "CAPPED", "INCOMPLETE SEARCH")) or "PARTIAL SEARCH" in msg or msg.startswith("The sources that answered")): return "partial" return "no_match" if n_items == 0 else "complete" def read_csv(path): if not os.path.exists(path): return [] with open(path, newline="", encoding="utf-8-sig") as f: return list(csv.DictReader(f)) def write_csv(path, rows, columns): tmp = path + ".tmp" with open(tmp, "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=columns, extrasaction="ignore") w.writeheader() w.writerows(rows) os.replace(tmp, path) # --- end of apify run core --- def norm_name(s): s = (s or "").upper().replace("&", " AND ") s = re.sub(r"\bPUBLIC LIMITED COMPANY\b", "PLC", s) s = re.sub(r"\bLIMITED\b", "LTD", s) return re.sub(r"[^A-Z0-9]+", "", s) def pad(n): return n.zfill(8) if n.isdigit() else n def company_fields(c): addr = ", ".join(x for x in (c.get("reg_address_line1"), c.get("reg_address_line2")) if x) prev = c.get("previous_names") if isinstance(prev, list): prev = "; ".join(p.get("name", str(p)) if isinstance(p, dict) else str(p) for p in prev) return {"company_number": c.get("company_number"), "company_name": c.get("company_name"), "company_status": c.get("company_status"), "company_category": c.get("company_category"), "incorporation_date": c.get("incorporation_date"), "registered_address": addr, "reg_post_town": c.get("reg_post_town"), "reg_postcode": c.get("reg_postcode"), "sic_codes": c.get("sic_codes"), "accounts_next_due": c.get("accounts_next_due"), "accounts_last_made_up": c.get("accounts_last_made_up"), "conf_stmt_next_due": c.get("conf_stmt_next_due"), "previous_names": prev or "", "companies_house_url": c.get("companies_house_url"), "snapshot_date": c.get("snapshot_date")} def main(): p = argparse.ArgumentParser(description=__doc__.split("\n")[0]) p.add_argument("--in", dest="inp", default="companies_to_check.csv") p.add_argument("--out", default="enriched_companies.csv") p.add_argument("--per-name", type=int, default=5, help="most register matches paid for per name searched") p.add_argument("--refresh", action="store_true", help="re-check lookups already in the output file") p.add_argument("--cap", type=float, default=2.00, help="maximum charge for each run, USD") p.add_argument("--resume-numbers", default=None, help="collect this existing numbers run instead of starting one") p.add_argument("--resume-names", default=None, help="collect this existing names run instead of starting one") p.add_argument("--start-new", action="store_true", help="after an uncertain start you have checked: start anew") a = p.parse_args() token = os.environ.get("APIFY_TOKEN") if not token: sys.exit("Set APIFY_TOKEN to your own Apify API token first.") state_path = a.out + ".state.json" state = load_state(state_path) jobs_state = state.setdefault("jobs", {}) save = lambda: save_state(state_path, state) out_rows = read_csv(a.out) unfinished = {k for k, j in jobs_state.items() if j.get("pending_run") or j.get("start_attempt")} if unfinished or a.resume_numbers or a.resume_names: plan = {k: j["lookups"] for k, j in jobs_state.items() if j.get("lookups")} # finish what was started if not plan: sys.exit("Nothing saved to resume: --resume-numbers/--resume-names finish a run this script started.") else: wanted = list(dict.fromkeys((r.get("lookup") or "").strip() for r in read_csv(a.inp))) done = {r["lookup"] for r in out_rows} todo = [w for w in wanted if w and (a.refresh or w not in done)] if not todo: print("Nothing to do: every lookup is already in", a.out, "(use --refresh to re-check)"); return numbers = [w for w in todo if NUMBER.match(w.upper().replace(" ", ""))] names = [w for w in todo if w not in numbers] plan = {k: v for k, v in (("numbers", numbers), ("names", names)) if v} state["jobs"] = jobs_state = {k: {"lookups": v} for k, v in plan.items()} save() def actor_input(kind, lookups): # Numbers and names go in two runs: within one run the Actor combines its filters (number AND name). if kind == "numbers": return {"companyNumbers": [w.upper().replace(" ", "") for w in lookups], "maxRecords": len(lookups)} return {"searchQueries": lookups, "maxRecords": a.per_name * len(lookups)} resume = {"numbers": a.resume_numbers, "names": a.resume_names} def do(kind): run = start_or_resume(token, ACTOR, actor_input(kind, plan[kind]), a.cap, jobs_state[kind], save, resume.get(kind), a.start_new, job=f"{kind} run") return run, *run_results(token, run) with ThreadPoolExecutor(2) as ex: done_runs = dict(zip(plan, ex.map(do, plan))) now = dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds") for k, (run, items, receipt) in done_runs.items(): msg = run.get("statusMessage") or "" if run["status"] != "SUCCEEDED" or msg.startswith("SOURCE UNAVAILABLE"): for j in jobs_state.values(): j.pop("pending_run", None) # nothing usable was bought; the next run starts fresh save() print(f"The {k} run {run['id']} did not complete; nothing written. Status:", run["status"], msg) sys.exit(2) results = [] if "numbers" in plan: run, items, receipt = done_runs["numbers"] capped = (run.get("statusMessage") or "").startswith(("CAPPED", "PARTIAL")) not_found = {pad(n) for n in receipt.get("company_numbers_not_on_register") or []} by_number = {c.get("company_number"): c for c in items} for w in plan["numbers"]: n = pad(w.upper().replace(" ", "")) row = {"lookup": w, "lookup_type": "number", "checked_at": now} if n in by_number: row.update(company_fields(by_number[n]), result="found") else: row["result"] = ("not_on_register" if n in not_found or not capped else "not_checked_limit_reached") results.append(row) if "names" in plan: run, items, receipt = done_runs["names"] capped = ((run.get("statusMessage") or "").startswith(("CAPPED", "PARTIAL")) or len(items) >= a.per_name * len(plan["names"])) for w in plan["names"]: q = w.lower() hits = [c for c in items if any(m.strip().lower() == f"name: {q}" for m in (c.get("match") or "").split(";"))] exact = [c for c in hits if norm_name(c.get("company_name")) == norm_name(w)] row = {"lookup": w, "lookup_type": "name", "checked_at": now} row["other_matches"] = "; ".join(f"{c.get('company_name')} ({c.get('company_number')})" for c in hits if c not in exact[:1]) if len(exact) == 1: row.update(company_fields(exact[0]), result="found_exact_name") elif len(exact) > 1: row["result"] = "several_exact_names" elif hits: row["result"] = "no_exact_name" else: row["result"] = "not_checked_limit_reached" if capped else "no_match" results.append(row) by_lookup = {r["lookup"]: r for r in out_rows} for r in results: r = {k: ("" if v is None else v) for k, v in r.items()} if r["lookup"] in by_lookup: by_lookup[r["lookup"]].clear(); by_lookup[r["lookup"]].update(r) else: out_rows.append(r); by_lookup[r["lookup"]] = r write_csv(a.out, out_rows, COLUMNS) # if this fails, the run IDs stay saved: rerun to finish state["jobs"] = {} # results saved: forget the runs save() tally = {} for r in results: tally[r["result"]] = tally.get(r["result"], 0) + 1 charged = sum((r[2] or {}).get("charged", len(r[1])) for r in done_runs.values()) print(f"{len(results)} lookups checked, {charged} companies charged: {tally}.") for k, (run, _, _) in done_runs.items(): print(f" {k} run {run['id']}: {run.get('statusMessage')}") if __name__ == "__main__": main()