`dryrun_apply.py`

119 lines
#!/usr/bin/env python3
"""
DRY-RUN collection-banner remediation planner for the LIVE DW Shopify store.
READ-ONLY. Produces a proposed before/after banner list. Writes NOTHING to Shopify.

For each MISSING / LOW-RES collection that has products:
  - paced fetch of /collections/<handle>/products.json (bot-filter safe: sequential + backoff)
  - pick the LARGEST product image with min-dimension >= MIN_DIM, skipping logo files
  - if none >= MIN_DIM -> flag NO_HIRES_SOURCE (needs a hand-picked asset; do NOT auto-fill)
Empty (0-product) collections are skipped (hide/populate decision, not an image fix).
"""
import json, os, time, urllib.request, csv

HERE = os.path.dirname(os.path.abspath(__file__))
HOST = "designerwallcoverings.com"
UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)"
MIN_DIM = 1200          # a banner needs at least this on the short edge to stay sharp
PER_COLLECTION = 24     # scan up to N products for the best image
SLEEP = 1.1             # polite pacing to dodge the 403 bot-filter
MAX_RETRY = 4

cols = json.load(open(os.path.join(HERE, "dw_collections.json")))
probe = {r["handle"]: r for r in json.load(open(os.path.join(HERE, "dw_col_probe.json")))}

def is_problem(c):
    if not c.get("image"):
        return "MISSING", None, None
    p = probe.get(c["handle"])
    if p and p["w"] and (p["w"] < 1000 or p["ht"] < 600):
        return f"LOWRES", p["w"], p["ht"]
    return None, None, None

def fetch_products(handle):
    url = f"https://{HOST}/collections/{handle}/products.json?limit={PER_COLLECTION}"
    for attempt in range(MAX_RETRY):
        try:
            r = urllib.request.urlopen(urllib.request.Request(url, headers={"User-Agent": UA, "Accept": "application/json"}), timeout=30)
            return json.load(r).get("products", [])
        except urllib.error.HTTPError as e:
            if e.code in (403, 429):
                time.sleep(2.0 * (attempt + 1))  # backoff on throttle
                continue
            return None
        except Exception:
            time.sleep(1.5)
    return None

def best_image(products):
    """largest product image with min-dim >= MIN_DIM, skipping logo files; returns (src,w,h) or None, plus best-any fallback."""
    best = None; best_any = None
    for p in products:
        for im in p.get("images", []):
            src = im.get("src", "")
            if "logo" in src.lower():
                continue
            w, h = im.get("width") or 0, im.get("height") or 0
            area = w * h
            if best_any is None or area > best_any[3]:
                best_any = (src, w, h, area)
            if w >= MIN_DIM and h >= MIN_DIM:
                if best is None or area > best[3]:
                    best = (src, w, h, area)
    return best, best_any

problems = []
for c in cols:
    kind, w, h = is_problem(c)
    if kind:
        problems.append((c, kind, w, h))

print(f"problem collections: {len(problems)}  (MIN_DIM={MIN_DIM}, per-collection scan={PER_COLLECTION})")

rows = []
n_ok = n_nosrc = n_empty = 0
for i, (c, kind, w, h) in enumerate(problems, 1):
    handle = c["handle"]; pc = c.get("products_count") or 0
    cur = c["image"]["src"] if c.get("image") else None
    cur_dim = f"{w}x{h}" if w else ("none" if kind == "MISSING" else "?")
    if pc == 0:
        rows.append({"handle": handle, "title": c["title"], "products": pc, "issue": kind,
                     "current_dim": cur_dim, "current_src": cur, "new_src": None,
                     "new_dim": None, "status": "EMPTY_SKIP"})
        n_empty += 1
        continue
    prods = fetch_products(handle)
    time.sleep(SLEEP)
    if prods is None:
        rows.append({"handle": handle, "title": c["title"], "products": pc, "issue": kind,
                     "current_dim": cur_dim, "current_src": cur, "new_src": None,
                     "new_dim": None, "status": "FETCH_FAIL"})
        continue
    best, best_any = best_image(prods)
    if best:
        rows.append({"handle": handle, "title": c["title"], "products": pc, "issue": kind,
                     "current_dim": cur_dim, "current_src": cur, "new_src": best[0],
                     "new_dim": f"{best[1]}x{best[2]}", "status": "READY"})
        n_ok += 1
    else:
        ba = f"{best_any[1]}x{best_any[2]}" if best_any else "none"
        rows.append({"handle": handle, "title": c["title"], "products": pc, "issue": kind,
                     "current_dim": cur_dim, "current_src": cur,
                     "new_src": best_any[0] if best_any else None, "new_dim": ba,
                     "status": "NO_HIRES_SOURCE"})
        n_nosrc += 1
    if i % 25 == 0:
        print(f"  ...{i}/{len(problems)}  ready={n_ok} no-source={n_nosrc}")

json.dump(rows, open(os.path.join(HERE, "proposal.json"), "w"), indent=1)
with open(os.path.join(HERE, "proposal.csv"), "w", newline="") as f:
    wcsv = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
    wcsv.writeheader(); wcsv.writerows(rows)

print("\n=== DRY-RUN SUMMARY (NO LIVE WRITES) ===")
print(f"  READY (good >= {MIN_DIM}px in-collection image found): {n_ok}")
print(f"  NO_HIRES_SOURCE (needs hand-picked asset):            {n_nosrc}")
print(f"  EMPTY_SKIP (0 products - hide/populate decision):     {n_empty}")
print(f"  FETCH_FAIL:                                           {sum(1 for r in rows if r['status']=='FETCH_FAIL')}")
print(f"  proposal.json / proposal.csv written to {HERE}")