baseline.py

07-Automation/partner_pipeline/baseline.py

1. """seed-baseline: load the roster workbook as per-customer baselines (one time)."""
3. from __future__ import annotations
5. import csv
6. import datetime as dt
7. import json
8. import re
9. import sys
10. from pathlib import Path
12. from .history import _cust, to_float, load_history, save_history
13. from .paths import HISTORY
16. # Fuzzy header → canonical field. Override any of these with --map "field=Header".
17. BASELINE_FIELDS = {
18. "user_id": ["id", "plinth id", "plinth_id", "customer id", "user id", "user_id", "customer_id"],
19. "email": ["email", "email address", "primary email", "e-mail", "main email"],
20. "additional_emails": ["additional emails", "additional_emails", "other emails",
21. "second email", "additional email", "additional email 1"],
22. "first_name": ["first name", "first_name", "first"],
23. "last_name": ["last name", "last_name", "last"],
24. "company": ["company", "company name", "organization"],
25. "phone": ["phone", "telephone", "phone number"],
26. "ltv": ["ltv", "total spend", "total_spend", "lifetime value", "total spent",
27. "total historical spend", "total spent (orders)"],
28. "orders_count": ["total orders", "orders", "order count", "total_orders_count",
29. "# of orders", "number of orders", "# orders"],
30. "repeat_revenue": ["repeat revenue", "repeat_revenue", "repeat revenue (plinth)", "rr"],
31. "first_order_date": ["first order date", "first order", "first_order_date"],
32. "last_order_date": ["last order date", "last order", "last_order_date"],
33. "user_group": ["user group", "user_group", "tier", "group"],
34. }
35. REQUIRED_BASELINE = ["user_id", "ltv"]
38. def _match_headers(headers: list[str], overrides: dict) -> dict:
39. """Return {canonical: column_index} using overrides first, then fuzzy match."""
40. norm = [re.sub(r"\s+", " ", (h or "").strip().lower()) for h in headers]
41. mapping = {}
42. for field, aliases in BASELINE_FIELDS.items():
43. if field in overrides:
44. want = overrides[field].strip().lower()
45. for i, h in enumerate(norm):
46. if h == want:
47. mapping[field] = i
48. break
49. else:
50. sys.exit(f"--map {field}={overrides[field]!r}: no such column. Headers: {headers}")
51. continue
52. for i, h in enumerate(norm):
53. if h in aliases:
54. mapping[field] = i
55. break
56. return mapping
59. def _read_sheet(path: Path, tab: str | None = None) -> list[list]:
60. if path.suffix.lower() in (".xlsx", ".xlsm", ".xltx"):
61. try:
62. from openpyxl import load_workbook
63. except ImportError:
64. sys.exit("Reading xlsx needs openpyxl: pip install -r 07-Automation/requirements.txt")
65. wb = load_workbook(path, read_only=True, data_only=True)
66. if tab:
67. if tab not in wb.sheetnames:
68. sys.exit(f"No tab named {tab!r}. Tabs in this workbook: {wb.sheetnames}")
69. ws = wb[tab]
70. else:
71. ws = wb.active
72. if len(wb.sheetnames) > 1:
73. print(f"NOTE: multi-tab workbook, reading active tab {ws.title!r}. "
74. f"Use --tab to pick another: {wb.sheetnames}")
75. return [[c for c in row] for row in ws.iter_rows(values_only=True)]
76. if tab:
77. sys.exit("--tab only applies to xlsx workbooks.")
78. with path.open(newline="", encoding="utf-8-sig") as f:
79. return [row for row in csv.reader(f)]
82. def cmd_seed_baseline(args):
83. rows = _read_sheet(Path(args.sheet), args.tab)
84. if not rows:
85. sys.exit("Empty sheet.")
86. overrides = dict(kv.split("=", 1) for kv in (args.map or []))
87. headers = [str(h) if h is not None else "" for h in rows[0]]
88. mapping = _match_headers(headers, overrides)
90. missing = [f for f in REQUIRED_BASELINE if f not in mapping]
91. print(f"Sheet: {args.sheet}: {len(rows) - 1} data rows")
92. print("Detected column mapping:")
93. for f, i in sorted(mapping.items()):
94. print(f" {f:<18} <- column {i}: {headers[i]!r}")
95. if missing:
96. sys.exit(f"MISSING required column(s): {missing}. "
97. f"Fix with --map \"{missing[0]}=<exact header>\". Headers: {headers}")
99. as_of = args.as_of
100. if not as_of:
101. sys.exit("--as-of YYYY-MM-DD is required: the date the sheet's totals run through "
102. "(the roster's totals run 'up until Monday', so use that Monday). The weekly delta "
103. "counts only orders AFTER this date, so it must be exact.")
104. dt.date.fromisoformat(as_of) # validate
106. if args.dry_run:
107. print("\nDRY RUN: first 3 rows as parsed:")
108. for row in rows[1:4]:
109. parsed = {f: row[i] if i < len(row) else None for f, i in mapping.items()}
110. print(" ", parsed)
111. print("\nNothing written. Re-run without --dry-run to seed.")
112. return
114. if HISTORY.exists() and not args.force:
115. sys.exit(f"History already exists: {HISTORY}\n"
116. "seed-baseline is a one-time operation. Use --force to re-seed "
117. "(this replaces baselines but keeps ingested orders).")
119. h = load_history(required=False) if not HISTORY.exists() else json.loads(
120. HISTORY.read_text(encoding="utf-8"))
121. seeded = 0
122. for row in rows[1:]:
123. def val(f):
124. i = mapping.get(f)
125. return row[i] if i is not None and i < len(row) else None
126. uid = val("user_id")
127. if uid in (None, ""):
128. continue
129. uid = str(uid).strip().split(".")[0] # xlsx floats: 40115.0 -> "40115"
130. c = _cust(h, uid, str(val("email") or ""))
131. for f in ("email", "first_name", "last_name", "company", "phone"):
132. v = val(f)
133. if v not in (None, ""):
134. c[f] = str(v).strip().lower() if f == "email" else str(v).strip()
135. ug = val("user_group")
136. if ug not in (None, ""):
137. c["store_user_group"] = str(ug).strip()
138. emails = [c["email"]] if c["email"] else []
139. extra = val("additional_emails")
140. if extra not in (None, ""):
141. emails += [e.strip().lower() for e in re.split(r"[;,|]", str(extra)) if e.strip()]
142. c["known_emails"] = sorted(set(emails + c.get("known_emails", [])))
143. c["baseline"] = {
144. "as_of": as_of,
145. "ltv": round(to_float(val("ltv")), 2),
146. "orders_count": int(to_float(val("orders_count"))),
147. "repeat_revenue": round(to_float(val("repeat_revenue")), 2),
148. "first_order_date": str(val("first_order_date") or "")[:10],
149. "last_order_date": str(val("last_order_date") or "")[:10],
150. "source": Path(args.sheet).name + (f" [{args.tab}]" if args.tab else ""),
151. }
152. seeded += 1
153. h["meta"]["baseline_source"] = Path(args.sheet).name
154. h["meta"]["baseline_as_of"] = as_of
155. save_history(h)
156. print(f"\nSeeded baselines for {seeded} customers (as_of={as_of}) -> {HISTORY}")