← back to Eur Recrawl
build_held_disposition.py
296 lines
#!/usr/bin/env python3
"""
Build held_disposition.csv for the EUR- reonboard held set.
- Parse Osborne DiscontinuedList / LimitedStockList + DG-brands limited/disco list (pdftotext -layout).
- Confirm PDFs carry NO prices.
- Join held products (368 disco-absent + 165 disco-colorway) via shopify_products.
- Classify disposition; query live roll price via Admin API (READ ONLY).
Output: held_disposition.csv + summary to stdout.
"""
import csv, re, subprocess, os, json, sys, time
from collections import defaultdict
BASE = os.path.dirname(os.path.abspath(__file__))
PDFS = os.path.join(BASE, "pdfs")
def pdftext(name):
p = subprocess.run(["pdftotext", "-layout", os.path.join(PDFS, name), "-"],
capture_output=True, text=True)
return p.stdout
# --- 1. Price presence check across all list PDFs ---
list_pdfs = ["DiscontinuedList.pdf", "LimitedStockList.pdf",
"DGBRANDSLIMITEDANDDISCONTINUEDLISTFEB24.pdf"]
price_hits = {}
for n in list_pdfs:
t = pdftext(n)
price_hits[n] = len(re.findall(r'[£$]|\bprice\b|\bGBP\b|\bEUR\b', t, re.I))
# --- 2. Parse codes + discontinued/limited colorways ---
# Osborne PDFs: rows are "PatternName CODE colorways" where CODE = (C?W|F)\d{4}, colorways = digits/ranges.
# DG PDF: rows are "BRAND Pattern (P?(CL|DG|NCW)\d+|F\d+) LIMITED_cols DISCONTINUED_cols"
CODE_RE = re.compile(r'\b((?:NCW|PCL|PDG|CW|W|F)\d{3,5})\b')
# colorway tokens: 2-digit numbers, ranges "04-05", "03 to 06", comma lists
def extract_cw(s):
cws = set()
# normalise "X to Y" -> range, "X-Y" -> range
for m in re.finditer(r'(\d{2})\s*(?:to|-)\s*(\d{2})', s):
a, b = int(m.group(1)), int(m.group(2))
if a <= b and b - a < 60:
for k in range(a, b+1):
cws.add(f"{k:02d}")
# standalone 2-digit
for m in re.finditer(r'\b(\d{2})\b', s):
cws.add(m.group(1))
return cws
def parse_osborne(name):
"""Return dict base_code -> set(discontinued/limited colorways). Also codes with no cw (whole pattern gone)."""
out = defaultdict(set)
whole = set()
for line in pdftext(name).splitlines():
# find each code on the line; text after it up to next code = its colorway blob
codes = list(CODE_RE.finditer(line))
if not codes:
continue
for i, cm in enumerate(codes):
code = cm.group(1)
start = cm.end()
end = codes[i+1].start() if i+1 < len(codes) else len(line)
blob = line[start:end]
cws = extract_cw(blob)
if cws:
out[code] |= cws
else:
whole.add(code)
return out, whole
def parse_dg(name):
"""DG PDF has LIMITED and DISCONTINUED columns. Capture both, tag by column via header positions is unreliable
in -layout; we treat ANY listed colorway as 'in the list'. Also track disco vs limited best-effort."""
out = defaultdict(set)
whole = set()
for line in pdftext(name).splitlines():
cm = CODE_RE.search(line)
if not cm:
continue
code = cm.group(1)
blob = line[cm.end():]
cws = extract_cw(blob)
if cws:
out[code] |= cws
else:
whole.add(code)
return out, whole
disco_osb, disco_osb_whole = parse_osborne("DiscontinuedList.pdf")
lim_osb, lim_osb_whole = parse_osborne("LimitedStockList.pdf")
dg_all, dg_whole = parse_dg("DGBRANDSLIMITEDANDDISCONTINUEDLISTFEB24.pdf")
# Merge: a "disco list" (Osborne disco + DG all -> DG list mixes limited+disco; we split below by re-reading columns)
# For DG we re-parse to separate LIMITED vs DISCONTINUED using column split on 2+ spaces.
dg_limited = defaultdict(set); dg_disco = defaultdict(set)
for line in pdftext("DGBRANDSLIMITEDANDDISCONTINUEDLISTFEB24.pdf").splitlines():
cm = CODE_RE.search(line)
if not cm: continue
code = cm.group(1)
tail = line[cm.end():]
# split into columns by runs of 2+ spaces
cols = [c for c in re.split(r'\s{2,}', tail) if c.strip()]
# heuristic: first col after code with digits = LIMITED, next = DISCONTINUED (per header order)
digit_cols = [c for c in cols if re.search(r'\d', c)]
if len(digit_cols) >= 2:
dg_limited[code] |= extract_cw(digit_cols[0])
dg_disco[code] |= extract_cw(digit_cols[1])
elif len(digit_cols) == 1:
# ambiguous single column; header spacing: if it sits far right treat as disco else limited.
# fall back: put in disco (conservative — "discontinued" is the stronger claim we care about)
dg_disco[code] |= extract_cw(digit_cols[0])
def base_and_cw(mfr):
"""Split held mfr_code into base + colorway. e.g. NCW4396-01 -> (NCW4396,'01'); W6650-2 -> (W6650,'02'); PDG716 -> (PDG716,None)"""
m = re.match(r'^([A-Za-z]+\d{3,5})(?:[-/ ]?0*(\d{1,3}))?$', mfr.strip())
if not m:
# try loose
mm = re.match(r'^([A-Za-z]+\d+)', mfr.strip())
return (mm.group(1) if mm else mfr.strip(), None)
base = m.group(1)
cw = m.group(2)
cw = f"{int(cw):02d}" if cw is not None else None
return base, cw
def in_disco(base, cw):
# Osborne disco (wallcovering W/CW, fabric F) + DG disco column
if base in disco_osb_whole or base in dg_whole:
return True, "whole-pattern"
if cw is None:
return (base in disco_osb or base in dg_disco), "base-listed(no-cw)"
if base in disco_osb and cw in disco_osb[base]:
return True, "cw-match"
if base in dg_disco and cw in dg_disco[base]:
return True, "cw-match"
if base in disco_osb or base in dg_disco:
return False, "base-in-list-cw-not"
return False, ""
def in_limited(base, cw):
if base in lim_osb_whole:
return True, "whole-pattern"
if cw is None:
return (base in lim_osb or base in dg_limited), "base-listed(no-cw)"
if base in lim_osb and cw in lim_osb[base]:
return True, "cw-match"
if base in dg_limited and cw in dg_limited[base]:
return True, "cw-match"
if base in lim_osb or base in dg_limited:
return False, "base-in-list-cw-not"
return False, ""
# --- 3. Load held rows (disco-absent + disco-colorway only) ---
held = []
with open(os.path.join(BASE, "reonboard_held.csv")) as f:
for row in csv.DictReader(f):
if row["basis"] in ("discontinued-absent", "HELD-cw-absent-in-disco-list"):
held.append(row)
# --- 4. Live roll prices via Admin API (READ ONLY) ---
def load_token():
for line in open(os.path.expanduser("~/Projects/secrets-manager/.env")):
if line.startswith("SHOPIFY_ADMIN_TOKEN="):
return line.split("=",1)[1].strip()
return None
TOKEN = load_token()
SHOP = "designer-laboratory-sandbox.myshopify.com"
API = f"https://{SHOP}/admin/api/2024-10/graphql.json"
# Pull shopify_id for each held product from mirror (pre-exported via psql -> held_mirror.json)
_mj = json.load(open(os.path.join(BASE, "held_mirror.json")))
mirror = {sku: (v[0], v[1]) for sku, v in _mj.items()}
def gql(query, variables):
import urllib.request
data = json.dumps({"query": query, "variables": variables}).encode()
req = urllib.request.Request(API, data=data, headers={
"Content-Type": "application/json",
"X-Shopify-Access-Token": TOKEN})
with urllib.request.urlopen(req, timeout=30) as r:
return json.load(r)
PROD_Q = """query($id: ID!){ product(id:$id){ id status
variants(first:20){ nodes{ sku title price selectedOptions{name value} } } } }"""
_PRICE_CACHE_F = os.path.join(BASE, "held_live_prices.json")
_price_cache = {}
if os.path.exists(_PRICE_CACHE_F):
_price_cache = json.load(open(_PRICE_CACHE_F))
def live_roll_price(gid):
"""Return (roll_price_float_or_None, status, note). Disk-cached by gid."""
if not gid:
return None, None, "no-gid"
if gid in _price_cache:
c = _price_cache[gid]
return (c[0], c[1], c[2])
try:
res = gql(PROD_Q, {"id": gid})
except Exception as e:
return None, None, f"api-err:{type(e).__name__}"
prod = (res.get("data") or {}).get("product")
if not prod:
return None, None, "not-found"
status = prod.get("status")
roll = None
for v in prod["variants"]["nodes"]:
sku = (v.get("sku") or "")
title = (v.get("title") or "")
opts = " ".join(o.get("value","") for o in v.get("selectedOptions") or [])
is_sample = "sample" in (sku+title+opts).lower()
if is_sample:
continue
try:
p = float(v["price"])
except (TypeError, ValueError):
continue
# pick the non-sample variant with a real price (the roll)
if roll is None or p > roll:
roll = p
_price_cache[gid] = [roll, status, "ok"]
return roll, status, "ok"
# --- 5. Classify + write ---
rows_out = []
disp_counts = defaultdict(int)
below_cost = []
LOW_THRESHOLD = 150.0
for i, r in enumerate(held):
mfr = r["mfr_code"]
base, cw = base_and_cw(mfr)
dmatch, dwhy = in_disco(base, cw)
lmatch, lwhy = in_limited(base, cw)
gid, mmfr = mirror.get(r["sku"], (None, None))
roll_price, status, note = live_roll_price(gid)
# base-present = the pattern appears somewhere in the Osborne/DG disco or limited lists
base_in_disco = (base in disco_osb) or (base in dg_disco) or (base in disco_osb_whole) or (base in dg_whole)
base_in_lim = (base in lim_osb) or (base in dg_limited) or (base in lim_osb_whole)
base_present = base_in_disco or base_in_lim
if lmatch:
disp = "limited-stock-keep-sample"
elif dmatch:
disp = "confirmed-discontinued"
elif not mfr.strip():
disp = "no-mfr-code-recheck"
elif base_present:
# the PATTERN is partly discontinued/limited but THIS colorway is not listed -> likely still live, price it
disp = "colorway-live-pattern-partial-recheck"
else:
disp = "absent-unexplained-recheck"
disp_counts[disp] += 1
if roll_price is not None and roll_price < LOW_THRESHOLD:
below_cost.append((r["roll_sku"], roll_price, status, disp))
rows_out.append({
"sku": r["roll_sku"],
"vendor": r["vendor"],
"mfr_code": mfr,
"held_reason": r["basis"],
"in_disco_list": "yes" if dmatch else "no",
"in_limited_list": "yes" if lmatch else "no",
"base_pattern_in_disco": "yes" if base_in_disco else "no",
"base_pattern_in_limited": "yes" if base_in_lim else "no",
"live_roll_price": f"{roll_price:.2f}" if roll_price is not None else "",
"live_status": status or "",
"disposition": disp,
})
if (i+1) % 50 == 0:
print(f"...{i+1}/{len(held)} live-price queried", file=sys.stderr)
time.sleep(0.12) # polite pacing
with open(os.path.join(BASE, "held_disposition.csv"), "w", newline="") as f:
w = csv.DictWriter(f, fieldnames=["sku","vendor","mfr_code","held_reason",
"in_disco_list","in_limited_list","base_pattern_in_disco","base_pattern_in_limited",
"live_roll_price","live_status","disposition"])
w.writeheader()
w.writerows(rows_out)
json.dump(_price_cache, open(_PRICE_CACHE_F, "w"))
# --- summary ---
print("\n==================== SUMMARY ====================")
print("PDF price presence (regex hits for £/$/price/GBP/EUR):")
for n, c in price_hits.items():
print(f" {n}: {c} hits -> {'HAS PRICES' if c>3 else 'NO per-item prices'}")
print(f"\nHeld set processed (disco-absent + disco-colorway): {len(held)}")
print("Disposition counts:")
for k, v in sorted(disp_counts.items(), key=lambda x:-x[1]):
print(f" {k}: {v}")
priced = [x for x in rows_out if x['live_roll_price']]
print(f"\nLive roll price present on: {len(priced)}/{len(held)} held products")
print(f"Below ${LOW_THRESHOLD:.0f} (suspicious/potential below-cost) live rolls: {len(below_cost)}")
for sku, p, st, disp in sorted(below_cost, key=lambda x:x[1])[:40]:
print(f" {sku} ${p:.2f} [{st}] {disp}")
print("\nWrote held_disposition.csv")