`authoritative_source.py`

128 lines
#!/usr/bin/env python3
"""
authoritative_source.py — TK-11029 authoritative-field mapping + SOURCE PROOF.

READ-ONLY. Re-fetches each Maya Romanoff collection page ($0 plain-fetch GET,
no browser/proxy/creds), extracts the spec-table Width value, classifies it with
widthlib, and (if the DB is reachable) proves the already-applied maya_catalog
widths equal the live authoritative source.

Emits: authoritative-source-proof-<ts>.json
Never writes the DB. Network egress = read-only GETs of public product pages.
"""
import urllib.request, re, json, subprocess, sys, datetime, os
from widthlib import classify_width, is_meta_pollution

SLUGS = [
    "island-weaves", "craze", "natural-elements", "braided-hemp", "ajiro-burst-of-happiness",
    "entwine-inlet-linen", "serigraph-cityscape", "beadazzled-leaf", "entwine-tranquil-linen",
    "bouquet", "entwine-serene-silk", "wisping-weaves-montauk", "beadazzled",
    "wisping-weaves-cape-may", "cozy-nestle", "cozy-bed-fellow", "coco-chenille",
    "beadazzled-geode", "hurly-burly-ii", "ajiro-sunburst",
]
UA = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
      "(KHTML, like Gecko) Chrome/120 Safari/537.36")
WIDTH_RX = re.compile(r'>\s*Width\s*</td>\s*<td[^>]*>\s*([^<]{1,120}?)\s*</td>', re.I)
HERE = os.path.dirname(os.path.abspath(__file__))


def fetch_width(slug, attempts=3):
    url = f"https://www.mayaromanoff.com/collections/{slug}"
    last = None
    for i in range(attempts):
        try:
            req = urllib.request.Request(url, headers={"User-Agent": UA})
            with urllib.request.urlopen(req, timeout=30) as r:
                html = r.read().decode("utf-8", "ignore")
                status = r.status
            break
        except urllib.error.HTTPError as e:
            # 404 is authoritative (discontinued) — do not retry
            return {"slug": slug, "http_status": e.code, "width": None, "error": f"HTTP {e.code}"}
        except Exception as e:
            last = str(e)
            import time as _t; _t.sleep(1.5 * (i + 1))
    else:
        return {"slug": slug, "http_status": None, "width": None, "error": f"transient: {last}"}
    m = WIDTH_RX.search(html)
    val = re.sub(r"\s+", " ", m.group(1).strip()) if m else None
    dec, wi = classify_width(val)
    return {"slug": slug, "http_status": status, "width": val,
            "decision": dec, "width_inches": wi}


def db_rows():
    """Current maya_catalog state per collection slug (read-only)."""
    sql = ("SELECT split_part(product_url,'/collections/',2) AS slug, "
           "count(*), "
           "count(*) FILTER (WHERE width LIKE '%device-width%'), "
           "string_agg(DISTINCT coalesce(width_inches::text,'NULL'), '|') "
           "FROM maya_catalog WHERE dw_sku LIKE 'DWMR-8%' GROUP BY slug")
    r = subprocess.run(["psql", "-h", "/tmp", "-d", "dw_unified", "-At", "-F", "\t", "-c", sql],
                       capture_output=True, text=True)
    if r.returncode != 0:
        return None
    out = {}
    for line in r.stdout.strip().split("\n"):
        if not line:
            continue
        slug, n, poll, wis = line.split("\t")
        out[slug] = {"rows": int(n), "polluted": int(poll), "db_width_inches": wis}
    return out


def main():
    ts = datetime.datetime.now(datetime.timezone.utc).isoformat()
    src = [fetch_width(s) for s in SLUGS]
    for s in src:
        print(f"  {s['slug']:28s} http={str(s.get('http_status')):>4}  "
              f"{s.get('decision','-'):<14} wi={s.get('width_inches')}  {s.get('width')}")

    db = db_rows()
    proof = []
    if db:
        for s in src:
            d = db.get(s["slug"])
            if not d:
                continue
            match = None
            if s.get("decision") == "SINGLE":
                # every recovered row for this slug should carry the parsed inch value
                match = (d["db_width_inches"] == s["width_inches"])
            elif s.get("decision") == "MULTI_WIDTH":
                match = (d["db_width_inches"] == "NULL")   # ambiguous by design
            elif s.get("http_status") == 404:
                match = (d["polluted"] > 0)                # correctly left flagged
            proof.append({"slug": s["slug"], "decision": s.get("decision"),
                          "http_status": s.get("http_status"),
                          "src_width_inches": s.get("width_inches"),
                          "db_width_inches": d["db_width_inches"],
                          "db_rows": d["rows"], "db_polluted": d["polluted"],
                          "db_matches_source": match})

    single_ok = all(p["db_matches_source"] for p in proof
                    if p["decision"] == "SINGLE" and p["db_matches_source"] is not None)
    out = {
        "ticket": "TK-11029", "generated_at": ts, "mode": "READ-ONLY source proof",
        "source": "mayaromanoff.com collection spec-table Width cell",
        "slugs_fetched": len(src),
        "http_200": sum(1 for s in src if s.get("http_status") == 200),
        "http_404": sum(1 for s in src if s.get("http_status") == 404),
        "single_slugs": sum(1 for s in src if s.get("decision") == "SINGLE"),
        "multi_width_slugs": sum(1 for s in src if s.get("decision") == "MULTI_WIDTH"),
        "authoritative_source": src,
        "db_reconciliation": proof,
        "all_single_slugs_match_db": bool(single_ok) if proof else None,
    }
    path = os.path.join(HERE, f"authoritative-source-proof-{ts.replace(':', '-')}.json")
    json.dump(out, open(path, "w"), indent=2)
    print(f"\nSOURCE PROOF -> {path}")
    print(f"  200s={out['http_200']} 404s={out['http_404']} "
          f"single={out['single_slugs']} multi={out['multi_width_slugs']} "
          f"all_single_match_db={out['all_single_slugs_match_db']}")
    return 0


if __name__ == "__main__":
    sys.exit(main())