`build_d3b_manifest.py`

125 lines
#!/usr/bin/env python3
"""
TK-11331 Decision-3b reprice manifest builder (READ-ONLY, deterministic).

Re-runs the EXACT d3a join+guardrail over the products d3a skipped as `no-momentum-match`,
now that the D3b recovery refresh backfilled alt_sku onto Momentum staging rows that had a
blank join key. Products that NOW map -> d3b-manifest; still-unmatched -> d3b-skips.

INPUTS (all read-only):
  data/d3a-skips.jsonl               d3a skip set (scope = reason == 'no-momentum-match')
  data/enum.jsonl                    live Shopify reads (status, variants+price, uom, width)
  data/candidates.tsv                pid|sku|width
  /tmp/mirror_x.tsv                  shopify_id\tdw_sku\tmfr_sku\tpattern_name\tvendor
  /tmp/momentum_d3b.tsv              alt_sku\tmomentum_sku\tcolor_number\tpattern_number\thw_price\tpattern_name\tcolor_name
                                     (REFRESHED momentum_colorways, post D3b alt_sku backfill)

JOIN METHODS (first hit wins, identical to d3a):
  1 exact-mfr-alt  2 tail-match-alt  3 code-map-momentum_sku  4 code-map-color_number
GUARDRAIL (all must hold, else skip+reason): exactly ONE momentum row for the key;
  hw_price non-null & >0; 15.0<=hw_price<=72.0; a sellable (non -sample) variant exists.

OUTPUTS (FILE only -- NO dw_unified/Shopify writes):
  data/d3b-manifest.jsonl   one row per newly-recovered+confidently-mapped product
  data/d3b-skips.jsonl      still-unmatched / guardrail-failed products + reason
"""
import json, collections, os
HERE = os.path.dirname(os.path.abspath(__file__)); DATA = os.path.join(HERE, "data")
MIRROR = "/tmp/mirror_x.tsv"; MOMENTUM = "/tmp/momentum_d3b.tsv"
NONYARD = {None, "Full Roll", "Sold Per None", "Sold Per EA"}
MUSTQUOTE = {"XCD-69430"}; BAND_LO, BAND_HI = 15.0, 72.0

mom = []
for l in open(MOMENTUM):
    p = l.rstrip("\n").split("\t")
    if len(p) < 7: continue
    alt, msku, cnum, pnum, hw, pat, col = p
    mom.append(dict(alt=alt.strip(), msku=msku.strip(), cnum=cnum.strip(),
                    hw=float(hw) if hw else None, pat=pat, col=col))
def idx(field):
    d = collections.defaultdict(list)
    for r in mom:
        if r[field]: d[r[field]].append(r)
    return d
by_alt, by_msku, by_cnum = idx("alt"), idx("msku"), idx("cnum")

mfr_by_gid = {}
for l in open(MIRROR):
    p = l.rstrip("\n").split("\t")
    if len(p) < 5: continue
    mfr_by_gid[p[0]] = p[2].strip()

enum = {json.loads(l)["pid"]: json.loads(l) for l in open(os.path.join(DATA, "enum.jsonl"))}
tail = lambda m: m.rsplit("_", 1)[-1] if "_" in m else m
def sellable(d):
    vs = [v for v in d["variants"] if not (v["sku"] or "").lower().endswith("-sample")]
    if not vs: return None
    vs.sort(key=lambda v: v.get("position", 99)); return vs[0]

# SCOPE = d3a no-momentum-match products
scope = [json.loads(l) for l in open(os.path.join(DATA, "d3a-skips.jsonl"))
         if json.loads(l)["reason"] == "no-momentum-match"]

man = open(os.path.join(DATA, "d3b-manifest.jsonl"), "w")
skp = open(os.path.join(DATA, "d3b-skips.jsonl"), "w")
rec = open(os.path.join(DATA, "d3b-recovered-needs-variant.jsonl"), "w")  # recovered price basis, but sample-only
meth = collections.Counter(); skipc = collections.Counter(); conf = 0; recovered_novariant = 0

for s in scope:
    pid = str(s["shopify_product_id"]); sku = s.get("sku"); width = s.get("width")
    d = enum.get(pid)
    if not d:
        skipc["not-in-enum"] += 1
        skp.write(json.dumps({**s, "d3b_reason": "product not in enum snapshot"}) + "\n"); continue
    gid = f"gid://shopify/Product/{pid}"; mfr = mfr_by_gid.get(gid, s.get("mfr_sku", "") or "")
    base = (sku or "").rsplit("-yard", 1)[0] if (sku or "").endswith("-yard") else (sku or "").rsplit("-", 1)[0]
    sv = sellable(d); cur = float(sv["price"]) if sv else None; uv = s.get("uom")
    rb = dict(shopify_product_id=pid, sku=sku, dw_sku=base, mfr_sku=mfr, current_price=cur, uom=uv, width=width)

    if base in MUSTQUOTE or (sku or "") in MUSTQUOTE:
        skipc["must-quote-mdc"] += 1
        skp.write(json.dumps({**rb, "d3b_reason": "must-quote-mdc"}) + "\n"); continue

    t = tail(mfr); hit = None; m = None
    if mfr and mfr in by_alt: hit, m = by_alt[mfr], "exact-mfr-alt"
    elif t and t in by_alt: hit, m = by_alt[t], "tail-match-alt"
    elif mfr and mfr in by_msku: hit, m = by_msku[mfr], "code-map-momentum_sku"
    elif mfr and mfr in by_cnum: hit, m = by_cnum[mfr], "code-map-color_number"

    if not hit:
        skipc["still-no-momentum-match"] += 1
        skp.write(json.dumps({**rb, "d3b_reason": "still-absent-from-live-feed (discontinued / true must-quote)"}) + "\n"); continue
    if len(hit) > 1:
        skipc["ambiguous-multi-row"] += 1
        skp.write(json.dumps({**rb, "d3b_reason": f"ambiguous:{len(hit)}-momentum-rows"}) + "\n"); continue
    r = hit[0]; hw = r["hw"]
    if hw is None or hw <= 0:
        skipc["hw-null-or-zero"] += 1
        skp.write(json.dumps({**rb, "d3b_reason": f"hw_price null/zero ({hw})", "source_momentum_pattern": r["pat"]}) + "\n"); continue
    if not (BAND_LO <= hw <= BAND_HI):
        skipc["out-of-band"] += 1
        skp.write(json.dumps({**rb, "d3b_reason": f"hw_price {hw} outside ${BAND_LO:.0f}-{BAND_HI:.0f}/yd band", "source_momentum_pattern": r["pat"]}) + "\n"); continue
    if sv is None:
        # Join+price+band all PASSED -> Momentum pattern genuinely RECOVERED, price basis known.
        # But this ACTIVE X-prefix product is SAMPLE-ONLY (no sellable variant) -> cannot be repriced.
        # Correct follow-on = ADD a Sold-Per-Yard variant @ hw_price (pos1), NOT a reprice.
        skipc["recovered-but-sample-only"] += 1; recovered_novariant += 1
        payload = {**rb, "d3b_reason": "recovered-price-basis-but-sample-only: add Sold-Per-Yard variant @ hw_price (NOT a reprice)",
                   "recovered_new_price": hw, "recovered_hw_price": hw, "join_method": m,
                   "source_momentum_pattern": r["pat"], "source_momentum_color": r["col"], "source_momentum_alt_sku": r["alt"]}
        skp.write(json.dumps(payload) + "\n"); rec.write(json.dumps(payload) + "\n"); continue

    conf += 1; meth[m] += 1
    man.write(json.dumps(dict(
        shopify_product_id=pid, variant_id=sv["id"], sku=sv["sku"], current_price=cur,
        source_momentum_pattern=r["pat"], source_momentum_color=r["col"], source_momentum_alt_sku=r["alt"],
        new_price=hw, hw_price=hw, join_method=m, mfr_sku=mfr, uom=uv, width=width,
        recovered_via="d3b-alt_sku-backfill")) + "\n")

man.close(); skp.close(); rec.close()
print("scope (d3a no-momentum-match):", len(scope))
print("d3b reprice-manifest rows (recovered + has sellable variant):", conf)
print("recovered price basis but SAMPLE-ONLY (needs add-yard-variant, not reprice):", recovered_novariant)
print("join-method breakdown:", dict(meth))
print("skip breakdown:", dict(skipc), "total skipped:", sum(skipc.values()))