roster_workbook.py

07-Automation/partner_pipeline/roster_workbook.py

1. """refresh-workbook: regenerate the Drive-swap roster workbook.
3. The vault copy of the roster is not connected to Google Drive (it is swapped in by
4. hand), so each run carries the human-maintained columns forward from the previous
5. generated copy (the base), refreshes only the Plinth-derived columns, appends newly
6. qualified partner clients, rebuilds the all-customer tab, and leaves the curated
7. working tab untouched. A human-maintained column on an existing row is never written.
8. """
10. from __future__ import annotations
12. import datetime as dt
13. import json
14. import sys
15. from pathlib import Path
17. from .config import NON_PARTNER, guard_or_exit
18. from .history import load_merge_registry
19. from .paths import OUT_DIR, TODAY
21. PARTNER_TAB = "Complete Partner Clients List"
22. ALL_TAB = "All Accounts"
25. def latest_updated_workbook(exclude: Path | None = None):
26. cands = sorted(OUT_DIR.glob("Complete-Partner-Clients-List_updated-*.xlsx"))
27. cands = [c for c in cands if exclude is None or c.resolve() != exclude.resolve()]
28. return cands[-1] if cands else None
31. def _as_date(s):
32. try:
33. return dt.date.fromisoformat(str(s)[:10])
34. except Exception:
35. return s or ""
38. def cmd_refresh_workbook(args):
39. guard_or_exit()
40. from openpyxl import load_workbook
41. from copy import copy as _copy
43. def _snap(cell):
44. # snapshot a cell's full visual style (survives row deletion)
45. return (_copy(cell.font), _copy(cell.fill), _copy(cell.border),
46. _copy(cell.alignment), cell.number_format, _copy(cell.protection))
48. def _apply(cell, snap):
49. cell.font, cell.fill, cell.border, cell.alignment = snap[0], snap[1], snap[2], snap[3]
50. cell.number_format = snap[4]
51. cell.protection = snap[5]
53. plan = json.loads(Path(args.plan).read_text(encoding="utf-8"))
54. rows = plan["rows"]
55. by_uid = {str(r["user_id"]): r for r in rows if r.get("user_id") not in (None, "")}
56. partner = [r for r in rows if r["assigned_tier"] not in NON_PARTNER]
57. absorbed_uids = set(load_merge_registry()[0])
58. stale_merged = []
60. out = OUT_DIR / f"Complete-Partner-Clients-List_updated-{TODAY}.xlsx"
61. base = Path(args.base) if args.base else latest_updated_workbook(exclude=out)
62. if not base or not base.exists():
63. sys.exit("No base workbook found (Complete-Partner-Clients-List_updated-*.xlsx in "
64. f"{OUT_DIR}); pass --base, or generate the first copy by hand.")
65. wb = load_workbook(base)
67. d = dt.date.fromisoformat(TODAY)
68. date_us = f"{d.month:02d}/{d.day:02d}/{str(d.year)[2:]}"
70. # Partner tab: refresh the Plinth columns, append newly qualified partners.
71. ws = wb[PARTNER_TAB]
72. hdr = [c.value for c in ws[1]]
74. def col(name, prefix=False):
75. for idx, h in enumerate(hdr):
76. if h is None:
77. continue
78. if (prefix and str(h).startswith(name)) or (not prefix and h == name):
79. return idx
80. return None
82. c_id = col("Id")
83. c_fn = col("First Name")
84. c_ln = col("Last Name")
85. c_em = col("Email")
86. c_co = col("Company")
87. c_ph = col("Phone")
88. c_ug = col("User Group")
89. c_cg = col("Correct Group")
90. c_ltvd = col("Plinth LTV", prefix=True) # dated lifetime column
91. c_tot = col("Total Spent (Orders)")
92. c_last = col("Last Order")
93. c_ord = col("# Orders")
94. c_rr = col("Repeat Revenue (Plinth)")
95. c_first = col("First Order")
97. if c_ltvd is not None: # rename dated header to this run
98. ws.cell(row=1, column=c_ltvd + 1).value = f"Plinth LTV {date_us}"
100. def _uid_of(v):
101. if v is None:
102. return None
103. if isinstance(v, float) and v.is_integer():
104. return str(int(v))
105. return str(v).strip()
107. # Per-column master style from an existing data row (row 2). row 2's Plinth cells are
108. # always populated and correctly formatted (currency, mm/dd/yyyy, Arial), so we clone
109. # it onto every Plinth cell we write (existing rows and appended rows), so formatting
110. # always matches the master, even on rows a prior run left unformatted. Plinth columns
111. # only on existing rows; human columns are never restyled and never revalued.
112. tmpl = [_snap(ws.cell(row=2, column=ci + 1)) for ci in range(len(hdr))] if ws.max_row >= 2 else None
113. plinth_cols = [c for c in (c_ug, c_ltvd, c_tot, c_last, c_ord, c_rr, c_first) if c is not None]
115. seen, refreshed = set(), 0
116. for row in ws.iter_rows(min_row=2):
117. uid = _uid_of(row[c_id].value)
118. if uid is None:
119. continue
120. seen.add(uid)
121. if uid in absorbed_uids:
122. stale_merged.append(uid) # merged-away account: leave the row alone,
123. continue # flag below for a human to strike by hand
124. r = by_uid.get(uid)
125. if not r:
126. continue
127. # Only Plinth-derived columns. Identity and all human columns are left untouched.
128. if c_ug is not None: row[c_ug].value = r.get("store_user_group") or ""
129. if c_ltvd is not None: row[c_ltvd].value = round(r["ltv"], 2)
130. if c_tot is not None: row[c_tot].value = round(r["ltv"], 2)
131. if c_last is not None: row[c_last].value = _as_date(r.get("last_order_date"))
132. if c_ord is not None: row[c_ord].value = r.get("total_orders_count") or 0
133. if c_rr is not None: row[c_rr].value = round(r.get("repeat_revenue", 0), 2)
134. if c_first is not None: row[c_first].value = _as_date(r.get("first_order_date"))
135. if tmpl: # normalize Plinth-cell formatting to the master
136. for ci in plinth_cols:
137. _apply(row[ci], tmpl[ci])
138. refreshed += 1
140. ncols, appended = len(hdr), 0
141. for r in partner:
142. uid = _uid_of(r.get("user_id"))
143. if uid is None or uid in seen:
144. continue
145. nr = [None] * ncols
146. def setc(ci, val):
147. if ci is not None:
148. nr[ci] = val
149. setc(c_id, int(uid) if uid.isdigit() else uid)
150. setc(c_fn, r.get("first_name") or "")
151. setc(c_ln, r.get("last_name") or "")
152. setc(c_em, r.get("email") or "")
153. setc(c_co, r.get("company") or "")
154. setc(c_ph, r.get("phone") or "")
155. setc(c_ug, r.get("store_user_group") or "")
156. setc(c_cg, r.get("assigned_tier") or "") # new row: record computed tier (no human value to preserve)
157. setc(c_ltvd, round(r["ltv"], 2))
158. setc(c_tot, round(r["ltv"], 2))
159. setc(c_last, _as_date(r.get("last_order_date")))
160. setc(c_first, _as_date(r.get("first_order_date")))
161. setc(c_ord, r.get("total_orders_count") or 0)
162. setc(c_rr, round(r.get("repeat_revenue", 0), 2))
163. ws.append(nr)
164. rr = ws.max_row
165. if tmpl:
166. for ci in range(ncols):
167. _apply(ws.cell(row=rr, column=ci + 1), tmpl[ci])
168. seen.add(uid)
169. appended += 1
171. # All Accounts: rebuild the all-customer base.
172. base_n = 0
173. if ALL_TAB in wb.sheetnames:
174. b = wb[ALL_TAB]
175. bcols = b.max_column
176. # snapshot the master row style per column before deleting (currency on LTV/RR etc.)
177. btmpl = [_snap(b.cell(row=2, column=c)) for c in range(1, bcols + 1)] if b.max_row >= 2 else None
178. if b.max_row > 1:
179. b.delete_rows(2, b.max_row - 1)
180. for r in rows:
181. uid = _uid_of(r.get("user_id"))
182. b.append([
183. int(uid) if (uid and uid.isdigit()) else (uid or ""),
184. r.get("first_name") or "", r.get("last_name") or "", r.get("email") or "",
185. round(r["ltv"], 2), round(r.get("repeat_revenue", 0), 2),
186. ])
187. base_n += 1
188. if btmpl: # re-apply the master per-column formatting to every rebuilt row
189. for rr in range(2, b.max_row + 1):
190. for c in range(1, bcols + 1):
191. _apply(b.cell(row=rr, column=c), btmpl[c - 1])
193. wb.save(out)
194. if stale_merged:
195. print(f"MERGE REGISTRY NOTE: partner-tab row(s) for absorbed account(s) "
196. f"{', '.join(stale_merged)} were left untouched: strike them by hand "
197. "(their spend now lives on the survivor's row).")
198. print(f"Workbook refreshed: base {base.name} -> {out.name}. "
199. f"Partner tab: {refreshed} rows refreshed, {appended} appended. "
200. f"All Accounts rebuilt to {base_n} customers. Dated col renamed 'Plinth LTV {date_us}'. "
201. f"Working tab and all human columns untouched.")
202. print("REMINDER: append one LEDGER entry for this change-set.")