`audit_last_30.py`

207 lines
#!/usr/bin/env python3
"""
Click-through audit of the last 30 Kravet products pushed to Shopify.

For each product, checks everything a human would see clicking through admin:
  - Title format (Pattern Color | Brand)
  - Status (ACTIVE / DRAFT / ARCHIVED)
  - Images (count, storefront URL, size > placeholder)
  - Variants: main DWKK-XXX + DWKK-XXX-Sample, price, cost
  - Metafields: manufacturer_sku (custom + dwc), color, width, compliance
  - SEO title & description
  - descriptionHtml length
  - Tags (incl. New Arrival)
  - Collection membership (New Arrivals)

Emits a pass/fail summary + detailed issue list per product.
"""
import json, subprocess, urllib.request, io, csv, sys

TOKEN=os.environ.get('SHOPIFY_ADMIN_TOKEN','')
URL='https://designer-laboratory-sandbox.myshopify.com/admin/api/2024-10/graphql.json'
SSH=['ssh','root@45.61.58.125']
NEW_ARRIVALS_GID='gid://shopify/Collection/167327760435'


def pg(sql):
    r = subprocess.run(SSH + [
        f'PGPASSWORD=DW2024! psql -h 127.0.0.1 -U dw_admin -d dw_unified --csv -t -c "{sql}"'],
        capture_output=True, text=True, check=True)
    return [row for row in csv.reader(io.StringIO(r.stdout)) if row]


def gql(q, v=None):
    body=json.dumps({"query":q,"variables":v or {}}).encode()
    req=urllib.request.Request(URL,data=body,method="POST",
        headers={"Content-Type":"application/json","X-Shopify-Access-Token":TOKEN})
    d=json.loads(urllib.request.urlopen(req,timeout=30).read())
    if "errors" in d: raise RuntimeError(d["errors"])
    return d["data"]


# Last 30 Kravet products I pushed
rows = pg("""
  SELECT dw_sku, shopify_product_id, mfr_sku, pattern_name, color_name, brand, imported_at::text
    FROM kravet_catalog
   WHERE dw_sku LIKE 'DWKK-%'
     AND imported_at IS NOT NULL
     AND imported_at >= now() - INTERVAL '2 days'
     AND shopify_product_id LIKE 'gid://%'
   ORDER BY imported_at DESC
   LIMIT 30
""")
if len(rows) < 30:
    print(f"note: only {len(rows)} rows with imported_at in last 2 days")

print(f"\nauditing {len(rows)} products ...\n")

Q="""query($id:ID!){
  product(id:$id){
    id title handle status vendor productType tags createdAt
    descriptionHtml
    seo{ title description }
    images(first:10){ nodes{ url width height } }
    media(first:10){ nodes{ status mediaContentType } }
    variants(first:10){ nodes{
      id sku title price
      inventoryItem{ unitCost{ amount } tracked }
    }}
    metafields(first:80){ nodes{ namespace key value } }
    collections(first:50){ nodes{ id title handle } }
  }
}"""

# Accept EITHER pre-Full-Monty keys (global.*) OR post-Full-Monty normalized keys
# (custom.*, specs.*). Full Monty's onboard_enrich rewrites under the normalized
# conventions but keeps values intact.
REQ_META_ALTS = {
    "manufacturer_sku":   [("custom","manufacturer_sku"), ("dwc","manufacturer_sku")],
    "width":              [("global","width"), ("custom","width"), ("specs","width")],
    "brand":              [("global","Brand"), ("custom","brand"), ("dwc","brand"), ("specs","brand")],
    "content":            [("global","Content"), ("custom","material"), ("specs","composition")],
    "prop_65":            [("global","Prop-65"), ("custom","prop_65_text")],
    "ca_tb117":           [("global","CA-TB117"), ("custom","ca_tb117_text")],
    "collection":         [("global","Collection"), ("custom","collection_name"), ("custom","collection_text"), ("specs","collection")],
}

def check(row):
    dw_sku, pid, mfr, pat, col, brn, imported = row
    p = gql(Q, {"id": pid})["product"]
    if not p:
        return {"dw_sku": dw_sku, "status": "NOT_FOUND", "issues": ["product missing on Shopify"]}

    issues = []

    # Title
    if "|" not in p["title"]:
        issues.append("title missing ' | Brand' suffix")
    if "wallpaper" in p["title"].lower():
        issues.append("title contains banned word 'wallpaper'")

    # Status
    if p["status"] != "ACTIVE":
        issues.append(f"status={p['status']} (expected ACTIVE)")

    # Images
    imgs = p["images"]["nodes"]
    media = p["media"]["nodes"]
    failed_media = [m for m in media if m.get("status") == "FAILED"]
    if len(imgs) == 0:
        issues.append("NO images")
    if failed_media:
        issues.append(f"{len(failed_media)} FAILED media")

    # Variants
    vs = p["variants"]["nodes"]
    main = next((v for v in vs if v["sku"] == dw_sku), None)
    sample = next((v for v in vs if v["sku"] == f"{dw_sku}-Sample"), None)
    if not main:
        issues.append("main variant DWKK-XXX missing")
    else:
        cost = (main["inventoryItem"].get("unitCost") or {}).get("amount")
        if not cost:
            issues.append("main variant has no cost")
        try:
            if float(main["price"]) <= 0: issues.append("main price <= 0")
        except: issues.append("main price unparseable")
    if not sample:
        issues.append("Sample variant missing")
    elif sample["price"] != "4.25":
        issues.append(f"Sample price={sample['price']} (expected 4.25)")

    # Metafields — each required concept can be satisfied by any of its alternative keys
    have = {(m["namespace"], m["key"]) for m in p["metafields"]["nodes"]}
    missing_concepts = []
    for concept, alts in REQ_META_ALTS.items():
        if not any(alt in have for alt in alts):
            missing_concepts.append(concept)
    if missing_concepts:
        issues.append(f"missing concepts: {missing_concepts}")

    # SEO
    if not p["seo"]["title"]:
        issues.append("no SEO title")
    if not p["seo"]["description"]:
        issues.append("no SEO description")

    # Description
    if not p["descriptionHtml"] or len(p["descriptionHtml"]) < 200:
        issues.append(f"thin description ({len(p.get('descriptionHtml') or '')} chars)")

    # Tags
    if "New Arrival" not in p["tags"]:
        issues.append("'New Arrival' tag missing")

    # Collection
    in_new_arrivals = any(c["id"] == NEW_ARRIVALS_GID for c in p["collections"]["nodes"])
    if not in_new_arrivals:
        issues.append("not in New Arrivals collection")

    return {
        "dw_sku": dw_sku,
        "pid_num": pid.split("/")[-1],
        "title": p["title"],
        "status": p["status"],
        "images": len(imgs),
        "variants": len(vs),
        "metafields": len(p["metafields"]["nodes"]),
        "issues": issues,
    }


results = [check(r) for r in rows]

# Summary
total = len(results)
pass_ = sum(1 for r in results if not r["issues"])
fail = total - pass_
print(f"=== AUDIT SUMMARY ===")
print(f"  total:  {total}")
print(f"  pass:   {pass_}")
print(f"  fail:   {fail}")

# Issue histogram
from collections import Counter
hist = Counter()
for r in results:
    for i in r["issues"]:
        # Collapse variable details
        import re
        canon = re.sub(r"=\S+", "=X", re.sub(r"\(\d+ chars\)", "(N chars)", i))
        canon = re.sub(r"'[^']+'", "'X'", canon)
        canon = re.sub(r"\[.*\]", "[...]", canon)
        hist[canon] += 1
print(f"\n=== ISSUE HISTOGRAM ===")
for issue, count in hist.most_common():
    print(f"  {count:3d}  {issue}")

print(f"\n=== PER-PRODUCT ===")
for r in results:
    mark = "✓" if not r["issues"] else "✗"
    print(f"  {mark} {r['dw_sku']:15s} [{r['status']:8s}] {r['images']}img {r['variants']}v {r['metafields']}mf  '{r['title'][:50]}'")
    for i in r["issues"]:
        print(f"       ! {i[:120]}")

json.dump(results, open("audit_last_30.json","w"), indent=2)
print(f"\nwrote audit_last_30.json")