audit_workbook.py

07-Automation/partner_pipeline/audit_workbook.py

1. """The per-funnel outreach audit workbook."""
3. from __future__ import annotations
5. import csv
6. import datetime as dt
7. from pathlib import Path
9. from .paths import TODAY, VAULT_ROOT
10. from .stages import AUDIT_FUNNELS, AUDIT_TAB
12. AUDIT_STALE_DAYS = 90
13. # Identity matches that do not place a deal in Plinth.
14. UNRESOLVED_MATCHES = ("UNRESOLVED", "email only (no Plinth match)")
17. def write_audit_workbook(rows: dict, fresh: int, reused: int, missing: list, wbk,
18. missing_funnels: list | None = None) -> Path:
19. outdir = VAULT_ROOT / "05-Outputs"
20. outdir.mkdir(parents=True, exist_ok=True)
21. stem = outdir / f"Bitrix_Pipeline_Outreach_Audit_{TODAY}"
22. try:
23. from openpyxl import Workbook
24. from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
25. from openpyxl.utils import get_column_letter as L
26. except ImportError:
27. out = stem.with_suffix(".csv")
28. with out.open("w", newline="", encoding="utf-8") as f:
29. w = csv.writer(f)
30. w.writerow(["funnel", "client", "company", "plinth_id", "email", "phone",
31. "tier", "ltv", "last_reached_out", "touch_type", "touches",
32. "first_reached_out", "current_stage", "owner", "deal_id", "matched_by", "notes"])
33. for cat, rs in rows.items():
34. for r in rs:
35. w.writerow([AUDIT_FUNNELS[cat], r["name"], r["company"], r["plinth"], r["email"],
36. r["phone"], r["tier"], r["ltv"],
37. r["last"].isoformat(sep=" ") if r["last"] else "", r["last_stage"],
38. r["count"], r["first"].isoformat(sep=" ") if r["first"] else "",
39. r["stage"], r["owner"], r["deal"], r["match"], r["note"]])
40. print("NOTE: openpyxl not installed: wrote CSV instead of xlsx.")
41. return out
43. AR = "Arial"
44. T = Font(name=AR, size=16, bold=True, color="1F3864")
45. SUB = Font(name=AR, size=10, italic=True, color="595959")
46. H = Font(name=AR, size=10, bold=True, color="FFFFFF")
47. HF = PatternFill("solid", fgColor="1F3864")
48. B = Font(name=AR, size=10)
49. BB = Font(name=AR, size=10, bold=True)
50. thin = Side(style="thin", color="D9D9D9")
51. BD = Border(left=thin, right=thin, top=thin, bottom=thin)
52. NEV = PatternFill("solid", fgColor="EDEDED")
53. STALE = PatternFill("solid", fgColor="FCE4E4")
54. WARN = PatternFill("solid", fgColor="FFF2CC")
55. today = dt.datetime.combine(dt.date.today(), dt.time())
57. wb = Workbook()
58. wb.remove(wb.active)
59. summary = []
60. for cat in [c for c in AUDIT_FUNNELS if c in rows]:
61. rs = rows[cat]
62. ws = wb.create_sheet(AUDIT_TAB[cat])
63. contacted = [r for r in rs if r["count"] > 0 and "TEST RECORD" not in r["note"]]
64. never = [r for r in rs if r["count"] == 0]
65. ws["A1"] = f"{AUDIT_FUNNELS[cat]} (Bitrix funnel {cat})"
66. ws["A1"].font = T
67. ws["A2"] = (f"{len(rs)} deals · {len(contacted)} with a verified outbound touch · "
68. f"{len(never)} never contacted. Sorted by last contact, newest first. "
69. f"Built from Bitrix stage history {TODAY}.")
70. ws["A2"].font = SUB
71. cols = ["Client", "Company", "Plinth ID", "Email", "Phone", "Partner tier",
72. "Plinth LTV", "Last reached out", "Touch type", "Touches", "First reached out",
73. "Days since", "Current stage", "Owner", "Bitrix deal", "Deal link",
74. "How matched to Plinth", "Notes"]
75. wd = [30, 28, 13, 32, 15, 15, 12, 17, 30, 9, 17, 10, 30, 15, 11, 46, 20, 60]
76. if cat == "16":
77. cols.insert(2, "Cart value")
78. wd.insert(2, 12)
79. for i, c in enumerate(cols, 1):
80. x = ws.cell(row=4, column=i, value=c)
81. x.font = H
82. x.fill = HF
83. x.border = BD
84. x.alignment = Alignment(horizontal="left", vertical="center", wrap_text=True)
85. for i, w in enumerate(wd, 1):
86. ws.column_dimensions[L(i)].width = w
87. ws.freeze_panes = ws.cell(row=5, column=1)
88. rs.sort(key=lambda r: (r["last"] is None, -(r["last"] or dt.datetime(1900, 1, 1)).timestamp()))
89. rn = 5
90. for r in rs:
91. days = (today - r["last"]).days if r["last"] else None
92. vals = [r["name"], r["company"], r["plinth"], r["email"], r["phone"], r["tier"], r["ltv"],
93. r["last"], r["last_stage"], r["count"], r["first"], days, r["stage"],
94. r["owner"], int(r["deal"]) if str(r["deal"]).isdigit() else r["deal"],
95. f"https://meridian.bitrix24.example/crm/deal/details/{r['deal']}/",
96. r["match"], r["note"]]
97. if cat == "16":
98. vals.insert(2, r.get("value", ""))
99. for i, v in enumerate(vals, 1):
100. c = ws.cell(row=rn, column=i, value=v)
101. c.font = B
102. c.border = BD
103. hdr = cols[i - 1]
104. if hdr in ("Last reached out", "First reached out"):
105. c.number_format = "mm/dd/yyyy hh:mm AM/PM"
106. if hdr in ("Plinth LTV", "Cart value"):
107. c.number_format = "$#,##0.00"
108. if hdr == "Touches":
109. c.alignment = Alignment(horizontal="center")
110. if r["count"] == 0:
111. for i in range(1, len(vals) + 1):
112. ws.cell(row=rn, column=i).fill = NEV
113. elif days is not None and days > AUDIT_STALE_DAYS:
114. ws.cell(row=rn, column=cols.index("Days since") + 1).fill = STALE
115. if r["note"]:
116. ws.cell(row=rn, column=len(vals)).fill = WARN
117. rn += 1
118. ws.auto_filter.ref = f"A4:{L(len(cols))}{rn - 1}"
119. lastc = max((r["last"] for r in rs if r["last"]), default=None)
120. summary.append({"pipe": AUDIT_FUNNELS[cat], "cat": cat, "deals": len(rs),
121. "contacted": len(contacted), "never": len(never), "last": lastc,
122. "stale": sum(1 for r in contacted if r["last"] and (today - r["last"]).days > AUDIT_STALE_DAYS),
123. "prem": sum(1 for r in rs if r["tier"] and r["tier"] != "not partner"),
124. "unres": sum(1 for r in rs if r["match"] in UNRESOLVED_MATCHES)})
126. s = wb.create_sheet("Summary", 0)
127. s["A1"] = f"Bitrix Outreach Audit: every pipeline, current as of {TODAY}"
128. s["A1"].font = T
129. s["A2"] = ("One sheet per pipeline. Each row is a client with their Plinth locating details and "
130. "the date they were last actually reached out to. Generated by "
131. "07-Automation/weekly_partner_pipeline.py outreach-audit.")
132. s["A2"].font = SUB
133. hdrs = ["Pipeline", "Cat", "Deals", "Clients contacted", "Never contacted",
134. "Partner clients", "Most recent outreach", f"Contacted >{AUDIT_STALE_DAYS} days ago",
135. "No Plinth match"]
136. for i, c in enumerate(hdrs, 1):
137. x = s.cell(row=4, column=i, value=c)
138. x.font = H
139. x.fill = HF
140. x.border = BD
141. x.alignment = Alignment(wrap_text=True, vertical="center")
142. for i, w in enumerate([34, 7, 9, 17, 16, 15, 22, 22, 14], 1):
143. s.column_dimensions[L(i)].width = w
144. r = 5
145. for x in summary:
146. for i, v in enumerate([x["pipe"], x["cat"], x["deals"], x["contacted"], x["never"],
147. x["prem"], x["last"], x["stale"], x["unres"]], 1):
148. c = s.cell(row=r, column=i, value=v)
149. c.font = B
150. c.border = BD
151. if i == 7:
152. c.number_format = "mm/dd/yyyy"
153. r += 1
154. s.cell(row=r, column=1, value="TOTAL").font = BB
155. for i in (3, 4, 5, 6, 8, 9):
156. c = s.cell(row=r, column=i, value=f"=SUM({L(i)}5:{L(i)}{r - 1})")
157. c.font = BB
158. for i in range(1, 10):
159. s.cell(row=r, column=i).border = BD
160. s.cell(row=r, column=i).fill = PatternFill("solid", fgColor="D9E2F3")
161. r += 2
162. s.cell(row=r, column=1, value="How to read this").font = BB
163. r += 1
164. for line in [
165. '"Last reached out" is the newest stage entry with a real outbound email, call or LinkedIn DM behind it. '
166. 'System reactions (No - Reply, No Reply Nth Email (Auto), Email Defunct, Changed Companies, Not Interested, '
167. 'Completed) are NOT contact and are excluded; see SEND_TOKENS in weekly_partner_pipeline.py.',
168. 'Grey rows are clients in a pipeline who have never been contacted. Red "Days since" is older than '
169. f'{AUDIT_STALE_DAYS} days.',
170. '"Touches" is best-effort: same-minute workflow re-entries collapse to one send, but a bulk re-stage days '
171. 'apart still counts. Trust "Last reached out" over "Touches".',
172. '"Plinth ID" is the key for the Plinth backend. "How matched to Plinth" records which route found it, so a '
173. 'weak match is auditable rather than invisible.',
174. 'Order, production and fulfillment funnels are excluded by design: customer email there is order handling, '
175. 'not outreach. Same for the inbound Leads funnel.',
176. f'Freshness: {fresh} deal histories were re-pulled from Bitrix this run; {reused} were reused from '
177. f'07-Automation/data/outreach_touch_cache.json (terminal stage, unchanged since last run).',
178. f'Plinth identity resolved against: {Path(wbk).name if wbk else "NO WORKBOOK FOUND"}.',
179. ]:
180. c = s.cell(row=r, column=1, value="- " + line)
181. c.font = B
182. r += 1
183. if missing:
184. r += 1
185. c = s.cell(row=r, column=1, value=f"- DROPPED: {len(missing)} deals had no history and no cache entry: "
186. f"{', '.join(missing[:30])}")
187. c.font = BB
188. c.fill = WARN
189. if missing_funnels:
190. r += 1
191. c = s.cell(row=r, column=1,
192. value="- INCOMPLETE RUN: these outreach funnels were NOT in this run's enumeration and are "
193. "missing from this workbook entirely: "
194. + ", ".join(f"{AUDIT_FUNNELS[x]} (cat {x})" for x in sorted(missing_funnels))
195. + ". Do not read this workbook as a full picture.")
196. c.font = BB
197. c.fill = PatternFill("solid", fgColor="FFC7CE")
198. out = stem.with_suffix(".xlsx")
199. wb.save(out)
200. return out