route.py

07-Automation/partner_pipeline/route.py

1. """route: the weekly outreach candidate lists."""
3. from __future__ import annotations
5. import csv
6. import datetime as dt
7. import json
8. import re
9. import sys
11. from .census import norm_key
12. from .classify import classify_one
13. from .config import NON_PARTNER, is_armed, load_config
14. from .history import apply_merges, compute_metrics, load_history, load_merge_registry
15. from .paths import CLIENTS_DIR, OUT_DIR, SUPPRESSION, TODAY, XML_MAP
16. from .stages import OPTOUT_STAGES
18. ROLE_NAME_RE = re.compile(
19. r"\b(test|testing|marketing|vendors?|department|dept|purchasing|expense|admin|office|"
20. r"accounts?|accounting|info|sales|creatives?|design|graphics|print|shop|store|team|"
21. r"staff|mngt|management|hr|billing)\b", re.I)
24. def cmd_route(args):
25. """Evaluate pipeline-routing-rules.md against the census. Read-only on the vault;
26. writes candidate artifacts to 05-Outputs/partner-runs/. Never touches Bitrix.
27. Rules: Winback first (partner, no order 12+ months, one touch per dormancy
28. spell); cat 24 = active never-contacted + tier crossings; Dormant queue
29. = active, previously contacted, no open deal, longest-since-outreach first. Opt-outs
30. (suppression file + any Not-Interested closure) never route anywhere."""
31. import xml.etree.ElementTree as ET
32. cfg = load_config()
33. armed, reasons = is_armed(cfg)
34. if not armed:
35. sys.exit("REFUSED: thresholds not set (" + "; ".join(reasons) + ")")
36. h = load_history()
37. metrics = compute_metrics(h, cfg)
38. aliases, _ = load_merge_registry()
39. metrics, _ = apply_merges(metrics, aliases)
40. supp_names, supp_emails, supp_domains = set(), set(), set()
41. if SUPPRESSION.exists():
42. try:
43. s = json.loads(SUPPRESSION.read_text(encoding="utf-8"))
44. supp_names = {norm_key(x) for x in s.get("names", [])}
45. supp_emails = {x.strip().lower() for x in s.get("emails", [])}
46. # Whole-domain never-contact list. Screening role-shaped names cannot see
47. # a domain: a directly partnered account whose relationship is handled
48. # in person, and staff mailboxes on the company's own domain, look like
49. # ordinary high-value clients and would otherwise be cold-emailed.
50. supp_domains = {x.strip().lower().lstrip("@") for x in s.get("domains", [])}
51. except Exception as ex:
52. sys.exit(f"Suppression list unreadable ({ex}): fix {SUPPRESSION}; "
53. "refusing to route without it.")
54. cutoff = (dt.date.today() - dt.timedelta(days=365)).isoformat()
55. root = ET.parse(XML_MAP).getroot()
56. pending = []
57. for c in root.findall("client"):
58. plinth = (c.findtext("plinth_id") or "").strip()
59. m = metrics.get(plinth)
60. if m is None:
61. cid = c.get("id") or ""
62. if cid.startswith("pl-"):
63. m = metrics.get(cid[3:])
64. if m is None:
65. continue
66. tier = classify_one(m, cfg, True)
67. partner = tier not in NON_PARTNER
68. name = (c.findtext("full_name") or "").strip()
69. email = (c.findtext("email") or "").strip().lower()
70. blocks = [dict(key=(p.get("name") or p.get("key")),
71. status=p.findtext("status") or "",
72. stage=p.findtext("stage") or "",
73. entered=p.findtext("entered_date") or "",
74. touch=p.findtext("last_touch") or "")
75. for p in c.findall("pipelines/pipeline")]
76. lcd = ""
77. fn = c.findtext("stub/file_name")
78. if fn and (CLIENTS_DIR / fn).exists():
79. mt = re.search(r"(?m)^last_contact_date: '?([\d-]+)'?",
80. (CLIENTS_DIR / fn).read_text(encoding="utf-8"))
81. if mt:
82. lcd = mt.group(1)
83. last = m["last_order_date"]
84. dates = [b["entered"] for b in blocks] + [b["touch"] for b in blocks] + [lcd]
85. row = dict(name=name, email=email, company=(c.findtext("company") or ""),
86. tier=tier, ltv=m["ltv"], last_order=last or "",
87. last_outreach=max([d for d in dates if d] or [""]),
88. plinth_id=plinth, bitrix_id=(c.findtext("bitrix_id") or ""),
89. chunk=c.get("id") or "",
90. never_contacted=(not blocks and not lcd),
91. role_name=bool(ROLE_NAME_RE.search(name)),
92. stages="; ".join(f"{b['key']}:{b['stage']}({b['status']})"
93. for b in blocks))
94. dormant = (not last) or last < cutoff
95. open_deal = any(b["status"] == "active" for b in blocks)
96. wb_done = any((b["entered"] or "0000") >= (last or "") or not b["entered"]
97. for b in blocks if b["key"] == "winback")
98. edom = email.rsplit("@", 1)[-1] if "@" in email else ""
99. optout = (norm_key(name) in supp_names or email in supp_emails
100. or (edom and edom in supp_domains)
101. or any(b["stage"] in OPTOUT_STAGES for b in blocks))
102. pending.append((row, dormant, open_deal, wb_done, optout, partner))
104. # Identity-level exclusion inheritance (the duplicate-twin case):
105. # duplicate chunks for one person split the evidence. A Not-Interested closure
106. # or open deal can sit on one twin while the blockless twin routes clean.
107. # Exclusion evidence (opt-out / open deal / winback-done) is unioned across
108. # every chunk sharing an email or a normalized full name, partner or not.
109. # Inheritance only ever excludes; eligibility is never inherited.
110. ev = {}
111. def _bump(k, oo, od, wb):
112. if not k:
113. return
114. cur = ev.get(k, (False, False, False))
115. ev[k] = (cur[0] or oo, cur[1] or od, cur[2] or wb)
116. for row, dormant, od, wb, oo, partner in pending:
117. _bump(row["email"], oo, od, wb)
118. _bump("n:" + norm_key(row["name"]), oo, od, wb)
119. winback, cat24, dormq, excluded = [], [], [], []
120. for row, dormant, od, wb, oo, partner in pending:
121. if not partner:
122. continue
123. e1 = ev.get(row["email"], (False, False, False))
124. e2 = ev.get("n:" + norm_key(row["name"]), (False, False, False))
125. i_oo = oo or e1[0] or e2[0]
126. i_od = od or e1[1] or e2[1]
127. i_wb = wb or e1[2] or e2[2]
128. why = []
129. if i_oo:
130. rdom = row["email"].rsplit("@", 1)[-1] if "@" in row["email"] else ""
131. if rdom and rdom in supp_domains:
132. why.append(f"never-contact domain @{rdom} "
133. "(relationship handled directly, no cold outreach)")
134. else:
135. why.append("opt-out (unsubscribe or Not-Interested closure)"
136. + ("" if oo else ", inherited from identity twin"))
137. if i_od and not why:
138. why.append(("open deal: " + row["stages"]) if od
139. else "open deal on identity twin (duplicate chunk)")
140. if dormant and i_wb and not why:
141. why.append("winback already done this dormancy spell"
142. + ("" if wb else " (identity twin)"))
143. if why:
144. excluded.append({**row, "reason": " | ".join(why)})
145. elif dormant:
146. winback.append(row)
147. elif row["never_contacted"]:
148. cat24.append(row)
149. else:
150. dormq.append(row)
151. winback.sort(key=lambda r: -r["ltv"])
152. cat24.sort(key=lambda r: -r["ltv"])
153. dormq.sort(key=lambda r: (r["last_outreach"] or "0000", -r["ltv"]))
154. OUT_DIR.mkdir(parents=True, exist_ok=True)
155. out_json = OUT_DIR / f"routing-candidates_{TODAY}.json"
156. out_json.write_text(json.dumps({"winback": winback, "cat24": cat24,
157. "dormant_queue": dormq, "excluded": excluded},
158. indent=1), encoding="utf-8")
159. # winback import CSV: person-named only; role-named to a review CSV
160. def _wcsv(path, rows):
161. with path.open("w", newline="", encoding="utf-8-sig") as f:
162. w = csv.writer(f)
163. w.writerow(["Deal Name", "Sales Funnel", "Deal Stage", "Contact: First Name",
164. "Contact: Last Name", "Contact: Email", "Contact: Company",
165. "User Group", "Comment"])
166. for r in rows:
167. parts = r["name"].split(" ", 1)
168. w.writerow([r["name"], "Winback", "Client", parts[0],
169. parts[1] if len(parts) > 1 else "", r["email"], r["company"],
170. r["tier"],
171. f"Lifetime {r['ltv']:.2f} USD | last order "
172. f"{r['last_order'] or 'unknown'} | Plinth id "
173. f"{r['plinth_id'] or 'n/a'} | routed {TODAY}"])
174. _wcsv(OUT_DIR / f"winback-import_{TODAY}.csv",
175. [r for r in winback if not r["role_name"]])
176. _wcsv(OUT_DIR / f"winback-role-review_{TODAY}.csv",
177. [r for r in winback if r["role_name"]])
178. print(f"ROUTE: winback {len(winback)} "
179. f"({len([r for r in winback if r['role_name']])} role-named held for review) | "
180. f"cat24 first-touch {len(cat24)} | dormant queue {len(dormq)} | "
181. f"excluded {len(excluded)}")
182. print(f"Artifacts: {out_json.name}, winback-import_{TODAY}.csv, "
183. f"winback-role-review_{TODAY}.csv in {OUT_DIR}")
184. print("Read-only on vault + Bitrix. Deal creation stays human-gated: winback CSV "
185. "-> bulk push after review; cat24/dormant -> push after review. "
186. "Tier crossings appear in classify output (user_group_correct changes).")