← back to Designer Wallcoverings
onboarding/sangetsu-lilycolor/scripts/sangetsu-build.py
191 lines
#!/usr/bin/env python3
"""sangetsu-build.py — explode Sangetsu pattern-staging into per-colorway
publishable records (Steve's decision: each colorway = its own product).
OFFLINE / non-gated / idempotent. Reads staging/sangetsu-staging.jsonl (623
patterns, colorways nested in colorway_skus) and emits one record per UNIQUE
colorway SKU (~19,463) into staging/sangetsu-unified-staging.jsonl.
HARD CONSTRAINTS (never violated):
- Goodrich / Sangetsu NEVER customer-facing. Brand = "Architectural Wallcoverings".
Scrub goodrich|sangetsu (+ the raw upstream makers) from title/description/handle.
- "Wallpaper" -> "Wallcovering" everywhere. Never "Unknown" in a title.
- Every record: cost_confirmed=false, settlement_status="pending" (-> DRAFT),
activation_ready=false, status="draft". Publish stays gated on cost + settlement.
- vendor_prefix = DWGO (assigned per-unique-SKU, dw_unified write stays GO-B).
- Licensed-brand collections (Trussardi/Pip Studio/Missoni/Versace/...) flagged
needs_brand_review (trademark/licensing = Steve's call).
Settlement is NOT faked here. All records are settlement_status="pending". A
labelled, NON-authoritative risk pre-screen (settlement_risk) estimates how many
patterns will need the real settlement skill, purely so Steve sees the scale.
"""
import sys, os, json, re, collections
HERE = os.path.dirname(os.path.abspath(__file__))
STAGING = os.path.join(os.path.dirname(HERE), "staging")
IN = os.path.join(STAGING, "sangetsu-staging.jsonl")
OUT = os.path.join(STAGING, "sangetsu-unified-staging.jsonl")
BRAND = "Architectural Wallcoverings"
# names that must never surface customer-facing (private-label upstream)
LEAK_RE = re.compile(r"\b(goodrich|sangetsu)\b", re.I)
# "Wallpaper" -> "Wallcovering" (case-preserving-ish)
def clean_wallpaper(s):
if not s:
return s
s = re.sub(r"WALLPAPERS", "WALLCOVERINGS", s)
s = re.sub(r"WALLPAPER", "WALLCOVERING", s)
s = re.sub(r"Wallpapers", "Wallcoverings", s)
s = re.sub(r"Wallpaper", "Wallcovering", s)
s = re.sub(r"wallpapers", "wallcoverings", s)
s = re.sub(r"wallpaper", "wallcovering", s)
return s
def scrub_leak(s):
if not s:
return s
return LEAK_RE.sub("", s).replace(" ", " ").strip()
# licensed-brand collections -> needs_brand_review (trademark call)
LICENSED = re.compile(
r"\b(trussardi|pip studio|missoni|versace|zoffany|sanderson|harlequin|"
r"morris|william morris|omexco|jab|arte|elitis)\b", re.I)
# settlement RISK pre-screen (TRIAGE ONLY — not the legal gate)
BOTANICAL = re.compile(
r"\b(palm|frond|leaf|leaves|foliage|tropical|jungle|botanical|floral|flower|"
r"vine|fern|banana|grape|bird|butterfly|butterflies|feather|nature|garden)\b", re.I)
# junk pattern-name cleanup (entities from the scrape)
def clean_name(s):
if not s:
return s
s = (s.replace("&", "&").replace("&", "&").replace("&", "&")
.replace("’", "'").replace("–", "-").replace("“", "")
.replace("”", "").replace(""", '"'))
s = re.sub(r"[_]+$", "", s).strip() # trailing underscores from scrape
s = re.sub(r"\s+", " ", s)
return s
_TITLE_BAD = ("unknown", "page not found", "404", "not found", "undefined", "null")
def main():
pats = [json.loads(l) for l in open(IN) if l.strip()]
n_out = 0
seen = set()
brand_review = 0
risk_hi = 0
no_img = 0
guard_fail = []
per_pattern_risk = {}
with open(OUT, "w") as fout:
for p in pats:
raw_name = p.get("pattern") or ""
name = clean_name(raw_name)
overview = clean_wallpaper(scrub_leak(p.get("overview") or "")) or ""
spec = p.get("spec") or {}
imgs = p.get("sku_images")
pat_licensed = bool(LICENSED.search(name))
risk = bool(BOTANICAL.search(name + " " + overview))
per_pattern_risk[name] = risk
if risk:
risk_hi += 1
# normalise image access: dict{sku:url} or list[...]
def img_for(sku):
if isinstance(imgs, dict):
return imgs.get(sku) or (list(imgs.values())[0] if imgs else None)
if isinstance(imgs, list) and imgs:
return imgs
return None
for x in (p.get("colorway_skus") or []):
sku = str(x).strip()
if not sku or sku in seen:
continue
seen.add(sku)
img = img_for(sku)
has_image = bool(img)
if not has_image:
no_img += 1
# title: "<Pattern> <ColorwaySKU> | Architectural Wallcoverings"
# (no per-colorway color name exists in the free source -> flag it)
pat_disp = name if name else sku # never blank; never "Unknown"
title = clean_wallpaper(scrub_leak(f"{pat_disp} {sku} | {BRAND}"))
# Guard: "wallpaper" banned anywhere; error-tokens only as WHOLE
# words in the PATTERN-NAME segment — NOT substring-matched into the
# SKU (e.g. "404" legitimately appears in KAM404 / MIN3404).
nameseg = pat_disp.lower()
bad = "wallpaper" in title.lower() or any(
re.search(r"(?<![0-9a-z])" + re.escape(t) + r"(?![0-9a-z])", nameseg)
for t in _TITLE_BAD)
if bad:
guard_fail.append((sku, title))
continue # never emit a bad title
rec = {
"source": "goodrich-sangetsu", # internal only, never shipped
"mfr_sku": sku,
"mfr_prefix": p.get("mfr_prefix") or "",
"pattern": pat_disp,
"book": pat_disp,
"vendor_prefix": "DWGO",
"brand": BRAND,
"title": title,
"description": overview,
"width": spec.get("width") or "",
"spec": spec,
"image_url": img,
"image_count": 1 if has_image else 0,
"source_url": p.get("source_url"),
# gates — all closed; publish needs cost AND settlement
"cost_confirmed": False,
"settlement_status": "pending",
"settlement_risk_triage": "review" if risk else "low",
"activation_ready": False,
"status": "draft",
"needs_color_name": True,
"needs_brand_review": pat_licensed,
"needs_settlement": True,
}
if pat_licensed:
brand_review += 1
fout.write(json.dumps(rec, ensure_ascii=False) + "\n")
n_out += 1
# leak audit on the emitted file (belt + suspenders)
leaks = 0
banned = 0
for l in open(OUT):
d = json.loads(l)
blob = (d["title"] + " " + d["description"]).lower()
if re.search(r"goodrich|sangetsu", blob):
leaks += 1
if re.search(r"\bwallpaper", blob):
banned += 1
print(json.dumps({
"patterns_in": len(pats),
"colorway_records_out": n_out,
"title_guard_blocked": len(guard_fail),
"records_no_image (-> DRAFT)": no_img,
"needs_brand_review (licensed collections)": brand_review,
"settlement_risk_triage patterns (NON-authoritative)": risk_hi,
"LEAK audit goodrich/sangetsu in title+desc": leaks,
"BANNED 'wallpaper' in title+desc": banned,
"out": os.path.relpath(OUT, os.path.dirname(HERE)),
}, indent=2))
if guard_fail:
print(" sample guard-blocked:", guard_fail[:3])
if __name__ == "__main__":
main()