`scripts/parse_brand_portal.py`

137 lines
#!/usr/bin/env python3
"""Parse a BRAND+category trade-portal PLP innerText dump -> JSONL rows (TK-10877).

Current SDG US trade PLP block format (one product):
  <Brand>                                  e.g. "Morris & Co"
  <Collection ... <code-num> <color>>      e.g. "Pure Morris North Fabrics 236810 Honeycomb"
  (<MFR_CODE>)                             e.g. "(DMPUR236810)"
  <Status>                                 Live | Discontinued | Limited Stock | New In | New | Coming Soon
  <Type>                                   Wallpaper | Fabric
  $<SSP>                                   higher price (retail suggested)
  $<Trade>                                 lower price (trade/wholesale)  <-- SSP == 2*Trade invariant
  <stock line>                             e.g. "30 Roll available"  (optional)
  Compare

PRICING: trade portal -> price shown is TRADE. trade_usd = trade price.
retail_usd = trade_usd / 0.65 / 0.85 (round 2dp). ssp_usd recorded when visible.
NEVER fabricate a price -> NULL on any miss.

Usage: parse_brand_portal.py <raw-file> <brand-match-regex> <product_type> <out.jsonl>
  brand-match-regex e.g. 'morris' (case-insensitive, matched against the brand line)
  product_type      'wallcovering' | 'fabric'  (canonical, stamped on every row)
"""
import re, sys, json, os

RAW, BRAND_RE, PTYPE, OUT = sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4]
brand_pat = re.compile(BRAND_RE, re.I)
CODE = re.compile(r'^\(([A-Z0-9][A-Z0-9/\-]{3,20})\)$')          # (DMPUR236810)
PRICE = re.compile(r'^\$([\d,]+(?:\.\d+)?)$')
STATUS = {'Live', 'Discontinued', 'Limited Stock', 'New In', 'New', 'Coming Soon', 'Made To Order'}
TYPE = {'Wallpaper', 'Fabric', 'Trimming', 'Wallpaper Accessory', 'Paint'}

def num(s): return float(s.replace(',', ''))

with open(RAW) as fh:
    L = [x.strip() for x in fh.read().split('\n')]

rows = []
seen = set()
dropped = []
n = len(L)
i = 0
while i < n:
    line = L[i]
    m = CODE.match(line)
    if not m:
        i += 1
        continue
    code = m.group(1)
    # brand + collection/color are on the two most-recent non-empty lines above the code
    prev = [x for x in L[max(0, i - 6):i] if x]
    if len(prev) < 2:
        i += 1
        continue
    brand = prev[-2]
    coll_color = prev[-1]
    if not brand_pat.search(brand):
        i += 1
        continue
    if code in seen:
        i += 1
        continue
    # forward window: status, type, prices, stock
    seg = [x for x in L[i + 1:i + 12] if x]
    status = next((x for x in seg if x in STATUS), None)
    ptype_seen = next((x for x in seg if x in TYPE), None)
    prices = [num(pm.group(1)) for x in seg for pm in [PRICE.match(x)] if pm]
    stock = next((x for x in seg if re.search(r'\b(available|out of stock|no stock)\b', x, re.I)), None)
    trade = ssp = None
    if len(prices) >= 2:
        ssp = max(prices[0], prices[1])
        trade = min(prices[0], prices[1])
    elif len(prices) == 1:
        # single price on a trade portal = trade price
        trade = prices[0]
    # parse collection + color out of coll_color: "<Collection> <codenum> <Color words>"
    pattern = coll_color
    color = None
    cm = re.search(r'^(.*?)\s+(\d{4,7})\s+(.+)$', coll_color)
    if cm:
        pattern = cm.group(1).strip()
        color = cm.group(3).strip()
    retail = round(trade / 0.65 / 0.85, 2) if trade is not None else None
    seen.add(code)
    rows.append({
        'base_code': code,
        'mfr_sku': code,
        'product_type': PTYPE,
        'pattern': pattern,
        'color': color,
        'collection': pattern,
        'width': None,
        'length': None,
        'ssp_usd': ssp,
        'trade_usd': trade,
        'retail_usd': retail,
        'image': None,
        'status': status,
        'stock': stock,
        'raw_label': coll_color,
    })
    if trade is None:
        dropped.append(f"{code} {coll_color} (no price)")
    i += 1

# append (resume-safe) — dedupe against existing checkpoint on base_code
existing = {}
if os.path.exists(OUT):
    with open(OUT) as fh:
        for ln in fh:
            ln = ln.strip()
            if ln:
                try:
                    r = json.loads(ln)
                    existing[r['base_code']] = r
                except Exception:
                    pass
for r in rows:
    existing[r['base_code']] = r
with open(OUT, 'w') as fh:
    for r in existing.values():
        fh.write(json.dumps(r) + '\n')

total = len(existing)
priced = sum(1 for r in existing.values() if r.get('trade_usd') is not None)
# SSP == 2*trade invariant check
inv_ok = inv_bad = 0
for r in existing.values():
    if r.get('ssp_usd') and r.get('trade_usd'):
        if abs(r['ssp_usd'] - 2 * r['trade_usd']) <= max(0.02, 0.01 * r['ssp_usd']):
            inv_ok += 1
        else:
            inv_bad += 1
print(f"[{BRAND_RE}/{PTYPE}] parsed {len(rows)} this run; checkpoint total={total} priced={priced}")
print(f"  SSP=2xTrade invariant: ok={inv_ok} bad={inv_bad} (of {inv_ok + inv_bad} with both prices)")
if dropped:
    print(f"  {len(dropped)} priced-miss (NULL trade): " + "; ".join(dropped[:5]))