`scripts/verify-batchA-tier2.py`

123 lines
#!/usr/bin/env python3
"""TK-12097 Batch A — verify the 99 tier-2 (colorway-unverified) kravet.com heroes.

Tier-2 = a low-res product whose kravet.com hero filename's COLOR matched but whose
PATTERN base differed (legacy Brunschwig/Kravet cross-reference codes). Very likely the
same pattern+color, but unproven. This confirms it the cheap, strong way:

  Fetch kravet.com/<naive_slug(mfr)> (follow redirects). A row is VERIFIED iff:
    (1) the page returns 200 and did NOT redirect away to /search or a category (the
        final URL still resolves to the requested pattern-color slug), AND
    (2) the page's isMain hero == the URL phase-2 recorded (same asset), AND
    (3) the requested COLOR suffix appears on the product page identity.
  Otherwise the row stays UNVERIFIED (dropped — never auto-swapped).

$0 (kravet.com reads only). Writes batchA-tier2-verified.json (safe to add to the map)
and batchA-tier2-still-unverified.json.
"""
import json, re, os, subprocess, sys
from concurrent.futures import ThreadPoolExecutor, as_completed

HERE = os.path.dirname(os.path.abspath(__file__))
PROJ = os.path.dirname(HERE)
D = os.path.join(PROJ, "data/tk12097")
UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126 Safari/537.36"


def naive_slug(mfr):
    b = re.sub(r'[./_]', '-', mfr.strip().lower())
    b = re.sub(r'-+', '-', b).strip('-')
    return re.sub(r'-0$', '', b)


def fetch(url):
    # return (final_url, http_code, html)
    r = subprocess.run(["curl", "-sL", "-A", UA, "--max-time", "30",
                        "-w", "\n%{url_effective} %{http_code}", url], capture_output=True)
    out = r.stdout
    idx = out.rfind(b"\n")
    meta = out[idx + 1:].decode("ascii", "replace").strip().split(" ")
    return (meta[0] if meta else url), (meta[1] if len(meta) > 1 else "000"), out[:idx].decode("utf-8", "replace")


def extract_main_hero(html):
    for m in re.finditer(r'\{"thumb":"(.*?)","img":"(.*?)","full":"(.*?)".*?(?:"isMain":(true|false))?', html):
        if m.group(4) == "true":
            return m.group(3).replace("\\/", "/")
    b = re.findall(r'https://cdn\.brandfolder\.io/[A-Za-z0-9/_.-]+\.(?:jpg|jpeg|png|auto)', html.replace("\\/", "/"))
    return b[0] if b else None


def base_of(u):
    return re.sub(r'\.(png|jpg|jpeg|auto)$', '', u.split("?", 1)[0], flags=re.I)


def color_suffix(mfr):
    # e.g. 8013149.161.0 -> 161 ; 34258.116.0 -> 116
    parts = re.sub(r'\.0$', '', mfr).split('.')
    return parts[-1] if len(parts) > 1 else ''


def verify(row):
    mfr = row["mfr_sku"]
    slug = naive_slug(mfr)
    url = "https://www.kravet.com/" + slug
    rec = {"shopify_id": row["shopify_id"], "vendor": row["vendor"], "mfr_sku": mfr,
           "dw_sku": row.get("dw_sku"), "rollback_url": row["rollback_url"],
           "recorded_hires_url": row["proposed_hires_url"], "hires_le": row.get("hires_le"),
           "slug": slug}
    try:
        final, code, html = fetch(url)
        rec["final_url"] = final
        rec["http"] = code
        if code != "200":
            rec.update(verified=False, reason=f"page {code}")
            return rec
        # (1) not redirected away to search/category
        redirected_away = ("/search" in final.lower()) or (slug not in final.lower().replace("www.kravet.com/", ""))
        # (2) page hero base == recorded hero base
        hero = extract_main_hero(html)
        hero_ok = bool(hero) and (base_of(hero).lower() == base_of(row["proposed_hires_url"]).lower())
        # (3) requested color suffix present on the page
        csuf = color_suffix(mfr)
        color_on_page = bool(csuf) and (csuf in html)
        rec.update(redirected_away=redirected_away, hero_matches=hero_ok, color_on_page=color_on_page,
                   page_hero=hero)
        if (not redirected_away) and hero_ok and color_on_page:
            rec.update(verified=True, reason="slug resolved to requested product; isMain hero == recorded; color present")
        else:
            rec.update(verified=False,
                       reason=f"redirected_away={redirected_away} hero_matches={hero_ok} color_on_page={color_on_page}")
    except Exception as e:
        rec.update(verified=False, reason="ERR:" + str(e)[:120])
    return rec


def main():
    tier2 = json.load(open(os.path.join(D, "batchA-tier2-colorway-unverified.json")))["rows"]
    print(f"verifying {len(tier2)} tier-2 rows against kravet.com", file=sys.stderr)
    out = []
    with ThreadPoolExecutor(max_workers=8) as ex:
        futs = {ex.submit(verify, r): r for r in tier2}
        for i, f in enumerate(as_completed(futs), 1):
            out.append(f.result())
            if i % 20 == 0:
                print(f"  {i}/{len(tier2)}", file=sys.stderr)
    ver = [r for r in out if r.get("verified")]
    unv = [r for r in out if not r.get("verified")]
    # verified rows -> apply-hires map schema (ready to fold into batchA-final-map)
    map_rows = [{
        "shopify_id": r["shopify_id"], "vendor": r["vendor"], "dw_sku": r["dw_sku"], "mfr_sku": r["mfr_sku"],
        "cur_width": r.get("cur_width"), "current_400px_url": r["rollback_url"], "rollback_url": r["rollback_url"],
        "proposed_hires_url": r["recorded_hires_url"], "hires_le": r["hires_le"],
        "hires_source": "kravet_com_brandfolder_hero_tier2_verified", "swappable_from_local_staging": True,
    } for r in ver]
    json.dump({"ticket": "TK-12097", "batch": "A-tier2", "verified": len(ver), "unverified": len(unv),
               "rows": map_rows}, open(os.path.join(D, "batchA-tier2-verified.json"), "w"), indent=1)
    json.dump({"rows": unv}, open(os.path.join(D, "batchA-tier2-still-unverified.json"), "w"), indent=1)
    print(json.dumps({"tier2_total": len(out), "verified": len(ver), "still_unverified": len(unv)}, indent=2))


if __name__ == "__main__":
    main()