`build_d3a_manifest.py`

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

Rebuilds the momentum_colorways -> ACTIVE X-prefix Shopify product join that a prior
session computed but never persisted (momentum_colorways.shopify_product_id is 100% NULL).

INPUTS (all read-only):
  data/candidates.tsv   pid|sku|width         (3,600 ACTIVE X-prefix products)
  data/enum.jsonl       live Shopify reads    (status, variants+price, uom, width, min)
  /tmp/mirror_x.tsv     dw_unified mirror dump: shopify_id\tdw_sku\tmfr_sku\tpattern_name\tvendor
  /tmp/momentum.tsv     momentum_colorways dump: alt_sku\tmomentum_sku\tcolor_number\tpattern_number\thw_price\tpattern_name\tcolor_name

  To regenerate the two /tmp dumps from the local mirror socket (host=/tmp):
    psql -h /tmp -d dw_unified -F $'\t' -A -t -c \
      "SELECT shopify_id, dw_sku, coalesce(mfr_sku,''), coalesce(pattern_name,''), coalesce(vendor,'') \
       FROM shopify_products WHERE dw_sku LIKE 'X%'" > /tmp/mirror_x.tsv
    psql -h /tmp -d dw_unified -F $'\t' -A -t -c \
      "SELECT coalesce(alt_sku,''), coalesce(momentum_sku,''), coalesce(color_number,''), \
              coalesce(pattern_number,''), coalesce(hw_price::text,''), pattern_name, color_name \
       FROM momentum_colorways" > /tmp/momentum.tsv

SCOPE / CLASSES: reprice-needed NON-YARD classes only, by uom value in
  {null, 'Full Roll', 'Sold Per None', 'Sold Per EA'}.  ('Sold Per Yard*' rows are already
  per-yard and are excluded.)

JOIN METHODS (first hit wins, deterministic order):
  1. exact-mfr-alt          mfr_sku == momentum.alt_sku
  2. tail-match-alt         segment after last '_' in mfr_sku == momentum.alt_sku
  3. code-map-momentum_sku  mfr_sku == momentum.momentum_sku
  4. code-map-color_number  mfr_sku == momentum.color_number

GUARDRAIL (all must hold, else EXCLUDE with reason):
  - exactly ONE momentum row for the matched key (else ambiguous:N)
  - hw_price non-null and > 0
  - 15.0 <= hw_price <= 72.0  (per-yard retail band)
  - a sellable (non "-sample") variant exists on the product
  - XCD-69430 is MUST-QUOTE/MDC -> hard-excluded

OUTPUTS (FILE artifacts only -- NO writes to dw_unified or Shopify):
  data/d3a-manifest.jsonl   one row per confidently-mapped product (reprice + rollback source)
  data/d3a-skips.jsonl      excluded 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.tsv"
NONYARD = {None, "Full Roll", "Sold Per None", "Sold Per EA"}
MUSTQUOTE = {"XCD-69430"}
BAND_LO, BAND_HI = 15.0, 72.0

# --- momentum key indexes ---
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")

# --- mirror mfr_sku by shopify gid ---
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()

# --- live Shopify enum ---
enum = {json.loads(l)["pid"]: json.loads(l) for l in open(os.path.join(DATA, "enum.jsonl"))}
cand = [l.rstrip("\n").split("|") for l in open(os.path.join(DATA, "candidates.tsv"))]

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]

man = open(os.path.join(DATA, "d3a-manifest.jsonl"), "w")
skp = open(os.path.join(DATA, "d3a-skips.jsonl"), "w")
meth = collections.Counter(); skipc = collections.Counter()
conf = 0; price_flags = []

for pid, sku, width in cand:
    d = enum.get(pid)
    if not d:
        continue
    uom = d.get("uom"); uv = (uom or {}).get("value") if uom else None
    if uv not in NONYARD:
        continue
    gid = f"gid://shopify/Product/{pid}"
    mfr = mfr_by_gid.get(gid, "")
    base = sku.rsplit("-yard", 1)[0] if sku.endswith("-yard") else sku.rsplit("-", 1)[0]
    sv = sellable(d)
    cur = float(sv["price"]) if sv else None
    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 in MUSTQUOTE:
        skipc["must-quote-mdc"] += 1
        skp.write(json.dumps({**rb, "reason": "must-quote-mdc (XCD-69430 excluded per directive)"}) + "\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["no-momentum-match"] += 1
        skp.write(json.dumps({**rb, "reason": "no-momentum-match"}) + "\n")
        continue
    if len(hit) > 1:
        skipc["ambiguous-multi-row"] += 1
        skp.write(json.dumps({**rb, "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, "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, "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:
        skipc["no-sellable-variant"] += 1
        skp.write(json.dumps({**rb, "reason": "no sellable variant found", "source_momentum_pattern": r["pat"]}) + "\n")
        continue

    conf += 1; meth[m] += 1
    if cur is not None and abs(cur - 4.25) > 0.001 and abs(cur - hw) > 0.001:
        price_flags.append((sku, cur, hw, r["pat"]))
    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,
    )) + "\n")

man.close(); skp.close()
print("confidently-mapped:", conf)
print("join-method breakdown:", dict(meth))
print("skip breakdown:", dict(skipc), "total skipped:", sum(skipc.values()))
print("price-flags (current != $4.25 and != hw_price):", len(price_flags))
for s, c, h, p in sorted(price_flags, key=lambda x: -x[1]):
    print(f"  {s}: current=${c} target_hw=${h} [{p}]")