history.py

07-Automation/partner_pipeline/history.py

1. """The local order history, the merge registry, and ingest."""
3. from __future__ import annotations
5. import json
6. import sys
7. from pathlib import Path
9. from .config import load_config
10. from .paths import DATA_DIR, HISTORY, MERGE_REGISTRY, NOW, TODAY
13. def load_history(required: bool = True) -> dict:
14. if not HISTORY.exists():
15. if required:
16. sys.exit(f"History store not found: {HISTORY}\n"
17. "Run seed-baseline first (or ingest, for a no-baseline start).")
18. return {"meta": {"schema": 1, "created": NOW, "baseline_source": None,
19. "last_ingest": None, "previous_computed": {}},
20. "customers": {}}
21. return json.loads(HISTORY.read_text(encoding="utf-8"))
24. def save_history(h: dict):
25. DATA_DIR.mkdir(parents=True, exist_ok=True)
26. tmp = HISTORY.with_suffix(".json.tmp")
27. tmp.write_text(json.dumps(h, indent=1), encoding="utf-8")
28. tmp.replace(HISTORY)
31. def _cust(h: dict, user_id, email="") -> dict:
32. key = str(user_id) if user_id not in (None, "", 0) else f"email:{(email or '').lower()}"
33. return h["customers"].setdefault(key, {
34. "user_id": user_id, "email": (email or "").lower(),
35. "first_name": "", "last_name": "", "company": "", "phone": "",
36. "store_user_group": "", "baseline": None, "orders": {},
37. })
40. def load_merge_registry() -> tuple[dict, list]:
41. """Return ({absorbed_key: survivor_key}, raw_entries). Tolerant by design:
42. a missing or malformed registry must never kill the scheduled run; it only
43. means merges don't apply this week."""
44. if not MERGE_REGISTRY.exists():
45. return {}, []
46. try:
47. reg = json.loads(MERGE_REGISTRY.read_text(encoding="utf-8"))
48. entries = reg.get("merges", [])
49. except (json.JSONDecodeError, OSError, AttributeError) as e:
50. print(f"WARNING: merge registry unreadable ({e}): continuing UNMERGED. "
51. f"Fix {MERGE_REGISTRY} and re-run.")
52. return {}, []
53. aliases: dict[str, str] = {}
54. survivors = set()
55. for m in entries:
56. s = str(m.get("survivor") or "").strip()
57. if not s:
58. print(f"WARNING: registry entry with no survivor skipped: {m}")
59. continue
60. survivors.add(s)
61. for a in m.get("absorbed") or []:
62. a = str(a).strip()
63. if not a or a == s:
64. continue
65. if a in aliases and aliases[a] != s:
66. print(f"WARNING: {a} absorbed by both {aliases[a]} and {s}: "
67. "keeping the first, fix the registry.")
68. continue
69. aliases[a] = s
70. bad = survivors & set(aliases)
71. if bad:
72. print(f"WARNING: {sorted(bad)} are both survivor and absorbed (chained "
73. "merges are not supported): those entries may not fold fully.")
74. return aliases, entries
77. def apply_merges(metrics: dict, aliases: dict) -> tuple[dict, list[str]]:
78. """Fold absorbed accounts' computed metrics into their survivors. Runs after
79. compute_metrics, before classification, so tiers see combined lifetime value
80. and absorbed accounts never classify, get stubs, or enter the XML census."""
81. folded = []
82. for a, s in aliases.items():
83. if a not in metrics:
84. continue # already folded / not in history: no-op
85. if s not in metrics:
86. print(f"WARNING: merge survivor {s} not in history: leaving "
87. f"absorbed account {a} unmerged.")
88. continue
89. av, sv = metrics.pop(a), metrics[s]
90. sv["ltv"] = round(sv["ltv"] + av["ltv"], 2)
91. sv["repeat_revenue"] = round(sv["repeat_revenue"] + av["repeat_revenue"], 2)
92. sv["total_orders_count"] += av["total_orders_count"]
93. sv["post_baseline_orders"] += av["post_baseline_orders"]
94. firsts = [d for d in (sv["first_order_date"], av["first_order_date"]) if d]
95. sv["first_order_date"] = min(firsts) if firsts else ""
96. sv["last_order_date"] = max(sv["last_order_date"], av["last_order_date"])
97. sv["has_baseline"] = sv["has_baseline"] or av["has_baseline"]
98. sv.setdefault("merged_from", []).append(a)
99. folded.append(f"{a}->{s}")
100. return metrics, folded
103. def _iter_orders(raw) -> list[dict]:
104. """Collect the order dicts from whatever shape the Plinth dump takes."""
105. out = []
107. def eat(obj):
108. if isinstance(obj, list):
109. for o in obj:
110. eat(o)
111. elif isinstance(obj, dict):
112. if "order_map" in obj and isinstance(obj["order_map"], dict):
113. for o in obj["order_map"].values():
114. out.append(o)
115. elif "order_ref" in obj:
116. out.append(obj)
117. else:
118. for v in obj.values():
119. if isinstance(v, dict) and "order_ref" in v:
120. out.append(v)
121. else:
122. eat(v)
124. eat(raw)
125. return out
128. def to_float(v) -> float:
129. try:
130. return float(str(v).replace("$", "").replace(",", "").strip() or 0)
131. except (TypeError, ValueError):
132. return 0.0
135. def _order_date(o) -> str | None:
136. d = o.get("closed_on") or ""
137. return d.split(" ")[0] if d else None
140. def cmd_ingest(args):
141. cfg = load_config()
142. suspect_at = int(cfg.get("suspect_row_count") or 50)
143. h = load_history(required=False)
144. added = updated = seen = 0
145. for path in args.orders:
146. raw = json.loads(Path(path).read_text(encoding="utf-8"))
147. orders = _iter_orders(raw)
148. seen += len(orders)
149. for o in orders:
150. oid = str(o.get("order_ref"))
151. if oid in (None, "None", ""):
152. continue
153. c = _cust(h, o.get("account_ref"), o.get("email"))
154. for f, k in (("first_name", "first_name"),
155. ("last_name", "last_name"),
156. ("company", "company_name"),
157. ("phone", "phone"),
158. ("store_user_group", "customer_group")):
159. v = o.get(k)
160. if v not in (None, ""):
161. c[f] = str(v)
162. em = (o.get("email") or "").strip().lower()
163. if em:
164. c["email"] = c["email"] or em
165. if em not in c.get("known_emails", []):
166. c.setdefault("known_emails", []).append(em)
167. rec = {
168. "grand_total": to_float(o.get("grand_total")),
169. "subtotal": to_float(o.get("subtotal")),
170. "date": _order_date(o),
171. "status": o.get("status_code"),
172. "last_seen": TODAY,
173. }
174. if oid in c["orders"]:
175. prev = c["orders"][oid]
176. rec["first_seen"] = prev.get("first_seen", TODAY)
177. if any(prev.get(k) != rec[k] for k in ("grand_total", "subtotal", "status", "date")):
178. updated += 1
179. c["orders"][oid] = rec
180. else:
181. rec["first_seen"] = TODAY
182. c["orders"][oid] = rec
183. added += 1
184. h["meta"]["last_ingest"] = NOW
185. save_history(h)
186. # The convergence loop reads this line: keep paging while "added" > 0.
187. print(f"INGEST: parsed={seen} added={added} updated={updated} "
188. f"customers={len(h['customers'])}")
189. if seen >= suspect_at * len(args.orders):
190. print(f"NOTE: page(s) at/over suspect_row_count ({suspect_at}): likely truncated. "
191. "Keep paging (offset) or narrow the date window until added=0.")
194. def compute_metrics(h: dict, cfg: dict) -> dict:
195. amount_field = cfg.get("amount_field", "grand_total")
196. counted = set(cfg.get("counted_order_statuses") or [])
197. excluded = set(cfg.get("excluded_order_statuses") or [])
198. out = {}
199. for key, c in h["customers"].items():
200. base = c.get("baseline")
201. as_of = (base or {}).get("as_of", "")
202. post = []
203. for oid, o in c["orders"].items():
204. if o.get("status") in excluded:
205. continue
206. if counted and o.get("status") not in counted:
207. continue
208. d = o.get("date")
209. # With a baseline, only orders strictly after as_of add to it.
210. # Undated orders (not yet finished) are treated as recent -> post.
211. if base and d and d <= as_of:
212. continue
213. post.append(o)
214. post_sum = round(sum(to_float(o.get(amount_field)) for o in post), 2)
215. post_dates = sorted(o["date"] for o in post if o.get("date"))
216. if base:
217. ltv = round(base["ltv"] + post_sum, 2)
218. n = base["orders_count"] + len(post)
219. first = base.get("first_order_date") or (post_dates[0] if post_dates else "")
220. last = max(filter(None, [base.get("last_order_date", ""),
221. post_dates[-1] if post_dates else ""])) if (
222. base.get("last_order_date") or post_dates) else ""
223. post_repeat = post_sum if base["orders_count"] > 0 else (
224. round(post_sum - to_float(post[0].get(amount_field)), 2) if post else 0.0)
225. repeat = round((base.get("repeat_revenue") or 0.0) + post_repeat, 2)
226. else:
227. ltv, n = post_sum, len(post)
228. first = post_dates[0] if post_dates else ""
229. last = post_dates[-1] if post_dates else ""
230. amts = [to_float(o.get(amount_field)) for o in
231. sorted(post, key=lambda x: x.get("date") or "9999")]
232. repeat = round(sum(amts[1:]), 2) if len(amts) > 1 else 0.0
233. out[key] = {
234. "user_id": c.get("user_id"), "email": c.get("email", ""),
235. "first_name": c.get("first_name", ""), "last_name": c.get("last_name", ""),
236. "company": c.get("company", ""), "phone": c.get("phone", ""),
237. "store_user_group": c.get("store_user_group", ""),
238. "has_baseline": bool(base), "post_baseline_orders": len(post),
239. "total_orders_count": n, "ltv": ltv, "repeat_revenue": repeat,
240. "first_order_date": first or "", "last_order_date": last or "",
241. }
242. return out