CRM clean up: fix the records, then find the money already in them.

"""
CRM clean up: four agents read a HubSpot export, fix what is safe to fix,
queue what needs a person, and then turn the clean book of customers into a
list of who a human should call next.

    python3 crm_cleanup.py data --out output --as-of 2026-11-06

Agents, in the order they run
  1. Duplicates   companies and contacts that are the same thing twice
  2. Fields       country, lifecycle stage, stale deals, missing data
  3. Owners       records owned by people who left, or by nobody
  4. Book         existing customers: risk, expansion, and who gets a human

Every proposed change gets a tier. Nothing is written to the CRM by this
script; it writes the change log and the API payloads, and a person approves.

  auto        reversible, rule based, old value logged (country format, lifecycle
              stage that contradicts a won deal)
  approve     one batch approval by the RevOps owner per run (merges, which
              cannot be undone, and owner changes, which move people's books)
  review      one by one, a person decides (same name but no shared domain)
  owner_task  goes to the record owner as a task (stale or empty deals)
  enrich      goes to the enrichment queue (missing industry, bounced champion)

Rules are set by hand at the top of this file. Standard library only.
"""
import argparse
import csv
import json
import os
import re
from collections import defaultdict
from datetime import date

# ---- rules set by hand ------------------------------------------------------

LEGAL_SUFFIXES = {"gmbh", "ag", "ltd", "bv", "oy", "sas", "sarl", "ab", "as", "aps", "sl", "inc", "llc", "plc", "se"}

COUNTRIES = {
    "de": "Germany", "deutschland": "Germany", "germany": "Germany",
    "uk": "United Kingdom", "gb": "United Kingdom", "great britain": "United Kingdom",
    "united kingdom": "United Kingdom",
    "nl": "Netherlands", "holland": "Netherlands", "netherlands": "Netherlands",
}

TERRITORY = {
    "Germany": "DACH", "Austria": "DACH", "Switzerland": "DACH",
    "United Kingdom": "UKI", "Ireland": "UKI",
    "Netherlands": "Benelux and France", "Belgium": "Benelux and France", "France": "Benelux and France",
    "Sweden": "Nordics", "Norway": "Nordics", "Denmark": "Nordics", "Finland": "Nordics",
}

LIFECYCLE_RANK = {"customer": 3, "opportunity": 2, "lead": 1}
OPEN_STAGES = {"appointmentscheduled", "qualifiedtobuy", "presentationscheduled",
               "decisionmakerboughtin", "contractsent"}
STALE_AFTER_DAYS = 30          # open deal whose close date passed more than this

# Book of business
TIER_STRATEGIC = 100000        # EUR ARR and up: a named owner, always a human
TIER_CORE = 10000              # EUR ARR and up: a human when there is a signal
RISK_USAGE_DROP = -0.20        # usage month 3 vs month 1
RISK_RENEWAL_DAYS = 90         # renewal this close ...
RISK_NO_TOUCH_DAYS = 60        # ... and nobody spoke to them for this long
EXPANSION_USAGE_RISE = 0.30
EXPANSION_SEAT_UTIL = 0.90
QUIET_DAYS = 180               # core and strategic: no human contact for this long
EXPANSION_SHARE = 0.25         # assumption: an upsell adds a quarter of current ARR

# The LLM steps. Designed, not running in this sample: the rules below stand in
# and anything they cannot decide goes to a person.
DUPLICATE_JUDGE_PROMPT = """You are checking two CRM company records that share a
name but not a domain. Using only the fields given and the public website of each
domain, answer SAME, DIFFERENT or UNSURE, then one sentence of evidence. SAME only
if both domains lead to the same legal entity. Never guess; UNSURE goes to a person."""

LONG_TAIL_CHECKIN_PROMPT = """Write a three sentence email from the account team to a
small customer. Use only these facts: product, usage trend, seats used, renewal date.
Ask one question about what they are trying to do next. No discount, no pricing, no
claims about features. If usage is falling, ask what changed. Plain text."""

# ---- helpers ----------------------------------------------------------------


def norm_domain(d):
    d = (d or "").strip().lower()
    d = re.sub(r"^https?://", "", d)
    d = d.split("/")[0]
    if d.startswith("www."):
        d = d[4:]
    return d


def norm_name(n):
    words = re.sub(r"[^a-z0-9 ]", "", (n or "").lower().replace(".", "")).split()
    return " ".join(w for w in words if w not in LEGAL_SUFFIXES)


def days_between(a, b):
    return (b - a).days


def parse_date(s):
    return date.fromisoformat(s) if s else None


def eur(x):
    return f"EUR {x:,.0f}"


def load(folder):
    def rows(name):
        with open(os.path.join(folder, name), newline="") as f:
            return list(csv.DictReader(f))
    owners = {o["owner_id"]: o for o in rows("owners.csv")}
    companies = rows("companies.csv")
    for c in companies:
        for k in ("arr_eur", "usage_m1", "usage_m2", "usage_m3", "seats_used", "seats_limit"):
            c[k] = int(c[k] or 0)
    contacts = rows("contacts.csv")
    deals = rows("deals.csv")
    for d in deals:
        d["amount_eur"] = int(d["amount_eur"] or 0)
    return owners, companies, contacts, deals


class Log:
    def __init__(self):
        self.rows = []

    def add(self, agent, tier, obj, rid, name, field, old, new, reason):
        self.rows.append({
            "change_id": f"ch{len(self.rows) + 1:03d}", "agent": agent, "tier": tier, "object": obj,
            "record_id": rid, "record_name": name, "field": field, "old": old, "new": new, "reason": reason,
        })

# ---- agent 1: duplicates ----------------------------------------------------


def pick_primary(group, deal_count):
    return sorted(group, key=lambda c: (-deal_count[c["company_id"]],
                                        -LIFECYCLE_RANK.get(c["lifecycle_stage"], 0),
                                        not norm_domain(c["domain"]), c["created"]))[0]


def agent_duplicates(companies, contacts, deals, log):
    deal_count = defaultdict(int)
    for d in deals:
        deal_count[d["company_id"]] += 1
    merged_into = {}            # loser company_id -> primary company_id (approve tier only)
    seen_pairs = set()

    by_domain = defaultdict(list)
    for c in companies:
        if norm_domain(c["domain"]):
            by_domain[norm_domain(c["domain"])].append(c)
    for dom, group in by_domain.items():
        if len(group) < 2:
            continue
        primary = pick_primary(group, deal_count)
        for c in group:
            if c is primary:
                continue
            reason = f"same domain {dom}"
            if c["arr_eur"] and primary["arr_eur"]:
                reason += (f"; ARR on both records ({eur(primary['arr_eur'])} + {eur(c['arr_eur'])}), "
                           "reports show the account at the smaller figure; check the ARR field after the merge")
            log.add("duplicates", "approve", "company", c["company_id"], c["name"], "merge",
                    c["company_id"], primary["company_id"], reason)
            merged_into[c["company_id"]] = primary["company_id"]
            seen_pairs.add(frozenset((c["company_id"], primary["company_id"])))

    by_name = defaultdict(list)
    for c in companies:
        by_name[norm_name(c["name"])].append(c)
    for name, group in by_name.items():
        if len(group) < 2:
            continue
        primary = pick_primary(group, deal_count)
        for c in group:
            if c is primary or frozenset((c["company_id"], primary["company_id"])) in seen_pairs:
                continue
            if not norm_domain(c["domain"]) or not norm_domain(primary["domain"]):
                reason = "same name, one record has no domain"
            else:
                reason = (f"same name on two domains ({norm_domain(primary['domain'])}, "
                          f"{norm_domain(c['domain'])}); could be two legal entities")
            log.add("duplicates", "review", "company", c["company_id"], c["name"], "merge",
                    c["company_id"], primary["company_id"], reason)

    by_email = defaultdict(list)
    for p in contacts:
        by_email[p["email"].strip().lower()].append(p)
    for email, group in by_email.items():
        if len(group) < 2:
            continue
        group.sort(key=lambda p: (not p["job_title"], p["contact_id"]))
        primary = group[0]
        for p in group[1:]:
            log.add("duplicates", "approve", "contact", p["contact_id"], email, "merge",
                    p["contact_id"], primary["contact_id"], "same email in a different case")
    return merged_into

# ---- agent 2: fields --------------------------------------------------------


def agent_fields(companies, contacts, deals, merged_into, as_of, log):
    won = {d["company_id"] for d in deals if d["stage"] == "closedwon"}
    for c in companies:
        if c["company_id"] in merged_into:
            continue
        key = c["country"].strip().lower()
        if key in COUNTRIES and COUNTRIES[key] != c["country"]:
            log.add("fields", "auto", "company", c["company_id"], c["name"], "country",
                    c["country"], COUNTRIES[key], "one spelling per country so territory rules and reports work")
        clean = norm_domain(c["domain"])
        if clean and clean != c["domain"]:
            log.add("fields", "auto", "company", c["company_id"], c["name"], "domain",
                    c["domain"], clean, "bare domain, no protocol or path")
        if c["company_id"] in won and c["lifecycle_stage"] != "customer":
            log.add("fields", "auto", "company", c["company_id"], c["name"], "lifecyclestage",
                    c["lifecycle_stage"], "customer", "has a closed won deal")
        if c["lifecycle_stage"] == "customer" and c["company_id"] not in won and not c["arr_eur"]:
            log.add("fields", "review", "company", c["company_id"], c["name"], "lifecyclestage",
                    "customer", "?", "says customer, no won deal and no ARR: churned, or never won?")
        if not c["industry"]:
            log.add("fields", "enrich", "company", c["company_id"], c["name"], "industry",
                    "", "", "missing industry, needed for segment reports and routing")
    company_name = {c["company_id"]: c["name"] for c in companies}
    for p in contacts:
        if p["bounced"] == "true" and p["company_id"] not in merged_into:
            log.add("fields", "enrich", "contact", p["contact_id"], p["email"], "replacement",
                    p["job_title"], "", f"email bounces at {company_name.get(p['company_id'], '?')}: "
                    "the person likely left; find who replaced them")
    for d in deals:
        if d["stage"] not in OPEN_STAGES:
            continue
        close = parse_date(d["close_date"])
        late = days_between(close, as_of) if close else 0
        if late > STALE_AFTER_DAYS:
            money = f" ({eur(d['amount_eur'])} in the forecast)" if d["amount_eur"] else ""
            log.add("fields", "owner_task", "deal", d["deal_id"], d["deal_name"], "close_date",
                    d["close_date"], "", f"open deal, close date passed {late} days ago{money}: "
                    "update the date or close it lost")
        if not d["amount_eur"]:
            log.add("fields", "owner_task", "deal", d["deal_id"], d["deal_name"], "amount",
                    "0", "", "open deal with no amount, invisible in the forecast")

# ---- agent 3: owners --------------------------------------------------------


def agent_owners(companies, deals, owners, merged_into, log):
    active = {k: o for k, o in owners.items() if o["active"] == "true"}
    am = sorted(k for k, o in active.items() if o["role"] == "AM")[0]
    ae_by_territory = {}
    for k in sorted(active):
        if active[k]["role"] == "AE":
            ae_by_territory.setdefault(active[k]["territory"], k)

    def new_owner(country, is_customer):
        if is_customer:
            return am, "customers go to account management"
        country = COUNTRIES.get(country.strip().lower(), country)
        t = TERRITORY.get(country)
        if t and t in ae_by_territory:
            return ae_by_territory[t], f"territory {t}"
        return None, f"no active owner for {country}"

    company = {c["company_id"]: c for c in companies}
    for c in companies:
        if c["company_id"] in merged_into:
            continue
        cur = c["owner_id"]
        if cur and cur in active:
            continue
        why = "owner has left" if cur else "no owner"
        to, rule = new_owner(c["country"], c["lifecycle_stage"] == "customer" or c["arr_eur"] > 0)
        tier = "approve" if to else "review"
        log.add("owners", tier, "company", c["company_id"], c["name"], "hubspot_owner_id",
                cur, to or "?", f"{why}; {rule}; {eur(c['arr_eur'])} ARR" if c["arr_eur"] else f"{why}; {rule}")
    for d in deals:
        if d["owner_id"] in active:
            continue
        cid = merged_into.get(d["company_id"], d["company_id"])
        c = company[cid]
        to, rule = new_owner(c["country"], c["lifecycle_stage"] == "customer" or c["arr_eur"] > 0)
        log.add("owners", "approve" if to else "review", "deal", d["deal_id"], d["deal_name"],
                "hubspot_owner_id", d["owner_id"], to or "?", f"owner has left; {rule}")

# ---- agent 4: book of business ----------------------------------------------


def agent_book(companies, contacts, merged_into, as_of):
    bounced = defaultdict(list)
    for p in contacts:
        if p["bounced"] == "true":
            bounced[p["company_id"]].append(p["job_title"] or "contact")
    arr = defaultdict(int)
    for c in companies:
        arr[merged_into.get(c["company_id"], c["company_id"])] += c["arr_eur"]
    book = []
    for c in companies:
        cid = c["company_id"]
        if cid in merged_into or not arr[cid]:
            continue
        a = arr[cid]
        tier = "strategic" if a >= TIER_STRATEGIC else "core" if a >= TIER_CORE else "long tail"
        change = (c["usage_m3"] - c["usage_m1"]) / c["usage_m1"] if c["usage_m1"] else 0.0
        util = c["seats_used"] / c["seats_limit"] if c["seats_limit"] else 0.0
        renewal = parse_date(c["renewal_date"])
        to_renewal = days_between(as_of, renewal) if renewal else None
        touch = parse_date(c["last_human_touch"])
        since_touch = days_between(touch, as_of) if touch else None

        risk, expand, why = [], [], []
        if change <= RISK_USAGE_DROP:
            risk.append(f"usage down {abs(change):.0%} in three months")
        if to_renewal is not None and to_renewal <= RISK_RENEWAL_DAYS and (since_touch is None or since_touch > RISK_NO_TOUCH_DAYS):
            risk.append(f"renews in {to_renewal} days, " + ("no human contact on record" if since_touch is None
                                                            else f"last human contact {since_touch} days ago"))
        if bounced[cid]:
            risk.append(f"{', '.join(bounced[cid])} email bounces")
        if change >= EXPANSION_USAGE_RISE:
            expand.append(f"usage up {change:.0%} in three months")
        if util >= EXPANSION_SEAT_UTIL and change > RISK_USAGE_DROP:
            expand.append(f"{c['seats_used']} of {c['seats_limit']} seats used")
        quiet = tier != "long tail" and (since_touch is None or since_touch > QUIET_DAYS)
        if a != c["arr_eur"]:
            why.append(f"ARR {eur(a)} after merging the duplicate record (was {eur(c['arr_eur'])})")

        if risk:
            signal, stake = "risk", a
        elif expand:
            signal, stake = "expansion", round(a * EXPANSION_SHARE)
        elif quiet:
            signal, stake = "quiet", 0
            why.append("no human contact on record" if since_touch is None else f"no human contact for {since_touch} days")
        else:
            signal, stake = "steady", 0

        if tier == "strategic":
            channel = "human"
            action = {"risk": "Owner call this week, then a written save plan",
                      "expansion": "Owner call this week to scope the next tier",
                      "quiet": "Owner call, nobody has spoken to them in months",
                      "steady": "Quarterly review, AI prepares the brief"}[signal]
        elif tier == "core":
            channel = "human" if signal in ("risk", "expansion", "quiet") else "ai, owner approves"
            action = {"risk": "Owner call this week",
                      "expansion": "Owner call to scope the upgrade",
                      "quiet": "Owner check in call",
                      "steady": "AI drafted check in, owner approves before it goes"}[signal]
        else:
            channel = "ai, human on reply" if signal in ("risk", "expansion") else "automated"
            action = {"risk": "AI check in asking what changed, a human picks up any reply",
                      "expansion": "AI upgrade offer, a human picks up any reply",
                      "steady": "Monthly usage summary email, no human"}.get(signal, "Monthly usage summary email, no human")
        book.append({
            "company_id": cid, "name": c["name"], "arr_eur": a, "tier": tier, "signal": signal,
            "eur_at_stake": stake, "channel": channel, "action": action,
            "evidence": "; ".join(risk + expand + why),
        })
    order = {"risk": 0, "expansion": 1, "quiet": 2, "steady": 3}
    book.sort(key=lambda r: (r["channel"] != "human", order[r["signal"]], -r["arr_eur"]))
    return book

# ---- outputs ----------------------------------------------------------------


def payloads(log):
    """The HubSpot CRM v3 calls a person approves. Built, never sent from here."""
    plural = {"company": "companies", "contact": "contacts", "deal": "deals"}
    batches = defaultdict(lambda: defaultdict(dict))
    merges = []
    for r in log.rows:
        if r["field"] == "merge" and r["tier"] == "approve":
            merges.append({"endpoint": f"POST /crm/v3/objects/{plural[r['object']]}/merge",
                           "body": {"primaryObjectId": r["new"], "objectIdToMerge": r["old"]}})
        elif r["tier"] in ("auto", "approve") and r["field"] != "merge":
            batches[(r["tier"], r["object"])][r["record_id"]][r["field"]] = r["new"]
    updates = []
    for (tier, obj), recs in sorted(batches.items()):
        updates.append({"tier": tier, "endpoint": f"POST /crm/v3/objects/{plural[obj]}/batch/update",
                        "body": {"inputs": [{"id": k, "properties": v} for k, v in recs.items()]}})
    return {"note": "Dry run. Nothing here has been sent.", "updates": updates, "merges": merges}


def write_outputs(log, book, deals, companies, owners, out, as_of):
    os.makedirs(out, exist_ok=True)
    with open(os.path.join(out, "changes.csv"), "w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=list(log.rows[0].keys()))
        w.writeheader()
        w.writerows(log.rows)
    with open(os.path.join(out, "book.csv"), "w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=list(book[0].keys()))
        w.writeheader()
        w.writerows(book)
    with open(os.path.join(out, "hubspot_payloads.json"), "w") as f:
        json.dump(payloads(log), f, indent=2)
    # The enrich tier, keyed the way a Clay Audiences import matches records:
    # domain for companies, email for people. Only these rows spend credits.
    domain = {c["company_id"]: norm_domain(c["domain"]) for c in companies}
    with open(os.path.join(out, "clay_enrich_queue.csv"), "w", newline="") as f:
        w = csv.writer(f)
        w.writerow(["record_id", "object", "match_key", "need", "reason"])
        for r in log.rows:
            if r["tier"] == "enrich":
                key = domain.get(r["record_id"], "") if r["object"] == "company" else r["record_name"]
                w.writerow([r["record_id"], r["object"], key, r["field"], r["reason"]])

    tiers = defaultdict(int)
    for r in log.rows:
        tiers[r["tier"]] += 1
    active = {k for k, o in owners.items() if o["active"] == "true"}
    stale = [r for r in log.rows if r["object"] == "deal" and r["field"] == "close_date"]
    stale_eur = sum(d["amount_eur"] for d in deals if d["deal_id"] in {r["record_id"] for r in stale})
    hidden = sum(c["arr_eur"] for c in companies
                 if any(r["record_id"] == c["company_id"] and r["field"] == "merge" and r["tier"] == "approve" for r in log.rows))
    unowned = sum(r["arr_eur"] for r in book
                  if next(c for c in companies if c["company_id"] == r["company_id"])["owner_id"] not in active)
    calls = [r for r in book if r["channel"] == "human" and r["signal"] != "steady"]
    summary = {
        "as_of": as_of.isoformat(),
        "records_in": {"companies": len(companies), "deals": len(deals)},
        "changes": len(log.rows),
        "by_tier": dict(sorted(tiers.items())),
        "by_agent": {a: sum(1 for r in log.rows if r["agent"] == a) for a in ("duplicates", "fields", "owners")},
        "customers_in_book": len(book),
        "book_arr_eur": sum(r["arr_eur"] for r in book),
        "arr_on_duplicate_records_eur": hidden,
        "customer_arr_with_no_active_owner_eur": unowned,
        "stale_open_pipeline_eur": stale_eur,
        "human_calls": len(calls),
        "arr_with_risk_signal_eur": sum(r["eur_at_stake"] for r in book if r["signal"] == "risk"),
        "expansion_potential_eur": sum(r["eur_at_stake"] for r in book if r["signal"] == "expansion"),
        "channels": {ch: sum(1 for r in book if r["channel"] == ch) for ch in sorted({r["channel"] for r in book})},
        "assumptions": {"expansion_share_of_arr": EXPANSION_SHARE},
    }
    with open(os.path.join(out, "summary.json"), "w") as f:
        json.dump(summary, f, indent=2)

    lines = [f"# Who a human calls next, as of {as_of.isoformat()}", "",
             "| # | Account | ARR | Signal | Evidence | Action |", "| --- | --- | --- | --- | --- | --- |"]
    for i, r in enumerate(calls, start=1):
        lines.append(f"| {i} | {r['name']} | {eur(r['arr_eur'])} | {r['signal']} | {r['evidence']} | {r['action']} |")
    lines += ["", f"{len(log.rows)} changes proposed: " + ", ".join(f"{v} {k}" for k, v in sorted(tiers.items())) + "."]
    with open(os.path.join(out, "call_list.md"), "w") as f:
        f.write("\n".join(lines) + "\n")
    print("\n".join(lines))
    print(json.dumps(summary, indent=2))


def run(folder, as_of):
    owners, companies, contacts, deals = load(folder)
    log = Log()
    merged_into = agent_duplicates(companies, contacts, deals, log)
    agent_fields(companies, contacts, deals, merged_into, as_of, log)
    agent_owners(companies, deals, owners, merged_into, log)
    book = agent_book(companies, contacts, merged_into, as_of)
    return log, book, owners, companies, deals


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("data", help="folder with owners, companies, contacts and deals csv")
    ap.add_argument("--out", default="output")
    ap.add_argument("--as-of", default="2026-11-06")
    a = ap.parse_args()
    as_of = date.fromisoformat(a.as_of)
    log, book, owners, companies, deals = run(a.data, as_of)
    write_outputs(log, book, deals, companies, owners, a.out, as_of)


if __name__ == "__main__":
    main()