← back to Reid Witlin Onboarding
build_batch.py
222 lines
#!/usr/bin/env python3
"""
Reid Witlin (rwltd.com) full-PL onboarding — DRY-RUN batch generator.
Reads dw_unified.rwltd_catalog orphan rows (not yet onboarded), dedupes by
mfr_sku (already unique), recovers the correct colorway image from
gallery_images, builds the live-matching Shopify product template
(vendor "Architectural Fabrics", "{Town} {Colorway}" title, $4.25 sample-only,
"quotes"+"Commercial" tags, DWRW-210xxx SKU continuing from 210314), and splits
into image-confident (ready) vs image-held (eyeball) piles.
OUTPUT ONLY — writes local CSVs + summary.json. Does NOT touch Shopify.
The live create is a separate, Steve-gated step.
"""
import json, csv, re, subprocess, os
OUT = os.path.dirname(os.path.abspath(__file__))
SKU_PREFIX = "DWDQ" # dedicated Reid Witlin prefix — avoids the DWRW/Rebel Walls collision
SKU_START = 100001 # DWDQ is unused anywhere; start a clean 6-digit band
VENDOR = "Architectural Fabrics"
SAMPLE_PRICE = "4.25"
def psql_json(sql):
"""Run a query, return list of dicts via JSON aggregation."""
wrapped = f"SELECT COALESCE(json_agg(t),'[]') FROM ({sql}) t;"
r = subprocess.run(["psql", "host=/tmp dbname=dw_unified", "-tA", "-c", wrapped],
capture_output=True, text=True)
if r.returncode != 0:
raise SystemExit(f"psql failed: {r.stderr}")
return json.loads(r.stdout.strip() or "[]")
def slugify(s):
return re.sub(r'-+', '-', re.sub(r'[^a-z0-9]+', '-', (s or '').lower())).strip('-')
def colorway_of(mfr_sku, pattern_name):
"""Colorway = mfr_sku with the pattern slug prefix stripped."""
pat = slugify(pattern_name)
slug = slugify(mfr_sku)
if pat and slug.startswith(pat + '-'):
return slug[len(pat)+1:]
# fallback: last hyphen segment
return slug.rsplit('-', 1)[-1]
def recover_image(colorway, gallery):
"""Find the gallery URL whose filename matches the colorway.
Returns (url, confidence) where confidence in {exact, prefix, none}.
Numeric colorways (e.g. "220", "1") are too weak to fuzzy-match against
hash/product-id filenames that happen to contain the same digits, so they
are matched ONLY as a whole bounded token, never via the prefix path.
"""
if not gallery:
return None, "none"
cw = colorway.replace('-', '')
is_numeric = bool(cw) and cw.isdigit()
best = None
for url in gallery:
if not isinstance(url, str) or not url.startswith('http'):
continue
fname = url.split('/')[-1].split('?')[0].lower()
stem = re.sub(r'\.(jpg|jpeg|png|webp|gif)$', '', fname)
stem_alnum = re.sub(r'[^a-z0-9]', '', stem)
if is_numeric:
# only accept the number as a whole token bounded by non-digits
if re.search(r'(?<!\d)' + re.escape(cw) + r'(?!\d)', stem):
return url, "exact"
continue
# alpha colorway: whole token appears in filename
if cw and cw in stem_alnum:
return url, "exact"
# prefix fuzzy: filename stem shares a 4-char lead with the colorway
# (handles vendor abbreviations: eucal~eucalyptus, cash~cashmere)
if cw and len(cw) >= 4 and len(stem_alnum) >= 3 and (
stem_alnum.startswith(cw[:4]) or cw.startswith(stem_alnum[:4])):
best = best or (url, "prefix")
return best if best else (None, "none")
def build_tags(row, colorway):
tags = set()
tags.add(VENDOR)
town = (row.get("virginia_town") or "").strip()
if town:
tags.add(town)
tags.add(f"{town} Collection")
# color tag = the COLORWAY, not row.color_name (that field is a per-pattern
# bucket set for the hero variant, so it mis-tags most colorways).
cwt = colorway.replace('-', ' ').title().strip()
if cwt:
tags.add(cwt)
# origin has no text-typed metafield -> carry as a tag (matches live "Made in X")
origin = (row.get("spec_origin") or "").strip()
if origin:
tags.add(f"Made in {origin}")
# quote-only + commercial posture (matches live Reid Witlin products)
tags.update(["quotes", "Commercial", "Showroom Line", "display_variant"])
st = (row.get("spec_type") or "").lower()
if "wallcovering" in st or "vinyl" in st:
tags.add("Commercial Wallcovering")
tags.add("Fabric")
# fold in AI-derived tags/styles if present
for key in ("ai_tags", "ai_styles"):
v = row.get(key)
if isinstance(v, list):
for t in v:
if isinstance(t, str) and t.strip():
tags.add(t.strip())
return sorted(tags)
def product_type(row):
st = (row.get("spec_type") or "").lower()
if "wallcovering" in st or "vinyl" in st:
return "Wallcovering"
return "Fabric"
def main():
rows = psql_json("""
SELECT mfr_sku, pattern_name, color_name, virginia_town, spec_type,
spec_width, spec_actual_width, spec_content, spec_repeat,
spec_flamecode, spec_origin, spec_cleaning, spec_finish,
spec_abrasion,
gallery_images, ai_tags, ai_styles, description, ai_description,
product_url
FROM rwltd_catalog
WHERE (shopify_product_id IS NULL OR shopify_product_id='')
AND mfr_sku IS NOT NULL AND mfr_sku<>''
ORDER BY mfr_sku
""")
ready, held = [], []
seen = set()
n = SKU_START
for row in rows:
mfr = row["mfr_sku"]
if mfr in seen:
continue
seen.add(mfr)
pat = row.get("pattern_name") or ""
town = (row.get("virginia_town") or "").strip()
cw = colorway_of(mfr, pat)
gallery = row.get("gallery_images") or []
img, conf = recover_image(cw, gallery)
colorway_disp = cw.replace('-', ' ').title()
title = f"{town} {colorway_disp}".strip() if town else f"{pat} {colorway_disp}".strip()
sku = f"{SKU_PREFIX}-{n}"
handle = slugify(f"{town or pat} {colorway_disp} architectural-fabrics")
rec = {
"sku": sku,
"title": title,
"handle": handle,
"vendor": VENDOR,
"product_type": product_type(row),
"price": SAMPLE_PRICE,
"mfr_sku": mfr,
"pattern": pat,
"town": town,
"colorway": colorway_disp,
"image_url": img or "",
"image_confidence": conf,
"n_gallery": len(gallery),
"width": row.get("spec_actual_width") or row.get("spec_width") or "",
"content": row.get("spec_content") or "",
"origin": row.get("spec_origin") or "",
"fire_rating": row.get("spec_flamecode") or "",
"finish": row.get("spec_finish") or "",
"abrasion": row.get("spec_abrasion") or "",
"care": row.get("spec_cleaning") or "",
"repeat": row.get("spec_repeat") or "",
"spec_type": row.get("spec_type") or "",
"style": ", ".join(row["ai_styles"]) if isinstance(row.get("ai_styles"), list) else "",
"tags": " | ".join(build_tags(row, cw)),
}
n += 1
if img and conf in ("exact", "prefix"):
ready.append(rec)
else:
held.append(rec)
# POST-PASS: any image assigned to >1 product is ambiguous -> demote all to held.
from collections import Counter
img_counts = Counter(r["image_url"] for r in ready if r["image_url"])
collided = {u for u, c in img_counts.items() if c > 1}
if collided:
keep, demoted = [], []
for r in ready:
if r["image_url"] in collided:
r["image_confidence"] = "collision-held"
r["image_url"] = "" # clear the untrusted image
demoted.append(r)
else:
keep.append(r)
ready = keep
held.extend(demoted)
def write_csv(path, recs):
if not recs:
open(path, "w").close(); return
cols = list(recs[0].keys())
with open(path, "w", newline="") as f:
w = csv.DictWriter(f, fieldnames=cols); w.writeheader(); w.writerows(recs)
write_csv(os.path.join(OUT, "targets_ready.csv"), ready)
write_csv(os.path.join(OUT, "targets_image_held.csv"), held)
exact = sum(1 for r in ready if r["image_confidence"] == "exact")
prefix = sum(1 for r in ready if r["image_confidence"] == "prefix")
summary = {
"total_orphans": len(rows),
"unique_products": len(seen),
"ready_to_onboard": len(ready),
" image_exact_match": exact,
" image_prefix_match": prefix,
"image_held_for_eyeball": len(held),
"sku_range": f"{SKU_PREFIX}-{SKU_START}..{SKU_PREFIX}-{n-1}",
"vendor": VENDOR,
"price": f"${SAMPLE_PRICE} sample-only (no cost -> quote-only, 'quotes' tag)",
"distinct_towns": len(set(r["town"] for r in ready + held if r["town"])),
}
json.dump(summary, open(os.path.join(OUT, "summary.json"), "w"), indent=2)
print(json.dumps(summary, indent=2))
if __name__ == "__main__":
main()