← back to Dw Five Field Step0
cost-relink-dryrun-batch2.py
206 lines
#!/usr/bin/env python3
"""
Part B — BATCH 2 cost-relink dry-run for the 28 'join-bug-relink (fuzzy-heavy)'
vendors. READ-ONLY. NO dw_unified writes. Owner: vp-dw-commerce, 2026-06-21.
Same DTD A+B+C confidence scheme as Part A (cost-relink-dryrun.py Steps 4-5):
exact = exact mfr_sku join to catalog cost col (>0)
fuzzy-high = Rule C: flagged pattern matches a catalog pattern whose colorways
ALL share ONE distinct cost -> unambiguous pattern->cost
review = Rule B: flagged pattern matches a catalog pattern with >1 distinct
cost -> ambiguous, surfaced as a range for human review
unresolved = no pattern match
Pricing mirrors the live drain:
Kravet-family -> MAP as-is (cost col already MAP/price_trade for those)
all others -> round(cost / 0.65 / 0.85, 2)
The catalog table + cost column per vendor come from out/vendor-classification.json
(the same auto-classifier that bucketed them). Produces:
out/cost-relink-diff-batch2.json full diff
out/cost-relink-batch2-summary.md per-vendor recoverable counts
"""
import json, os, re, subprocess, sys, datetime
from collections import defaultdict
ROOT = os.path.dirname(os.path.abspath(__file__))
OUT = os.path.join(ROOT, "out")
CLASS = os.path.join(OUT, "vendor-classification.json")
def psql(sql):
env = dict(os.environ); env.pop("PGPASSWORD", None); env.pop("DATABASE_URL", None)
r = subprocess.run(["psql", "-d", "dw_unified", "-tAF", "\t", "-f", "-"],
input=sql, capture_output=True, text=True, env=env)
if r.returncode != 0:
raise RuntimeError(f"psql failed: {r.stderr.strip()[:400]}")
return [ln.split("\t") for ln in r.stdout.split("\n") if ln != ""]
def q(v):
return "'" + str(v).replace("'", "''") + "'"
KRAVET_FAM = {
'Kravet','Kravet Couture','Kravet Design','Kravet Contract','Kravet Basics',
'Lee Jofa','Lee Jofa Modern','Groundworks','Brunschwig & Fils','Cole & Son',
'GP & J Baker','Colefax And Fowler','Colefax and Fowler','Clarke And Clarke',
'Clarke & Clarke','Mulberry','Mulberry Home','Threads','Baker Lifestyle',
'Andrew Martin','Nicolette Mayer','Aerin','Barclay Butera','Thom Filicia',
'Gaston Y Daniela','Gaston y Daniela','Donghia'}
def norm(s):
if s is None: return ''
s = s.lower().strip()
s = re.sub(r'\bcolou?rway\b', '', s)
s = re.sub(r'[^a-z0-9]+', ' ', s).strip()
return s
def roll_price(cost):
return round(cost / 0.65 / 0.85, 2)
def relink_vendor(vendor, table, col):
kf = vendor in KRAVET_FAM
flagged = psql(f"""
SELECT COALESCE(NULLIF(g.dw_sku,''), g.mfr_sku), g.mfr_sku,
replace(COALESCE(sp.pattern_name,''), E'\\t', ' '),
replace(COALESCE(sp.title,''), E'\\t', ' ')
FROM active_five_field_gaps g
LEFT JOIN shopify_products sp ON sp.dw_sku=g.dw_sku AND NULLIF(g.dw_sku,'') IS NOT NULL
WHERE g.priceable IS NOT TRUE AND g.vendor={q(vendor)};""")
cat = psql(f"""
SELECT upper(mfr_sku),
replace(COALESCE(pattern_name,''), E'\\t', ' '),
replace(COALESCE(color_name,''), E'\\t', ' '),
{col}
FROM {table} WHERE ({col})::numeric > 0;""")
cat_by_mfr = {}
pat_costs = defaultdict(set)
for row in cat:
if len(row) < 4: continue
mfr, pat, colr, cost = row[0], row[1], row[2], row[3]
try: cost = float(cost)
except: continue
cat_by_mfr[mfr] = (cost, pat, colr)
pat_costs[norm(pat)].add(round(cost, 2))
diff = []
n_exact = n_high = n_review = n_unres = 0
for fr in flagged:
if len(fr) < 4: continue
dw, mfr, pat, title = fr[0], fr[1], fr[2], fr[3]
np_ = norm(pat)
row = {'dw_sku': dw, 'vendor': vendor, 'mfr_sku': mfr, 'flagged_pattern': pat}
ex = cat_by_mfr.get((mfr or '').upper())
if ex:
cost, cpat, ccolr = ex
price = cost if kf else roll_price(cost)
row.update({'match_confidence': 'exact', 'matched_pattern': cpat,
'matched_color': ccolr, 'sourced_cost': round(cost, 2),
'computed_price': price,
'price_basis': 'MAP' if kf else 'cost/0.65/0.85'})
diff.append(row); n_exact += 1; continue
if np_ and np_ in pat_costs:
costs = pat_costs[np_]
if len(costs) == 1:
cost = next(iter(costs))
price = cost if kf else roll_price(cost)
row.update({'match_confidence': 'fuzzy-high', 'matched_pattern': pat,
'matched_color': '(pattern-consistent cost)',
'sourced_cost': round(cost, 2), 'computed_price': price,
'price_basis': 'MAP' if kf else 'cost/0.65/0.85',
'rule': 'C: single cost across colorways'})
diff.append(row); n_high += 1; continue
else:
lo, hi = min(costs), max(costs)
plo = lo if kf else roll_price(lo)
phi = hi if kf else roll_price(hi)
row.update({'match_confidence': 'review', 'matched_pattern': pat,
'matched_color': '(multi-cost pattern)',
'sourced_cost_range': [round(lo, 2), round(hi, 2)],
'computed_price_range': [plo, phi],
'price_basis': 'MAP' if kf else 'cost/0.65/0.85',
'rule': 'B: pattern has >1 distinct cost'})
diff.append(row); n_review += 1; continue
row.update({'match_confidence': 'unresolved', 'matched_pattern': None,
'sourced_cost': None, 'computed_price': None})
diff.append(row); n_unres += 1
return diff, {'flagged': len(flagged), 'exact': n_exact, 'fuzzy_high': n_high,
'review': n_review, 'unresolved': n_unres,
'catalog_cost_col': f'{table}.{col}', 'kravet_family': kf}
def main():
cls = json.load(open(CLASS))
vendors = [v for v in cls['vendors'] if v['bucket'] == 'join-bug-relink (fuzzy-heavy)']
assert len(vendors) == 28, f"expected 28 fuzzy-heavy, got {len(vendors)}"
all_diff = []
summary = {}
for v in vendors:
vendor, table, col = v['vendor'], v['catalog_table'], v['cost_col']
if not table or not col:
summary[vendor] = {'error': 'no catalog table/col', 'flagged': v['flagged']}
continue
try:
diff, s = relink_vendor(vendor, table, col)
except Exception as e:
summary[vendor] = {'error': str(e)[:200], 'flagged': v['flagged']}
continue
for d in diff: all_diff.append(d)
summary[vendor] = s
print(f" {vendor:<30} exact={s['exact']:<4} fuzzy-high={s['fuzzy_high']:<4} "
f"review={s['review']:<4} unres={s['unresolved']}", file=sys.stderr)
tot = {
'exact': sum(1 for d in all_diff if d.get('match_confidence') == 'exact'),
'fuzzy_high': sum(1 for d in all_diff if d.get('match_confidence') == 'fuzzy-high'),
'review': sum(1 for d in all_diff if d.get('match_confidence') == 'review'),
'unresolved': sum(1 for d in all_diff if d.get('match_confidence') == 'unresolved'),
'rows': len(all_diff),
}
tot['recoverable_high_confidence'] = tot['exact'] + tot['fuzzy_high']
payload = {
'generatedAt': datetime.datetime.now(datetime.timezone.utc).isoformat(),
'note': 'READ-ONLY dry-run. NO dw_unified writes. Batch-2 review diff.',
'confidence_scheme': 'DTD A+B+C (same as Part A batch 1)',
'pricing': 'Kravet-family -> MAP; others -> round(cost/0.65/0.85, 2)',
'vendor_count': len(vendors),
'totals': tot,
'per_vendor': summary,
'diff': all_diff,
}
json.dump(payload, open(os.path.join(OUT, 'cost-relink-diff-batch2.json'), 'w'), indent=2)
# ---- markdown summary ----
lines = []
lines.append("# Cost-Relink BATCH 2 — Fuzzy-Heavy Vendors Review Diff")
lines.append("")
lines.append(f"Generated: {payload['generatedAt']} · READ-ONLY · $0 (local SQL) · NO writes")
lines.append(f"Confidence: DTD A+B+C (same scheme as batch 1)")
lines.append("")
lines.append("## Headline")
lines.append(f"- Vendors: {len(vendors)} (the 'join-bug-relink fuzzy-heavy' bucket).")
lines.append(f"- Rows examined: {tot['rows']}.")
lines.append(f"- RECOVERABLE high-confidence: **{tot['recoverable_high_confidence']}** "
f"(exact {tot['exact']} + fuzzy-high {tot['fuzzy_high']}).")
lines.append(f"- Review (ambiguous, needs human): {tot['review']}.")
lines.append(f"- Unresolved (rescrape, not relink): {tot['unresolved']}.")
lines.append("")
lines.append("## Per-vendor (sorted by recoverable)")
lines.append("| Vendor | Flagged | Exact | Fuzzy-High | Recoverable | Review | Unresolved | Kravet? |")
lines.append("|---|---|---|---|---|---|---|---|")
rows = []
for vendor, s in summary.items():
if 'error' in s:
rows.append((0, f"| {vendor} | {s.get('flagged','?')} | ERR | ERR | 0 | 0 | 0 | — |"))
continue
rec = s['exact'] + s['fuzzy_high']
rows.append((rec, f"| {vendor} | {s['flagged']} | {s['exact']} | {s['fuzzy_high']} | "
f"**{rec}** | {s['review']} | {s['unresolved']} | "
f"{'yes' if s['kravet_family'] else 'no'} |"))
for _, line in sorted(rows, key=lambda x: -x[0]):
lines.append(line)
open(os.path.join(OUT, 'cost-relink-batch2-summary.md'), 'w').write("\n".join(lines) + "\n")
print(json.dumps(tot, indent=2))
if __name__ == "__main__":
main()