From 39694ee2e5e1dc1b31aa52d03506c3644ce5ff0f Mon Sep 17 00:00:00 2001 From: Prad Nukala Date: Wed, 26 Aug 2026 11:54:39 -0400 Subject: [PATCH] feat(tools): add Twenty CRM sync scripts and update .gitignore rules --- tools/gather_recent_context.py | 211 +++++++++++++++++++++++ tools/security_guards.py | 13 +- tools/twenty_client.py | 100 +++++++++++ tools/twenty_enrich.py | 295 +++++++++++++++++++++++++++++++++ 4 files changed, 612 insertions(+), 7 deletions(-) create mode 100755 tools/gather_recent_context.py create mode 100644 tools/twenty_client.py create mode 100755 tools/twenty_enrich.py diff --git a/tools/gather_recent_context.py b/tools/gather_recent_context.py new file mode 100755 index 0000000..b08225d --- /dev/null +++ b/tools/gather_recent_context.py @@ -0,0 +1,211 @@ +#!/usr/bin/env python3 +"""Sync job-opportunity email context into Twenty CRM as Notes. + +Called from the /gather-recent-context command. The command owns the email +side (Fastmail MCP: search_email / read_thread) and writes the classified +threads to a JSON file; this script owns everything deterministic: + + 1. Match each email thread to an opportunity on an ACTIVE loop + (stage not in the final set) by normalized company name, breaking ties + with role-title token overlap against the opportunity name. + 2. Build one "Email Context — " note per matched opportunity with + a dated markdown timeline of the thread summaries. + 3. Dedup by thread id against job_scraper/context_sync.json so a re-run + never writes the same conversation twice. + 4. Apply via Twenty REST (TWENTY_API_KEY) with --apply, or emit + job_scraper/context_plan.json for the command to apply through the + Twenty MCP session. + +Input emails JSON (written by the command): + {"threads": [{"thread_id": "...", "company": "Acme", "role": "Backend Engineer"|null, + "date": "YYYY-MM-DD", "from": "recruiter@acme.com", + "subject": "...", "summary": "1-3 sentence summary", + "signal": "interview|assessment|offer|rejection|scheduling|other"}]} + +Usage: + python3 tools/gather_recent_context.py --emails # plan only + python3 tools/gather_recent_context.py --emails --apply # plan + REST apply + ... --snapshot crm.json # offline CRM state instead of REST fetch +""" +from __future__ import annotations + +import argparse +import json +import sys +from datetime import date +from pathlib import Path + +from twenty_enrich import norm_company, norm_title + +REPO = Path(__file__).resolve().parent.parent +STATE = REPO / "job_scraper" / "context_sync.json" +PLAN_OUT = REPO / "job_scraper" / "context_plan.json" + +# Loops still in play: everything before a terminal outcome. +ACTIVE_STAGES = {"RANKED", "DRAFTED", "APPLIED", "INTERVIEW", "OFFER"} + + +def load_state(path: Path = STATE) -> dict: + if path.exists(): + return json.loads(path.read_text(encoding="utf-8")) + return {"processed_thread_ids": [], "last_run": None} + + +def match_thread(thread: dict, opportunities: list[dict], + company_name_by_id: dict[str, str]) -> dict | None: + """Best active-loop opportunity for a thread, or None. + + Opportunity names are role-only (the ' — Company' suffix is stripped once + the company relation is set - see tools/twenty_enrich.py), so the company + match goes through ``companyId`` -> Company.name. Among a company's active + opportunities, prefer the one whose role name shares the most title tokens + with the thread's role (when the thread names one); a company with exactly + one active loop matches unconditionally. + """ + key = norm_company(thread.get("company", "")) + if not key: + return None + candidates = [o for o in opportunities + if o.get("stage") in ACTIVE_STAGES + and norm_company(company_name_by_id.get(o.get("companyId") or "", "")) == key] + if not candidates: + return None + if len(candidates) == 1 or not thread.get("role"): + return candidates[0] + role_tokens = set(norm_title(thread["role"]).split()) + + def overlap(o): + return len(role_tokens & set(norm_title(o.get("name", "")).split())) + + return max(candidates, key=overlap) + + +def build_note_markdown(threads: list[dict]) -> str: + """Dated timeline, newest first. Only classified thread data - no bodies.""" + lines = [] + for t in sorted(threads, key=lambda t: t.get("date") or "", reverse=True): + sig = t.get("signal", "other") + lines.append(f"## {t.get('date', 'undated')} — {sig}") + lines.append(f"**From:** {t.get('from', 'unknown')} · **Subject:** {t.get('subject', '')}") + lines.append("") + lines.append(t.get("summary", "").strip()) + lines.append("") + return "\n".join(lines).strip() + + +def build_plan(threads: list[dict], opportunities: list[dict], + companies: list[dict], state: dict) -> dict: + """Notes plan grouped one note per opportunity, dedup'd by thread id.""" + company_name_by_id = {c["id"]: c.get("name", "") for c in companies} + seen_ids = set(state.get("processed_thread_ids", [])) + fresh = [t for t in threads if t.get("thread_id") and t["thread_id"] not in seen_ids] + by_opp: dict[str, dict] = {} + unmatched = [] + for t in fresh: + o = match_thread(t, opportunities, company_name_by_id) + if o is None: + unmatched.append({"thread_id": t["thread_id"], + "company": t.get("company"), "subject": t.get("subject")}) + continue + slot = by_opp.setdefault(o["id"], {"opportunity_id": o["id"], + "opportunity_name": o.get("name"), + "company": company_name_by_id.get(o.get("companyId") or "", ""), + "threads": []}) + slot["threads"].append(t) + + notes = [] + for slot in by_opp.values(): + company = slot["company"] or "Unknown" + notes.append({ + "opportunity_id": slot["opportunity_id"], + "title": f"Email Context — {company} ({date.today().isoformat()})", + "markdown": build_note_markdown(slot["threads"]), + "thread_ids": [t["thread_id"] for t in slot["threads"]], + }) + return {"notes": notes, "unmatched_threads": unmatched, + "skipped_already_processed": len(threads) - len(fresh)} + + +def apply_plan(plan: dict, client) -> None: + """Create the notes + note targets via REST, then update dedup state.""" + for n in plan["notes"]: + created = client.create_batch("notes", [{ + "title": n["title"], + "bodyV2": {"markdown": n["markdown"]}, + "position": "last", + }]) + note_id = created[0]["id"] + client.create_batch("noteTargets", [{ + "noteId": note_id, + "targetOpportunityId": n["opportunity_id"], + "position": "last", + }]) + n["note_id"] = note_id + + +def commit_state(plan: dict, state: dict, path: Path = STATE) -> None: + done = set(state.get("processed_thread_ids", [])) + for n in plan["notes"]: + done.update(n["thread_ids"]) + state["processed_thread_ids"] = sorted(done) + state["last_run"] = date.today().isoformat() + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(state, indent=2), encoding="utf-8") + + +def main(argv=None) -> int: + ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + ap.add_argument("--emails", type=Path, + help="classified threads JSON written by /gather-recent-context") + ap.add_argument("--snapshot", type=Path, + help="CRM opportunities JSON (offline; else fetched via REST)") + ap.add_argument("--apply", action="store_true", + help="create the notes via Twenty REST (needs TWENTY_API_KEY)") + ap.add_argument("--commit-plan", type=Path, + help="mark an already-applied plan's thread ids as processed and exit " + "(used after the command applies the plan through the Twenty MCP session)") + ap.add_argument("--out", type=Path, default=PLAN_OUT, help="plan output path") + args = ap.parse_args(argv) + + if args.commit_plan: + plan = json.loads(args.commit_plan.read_text(encoding="utf-8")) + commit_state(plan, load_state()) + print(f"state committed for {len(plan['notes'])} notes.") + return 0 + if not args.emails: + ap.error("--emails is required unless --commit-plan is given") + + threads = json.loads(args.emails.read_text(encoding="utf-8"))["threads"] + client = None + if args.snapshot: + snap = json.loads(args.snapshot.read_text(encoding="utf-8")) + opportunities, companies = snap["opportunities"], snap["companies"] + else: + from twenty_client import TwentyClient + client = TwentyClient() + opportunities = client.list_all("opportunities") + companies = client.list_all("companies") + + state = load_state() + plan = build_plan(threads, opportunities, companies, state) + + if args.apply and plan["notes"]: + if client is None: + from twenty_client import TwentyClient + client = TwentyClient() + apply_plan(plan, client) + commit_state(plan, state) + plan["applied"] = True + + args.out.parent.mkdir(parents=True, exist_ok=True) + args.out.write_text(json.dumps(plan, indent=2), encoding="utf-8") + print(f"plan -> {args.out}") + print(f"notes: {len(plan['notes'])} · unmatched threads: {len(plan['unmatched_threads'])}" + f" · already processed: {plan['skipped_already_processed']}") + if args.apply: + print("applied via REST; state updated." if plan["notes"] else "nothing to apply.") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/security_guards.py b/tools/security_guards.py index 2111781..dd0bb9c 100755 --- a/tools/security_guards.py +++ b/tools/security_guards.py @@ -45,13 +45,18 @@ ALLOWED_PERMISSIONS = { } # Personal-data ignore rules that must never disappear from .gitignore. +# Fork policy: documents/ (cv, linkedin, diplomas, references, postings, +# sources) is deliberately tracked in git - this repo is a personal fork, +# not the upstream template. Only generated/derived personal state and +# secrets stay ignored. REQUIRED_IGNORE_RULES = [ "salary_data.json", # Depth-independent: the job-scraper skill resolves `job_scraper/` relative # to its own directory, so the state file lands under .claude/skills/... and # a repo-rooted rule silently fails to match it. "**/job_scraper/seen_jobs.json", - "**/job_scraper/notion_sync.json", + "**/job_scraper/twenty_sync.json", + "**/job_scraper/context_sync.json", "**/job_scraper/*.md", "*_BehavioralReport.pdf", "linkedin_Profile.pdf", @@ -62,15 +67,9 @@ REQUIRED_IGNORE_RULES = [ "cover_letters/cover_*.*", # /apply also recognizes the uppercase Cover_* naming variant. "cover_letters/Cover_*.*", - "documents/cv/**", - "documents/linkedin/**", - "documents/diplomas/**", - "documents/references/**", "documents/applications/**", - "documents/postings/**", "documents/interview/**", "job_search_tracker.csv", - "gmail_sync/", "reports/", "upskill/*.md", # Depth-independent twin of the rule above. The upskill *skill* resolves diff --git a/tools/twenty_client.py b/tools/twenty_client.py new file mode 100644 index 0000000..d8e7914 --- /dev/null +++ b/tools/twenty_client.py @@ -0,0 +1,100 @@ +"""Minimal Twenty CRM REST client (stdlib only). + +Shared by tools/twenty_enrich.py and tools/gather_recent_context.py. + +Auth: set TWENTY_API_KEY (workspace API key from Settings -> API & Webhooks) +and optionally TWENTY_BASE_URL (default: https://crm.prad.nu). + +Both consumer scripts also run keyless: they compute a *plan* JSON from local +data plus an optional CRM snapshot, and the /sync-twenty or +/gather-recent-context command applies the plan through the Twenty MCP +server, which is session-authenticated via OAuth. The REST path exists so the +whole loop can also run headless (cron, CI) with an API key. +""" +from __future__ import annotations + +import json +import os +import time +import urllib.error +import urllib.parse +import urllib.request + +DEFAULT_BASE_URL = "https://crm.prad.nu" +PAGE_LIMIT = 60 # Twenty REST caps page size at 60 + + +class TwentyError(RuntimeError): + pass + + +class TwentyClient: + def __init__(self, base_url: str | None = None, api_key: str | None = None): + self.base_url = (base_url or os.environ.get("TWENTY_BASE_URL") or DEFAULT_BASE_URL).rstrip("/") + self.api_key = api_key or os.environ.get("TWENTY_API_KEY") + if not self.api_key: + raise TwentyError( + "TWENTY_API_KEY is not set. Either export it, or run without --apply " + "to emit a plan JSON and apply it through the Twenty MCP session." + ) + + def _request(self, method: str, path: str, body: dict | None = None, + params: dict | None = None, retries: int = 3): + url = f"{self.base_url}{path}" + if params: + url += "?" + urllib.parse.urlencode(params) + data = json.dumps(body).encode() if body is not None else None + last_err = None + for attempt in range(retries): + req = urllib.request.Request(url, data=data, method=method, headers={ + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json", + "Accept": "application/json", + }) + try: + with urllib.request.urlopen(req, timeout=60) as resp: + return json.loads(resp.read().decode()) + except urllib.error.HTTPError as e: # 4xx never retried: same input, same answer + detail = e.read().decode(errors="replace")[:500] + if e.code == 429 or e.code >= 500: + last_err = TwentyError(f"{method} {path} -> {e.code}: {detail}") + time.sleep(1.5 * (attempt + 1)) + continue + raise TwentyError(f"{method} {path} -> {e.code}: {detail}") from e + except urllib.error.URLError as e: + last_err = TwentyError(f"{method} {path} -> {e.reason}") + time.sleep(1.5 * (attempt + 1)) + raise last_err + + # -- records ------------------------------------------------------------ + + def list_all(self, plural: str, order_by: str = "id") -> list[dict]: + """Every record of an object, cursor-paged.""" + out: list[dict] = [] + cursor = None + while True: + params = {"limit": PAGE_LIMIT, "order_by": order_by} + if cursor: + params["starting_after"] = cursor + payload = self._request("GET", f"/rest/{plural}", params=params) + page = payload.get("data", {}).get(plural, []) + out.extend(page) + page_info = payload.get("pageInfo", {}) + if not page_info.get("hasNextPage") or not page: + break + cursor = page_info.get("endCursor") or page[-1]["id"] + return out + + def create_batch(self, plural: str, records: list[dict]) -> list[dict]: + """Create records in chunks of PAGE_LIMIT via the batch endpoint.""" + created: list[dict] = [] + for i in range(0, len(records), PAGE_LIMIT): + payload = self._request("POST", f"/rest/batch/{plural}", + body=records[i:i + PAGE_LIMIT]) + created.extend(payload.get("data", {}).get(plural, [])) + return created + + def update_one(self, plural: str, record_id: str, patch: dict) -> dict: + payload = self._request("PATCH", f"/rest/{plural}/{record_id}", body=patch) + data = payload.get("data", {}) + return next(iter(data.values())) if data else payload diff --git a/tools/twenty_enrich.py b/tools/twenty_enrich.py new file mode 100755 index 0000000..0617756 --- /dev/null +++ b/tools/twenty_enrich.py @@ -0,0 +1,295 @@ +#!/usr/bin/env python3 +"""Enrich Twenty CRM opportunities and companies from local job-search data. + +Encodes the deterministic layers of the /sync-twenty enrichment pipeline: + + Opportunity salary (``amount``), in source priority order: + 1. seen_jobs.json ``salary_max_stated`` (stated in the posting at rank time) + 2. jobs.db postings salary_max (same company+title, stated on an ATS board) + 3. jobs.db companies levels_swe_median (levels.fyi company median - labeled estimate) + + Company fields (``domainName``, ``linkedinLink``, ``address``, ``annualRevenue``): + - jobs.db companies row matched on normalized name, homepage domain joined + through postings.company_domain. Stringified NaN / empty strings are absent. + +Anything these layers cannot ground is emitted in the plan's ``unresolved`` +section for the calling command to escalate to web research (scout subagents / +Perplexity) - this script never guesses. + +Usage: + python3 tools/twenty_enrich.py # plan -> job_scraper/enrichment_plan.json + python3 tools/twenty_enrich.py --apply # plan + apply via REST (needs TWENTY_API_KEY) + python3 tools/twenty_enrich.py --snapshot f.json # offline: CRM state from a JSON snapshot + +Snapshot shape (what /sync-twenty exports through MCP): + {"opportunities": [{id, url:{primaryLinkUrl}, amount, closeDate, ...}], + "companies": [{id, name, domainName, linkedinLink, address, ...}]} + +The plan JSON is idempotent to apply: every entry names the target record id +and only fields that are currently empty. +""" +from __future__ import annotations + +import argparse +import json +import re +import sqlite3 +import sys +from pathlib import Path + +REPO = Path(__file__).resolve().parent.parent +SEEN_JOBS = REPO / "job_scraper" / "seen_jobs.json" +JOBS_DB = REPO / "documents" / "postings" / "jobs.db" +PLAN_OUT = REPO / "job_scraper" / "enrichment_plan.json" + +_SUFFIXES = re.compile(r"\b(inc|llc|ltd|corp|corporation|co|labs|lab|hq|the)\b") + + +def norm_company(s: str) -> str: + """Normalized company key: lowercase, punctuation collapsed, suffixes stripped.""" + s = re.sub(r"[^a-z0-9]+", " ", (s or "").lower()).strip() + s = _SUFFIXES.sub("", s).strip() + return re.sub(r"\s+", " ", s) + + +def norm_title(s: str) -> str: + return re.sub(r"[^a-z0-9]+", " ", (s or "").lower()).strip() + + +def clean_domain(raw: str | None) -> str | None: + """Bare registrable domain or None. Rejects the stringified-NaN artifact.""" + if not raw: + return None + d = str(raw).strip().lower() + for prefix in ("http://", "https://", "www."): + d = d.removeprefix(prefix) + d = d.rstrip("/") + if not d or d == "nan" or "." not in d: + return None + return d + + +def _has_link(v) -> bool: + return bool(v and v.get("primaryLinkUrl")) + + +def _has_addr(v) -> bool: + return bool(v and (v.get("addressCity") or v.get("addressCountry"))) + + +def _has_amount(v) -> bool: + return bool(v and v.get("amountMicros")) + + +def load_local_sources(seen_path: Path = SEEN_JOBS, db_path: Path = JOBS_DB): + """Ranked jobs plus the jobs.db lookup tables the plan builders consume.""" + seen = json.loads(seen_path.read_text(encoding="utf-8"))["seen"] + ranked = [v for v in seen.values() if v.get("status") == "ranked"] + + db_companies: dict[str, dict] = {} + domains: dict[str, str] = {} + postings_salary: dict[tuple[str, str], int] = {} + levels: dict[str, int] = {} + if db_path.exists(): + con = sqlite3.connect(db_path) + con.row_factory = sqlite3.Row + for r in con.execute( + "SELECT company_id, company_name, linkedin_url, country, state, apollo_city," + " organization_revenue, levels_swe_median_usd, levels_median_comp_usd FROM companies" + ): + key = norm_company(r["company_name"]) + db_companies.setdefault(key, dict(r)) + lv = r["levels_swe_median_usd"] or r["levels_median_comp_usd"] + if lv and lv > 0: + levels.setdefault(key, int(lv)) + for r in con.execute( + "SELECT company_id, company_domain, COUNT(*) n FROM postings" + " WHERE company_domain IS NOT NULL AND company_domain != ''" + " GROUP BY company_id, company_domain ORDER BY n" + ): + domains[r["company_id"]] = r["company_domain"] # last row = most frequent + for r in con.execute( + "SELECT company_name, title, salary_max, salary_currency FROM postings" + " WHERE salary_max IS NOT NULL AND salary_max > 0" + ): + if (r["salary_currency"] or "USD").upper() in ("USD", "$", ""): + key = (norm_company(r["company_name"]), norm_title(r["title"])) + postings_salary.setdefault(key, int(float(r["salary_max"]))) + con.close() + return ranked, db_companies, domains, postings_salary, levels + + +def build_salary_plan(ranked, opportunities, postings_salary, levels): + """(updates, unresolved): amount patches for opportunities missing one.""" + by_url = {} + for o in opportunities: + u = (o.get("url") or {}).get("primaryLinkUrl") + if u: + by_url[u] = o + updates, unresolved = [], [] + for v in ranked: + o = by_url.get(v.get("url")) + if not o or _has_amount(o.get("amount")): + continue + amount = source = None + if v.get("salary_max_stated"): + amount, source = int(v["salary_max_stated"]), "seen_jobs.salary_max_stated" + else: + key = (norm_company(v.get("company", "")), norm_title(v.get("title", ""))) + if key in postings_salary: + amount, source = postings_salary[key], "jobs.db postings" + elif norm_company(v.get("company", "")) in levels: + amount = levels[norm_company(v.get("company", ""))] + source = "levels.fyi company SWE median (estimate)" + if amount: + updates.append({ + "id": o["id"], + "source": source, + "patch": {"amount": {"amountMicros": amount * 1_000_000, "currencyCode": "USD"}}, + }) + else: + unresolved.append({"id": o["id"], "url": v["url"], + "company": v.get("company"), "title": v.get("title")}) + return updates, unresolved + + +def build_company_plan(companies, db_companies, domains): + """(updates, unresolved): fill ONLY empty company fields from jobs.db.""" + updates, unresolved = [], [] + for c in companies: + key = norm_company(c.get("name", "")) + m = db_companies.get(key) or {} + patch = {} + if not _has_link(c.get("domainName")): + dom = clean_domain(domains.get(m.get("company_id", ""))) + if dom: + patch["domainName"] = {"primaryLinkUrl": f"https://{dom}", + "primaryLinkLabel": c["name"]} + if not _has_link(c.get("linkedinLink")): + li = (m.get("linkedin_url") or "").strip() + if li: + patch["linkedinLink"] = {"primaryLinkUrl": li.replace("http://", "https://"), + "primaryLinkLabel": c["name"]} + if not _has_addr(c.get("address")): + addr = {k: v for k, v in [("addressCity", m.get("apollo_city")), + ("addressState", m.get("state")), + ("addressCountry", m.get("country"))] if v} + if addr: + patch["address"] = addr + if patch: + updates.append({"id": c["id"], "name": c["name"], "patch": patch}) + # Fields still empty after the local layer -> web-research escalation + missing = [] + if not _has_link(c.get("domainName")) and "domainName" not in patch: + missing.append("domain") + if not _has_link(c.get("linkedinLink")) and "linkedinLink" not in patch: + missing.append("linkedin") + if not _has_addr(c.get("address")) and "address" not in patch: + missing.append("address") + if missing: + unresolved.append({"id": c["id"], "name": c["name"], "missing": missing}) + return updates, unresolved + + +def build_name_cleanup_plan(opportunities): + """Strip the ' — ' suffix from opportunity names. + + Runs LAST, and only for opportunities whose company relation is already + set: the suffix is the human-readable fallback while the record is + unlinked, and becomes redundant once ``companyId`` points at the Company. + An unlinked opportunity keeps its suffixed name so it stays identifiable. + """ + updates = [] + for o in opportunities: + name = o.get("name") or "" + if o.get("companyId") and " — " in name: + role = name.split(" — ")[0].strip() + if role and role != name: + updates.append({"id": o["id"], "patch": {"name": role}}) + return updates + + +def coverage(opportunities, companies) -> dict: + n, m = len(companies), len(opportunities) + return { + "companies": { + "total": n, + "domain": sum(1 for c in companies if _has_link(c.get("domainName"))), + "linkedin": sum(1 for c in companies if _has_link(c.get("linkedinLink"))), + "address": sum(1 for c in companies if _has_addr(c.get("address"))), + }, + "opportunities": { + "total": m, + "salary": sum(1 for o in opportunities if _has_amount(o.get("amount"))), + "company_link": sum(1 for o in opportunities if o.get("companyId")), + "score": sum(1 for o in opportunities if o.get("score") is not None), + }, + } + + +def main(argv=None) -> int: + ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + ap.add_argument("--snapshot", type=Path, + help="CRM state JSON (offline mode; else fetched via REST)") + ap.add_argument("--apply", action="store_true", + help="apply the plan via Twenty REST (needs TWENTY_API_KEY)") + ap.add_argument("--out", type=Path, default=PLAN_OUT, help="plan output path") + args = ap.parse_args(argv) + + ranked, db_companies, domains, postings_salary, levels = load_local_sources() + + client = None + if args.snapshot: + snap = json.loads(args.snapshot.read_text(encoding="utf-8")) + opportunities, companies = snap["opportunities"], snap["companies"] + else: + from twenty_client import TwentyClient # noqa: deferred import keeps offline mode key-free + client = TwentyClient() + opportunities = client.list_all("opportunities") + companies = client.list_all("companies") + + sal_updates, sal_unresolved = build_salary_plan(ranked, opportunities, postings_salary, levels) + comp_updates, comp_unresolved = build_company_plan(companies, db_companies, domains) + name_updates = build_name_cleanup_plan(opportunities) + + plan = { + "coverage_before": coverage(opportunities, companies), + "opportunity_updates": sal_updates, + "company_updates": comp_updates, + "name_cleanup": name_updates, + "unresolved": { + "opportunities_without_salary": sal_unresolved, + "companies_missing_fields": comp_unresolved, + }, + } + + if args.apply: + if client is None: + from twenty_client import TwentyClient + client = TwentyClient() + for u in sal_updates: + client.update_one("opportunities", u["id"], u["patch"]) + for u in comp_updates: + client.update_one("companies", u["id"], u["patch"]) + # Name cleanup LAST: it must observe the post-link state. + for u in name_updates: + client.update_one("opportunities", u["id"], u["patch"]) + plan["applied"] = {"opportunities": len(sal_updates), "companies": len(comp_updates), + "names_cleaned": len(name_updates)} + + args.out.parent.mkdir(parents=True, exist_ok=True) + args.out.write_text(json.dumps(plan, indent=2), encoding="utf-8") + cov = plan["coverage_before"] + print(f"plan -> {args.out}") + print(f"opportunity salary updates: {len(sal_updates)} " + f"(coverage {cov['opportunities']['salary']}/{cov['opportunities']['total']} before)") + print(f"company field updates: {len(comp_updates)} · name cleanups: {len(name_updates)}") + print(f"unresolved -> web research: {len(sal_unresolved)} postings, " + f"{len(comp_unresolved)} companies") + if args.apply: + print("applied via REST.") + return 0 + + +if __name__ == "__main__": + sys.exit(main())