run_sample.py

07-Automation/sample/run_sample.py

1. #!/usr/bin/env python3
2. """
3. run_sample.py: replay the sample's two Monday runs of the weekly pipeline.
5. Everything the vault ships under 03-Clients/, 05-Outputs/ (runs and audit) and
6. 07-Automation/data/ came out of this replay, apart from the research added by hand
7. to ten stubs, which a replay does not regenerate. It does what the scheduled Claude
8. session does in the real run (see scheduled-task.md), with the MCP calls stood
9. in for by files from sample/inputs/:
11. 0. once: seed the baseline from the roster workbook in two passes, copy it as
12. the first Drive-swap base, and start an empty census
13. 1. per run date, oldest first:
14. a. pull each day of the trailing window from Plinth and ingest it, paging
15. until a page adds nothing (the convergence loop; a missing next page
16. stands for a response that adds nothing)
17. b. classify, verify (exit 2 stops the run), write-stubs, refresh-xml,
18. refresh-workbook
19. c. reconcile-deals and route against that week's Bitrix enumeration
20. d. outreach-pull-list, then attach stage histories and contacts only for
21. the deals it asks for (the get_stage_history / get_record step), then
22. outreach-audit
24. The clock is pinned to each run date with freezegun, because the driver dates
25. its outputs by today's date. That is the only thing this script changes.
27. Usage (from the vault root, on a copy with the generated state removed; the
28. README's "Replay the sample" has the exact commands):
29. pip install -r 07-Automation/requirements.txt
30. python3 07-Automation/sample/generate_sample.py
31. python3 07-Automation/sample/run_sample.py
32. """
34. from __future__ import annotations
36. import argparse
37. import datetime as dt
38. import json
39. import re
40. import shutil
41. import subprocess
42. import sys
43. from pathlib import Path
45. HERE = Path(__file__).resolve().parent
46. VAULT = HERE.parent.parent
47. DRIVER = HERE.parent / "weekly_partner_pipeline.py"
49. CENSUS = "03-Clients/partner-client-master-list.xml"
50. CENSUS_ROOT = "partner_client_master_list"
51. ALL_TAB = "All Accounts"
52. PARTNER_TAB = "Complete Partner Clients List"
53. BASE_PREFIX = "Complete-Partner-Clients-List_updated-"
54. ROSTER_GLOB = "Partner-Clients-Roster_*.xlsx"
56. BASELINE_AS_OF = dt.date(2026, 9, 7)
57. RUN_DATES = [dt.date(2026, 9, 14), dt.date(2026, 9, 21)]
58. PULL_WINDOW_DAYS = 8
59. SUSPECT_ROWS = 50
61. BOOT = ("import sys, runpy, freezegun\n"
62. "freezegun.freeze_time(sys.argv[1]).start()\n"
63. "sys.argv = sys.argv[2:]\n"
64. "runpy.run_path(sys.argv[0], run_name='__main__')\n")
67. class Run:
68. def __init__(self, vault: Path, inputs: Path, driver: Path):
69. self.vault, self.inputs, self.driver = vault, inputs, driver
70. self.out = vault / "05-Outputs" / "partner-runs"
71. self.log: list[str] = []
73. def drive(self, when: str, *args: str, ok=(0,)) -> str:
74. cmd = [sys.executable, "-c", BOOT, when, str(self.driver), *args]
75. p = subprocess.run(cmd, cwd=self.vault, capture_output=True, text=True)
76. # Logs are kept in the vault, so print paths relative to it.
77. text = (p.stdout + p.stderr).replace(f"{self.vault}/", "")
78. shown = " ".join(a if " " not in a else f'"{a}"' for a in args)
79. self.log.append(f"$ weekly_partner_pipeline.py {shown}\n{text.rstrip()}\n")
80. print(f" {args[0]:<20} exit {p.returncode}")
81. if p.returncode not in ok:
82. print(text)
83. sys.exit(f"stopped: {args[0]} exited {p.returncode}")
84. return text
86. def seed(self):
87. roster = sorted((self.inputs).glob(ROSTER_GLOB))[-1]
88. data = self.vault / "07-Automation" / "data"
89. data.mkdir(parents=True, exist_ok=True)
90. for f in ("merge_registry.json", "suppression_list.json"):
91. if not (data / f).exists():
92. shutil.copy2(self.inputs / f, data / f)
93. rel = roster.relative_to(self.vault) if roster.is_relative_to(self.vault) else roster
94. when = f"{BASELINE_AS_OF.isoformat()} 18:00:00"
95. self.drive(when, "seed-baseline", "--sheet", str(rel), "--tab", ALL_TAB,
96. "--as-of", BASELINE_AS_OF.isoformat())
97. self.drive(when, "seed-baseline", "--sheet", str(rel), "--tab", PARTNER_TAB,
98. "--as-of", BASELINE_AS_OF.isoformat(), "--force")
99. self.out.mkdir(parents=True, exist_ok=True)
100. shutil.copy2(roster, self.out / f"{BASE_PREFIX}{BASELINE_AS_OF.isoformat()}.xlsx")
101. (self.vault / "03-Clients" / "client-stubs").mkdir(parents=True, exist_ok=True)
102. census = self.vault / CENSUS
103. if not census.exists():
104. census.parent.mkdir(parents=True, exist_ok=True)
105. census.write_text("<?xml version='1.0' encoding='UTF-8'?>\n"
106. f'<{CENSUS_ROOT} schema_version="3.0" />\n', encoding="utf-8")
108. def pull(self, run_date: dt.date, when: str):
109. raw = self.out / "raw" / run_date.isoformat()
110. raw.mkdir(parents=True, exist_ok=True)
111. for back in range(PULL_WINDOW_DAYS - 1, -1, -1):
112. day = run_date - dt.timedelta(days=back)
113. page = 0
114. while True:
115. src = self.inputs / "plinth" / f"{day.isoformat()}_p{page}.json"
116. if not src.exists():
117. break
118. dst = raw / src.name
119. shutil.copy2(src, dst)
120. text = self.drive(when, "ingest", "--orders", str(dst.relative_to(self.vault)))
121. m = re.search(r"parsed=(\d+) added=(\d+)", text)
122. parsed, added = (int(m.group(1)), int(m.group(2))) if m else (0, 0)
123. # Keep paging while a page adds orders or looks truncated.
124. if added == 0 and parsed < SUSPECT_ROWS:
125. break
126. page += 1
128. def week(self, run_date: dt.date):
129. d = run_date.isoformat()
130. when = f"{d} 06:00:00"
131. print(f"run {d}")
132. self.pull(run_date, when)
133. self.drive(when, "classify")
134. self.drive(when, "verify")
135. plan = f"05-Outputs/partner-runs/partner-classification_{d}.plan.json"
136. self.drive(when, "write-stubs", "--plan", plan)
137. self.drive(when, "refresh-xml", "--plan", plan)
138. self.drive(when, "refresh-workbook", "--plan", plan)
139. deals_in = self.inputs / "bitrix" / f"bitrix-deals_{d}.json"
140. deals = self.out / f"bitrix-deals_{d}.json"
141. shutil.copy2(deals_in, deals)
142. rel = str(deals.relative_to(self.vault))
143. self.drive(when, "reconcile-deals", "--deals", rel)
144. self.drive(when, "route")
145. self.drive(when, "outreach-pull-list", "--deals", rel)
146. pl = json.loads((self.out / f"outreach-pull-list_{d}.json").read_text(encoding="utf-8"))
147. hist = json.loads((self.inputs / "bitrix" / "stage-histories.json").read_text(encoding="utf-8"))
148. cons = json.loads((self.inputs / "bitrix" / "contacts.json").read_text(encoding="utf-8"))
149. need_h, need_c = set(pl["need_history"]), set(pl["need_contact"])
150. body = json.loads(deals.read_text(encoding="utf-8"))
151. for x in body["deals"]:
152. did = x["deal_id"]
153. if did in need_h:
154. x["history"] = [e for e in hist.get(did, []) if e["at"][:10] <= d]
155. if did in need_c and did in cons:
156. x["contact"] = cons[did]
157. deals.write_text(json.dumps(body, indent=1), encoding="utf-8")
158. self.drive(when, "outreach-audit", "--deals", rel)
159. (self.out / f"run-log_{d}.txt").write_text("\n".join(self.log), encoding="utf-8")
160. self.log = []
163. def main():
164. ap = argparse.ArgumentParser(description="Replay the sample's two weekly runs.")
165. ap.add_argument("--vault", default=str(VAULT))
166. ap.add_argument("--inputs", default=str(HERE / "inputs"))
167. ap.add_argument("--driver", default=str(DRIVER))
168. a = ap.parse_args()
169. r = Run(Path(a.vault).resolve(), Path(a.inputs).resolve(), Path(a.driver).resolve())
170. if (r.vault / "07-Automation" / "data" / "order_history.json").exists():
171. sys.exit("order_history.json already exists: this replay expects an unseeded vault "
172. "(see the README, 'Replay the sample').")
173. print("seed")
174. r.seed()
175. (r.out / f"run-log_{BASELINE_AS_OF.isoformat()}-seed.txt").write_text("\n".join(r.log), encoding="utf-8")
176. r.log = []
177. for day in RUN_DATES:
178. r.week(day)
179. print("done")
182. if __name__ == "__main__":
183. main()