`dwjj-mfr-backfill/write-enrichment.py`

46 lines
#!/usr/bin/env python3
"""Write JF enrichment specs (removability/paste/coverage/yards/origin/washable) onto the 58
DWJJ products as internal global.* metafields. DRY-RUN by default; --live to write; --test=N to limit.
Images deliberately excluded (sibling-colorway bleed risk)."""
import json
import os
import subprocess
import sys
DRY = '--live' not in sys.argv
TEST = next((int(a.split('=')[1]) for a in sys.argv if a.startswith('--test=')), None)
os.chdir(os.path.expanduser('~/Projects/Designer-Wallcoverings'))
TOKEN = subprocess.check_output("grep -E '^SHOPIFY_ADMIN_TOKEN=' .env|head -1|cut -d= -f2|tr -d '\"'", shell=True, text=True).strip()
DOMAIN = "designer-laboratory-sandbox.myshopify.com"
def gql(q, v=None):
    body = {"query": q}
    if v: body["variables"] = v
    cmd = [
        "curl", "-s", "-X", "POST",
        f"https://{DOMAIN}/admin/api/2024-10/graphql.json",
        "-H", f"X-Shopify-Access-Token: {TOKEN}",
        "-H", "Content-Type: application/json",
        "-d", json.dumps(body),
    ]
    p = subprocess.run(cmd, capture_output=True, text=True)
    return json.loads(p.stdout)
# staged record field -> Shopify global metafield key
FIELDS = {"removability":"Removability","paste":"Paste","coverage":"Coverage","yards_dr":"Yards Per Double Roll","origin":"Country of Origin","washable":"Washable"}
SPECS = os.path.expanduser('~/Projects/jf-fabrics-recrawl/dwjj-mfr-backfill/dwjj_specs.jsonl')
with open(SPECS, encoding='utf-8') as fh:
    recs = [json.loads(l) for l in fh if l.strip()]
targets = [r for r in recs if r.get('ok') and any(r.get(f) for f in FIELDS)]
if TEST: targets = targets[:TEST]
M = '''mutation($mf:[MetafieldsSetInput!]!){ metafieldsSet(metafields:$mf){ metafields{ key } userErrors{ field message } } }'''
print(f"{'DRY-RUN' if DRY else 'LIVE'} — {len(targets)} products, up to {len(FIELDS)} enrichment fields each")
errs = 0; wrote = 0
for r in targets:
    mfs = [{"ownerId":r['gid'],"namespace":"global","key":k,"type":"single_line_text_field","value":str(r[f])}
           for f,k in FIELDS.items() if r.get(f)]
    if DRY:
        print(f"  {r['base']:<10} -> " + ", ".join(f"{m['key']}={m['value']}" for m in mfs)); continue
    d = gql(M, {"mf": mfs})
    ue = d.get('data',{}).get('metafieldsSet',{}).get('userErrors',[]) if 'data' in d else [{'message':json.dumps(d)[:120]}]
    if ue: errs += 1; print(f"  ERR {r['base']}: {ue}")
    else: wrote += 1; print(f"  OK  {r['base']} ({len(mfs)} fields)")
print(f"done. wrote={wrote} errors={errs}" if not DRY else "(dry-run, nothing written)")