`price_update_pass.py`

147 lines
#!/usr/bin/env python3
"""
Price-update pass for Kravet products whose sheet MAP has changed.

Source: kravet_diff_update_2026_04_20 where change_flag='retail_changed'
Updates: retail price (= sheet MAP) and cost (= MAP/2 + WHLS × Tariff%)
         on the main DWKK-XXX variant (the "Sold Per..." variant, NOT Sample).

Pipeline per product:
  1) query product variants, find main variant (sku = DWKK-XXX exactly)
  2) productVariantsBulkUpdate with new price + new cost
  3) write kravet_catalog.price_retail/cost_price

Concurrency: 8 threads.
Resume-safe: tracks updated products in JSONL; re-run skips already-done.

Usage:
  python3 price_update_pass.py --limit 100      # pilot
  python3 price_update_pass.py                  # all 4,570
  python3 price_update_pass.py --dry-run        # no Shopify writes
"""
import argparse, json, os, subprocess, sys, time, urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed
import io, csv, threading

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']
OUT_JSONL='price_update_results.jsonl'

_sem = threading.Semaphore(8)
_lock = threading.Lock()


def pg_rows(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 pg_exec(sql):
    subprocess.run(SSH + [
        f'PGPASSWORD=DW2024! psql -h 127.0.0.1 -U dw_admin -d dw_unified -c "{sql}"'],
        capture_output=True, text=True, check=True)


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"]


GET_VARS="""query($id:ID!){ product(id:$id){ variants(first:5){ nodes{ id sku price inventoryItem{ id unitCost{ amount } } } } } }"""
UPD="""mutation($pid:ID!,$vs:[ProductVariantsBulkInput!]!){ productVariantsBulkUpdate(productId:$pid,variants:$vs){ userErrors{field message} } }"""


def process_one(row, dry):
    # Row from joined diff ↔ variant_map: (dw_sku, variant_gid, product_gid,
    #                                       current_price, sheet_map, sheet_cost)
    dw_sku, variant_gid, product_gid, current_price, sheet_map, sheet_cost = row
    try:
        try:
            new_price = float(sheet_map)
            new_cost  = float(sheet_cost) if sheet_cost else None
            cur_price = float(current_price) if current_price else None
        except (ValueError, TypeError):
            return (dw_sku, "bad_numbers", None)
        if cur_price is not None and abs(cur_price - new_price) < 0.005:
            return (dw_sku, "nochange", {"price": cur_price})

        variant_input = {"id": variant_gid, "price": f"{new_price:.2f}"}
        if new_cost is not None:
            variant_input["inventoryItem"] = {"cost": f"{new_cost:.2f}"}
        if dry:
            return (dw_sku, "dry", {"from": cur_price, "to": new_price, "cost": new_cost})

        with _sem:
            r = gql(UPD, {"pid": product_gid, "vs": [variant_input]})
        errs = r["productVariantsBulkUpdate"]["userErrors"]
        if errs:
            return (dw_sku, f"shopify_err:{str(errs)[:120]}", None)

        esc = dw_sku.replace("'", "''")
        pg_exec(f"UPDATE kravet_catalog SET price_retail={new_price}, "
                f"cost_price={new_cost if new_cost is not None else 'NULL'}, "
                f"updated_at=now() WHERE dw_sku='{esc}'")
        return (dw_sku, "ok", {"from": cur_price, "to": new_price, "cost": new_cost})
    except Exception as e:
        return (dw_sku, f"exc:{type(e).__name__}:{str(e)[:100]}", None)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--limit", type=int, default=0)
    ap.add_argument("--dry-run", action="store_true")
    ap.add_argument("--threads", type=int, default=8)
    args = ap.parse_args()

    # Resume: skip SKUs already marked done in JSONL
    done = set()
    if os.path.exists(OUT_JSONL):
        with open(OUT_JSONL) as f:
            for line in f:
                try: d = json.loads(line); done.add(d["dw_sku"])
                except Exception: pass

    # Join diff ↔ authoritative variant map.
    # Recompute "change" against Shopify's actual current price (map.price),
    # not catalog.price_retail which had drift.
    sql = """SELECT u.dw_sku, m.variant_gid, m.product_gid, m.price::text,
                    u.sheet_map::text, u.sheet_cost::text
               FROM kravet_diff_update_2026_04_20 u
               JOIN kravet_dwkk_variant_map m ON m.dw_sku = u.dw_sku
              WHERE u.change_flag='retail_changed'"""
    if args.limit: sql += f" LIMIT {args.limit}"
    rows = pg_rows(sql)
    rows = [r for r in rows if r[0] not in done]
    print(f"[scope] {len(rows)} rows to update (resumed: {len(done)})")

    out = open(OUT_JSONL, "a", buffering=1)
    ok = err = nochange = dry = 0
    t0 = time.time()
    with ThreadPoolExecutor(max_workers=args.threads) as ex:
        futs = [ex.submit(process_one, r, args.dry_run) for r in rows]
        for i, f in enumerate(as_completed(futs), 1):
            dw_sku, status, data = f.result()
            out.write(json.dumps({"dw_sku": dw_sku, "status": status, "data": data}) + "\n")
            if status == "ok":         ok += 1
            elif status == "nochange": nochange += 1
            elif status == "dry":      dry += 1
            else:                      err += 1
            if i % 100 == 0 or i == len(rows):
                rate = i / max(1, time.time() - t0)
                eta = (len(rows) - i) / max(1, rate)
                print(f"  [{i:,}/{len(rows):,}] ok={ok} nc={nochange} err={err} dry={dry}  "
                      f"{rate:.1f}/s  eta {eta/60:.0f}m", flush=True)
    out.close()
    print(f"\n[done] ok={ok} nochange={nochange} err={err} dry={dry}  {time.time()-t0:.0f}s")


if __name__ == "__main__":
    main()