`scripts/preflight-measure.py`

114 lines
#!/usr/bin/env python3
"""TK-11658 Cycle 2 pre-flight: MEASURE (do not trust) every recovered hi-res URL
against Shopify media ceilings: HTTP 200 + long-edge <= 5000px + bytes <= 20MB.
Method: ranged GET (first N KB) -> real HTTP status, true total bytes (content-range),
real pixel dims (PIL). ~64KB/row. $0 (network only)."""
import json, io, sys, concurrent.futures as cf
from urllib.request import Request, urlopen
from urllib.error import HTTPError, URLError
from PIL import Image

SRC = "data/kravet-hires-phase2-swap-map.json"
OUT = "data/phase2-preflight-measured.json"
SHOP_PX = 5000            # Shopify long-edge ceiling
SHOP_BYTES = 20*1024*1024 # Shopify 20MB ceiling
UA = "Mozilla/5.0 (dw-preflight TK-11658)"

def fetch_head_bytes(url, nbytes):
    req = Request(url, headers={"User-Agent": UA, "Range": f"bytes=0-{nbytes-1}"})
    with urlopen(req, timeout=45) as r:
        code = r.status
        cr = r.headers.get("Content-Range", "")
        total = None
        if "/" in cr:
            tail = cr.rsplit("/",1)[1].strip()
            if tail.isdigit(): total = int(tail)
        if total is None:
            gcl = r.headers.get("x-goog-stored-content-length")
            if gcl and gcl.isdigit(): total = int(gcl)
        data = r.read()
    return code, total, data

def measure(row):
    url = row.get("proposed_hires_url") or ""
    res = {"mfr_sku": row.get("mfr_sku"), "vendor": row.get("vendor"),
           "shopify_id": row.get("shopify_id"), "url": url,
           "claim_le": row.get("recovered_long_edge"),
           "claim_bytes": row.get("recovered_bytes")}
    if not url:
        res.update(status="NO_URL", ok=False); return res
    try:
        code, total, data = fetch_head_bytes(url, 65536)
        dims = None
        # try progressively larger reads if header truncated
        for rng in (65536, 262144, 1048576):
            try:
                if rng > len(data):
                    code2, total2, data = fetch_head_bytes(url, rng)
                    if total2: total = total2
                im = Image.open(io.BytesIO(data)); im.draft(None,None)
                dims = im.size
                break
            except Exception:
                continue
        res.update(http=code, meas_bytes=total,
                   meas_w=(dims[0] if dims else None),
                   meas_h=(dims[1] if dims else None))
        if code not in (200,206):
            res.update(status=f"HTTP_{code}", ok=False); return res
        if dims is None:
            res.update(status="NO_DIMS", ok=False); return res
        le = max(dims)
        res["meas_le"] = le
        over_px = le > SHOP_PX
        over_bytes = (total is not None and total > SHOP_BYTES)
        unknown_bytes = total is None
        if over_px or over_bytes:
            res.update(status="OVER_LIMIT", ok=False,
                       over_px=over_px, over_bytes=over_bytes)
        elif unknown_bytes:
            res.update(status="BYTES_UNKNOWN", ok=False)  # not-measured != pass
        else:
            res.update(status="PASS", ok=True)
    except HTTPError as e:
        res.update(status=f"HTTP_{e.code}", ok=False)
    except (URLError, Exception) as e:
        res.update(status="ERR", ok=False, err=str(e)[:120])
    return res

def main():
    d = json.load(open(SRC))
    rows = d["rows"]
    results = [None]*len(rows)
    done = 0
    with cf.ThreadPoolExecutor(max_workers=16) as ex:
        futs = {ex.submit(measure, r): i for i,r in enumerate(rows)}
        for f in cf.as_completed(futs):
            i = futs[f]; results[i] = f.result(); done += 1
            if done % 100 == 0: print(f"  measured {done}/{len(rows)}", file=sys.stderr)
    paths = {"clean":[], "flagged":[]}
    for r in results:
        (paths["clean"] if r["ok"] else paths["flagged"]).append(r)
    summary = {}
    for r in results:
        summary[r["status"]] = summary.get(r["status"],0)+1
    out = {"ticket":"TK-11658","measured_at_utc":None,
           "method":"ranged GET (64KB-1MB) content-range total + PIL dims; MEASURED not claimed",
           "ceilings":{"long_edge_px":SHOP_PX,"bytes":SHOP_BYTES},
           "total":len(results),"measured":sum(1 for r in results if r.get("http")),
           "pass":len(paths["clean"]),"flagged":len(paths["flagged"]),
           "status_breakdown":summary,"rows":results}
    import datetime
    out["measured_at_utc"]=datetime.datetime.utcnow().isoformat()+"Z"
    json.dump(out, open(OUT,"w"), indent=2)
    print(json.dumps({"total":out["total"],"measured":out["measured"],
                      "pass":out["pass"],"flagged":out["flagged"],
                      "status_breakdown":summary}, indent=2))
    # max measured dims/bytes
    les=[r["meas_le"] for r in results if r.get("meas_le")]
    bs=[r["meas_bytes"] for r in results if r.get("meas_bytes")]
    if les: print("MAX measured long-edge:",max(les),"px")
    if bs: print("MAX measured bytes:",max(bs),"(",round(max(bs)/1024/1024,2),"MB)")

if __name__=="__main__": main()