audit.py

07-Automation/partner_pipeline/audit.py

1. """outreach-pull-list and outreach-audit: who was actually contacted, and when."""
3. from __future__ import annotations
5. import datetime as dt
6. import json
7. import sys
8. from pathlib import Path
10. from .audit_workbook import UNRESOLVED_MATCHES, write_audit_workbook
11. from .census import norm_key
12. from .paths import DATA_DIR, NOW, OUT_DIR, TODAY, TOUCH_CACHE, XML_MAP
13. from .roster_workbook import ALL_TAB, PARTNER_TAB, latest_updated_workbook
14. from .stages import AUDIT_FUNNELS, SEND_TOKENS, TEST_DEALS, audit_stage_name, is_terminal
16. SAME_SEND_GAP_MIN = 2 # two entries into the same stage this close = one send
19. def _reusable(cat: str, cached: dict, stage_id: str) -> bool:
20. """Can this deal's cached touch history be reused instead of re-pulled?
22. Two cases, both requiring the current stage to be unchanged since the cache was
23. written (the caller checks that):
25. 1. Terminal stage: the deal is closed, nothing more can happen to it.
26. 2. Zero recorded touches: the deal has never entered a send stage, so it cannot
27. be sitting in one now, and a send would have moved it. This is what keeps the
28. big parked cohorts cheap: Winback "Client", Abandoned Carts "New Abandoned
29. Cart", cat 24 "LinkedIn Research & Connection", cat 36 "Active Check-In List"
30. hold most of the deals, and nobody has touched them.
32. Residual risk in both cases is the same: a deal that entered a send stage and
33. returned to its original stage between two runs would be missed. It self-corrects
34. the next time its stage actually changes. Deals with touches sitting in a send
35. stage are always re-pulled, because a repeat send re-enters the same stage and
36. that is a real pattern (cat 26 Follow-Up Email).
37. """
38. if is_terminal(cat, stage_id):
39. return True
40. return int(cached.get("count") or 0) == 0
43. def _parse_ts(s):
44. if not s or str(s).strip().upper() in ("", "NONE", "NULL"):
45. return None
46. s = str(s).strip().replace("T", " ")[:16]
47. try:
48. return dt.datetime.strptime(s, "%Y-%m-%d %H:%M")
49. except ValueError:
50. try:
51. return dt.datetime.strptime(s[:10], "%Y-%m-%d")
52. except ValueError:
53. return None
56. def compute_touches(cat: str, history: list) -> dict:
57. """Reduce a raw stage history to verified outbound contact.
59. history: [{"stage": "C36:EXECUTING", "at": "2026-07-29 16:40"}, ...] any order.
60. Returns {first, last, last_stage, count} with datetimes (or None).
62. Only SEND_TOKENS count. Repeat entries into the SAME stage within
63. SAME_SEND_GAP_MIN minutes collapse to one send: Bitrix workflows re-enter a
64. stage several times in the same minute and that is one email, not four.
65. Re-entries days apart DO count separately; the `count` is therefore a
66. best-effort figure while `last` is the reliable one.
67. """
68. sends = []
69. for h in history or []:
70. stage = h.get("stage") or ""
71. # A deal can be moved between funnels, so its history legitimately contains
72. # stages from other categories (e.g. an Abandoned Cart that converted and went
73. # on through Pre-Production/Production carries C18:* and C20:* entries).
74. # Token-only matching would then count C20:PREPARATION as an abandoned-cart
75. # email, because "PREPARATION" is a send token in cat 16. Require the category
76. # prefix to match the funnel we are auditing; a bare token with no prefix is
77. # assumed to belong to this funnel.
78. if ":" in stage:
79. h_cat, token = stage.split(":", 1)
80. h_cat = h_cat.lstrip("Cc")
81. if h_cat != cat:
82. continue
83. else:
84. token = stage
85. if token not in SEND_TOKENS.get(cat, set()):
86. continue
87. ts = _parse_ts(h.get("at"))
88. if ts is None:
89. continue
90. sends.append((ts, token, stage))
91. sends.sort()
92. kept = []
93. for ts, token, stage in sends:
94. if kept and kept[-1][1] == token and (ts - kept[-1][0]).total_seconds() <= SAME_SEND_GAP_MIN * 60:
95. continue
96. kept.append((ts, token, stage))
97. if not kept:
98. return {"first": None, "last": None, "last_stage": "", "count": 0}
99. return {"first": kept[0][0], "last": kept[-1][0],
100. "last_stage": audit_stage_name(cat, kept[-1][2]), "count": len(kept)}
103. def _load_touch_cache() -> dict:
104. if TOUCH_CACHE.exists():
105. try:
106. return json.loads(TOUCH_CACHE.read_text(encoding="utf-8"))
107. except Exception as e:
108. print(f"WARNING: touch cache unreadable ({e}): treating every deal as new.")
109. return {}
112. def _build_plinth_index(workbook: Path | None):
113. """Identity lookup for the roster owner: Plinth id, email, phone, company, tier, LTV.
115. Three sources, most specific first:
116. 1. the XML census -> bitrix_deal_id -> plinth_id (exact, by deal)
117. 2. partner tab of the Drive-swap workbook -> phone/company/tier/LTV
118. 3. All Accounts of the same workbook = the full Plinth customer list (every account),
119. which is the only place non-partner people (cart guests, dormant winback)
120. can be found at all.
121. """
122. import xml.etree.ElementTree as ET
123. by_deal, prem, prem_email, prem_name, plinth_email, plinth_name = {}, {}, {}, {}, {}, {}
124. if XML_MAP.exists():
125. root = ET.parse(XML_MAP).getroot()
126. for c in root.findall("client"):
127. rec = {"plinth": (c.findtext("plinth_id") or "").strip(),
128. "email": (c.findtext("email") or "").strip(),
129. "company": (c.findtext("company") or "").strip(),
130. "tier": (c.findtext("user_group") or "").strip()}
131. for blk in c.iter("pipeline"):
132. d = (blk.findtext("bitrix_deal_id") or "").strip()
133. if d:
134. by_deal[d] = rec
135. if workbook and workbook.exists():
136. try:
137. from openpyxl import load_workbook
138. wb = load_workbook(workbook, read_only=True, data_only=True)
139. if PARTNER_TAB in wb.sheetnames:
140. for i, row in enumerate(wb[PARTNER_TAB].iter_rows(values_only=True)):
141. if i == 0 or not row or row[0] is None:
142. continue
143. sid = str(int(row[0])) if isinstance(row[0], (int, float)) else str(row[0]).strip()
144. rec = {"plinth": sid, "email": row[3], "company": row[5], "phone": row[6],
145. "tier": row[10] or row[9], "ltv": row[13]}
146. prem[sid] = rec
147. for e in (row[3], row[4]):
148. if e and "@" in str(e):
149. prem_email.setdefault(str(e).strip().lower(), rec)
150. n = norm_key(f"{row[1]}{row[2]}")
151. if n:
152. prem_name.setdefault(n, rec)
153. if ALL_TAB in wb.sheetnames:
154. for i, row in enumerate(wb[ALL_TAB].iter_rows(values_only=True)):
155. if i == 0 or not row or row[0] is None:
156. continue
157. sid = str(int(row[0])) if isinstance(row[0], (int, float)) else str(row[0]).strip()
158. rec = {"plinth": sid, "email": row[3], "ltv": row[4]}
159. if row[3] and "@" in str(row[3]):
160. plinth_email.setdefault(str(row[3]).strip().lower(), rec)
161. n = norm_key(f"{row[1]}{row[2]}")
162. if n:
163. plinth_name.setdefault(n, rec)
164. wb.close()
165. except ImportError:
166. print("NOTE: openpyxl missing: Plinth identity columns will be blank.")
167. return by_deal, prem, prem_email, prem_name, plinth_email, plinth_name
170. def resolve_plinth_identity(idx, deal_id, name, email=None) -> dict:
171. by_deal, prem, prem_email, prem_name, plinth_email, plinth_name = idx
172. out = {"plinth": "", "email": email or "", "company": "", "phone": "",
173. "tier": "", "ltv": None, "match": "UNRESOLVED"}
174. c = by_deal.get(str(deal_id))
175. if c and c["plinth"]:
176. out.update(plinth=c["plinth"], email=c["email"] or out["email"],
177. company=c["company"], tier=c["tier"], match="deal->census")
178. p = prem.get(c["plinth"])
179. if p:
180. out["phone"] = p.get("phone") or ""
181. out["ltv"] = p.get("ltv")
182. out["company"] = out["company"] or (p.get("company") or "")
183. out["email"] = out["email"] or (p.get("email") or "")
184. out["tier"] = out["tier"] or (p.get("tier") or "")
185. return out
186. if email:
187. e = str(email).strip().lower()
188. p = prem_email.get(e)
189. if p:
190. return {"plinth": p["plinth"], "email": p.get("email") or email,
191. "company": p.get("company") or "", "phone": p.get("phone") or "",
192. "tier": p.get("tier") or "", "ltv": p.get("ltv"),
193. "match": "email->partner"}
194. o = plinth_email.get(e)
195. if o:
196. return {"plinth": o["plinth"], "email": o.get("email") or email, "company": "",
197. "phone": "", "tier": "not partner", "ltv": o.get("ltv"),
198. "match": "email->Plinth census"}
199. n = norm_key(name)
200. if n and n in prem_name:
201. p = prem_name[n]
202. return {"plinth": p["plinth"], "email": p.get("email") or "", "company": p.get("company") or "",
203. "phone": p.get("phone") or "", "tier": p.get("tier") or "",
204. "ltv": p.get("ltv"), "match": "name->partner"}
205. if n and n in plinth_name:
206. o = plinth_name[n]
207. return {"plinth": o["plinth"], "email": o.get("email") or "", "company": "", "phone": "",
208. "tier": "not partner", "ltv": o.get("ltv"), "match": "name->Plinth census"}
209. if email:
210. out["match"] = "email only (no Plinth match)"
211. return out
214. def cmd_outreach_pull_list(args):
215. """Decide which deals actually need a fresh stage-history pull this week.
217. Reads the weekly enumeration JSON and the touch cache, and writes the deal ids
218. that must be re-pulled. A deal is skipped only when it is already cached, its
219. stage has not changed since the cache was written, and `_reusable` says nothing
220. more can have happened to it. Everything else is pulled.
221. """
222. deals = []
223. for path in args.deals:
224. deals += json.loads(Path(path).read_text(encoding="utf-8")).get("deals", [])
225. cache = _load_touch_cache()
226. prev_unresolved = set()
227. for f in sorted(OUT_DIR.glob("outreach-unresolved_*.json"))[-1:]:
228. try:
229. prev_unresolved = {str(x["deal_id"]) for x in
230. json.loads(f.read_text(encoding="utf-8")).get("deals", [])}
231. except Exception:
232. pass
233. need, skip, need_contact = [], 0, []
234. for d in deals:
235. cat = str(d.get("cat"))
236. if cat not in AUDIT_FUNNELS:
237. continue
238. did = str(d.get("deal_id"))
239. c = cache.get(did)
240. if c and c.get("stage") == d.get("stage") and _reusable(cat, c, d.get("stage") or ""):
241. skip += 1
242. else:
243. need.append(did)
244. # Abandoned Carts titles are "Cart - 12345", so those always need a contact
245. # lookup; everyone else only if a previous run could not place them in Plinth.
246. if not (c or {}).get("email") and (cat == "16" or did in prev_unresolved):
247. need_contact.append(did)
248. OUT_DIR.mkdir(parents=True, exist_ok=True)
249. out = OUT_DIR / f"outreach-pull-list_{TODAY}.json"
250. out.write_text(json.dumps({"generated": NOW, "need_history": need,
251. "need_contact": need_contact,
252. "reused_from_cache": skip}, indent=2), encoding="utf-8")
253. print(f"PULL-LIST: {len(need)} deals need stage history, {skip} reused from cache "
254. f"(cache holds {len(cache)}). {len(need_contact)} need a get_record contact lookup.")
255. print(f"wrote {out}")
256. if not cache:
257. print("NOTE: cache is empty: this is a first/full run, expect every deal in the list.")
258. return 0
261. def cmd_outreach_audit(args):
262. """Stage F5: build the per-pipeline outreach audit workbook.
264. Input: enumeration JSON for every audit funnel, where any deal that was
265. re-pulled this week also carries a "history" array. Deals without "history"
266. are served from the touch cache. Read-only on Bitrix and on the vault data;
267. writes one workbook to 05-Outputs/ and updates the cache.
268. """
269. deals = []
270. for path in args.deals:
271. deals += json.loads(Path(path).read_text(encoding="utf-8")).get("deals", [])
272. if not deals:
273. sys.exit("No deals in the input JSON: nothing to audit.")
274. wbk = Path(args.workbook) if args.workbook else latest_updated_workbook()
275. if not wbk or not Path(wbk).exists():
276. print("WARNING: no Drive-swap workbook found: Plinth phone/tier/LTV will be sparse.")
277. wbk = None
278. idx = _build_plinth_index(Path(wbk) if wbk else None)
279. cache = _load_touch_cache()
281. rows, fresh, reused, missing = {}, 0, 0, []
282. for d in deals:
283. cat = str(d.get("cat"))
284. if cat not in AUDIT_FUNNELS:
285. continue
286. did = str(d.get("deal_id"))
287. cached = cache.get(did) or {}
288. # Precedence: this week's contact pull > whatever we cached earlier > the deal
289. # title. The title is last on purpose: in Abandoned Carts it is "Cart - 12345",
290. # which is useless to whoever works the sheet, while the cached contact name is
291. # the real person.
292. con = d.get("contact") or {}
293. name = con.get("name") or cached.get("name") or d.get("title") or ""
294. email = con.get("email") or cached.get("email") or ""
295. phone = con.get("phone") or cached.get("phone") or ""
296. if "history" in d and d["history"] is not None:
297. t = compute_touches(cat, d["history"])
298. fresh += 1
299. cache[did] = {"cat": cat, "stage": d.get("stage"), "name": name,
300. "owner": d.get("owner", ""), "email": email, "phone": phone,
301. "first": t["first"].strftime("%Y-%m-%d %H:%M") if t["first"] else None,
302. "last": t["last"].strftime("%Y-%m-%d %H:%M") if t["last"] else None,
303. "last_stage": t["last_stage"], "count": t["count"],
304. "fetched": TODAY}
305. else:
306. c = cache.get(did)
307. if not c:
308. missing.append(did)
309. continue
310. t = {"first": _parse_ts(c.get("first")), "last": _parse_ts(c.get("last")),
311. "last_stage": c.get("last_stage", ""), "count": c.get("count", 0)}
312. reused += 1
313. ident = resolve_plinth_identity(idx, did, name, email or None)
314. note = "TEST RECORD - exclude from metrics" if did in TEST_DEALS else ""
315. rows.setdefault(cat, []).append({
316. "deal": did, "name": name, "owner": d.get("owner", "") or cache.get(did, {}).get("owner", ""),
317. "stage": audit_stage_name(cat, d.get("stage") or ""),
318. "stage_date": _parse_ts(d.get("stage_date")), "first": t["first"], "last": t["last"],
319. "last_stage": t["last_stage"], "count": t["count"], "value": d.get("value", ""),
320. "plinth": ident["plinth"], "email": ident["email"] or email,
321. "company": ident["company"], "phone": ident["phone"] or phone,
322. "tier": ident["tier"], "ltv": ident["ltv"], "match": ident["match"], "note": note})
324. if missing:
325. print(f"WARNING: {len(missing)} deals had neither history nor a cache entry and were "
326. f"DROPPED from the audit: {', '.join(missing[:20])}"
327. f"{' …' if len(missing) > 20 else ''}")
329. # A funnel missing from the enumeration would otherwise produce a clean-looking
330. # workbook with no hint that part of the outreach is absent (an enumeration built
331. # from an older prompt covers only 24/26/36). The cache remembers which funnels
332. # were audited before, so compare against it and warn loudly.
333. cached_cats = {}
334. for v in cache.values():
335. cached_cats[v.get("cat")] = cached_cats.get(v.get("cat"), 0) + 1
336. absent = [c for c, n in cached_cats.items()
337. if c in AUDIT_FUNNELS and n > 0 and not rows.get(c)]
338. if absent:
339. names = ", ".join(f"{AUDIT_FUNNELS[c]} (cat {c}, {cached_cats[c]} known deals)"
340. for c in sorted(absent))
341. print("=" * 78)
342. print(f"WARNING: {len(absent)} outreach funnel(s) present in the cache are MISSING "
343. f"from this enumeration and are absent from the workbook:")
344. print(f" {names}")
345. print(" The enumeration must cover all four in-scope funnels (24/26/36/16). If the "
346. "scheduled\n prompt is the stale pre-2026-08-10 version it only pulls "
347. "24/26/36: see scheduled-task.md.")
348. print("=" * 78)
350. # Deals we could not tie to a Plinth customer are the ones nobody can action.
351. # Almost always this is a person whose Bitrix deal name does not match their Plinth
352. # name (for example a spouse's name on a shared household mailbox) and whose deal
353. # carries no contact email yet. Writing them out means one get_record pass next run
354. # closes the gap: once a contact email lands in the cache it is reused every week.
355. unresolved = [{"deal_id": r["deal"], "cat": cat, "name": r["name"], "funnel": AUDIT_FUNNELS[cat]}
356. for cat, rs in rows.items() for r in rs
357. if r["match"] in UNRESOLVED_MATCHES]
358. if unresolved:
359. OUT_DIR.mkdir(parents=True, exist_ok=True)
360. up = OUT_DIR / f"outreach-unresolved_{TODAY}.json"
361. up.write_text(json.dumps({"generated": NOW, "count": len(unresolved),
362. "note": "run get_record on these deals and add "
363. "contact{name,email,phone} to next week's enumeration",
364. "deals": unresolved}, indent=2), encoding="utf-8")
365. print(f"UNRESOLVED: {len(unresolved)} deals have no Plinth match: listed in {up.name}")
366. DATA_DIR.mkdir(parents=True, exist_ok=True)
367. TOUCH_CACHE.write_text(json.dumps(cache, indent=2, sort_keys=True), encoding="utf-8")
368. out = write_audit_workbook(rows, fresh, reused, missing, wbk, absent)
369. tot = sum(len(v) for v in rows.values())
370. contacted = sum(1 for v in rows.values() for r in v
371. if r["count"] > 0 and "TEST RECORD" not in r["note"])
372. print(f"AUDIT: {tot} deals across {len(rows)} funnels: {contacted} with a verified touch, "
373. f"{tot - contacted} never contacted. {fresh} histories pulled fresh, {reused} reused.")
374. print(f"wrote {out}")
375. return 0