← back to Dw Five Field Step0
cost-relink-dryrun.py
309 lines
#!/usr/bin/env python3
"""
No-cost cost-sourcing — SAFE first pass (Steps 2-5). READ-ONLY, NO DB WRITES.
Owner: vp-dw-commerce. Plan: ~/.claude/yolo-queue/pending-approval/no-cost-vendor-cost-sourcing.md
Produces, all read-only:
Step 2 vendor-classification.json — auto-bucket all flagged vendors by
(cost_col_exists, cost_col_populated_pct, exact_join_pct, discount).
Step 3 exact-sku relink (dry-run) — exact mfr_sku join flagged -> catalog cost.
Step 4 fuzzy pattern+color relink (dry-run) for the 4 cost-already-in-DB vendors
(Scalamandre, Rebel Walls, Osborne, Coordonné), DTD A+B+C confidence scheme:
A tier by exactness: exact-pattern AND exact-color = HIGH;
pattern-exact + color-fuzzy/missing = REVIEW; pattern-fuzzy = LOW/exclude.
B unique catalog cost required for HIGH; multi-cost candidate -> REVIEW.
C pattern whose colorways all share ONE cost -> pattern-only match = HIGH.
Step 5 review diff -> out/cost-relink-diff.json + human summary.
Pricing mirrors the live activation drain (final-split.js):
Kravet-family -> MAP as-is (kravet_price)
all others -> round(catalog_cost / 0.65 / 0.85, 2)
NOTHING is written to dw_unified. The relink WRITE stays gated for Steve.
"""
import json, re, subprocess, os, sys
from collections import defaultdict
OUT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "out")
os.makedirs(OUT, exist_ok=True)
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]}\nSQL: {sql[:200]}")
return [ln.split("\t") for ln in r.stdout.split("\n") if ln != ""]
def q(v):
"""SQL string literal with single quotes escaped."""
return "'" + str(v).replace("'", "''") + "'"
# ---- cost-column priority per the cost-map / catalog reality ----
# basis: 'cost' = column IS our cost; 'retail' = net = value*(1-discount)
# For the drain-mirror we feed catalog_cost = value (cost basis) into /0.65/0.85.
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'}
# vendor (Shopify name) -> (catalog table, cost column)
COST_VENDORS = {
'Scalamandre Wallpaper': ('scalamandre_catalog', 'price_trade'),
'Rebel Walls': ('rebel_walls_catalog', 'price_retail'),
'Osborne & Little': ('osborne_catalog', 'price_retail'),
'Coordonné': ('coordonne_catalog', 'price_retail'),
}
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)
# ======================================================================
# STEP 2 — auto-classifier over all flagged vendors
# ======================================================================
def classify():
# all flagged vendors + counts
vend = psql("""
SELECT vendor, count(*)
FROM active_five_field_gaps WHERE priceable IS NOT TRUE
GROUP BY vendor ORDER BY count(*) DESC;""")
vend = [(v, int(n)) for v, n in vend]
# catalog tables present + their price/cost columns
cols = psql("""
SELECT table_name, string_agg(column_name, ',')
FROM information_schema.columns
WHERE table_name LIKE '%\\_catalog'
AND (column_name ILIKE '%price%' OR column_name ILIKE '%cost%')
GROUP BY table_name;""")
table_cols = {t: c.split(',') for t, c in cols}
# discounts
disc = {v: (float(d) if d.strip() else None)
for v, d in psql("SELECT vendor_name, COALESCE(vendor_discount_pct::text,'') FROM vendor_registry;")}
# vendor -> guessed catalog table (slug match)
def slug(v):
return re.sub(r'[^a-z0-9]+', '_', v.lower()).strip('_')
def find_table(v):
s = slug(v)
cand = [s + '_catalog', s.replace('_wallpaper', '') + '_catalog',
s.replace('_fabrics', '') + '_catalog']
# known aliases
alias = {'scalamandre_wallpaper': 'scalamandre_catalog',
'rebel_walls': 'rebel_walls_catalog',
'osborne_little': 'osborne_catalog',
'osborne_&_little': 'osborne_catalog',
'coordonn': 'coordonne_catalog', 'coordonne': 'coordonne_catalog',
'phillip_jeffries': 'pj_catalog', 'phillipe_romano': 'phillipe_romano_catalog',
'china_seas': 'china_seas_catalog', 'koroseal': 'koroseal_catalog',
'versa_designed_surfaces': 'versa_catalog', 'hollywood_wallcoverings': 'hollywood_catalog',
'los_angeles_fabrics': 'la_walls_catalog', 'designers_guild': 'designers_guild_catalog'}
if s in alias: return alias[s]
for c in cand:
if c in table_cols: return c
# fuzzy: first table containing the slug stem
stem = s.split('_')[0]
for t in table_cols:
if t.startswith(stem): return t
return None
COST_COL_HINTS = ['price_trade','price_retail','cost','cost_price','net_cost',
'our_price','price','trade_price','retail_price','roll_price',
'dw_sell_price','master_cost','price_usd','price_gbp']
out = []
for v, n in vend:
t = find_table(v)
cost_col_exists = bool(t and table_cols.get(t))
populated_pct = 0.0; chosen_col = None; total = 0
exact_pct = 0.0; exact_n = 0
if t and table_cols.get(t):
# pick the cost column with highest >0 population
best = (0.0, None, 0)
for col in table_cols[t]:
if col not in COST_COL_HINTS: continue
try:
r = psql(f"SELECT count(*), count(NULLIF({col},0)) FROM {t};")
tot, pop = int(r[0][0]), int(r[0][1])
pct = (pop / tot * 100) if tot else 0
if pct > best[0]: best = (pct, col, tot)
except Exception:
continue
populated_pct, chosen_col, total = best
# exact mfr_sku join % against the flagged set
if chosen_col:
try:
r = psql(f"""
SELECT count(*) FROM active_five_field_gaps g
JOIN {t} c ON upper(c.mfr_sku)=upper(g.mfr_sku)
WHERE g.priceable IS NOT TRUE AND g.vendor={q(v)}
AND ({chosen_col})>0;""")
exact_n = int(r[0][0]); exact_pct = exact_n / n * 100 if n else 0
except Exception:
exact_n = 0
d = disc.get(v)
# bucket logic
if v == 'Phillipe Romano':
bucket = 'upstream-portal'
elif not cost_col_exists or populated_pct == 0 and chosen_col is None:
bucket = 'scraper-build-cost-field' if not cost_col_exists else 'public-PDP-rescrape'
elif populated_pct >= 60 and exact_pct >= 5:
bucket = 'join-bug-relink'
elif populated_pct >= 30:
bucket = 'join-bug-relink (fuzzy-heavy)'
elif populated_pct > 0:
bucket = 'public-PDP-rescrape'
else:
bucket = 'public-PDP-rescrape' if cost_col_exists else 'scraper-build-cost-field'
# cost-less tables (no usable col found at all) -> price-list / scraper-build
if cost_col_exists and chosen_col is None:
bucket = 'price-list-email-or-scraper-build'
out.append({
'vendor': v, 'flagged': n, 'catalog_table': t,
'cost_col': chosen_col, 'cost_col_populated_pct': round(populated_pct, 1),
'exact_join_n': exact_n, 'exact_join_pct': round(exact_pct, 1),
'discount_pct': d, 'bucket': bucket,
})
return out
# ======================================================================
# STEPS 3-5 — exact + fuzzy relink dry-run for the 4 cost-in-DB vendors
# ======================================================================
def relink():
diff = []
summary = {}
for vendor, (table, col) in COST_VENDORS.items():
kf = vendor in KRAVET_FAM
# ---- pull flagged set with pattern_name from shopify_products ----
# LEFT join sp so flagged rows with NULL/blank dw_sku (real mfr_sku only)
# survive — their exact mfr_sku match still resolves a cost.
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)};""")
# ---- pull catalog cost rows (mfr_sku, pattern, color, cost) ----
cat = psql(f"""
SELECT upper(mfr_sku),
replace(COALESCE(pattern_name,''), E'\\t', ' '),
replace(COALESCE(color_name,''), E'\\t', ' '),
{col}
FROM {table} WHERE {col}>0;""")
cat_by_mfr = {}
cat_by_pc = defaultdict(list) # (npattern, ncolor) -> [cost,...]
pat_costs = defaultdict(set) # npattern -> {cost,...} (Rule C)
for mfr, pat, colr, cost in cat:
cost = float(cost)
cat_by_mfr[mfr] = (cost, pat, colr)
np_, nc_ = norm(pat), norm(colr)
cat_by_pc[(np_, nc_)].append(cost)
pat_costs[np_].add(round(cost, 2))
n_exact = n_high = n_review = n_low = n_unres = 0
for dw, mfr, pat, title in flagged:
np_ = norm(pat)
row = {'dw_sku': dw, 'vendor': vendor, 'mfr_sku': mfr,
'flagged_pattern': pat}
# ---- Step 3: EXACT mfr_sku join ----
ex = cat_by_mfr.get((mfr or '').upper())
if ex:
cost, cpat, ccolr = ex
price = cost if kf else roll_price(cost)
row.update({'matched_mfr_sku': mfr.upper(), '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
# ---- Step 4: FUZZY pattern+color (DTD A+B+C) ----
# try to extract a color token from title or pattern is just pattern;
# flagged products rarely carry a clean color -> rely on pattern, then C.
ncolr = '' # flagged color not reliably present; pattern-led match
# A: exact-pattern AND exact-color (only if we had a color — usually we don't)
# so the live path is: pattern match, color via Rule C consistency.
if np_ and np_ in pat_costs:
costs = pat_costs[np_]
if len(costs) == 1:
# Rule C: pattern's colorways all share ONE cost -> HIGH
cost = next(iter(costs))
price = cost if kf else roll_price(cost)
row.update({'matched_mfr_sku': None, '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:
# B: multiple distinct costs for the pattern -> REVIEW (ambiguous)
lo, hi = min(costs), max(costs)
plo = lo if kf else roll_price(lo)
phi = hi if kf else roll_price(hi)
row.update({'matched_mfr_sku': None, '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
# ---- no pattern match -> unresolved (LOW/exclude) ----
row.update({'match_confidence': 'unresolved', 'matched_pattern': None,
'sourced_cost': None, 'computed_price': None})
diff.append(row); n_unres += 1
summary[vendor] = {'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}
return diff, summary
def main():
print('STEP 2: classifying all flagged vendors ...', file=sys.stderr)
cls = classify()
with open(os.path.join(OUT, 'vendor-classification.json'), 'w') as f:
json.dump({'generatedAt': __import__('datetime').datetime.utcnow().isoformat() + 'Z',
'vendor_count': len(cls), 'vendors': cls}, f, indent=2)
print('STEPS 3-5: relink dry-run for the 4 cost-in-DB vendors ...', file=sys.stderr)
diff, summary = relink()
resolved = [d for d in diff if d.get('computed_price') is not None]
prices = [d['computed_price'] for d in resolved if isinstance(d.get('computed_price'), (int, float))]
payload = {
'generatedAt': __import__('datetime').datetime.utcnow().isoformat() + 'Z',
'note': 'READ-ONLY dry-run. NO dw_unified writes. Relink WRITE is gated for Steve.',
'confidence_scheme': 'DTD A+B+C (3/3 unanimous 2026-06-21)',
'pricing': 'Kravet-family -> MAP; others -> round(cost/0.65/0.85, 2)',
'per_vendor': summary,
'totals': {
'rows': len(diff),
'exact': sum(1 for d in diff if d.get('match_confidence') == 'exact'),
'fuzzy_high': sum(1 for d in diff if d.get('match_confidence') == 'fuzzy-high'),
'review': sum(1 for d in diff if d.get('match_confidence') == 'review'),
'unresolved': sum(1 for d in diff if d.get('match_confidence') == 'unresolved'),
'price_min': min(prices) if prices else None,
'price_max': max(prices) if prices else None,
},
'diff': diff,
}
with open(os.path.join(OUT, 'cost-relink-diff.json'), 'w') as f:
json.dump(payload, f, indent=2)
print(json.dumps(payload['totals'], indent=2))
print(json.dumps(payload['per_vendor'], indent=2))
if __name__ == '__main__':
main()