jobyssey/tools/gather_recent_context.py

#!/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 — <Company>" 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 <emails.json>            # plan only
  python3 tools/gather_recent_context.py --emails <emails.json> --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())