← back to Eur Recrawl
verify_reonboard.py
232 lines
#!/usr/bin/env python3
"""
verify_reonboard.py — READ-ONLY QA of the EUR- reonboard live write.
For each row in reonboard.csv, query the LIVE Shopify Admin API and check:
(a) a roll variant with sku == roll_sku exists,
(b) its price == target retail (within 1 cent),
(c) the roll price is NOT below trade_cost (no below-cost survivors),
(d) the $4.25 Sample variant is still present.
Batches multiple SKUs per GraphQL call (sku:X OR sku:Y ...) and is polite.
Writes verify-report.json + prints a per-brand summary.
READ-ONLY: only issues GraphQL `products(query:...)` reads. No mutations.
"""
import os, csv, json, time, urllib.request
from collections import defaultdict
HERE = os.path.dirname(os.path.abspath(__file__))
BATCH = int(os.environ.get("BATCH", "20"))
LIMIT = int(os.environ.get("LIMIT", "0")) or None
PRICE_TOL = 0.01 # within 1 cent
SAMPLE_PRICE = 4.25
SAMPLE_TOL = 0.01
def _tok():
for l in open(os.path.expanduser("~/Projects/secrets-manager/.env")):
if l.startswith("SHOPIFY_ADMIN_TOKEN="):
return l.split("=", 1)[1].strip().strip('"')
raise SystemExit("token not found")
TOKEN = os.environ.get("AT") or _tok()
URL = "https://designer-laboratory-sandbox.myshopify.com/admin/api/2024-10/graphql.json"
QUERY = """query($q:String!,$n:Int!){
products(first:$n, query:$q){
edges{node{
id title
variants(first:20){edges{node{sku price selectedOptions{name value}}}}
}}
}}"""
def gql(q, v=None):
b = json.dumps({"query": q, "variables": v or {}}).encode()
req = urllib.request.Request(URL, b, {"X-Shopify-Access-Token": TOKEN,
"Content-Type": "application/json"})
for a in range(8):
try:
d = json.load(urllib.request.urlopen(req, timeout=90))
if "errors" in d and any("THROTTLED" in str(e) for e in d["errors"]):
time.sleep(2 * (a + 1)); continue
return d
except Exception:
time.sleep(2 * (a + 1))
raise RuntimeError("gql failed after retries")
def main():
rows = list(csv.DictReader(open(os.path.join(HERE, "reonboard.csv"))))
if LIMIT:
rows = rows[:LIMIT]
by_roll = {r["roll_sku"]: r for r in rows}
# per-row verification results, keyed by roll_sku
result = {} # roll_sku -> dict(status, detail...)
def new_rec(r):
return {
"roll_sku": r["roll_sku"], "vendor": r["vendor"], "mfr_code": r["mfr_code"],
"trade": float(r["trade_price"]), "expected": float(r["retail"]),
"found_roll": False, "live_roll_price": None,
"sample_present": False, "live_sample_price": None,
"categories": [], # e.g. missing_roll, price_mismatch, below_cost, sample_disturbed
}
for rs, r in by_roll.items():
result[rs] = new_rec(r)
# Build batched queries. Each product has a Sample sku (roll_sku + "-Sample")
# and the roll sku itself. Query by the roll sku token; the returned product
# node carries ALL its variants so we see both roll + sample in one shot.
roll_skus = list(by_roll.keys())
total = len(roll_skus)
print(f"Verifying {total} EUR- products (batch={BATCH})...")
processed = 0
for start in range(0, total, BATCH):
chunk = roll_skus[start:start + BATCH]
# match products that carry any of these roll skus as a variant sku
q = " OR ".join(f"sku:{s}" for s in chunk)
d = gql(QUERY, {"q": q, "n": len(chunk) * 2 + 5})
edges = ((d.get("data") or {}).get("products") or {}).get("edges", [])
for e in edges:
node = e["node"]
variants = [v["node"] for v in node["variants"]["edges"]]
vskus = {v["sku"]: v for v in variants}
# figure out which chunk roll_sku this product belongs to
for rs in chunk:
if rs in vskus:
rec = result[rs]
roll_v = vskus[rs]
rec["found_roll"] = True
try:
rec["live_roll_price"] = float(roll_v["price"])
except (TypeError, ValueError):
rec["live_roll_price"] = None
# sample variant = sku rs + "-Sample" OR a variant with Sample option
sample_v = vskus.get(rs + "-Sample")
if not sample_v:
for v in variants:
opts = " ".join(o["value"] for o in v.get("selectedOptions", []))
if "sample" in opts.lower() or (v["sku"] or "").lower().endswith("sample"):
sample_v = v; break
if sample_v:
rec["sample_present"] = True
try:
rec["live_sample_price"] = float(sample_v["price"])
except (TypeError, ValueError):
rec["live_sample_price"] = None
processed += len(chunk)
if start % (BATCH * 10) == 0 or processed >= total:
print(f" ...{processed}/{total}")
time.sleep(0.3) # polite
# Classify each row
for rs, rec in result.items():
cats = rec["categories"]
if not rec["found_roll"]:
cats.append("missing_roll")
else:
lp, exp, trade = rec["live_roll_price"], rec["expected"], rec["trade"]
if lp is None:
cats.append("price_unreadable")
else:
if abs(lp - exp) > PRICE_TOL:
cats.append("price_mismatch")
if lp < trade - 0.001:
cats.append("below_cost")
# sample check
if not rec["sample_present"]:
cats.append("sample_missing")
elif rec["live_sample_price"] is not None and abs(rec["live_sample_price"] - SAMPLE_PRICE) > SAMPLE_TOL:
cats.append("sample_disturbed")
rec["correct"] = (rec["found_roll"]
and "price_mismatch" not in cats
and "below_cost" not in cats
and "price_unreadable" not in cats
and rec["sample_present"]
and "sample_disturbed" not in cats)
# Aggregate overall + by brand
brands = defaultdict(lambda: {
"total": 0, "correct": 0,
"missing_roll": [], "price_mismatch": [], "below_cost": [],
"price_unreadable": [], "sample_missing": [], "sample_disturbed": [],
})
overall = {"total": total, "correct": 0,
"missing_roll": 0, "price_mismatch": 0, "below_cost": 0,
"price_unreadable": 0, "sample_missing": 0, "sample_disturbed": 0}
mismatch_details = []
below_cost_details = []
sample_issue_details = []
missing_details = []
for rs, rec in result.items():
b = brands[rec["vendor"]]
b["total"] += 1
overall["total"] = overall["total"] # already set
if rec["correct"]:
b["correct"] += 1; overall["correct"] += 1
for cat in ("missing_roll", "price_mismatch", "below_cost",
"price_unreadable", "sample_missing", "sample_disturbed"):
if cat in rec["categories"]:
overall[cat] += 1
b[cat].append(rs)
if "price_mismatch" in rec["categories"]:
mismatch_details.append({"sku": rs, "vendor": rec["vendor"],
"expected": rec["expected"], "live": rec["live_roll_price"]})
if "below_cost" in rec["categories"]:
below_cost_details.append({"sku": rs, "vendor": rec["vendor"],
"trade": rec["trade"], "live": rec["live_roll_price"]})
if "missing_roll" in rec["categories"]:
missing_details.append({"sku": rs, "vendor": rec["vendor"]})
if "sample_missing" in rec["categories"] or "sample_disturbed" in rec["categories"]:
sample_issue_details.append({"sku": rs, "vendor": rec["vendor"],
"sample_present": rec["sample_present"],
"live_sample_price": rec["live_sample_price"]})
report = {
"generated_at": time.strftime("%Y-%m-%d %H:%M:%S %Z"),
"note": "READ-ONLY verification; live write may have been in progress.",
"overall": overall,
"by_brand": {v: {
"total": d["total"], "correct": d["correct"],
"missing_roll": len(d["missing_roll"]),
"price_mismatch": len(d["price_mismatch"]),
"below_cost": len(d["below_cost"]),
"price_unreadable": len(d["price_unreadable"]),
"sample_missing": len(d["sample_missing"]),
"sample_disturbed": len(d["sample_disturbed"]),
} for v, d in brands.items()},
"mismatch_details": mismatch_details,
"below_cost_details": below_cost_details,
"missing_details": missing_details,
"sample_issue_details": sample_issue_details,
}
json.dump(report, open(os.path.join(HERE, "verify-report.json"), "w"), indent=2)
# Print summary
print("\n" + "=" * 62)
print("EUR- REONBOARD VERIFICATION (READ-ONLY)")
print("=" * 62)
o = overall
print(f"TOTAL rows: {o['total']}")
print(f"VERIFIED CORRECT: {o['correct']} ({100*o['correct']//max(o['total'],1)}%)")
print(f" missing roll: {o['missing_roll']}")
print(f" price mismatch: {o['price_mismatch']}")
print(f" BELOW COST: {o['below_cost']}")
print(f" price unreadable: {o['price_unreadable']}")
print(f" sample missing: {o['sample_missing']}")
print(f" sample disturbed: {o['sample_disturbed']}")
print("-" * 62)
print(f"{'BRAND':<28}{'tot':>5}{'ok':>5}{'miss':>5}{'mism':>5}{'<cost':>6}{'smpl':>5}")
for v, d in sorted(brands.items()):
smpl = len(d["sample_missing"]) + len(d["sample_disturbed"])
print(f"{v:<28}{d['total']:>5}{d['correct']:>5}{len(d['missing_roll']):>5}"
f"{len(d['price_mismatch']):>5}{len(d['below_cost']):>6}{smpl:>5}")
print("=" * 62)
print("report -> verify-report.json")
if __name__ == "__main__":
main()