stubs.py

07-Automation/partner_pipeline/stubs.py

1. """Client stubs: index, create, upgrade, and mirror pipeline blocks."""
3. from __future__ import annotations
5. import json
6. import re
7. from pathlib import Path
9. import yaml
11. from .config import NON_PARTNER, guard_or_exit
12. from .history import to_float
13. from .paths import CLIENTS_DIR, TODAY
16. def safe_stub_name(first, last, email="") -> str:
17. """Filesystem-safe stub filename. Names come from Plinth order fields and can
18. contain slashes or shipping-label junk that would otherwise break the path."""
19. raw = f"{(first or '').strip()} {(last or '').strip()}".strip() or (email or "").strip()
20. safe = re.sub(r'[/\\:*?"<>|\n\r\t]+', '-', raw) # kill path/illegal chars
21. safe = re.sub(r'\s+', ' ', safe).strip(' .-') # tidy whitespace and edges
22. return safe[:80].strip(' .-') or "unknown"
25. def _read_frontmatter(path: Path) -> dict:
26. try:
27. text = path.read_text(encoding="utf-8")
28. except Exception:
29. return {}
30. m = re.match(r"^---\n(.*?)\n---", text, re.DOTALL)
31. if not m:
32. return {}
33. try:
34. return yaml.safe_load(m.group(1)) or {}
35. except yaml.YAMLError:
36. return {}
39. def index_existing_stubs() -> dict:
40. by_email, by_name = {}, {}
41. if not CLIENTS_DIR.exists():
42. return {"by_email": by_email, "by_name": by_name}
43. for p in CLIENTS_DIR.glob("*.md"):
44. fm = _read_frontmatter(p)
45. if not fm:
46. continue
47. rec = {"path": p, "fm": fm}
48. email = (str(fm.get("email") or "")).strip().lower()
49. if email:
50. by_email[email] = rec
51. fn = (str(fm.get("first_name") or "")).strip().lower()
52. ln = (str(fm.get("last_name") or "")).strip().lower()
53. if fn or ln:
54. by_name[f"{fn} {ln}".strip()] = rec
55. return {"by_email": by_email, "by_name": by_name}
58. def stub_status(c: dict, idx: dict) -> tuple[str, Path | None]:
59. email = c["email"]
60. name = f"{c['first_name']} {c['last_name']}".strip().lower()
61. rec = idx["by_email"].get(email) or idx["by_name"].get(name)
62. if not rec:
63. return "new", None
64. fm = rec["fm"]
65. changed = (
66. to_float(fm.get("ltv")) != c["ltv"]
67. or int(fm.get("total_orders_count") or 0) != c["total_orders_count"]
68. )
69. return ("exists-changed" if changed else "exists"), rec["path"]
72. def cmd_write_stubs(args):
73. guard_or_exit()
74. plan = json.loads(Path(args.plan).read_text(encoding="utf-8"))
75. idx = index_existing_stubs()
76. created, updated, flagged = [], [], []
77. for r in plan["rows"]:
78. if r["assigned_tier"] in NON_PARTNER:
79. continue
80. email = r["email"]
81. name = f"{(r.get('first_name') or '')} {(r.get('last_name') or '')}".strip()
82. rec = idx["by_email"].get(email) or idx["by_name"].get(name.lower())
83. if rec:
84. _upgrade_stub(rec["path"], r)
85. updated.append(rec["path"].name)
86. else:
87. path = _create_stub(r)
88. created.append(path.name)
89. if re.search(r'[/\\:]|c/o|booth|attn', name, re.IGNORECASE):
90. flagged.append(f"{name} -> {path.name}")
91. print(f"Stubs created: {len(created)}; upgraded: {len(updated)}")
92. if flagged:
93. print(f"\nFLAGGED: {len(flagged)} stub(s) created from messy Plinth name fields "
94. "(filename sanitized; review and rename or delete if not a real client):")
95. for f in flagged:
96. print(" ", f)
97. print("REMINDER: append one LEDGER entry for this change-set (attribution by session email).")
100. def _create_stub(r: dict) -> Path:
101. display = f"{(r.get('first_name') or '')} {(r.get('last_name') or '')}".strip() or r["email"]
102. fname = safe_stub_name(r.get("first_name"), r.get("last_name"), r["email"])
103. path = CLIENTS_DIR / f"{fname}.md"
104. if path.exists(): # don't clobber an existing same-name stub
105. path = CLIENTS_DIR / f"{fname} ({r.get('user_id', '')}).md"
106. fm = {
107. "type": "client",
108. "bitrix_id": "",
109. "first_name": r["first_name"],
110. "last_name": r["last_name"],
111. "email": r["email"],
112. "user_group": "",
113. "user_group_correct": r["assigned_tier"],
114. "industry": "",
115. "company": r.get("company", ""),
116. "phone": r.get("phone", ""),
117. "ltv": f"${r['ltv']:,.2f}",
118. "repeat_revenue": f"${r['repeat_revenue']:,.2f}",
119. "title": "",
120. "company_website": "",
121. "total_orders_count": r["total_orders_count"],
122. "last_order_date": r["last_order_date"],
123. "source": "weekly_partner_pipeline",
124. "last_modified": TODAY,
125. "tags": ["client", "partner-pipeline"],
126. }
127. body = (
128. f"\n# {display}\n\n## Snapshot\n| Field | Value |\n|-------|-------|\n"
129. f"| Title | [TBD] |\n| Company | {r.get('company','')} |\n"
130. f"| Email | {r['email']} |\n| Phone | {r.get('phone','')} |\n\n"
131. "## Order History\n"
132. f"- Total orders: {r['total_orders_count']}\n"
133. f"- Lifetime value: ${r['ltv']:,.2f}\n"
134. f"- Repeat revenue: ${r['repeat_revenue']:,.2f}\n"
135. f"- Last order: {r['last_order_date']}\n\n"
136. "## Research\n<!-- Enriched via client-enrich. Never overwrite Human Notes. -->\n"
137. "### Company\n[TBD]\n### Person / LinkedIn\n[TBD]\n### Recent News & Signals\n[TBD]\n### Sources\n\n"
138. "## Human Notes\n<!-- Matt's space. Automation never writes here. -->\n\n"
139. "## Outreach Log\n| Date | Pipeline | Channel | Template | Outcome | Next step |\n"
140. "|------|----------|---------|----------|---------|-----------|\n"
141. )
142. path.write_text("---\n" + yaml.safe_dump(fm, sort_keys=False, allow_unicode=True) + "---\n" + body,
143. encoding="utf-8")
144. return path
147. def _upgrade_stub(path: Path, r: dict):
148. """Update only computed frontmatter + last_modified. Never touch Human Notes/Research."""
149. text = path.read_text(encoding="utf-8")
150. m = re.match(r"^---\n(.*?)\n---(.*)$", text, re.DOTALL)
151. if not m:
152. return
153. fm = yaml.safe_load(m.group(1)) or {}
154. fm["ltv"] = f"${r['ltv']:,.2f}"
155. fm["repeat_revenue"] = f"${r['repeat_revenue']:,.2f}"
156. fm["total_orders_count"] = r["total_orders_count"]
157. fm["last_order_date"] = r["last_order_date"]
158. fm["user_group_correct"] = r["assigned_tier"]
159. fm["last_modified"] = TODAY
160. path.write_text("---\n" + yaml.safe_dump(fm, sort_keys=False, allow_unicode=True) + "---" + m.group(2),
161. encoding="utf-8")
164. def mirror_stub_pipeline(stub_path: Path, key: str, stage: str, status: str, entered: str):
165. """Copy a census pipeline block into the stub's `outreach_pipelines` frontmatter.
166. Returns True when the stub changed. A failure warns and never stops the run."""
167. try:
168. txt = stub_path.read_text(encoding="utf-8")
169. m = re.match(r"^---\n(.*?\n)---(.*)$", txt, re.DOTALL)
170. if not m:
171. return False
172. fm = yaml.safe_load(m.group(1)) or {}
173. op = fm.get("outreach_pipelines") or {}
174. e = op.get(key) or {}
175. if e.get("stage") == stage and e.get("status") == status:
176. return False
177. e.update({"stage": stage, "status": status})
178. if not e.get("entered"):
179. e["entered"] = entered or None
180. for f in ("last_touch", "next_step", "next_date"):
181. e.setdefault(f, None)
182. op[key] = e
183. fm["outreach_pipelines"] = op
184. fm["last_modified"] = TODAY
185. stub_path.write_text("---\n" + yaml.safe_dump(fm, sort_keys=False,
186. allow_unicode=True) + "---" + m.group(2), encoding="utf-8")
187. return True
188. except Exception as ex:
189. print(f"WARNING: stub mirror failed for {stub_path.name}: {ex}")
190. return False