`exec/ws5_activate.py`

159 lines
#!/usr/bin/env python3
"""WS5 - attach Elitis feed images to the Needs-Image drafts and flip them ACTIVE.
Images are PATTERN-level (Elitis publishes no per-colorway imagery) - Steve-approved
2026-07-11 (canary-then-batch). $0: Shopify fetches the CDN URL itself; Admin API free.

Per draft (from ws3_drafts ledger, status OK) matched to elitis_feed.json by
accent-normalized pattern:
  1. rank images (drop swatch-grid/nuancier; prefer ambiance/mural/detail), take top 3.
  2. productCreateMedia(originalSource=CDN url) -> wait until READY.
  3. strip 'Needs-Image' tag + set status ACTIVE.
Idempotent: skips drafts that already have an image or are already ACTIVE."""
import sys
import os
import csv
import json
import time
import unicodedata
sys.path.insert(0, "lib")
from shopify import gql

ROOT = ".."
FEED = os.path.join(ROOT, "elitis_feed.json")
DRAFTS = os.path.join("ledger", "ws3_drafts.csv")
LEDGER = os.path.join("ledger", "ws5_activate.csv")

M_MEDIA = """mutation($pid:ID!,$media:[CreateMediaInput!]!){
  productCreateMedia(productId:$pid, media:$media){
    media{ ... on MediaImage { id status } } mediaUserErrors{ field message } } }"""
M_TAGSRM = """mutation($id:ID!,$tags:[String!]!){ tagsRemove(id:$id, tags:$tags){ userErrors{ message } } }"""
M_UPD = """mutation($input:ProductInput!){ productUpdate(input:$input){
  product{ id status } userErrors{ field message } } }"""
Q_MEDIA = """query($id:ID!){ product(id:$id){ status
  media(first:10){ edges{ node{ ... on MediaImage { id status } } } } } }"""

DROP = ("grille", "nuancier", "pastille", "swatch", "-color", "colour")
BOOST = ("ambiance", "detail", "mural", "room", "deco")


def norm(s):
    s = unicodedata.normalize("NFKD", s or "")
    s = "".join(c for c in s if not unicodedata.combining(c))
    return " ".join(s.casefold().split())


def rank_images(urls):
    def score(u):
        lu = u.lower()
        s = 0
        if any(d in lu for d in DROP):
            s -= 10
        if any(b in lu for b in BOOST):
            s += 3
        return s
    return sorted(urls, key=score, reverse=True)


def gid(pid):
    return f"gid://shopify/Product/{pid}"


def log(pid, pat, dw, status):
    new = not (os.path.exists(LEDGER) and os.path.getsize(LEDGER) > 0)
    with open(LEDGER, "a", newline="") as f:
        w = csv.writer(f)
        if new:
            w.writerow(["ts", "product_id", "pattern", "dw_sku", "status"])
        w.writerow([time.strftime("%Y-%m-%dT%H:%M:%S"), pid, pat, dw, status])


def already_done(pid):
    d = gql(Q_MEDIA, {"id": gid(pid)})["product"]
    has_img = len(d["media"]["edges"]) > 0
    return d["status"], has_img


def wait_ready(pid, timeout=40):
    t0 = time.time()
    while time.time() - t0 < timeout:
        d = gql(Q_MEDIA, {"id": gid(pid)})["product"]
        st = [e["node"].get("status") for e in d["media"]["edges"] if e["node"]]
        if st and all(s == "READY" for s in st):
            return True
        if any(s == "FAILED" for s in st):
            return False
        time.sleep(2)
    return bool(st)  # timed out but media exists


def activate_one(row, feed, verbose=False):
    pid = row["product_id"]
    pat = row["pattern"]
    dw = row["dw_sku"]
    f = feed.get(norm(pat))
    if not f:
        log(pid, pat, dw, "SKIP:no-feed-match")
        return "SKIP:no-feed-match"
    status, has_img = already_done(pid)
    if has_img and status == "ACTIVE":
        log(pid, pat, dw, "SKIP:already-active-with-image")
        return "SKIP:already-active-with-image"
    imgs = rank_images(f["images"])[:3] if not has_img else []
    if imgs:
        media = [{"originalSource": u, "mediaContentType": "IMAGE",
                  "alt": row.get("colorway") or pat} for u in imgs]
        r = gql(M_MEDIA, {"pid": gid(pid), "media": media})
        ue = r["productCreateMedia"]["mediaUserErrors"]
        if ue:
            log(pid, pat, dw, "FAIL:media:" + json.dumps(ue)[:120])
            return "FAIL:media"
        wait_ready(pid)
    # strip Needs-Image, go ACTIVE
    gql(M_TAGSRM, {"id": gid(pid), "tags": ["Needs-Image"]})
    r = gql(M_UPD, {"input": {"id": gid(pid), "status": "ACTIVE"}})
    ue = r["productUpdate"]["userErrors"]
    if ue:
        log(pid, pat, dw, "FAIL:activate:" + json.dumps(ue)[:120])
        return "FAIL:activate"
    log(pid, pat, dw, "OK")
    if verbose:
        print(json.dumps({"pid": pid, "pattern": pat, "colorway": row.get("colorway"),
                          "dw_sku": dw, "images_attached": len(imgs),
                          "img_urls": imgs}, indent=2, ensure_ascii=False))
    return "OK"


def load():
    feed = json.load(open(FEED))
    rows = [r for r in csv.DictReader(open(DRAFTS)) if r["status"] == "OK"]
    return feed, rows


if __name__ == "__main__":
    feed, rows = load()
    if "--canary" in sys.argv:
        pid = sys.argv[sys.argv.index("--canary") + 1]
        row = next(r for r in rows if r["product_id"] == pid)
        print("=== WS5 CANARY:", row["pattern"], row["colorway"], "===")
        print("RESULT:", activate_one(row, feed, verbose=True))
        sys.exit(0)
    done = set()
    if os.path.exists(LEDGER):
        for r in csv.DictReader(open(LEDGER)):
            if r["status"].startswith("OK") or r["status"].startswith("SKIP:already"):
                done.add(r["product_id"])
    n_ok = n_skip = n_fail = 0
    todo = [r for r in rows if r["product_id"] not in done]
    print(f"WS5 batch: {len(todo)} drafts to activate ({len(done)} already done)")
    for i, r in enumerate(todo):
        st = activate_one(r, feed)
        if st == "OK":
            n_ok += 1
        elif st.startswith("FAIL"):
            n_fail += 1
        else:
            n_skip += 1
        if (i + 1) % 20 == 0:
            print(f"  ...{i+1}/{len(todo)} ok={n_ok} skip={n_skip} fail={n_fail}")
    print(f"WS5 DONE: ok={n_ok} skip={n_skip} fail={n_fail}")