← back to Maya Width Fix
pipeline.py
138 lines
#!/usr/bin/env python3
"""
pipeline.py — TK-11029 deterministic DRY-RUN correction pipeline for
maya_catalog.width meta-tag pollution.
DRY-RUN ONLY. Produces, deterministically and reproducibly, the exact per-row
correction plan by joining current DB rows to the authoritative mayaromanoff.com
source (via a source-proof JSON, or a fresh read-only re-scrape), classifying each
with widthlib, and emitting:
- dryrun-plan-<ts>.json the proposed UPDATE plan (NOT applied)
- restore-map-dryrun-<ts>.json rollback artifact: current state of every row the
plan would touch (finalizer applies UPDATEs, then
restore.py <this-map> is the one-command undo)
This script performs NO database writes and NO Shopify/deploy/send actions.
Applying the plan is the finalizer's (/root) gated step — see README.md.
Modes:
(default) scope = currently-polluted rows only (width LIKE '%device-width%')
--simulate-all scope = ALL 224 DWMR-8 rows, re-derived from source (full-line
deterministic plan; proves the correction logic end-to-end)
--source PATH use an existing authoritative-source-proof-*.json (skip network)
"""
import subprocess, json, datetime, os, sys, glob
from widthlib import classify_width
HERE = os.path.dirname(os.path.abspath(__file__))
PSQL = ["psql", "-h", "/tmp", "-d", "dw_unified", "-At", "-F", "\t", "-c"]
def db(sql):
r = subprocess.run(PSQL + [sql], capture_output=True, text=True)
if r.returncode != 0:
print("SQL ERR:", r.stderr, file=sys.stderr); sys.exit(1)
return [ln.split("\t") for ln in r.stdout.strip().split("\n") if ln]
def load_source(path=None):
"""slug -> {decision, width, width_inches, http_status} from a source-proof JSON."""
if not path:
cands = sorted(glob.glob(os.path.join(HERE, "authoritative-source-proof-*.json")))
if not cands:
print("No source-proof JSON found; run authoritative_source.py first "
"or pass --source.", file=sys.stderr)
sys.exit(2)
path = cands[-1]
data = json.load(open(path))
smap = {s["slug"]: s for s in data["authoritative_source"]}
return smap, path
def main():
args = sys.argv[1:]
simulate_all = "--simulate-all" in args
src_path = None
if "--source" in args:
src_path = args[args.index("--source") + 1]
if "--apply" in args:
print("REFUSED: pipeline.py is DRY-RUN ONLY. Applying maya_catalog writes is a\n"
"canonical-DB change owned by the finalizer (/root) under approval. "
"See README.md.", file=sys.stderr)
return 3
ts = datetime.datetime.now(datetime.timezone.utc).isoformat()
smap, used_src = load_source(src_path)
where = "" if simulate_all else "AND width LIKE '%device-width%'"
rows = db(f"""SELECT dw_sku, mfr_sku, width, coalesce(width_inches::text,''),
split_part(product_url,'/collections/',2)
FROM maya_catalog WHERE dw_sku LIKE 'DWMR-8%' {where} ORDER BY dw_sku""")
plan, restore = [], []
tally = {"FIX_SINGLE": 0, "KEEP_MULTI_WIDTH": 0, "UNRECOVERABLE_404": 0,
"SOURCE_MISSING": 0, "NO_CHANGE": 0}
for dw_sku, mfr_sku, cur_w, cur_wi, slug in rows:
s = smap.get(slug)
restore.append({"dw_sku": dw_sku, "old_width": cur_w, "old_width_inches": cur_wi})
if s is None:
decision, new_w, new_wi = "SOURCE_MISSING", None, None
elif s.get("http_status") == 404 or s.get("width") is None:
decision, new_w, new_wi = "UNRECOVERABLE_404", None, None
else:
dec, wi = classify_width(s["width"])
if dec == "SINGLE":
decision, new_w, new_wi = "FIX_SINGLE", s["width"], wi
elif dec == "MULTI_WIDTH":
decision, new_w, new_wi = "KEEP_MULTI_WIDTH", s["width"], None
else:
decision, new_w, new_wi = "SOURCE_MISSING", None, None
# NO_CHANGE if the DB already holds exactly what we'd write
if new_w is not None and cur_w == new_w and (cur_wi or "") == (new_wi or ""):
decision = "NO_CHANGE"
tally[decision] = tally.get(decision, 0) + 1
entry = {"dw_sku": dw_sku, "mfr_sku": mfr_sku, "slug": slug,
"current_width": cur_w, "current_width_inches": cur_wi,
"decision": decision, "proposed_width": new_w,
"proposed_width_inches": new_wi}
# only rows that would actually change are "actionable"
if decision in ("FIX_SINGLE", "KEEP_MULTI_WIDTH"):
entry["sql"] = _update_sql(dw_sku, new_w, new_wi)
plan.append(entry)
actionable = [p for p in plan if p["decision"] in ("FIX_SINGLE", "KEEP_MULTI_WIDTH")]
out = {
"ticket": "TK-11029", "generated_at": ts, "mode": "DRY-RUN (no writes)",
"scope": "all-224-simulated" if simulate_all else "currently-polluted-only",
"source_proof_used": os.path.basename(used_src),
"rows_in_scope": len(rows), "actionable_updates": len(actionable),
"tally": tally, "plan": plan,
}
plan_path = os.path.join(HERE, f"dryrun-plan-{ts.replace(':', '-')}.json")
json.dump(out, open(plan_path, "w"), indent=2)
rm = {"ticket": "TK-11029", "generated_at": ts, "table": "maya_catalog",
"columns": "width,width_inches", "restore_map": restore,
"note": "finalizer applies dryrun-plan then `python3 restore.py <this>` = undo"}
rm_path = os.path.join(HERE, f"restore-map-dryrun-{ts.replace(':', '-')}.json")
json.dump(rm, open(rm_path, "w"), indent=2)
print(f"scope={out['scope']} rows={len(rows)} actionable={len(actionable)} tally={tally}")
print(f"DRY-RUN PLAN -> {plan_path}")
print(f"RESTORE-MAP -> {rm_path}")
if not simulate_all and tally.get("UNRECOVERABLE_404"):
print(f"NOTE: {tally['UNRECOVERABLE_404']} residual row(s) unrecoverable from web "
f"(collection 404) -> needs Maya rep/PDF; DO NOT guess.")
return 0
def _update_sql(dw_sku, w, wi):
esc = lambda s: (s or "").replace("'", "''")
wi_clause = f"width_inches = {wi}" if wi else "width_inches = NULL"
return (f"UPDATE maya_catalog SET width='{esc(w)}', {wi_clause}, updated_at=now() "
f"WHERE dw_sku='{esc(dw_sku)}';")
if __name__ == "__main__":
sys.exit(main())