← back to Dw Material Reclassify
classify.py
189 lines
#!/usr/bin/env python3
"""
Wave-1 material reclassify — DRY RUN (writes nothing, calls no paid API).
Target set: ACTIVE Wallcovering products tagged 'New Arrival' whose only material
signal is the generic Textured/Woven catch-all (Appendix A bucket 13) — i.e. no
specific fiber tag. Goal: resolve them into real fiber buckets.
Two free tiers first, vision only for the true residual:
Tier 1 explicit "Material:" line inside body_html -> authoritative, free
Tier 2 high-confidence fiber phrase in TITLE/pattern_name (not body metaphors) -> free
Vision everything left -> counted + costed here, NOT called in dry-run
Run: python3 classify.py # summary
python3 classify.py --samples # + evidence samples
"""
import json, re, subprocess, sys
VISION_RATE = 0.0006 # $/product-image, Gemini 2.0 Flash (per cost-tracker pricing)
# Canonical buckets + the fiber keywords that map to them (Appendix A precedence order:
# most specific first; generic woven/textured is NOT a resolver here — it's what we're
# trying to get OUT of).
BUCKETS = [
("Cork", [r"cork"]),
("Grasscloth", [r"grass ?cloth"]),
("Paperweave", [r"paper ?weave"]),
("Natural Fiber", [r"sisal", r"jute", r"hemp", r"abaca", r"raffia",
r"seagrass", r"arrowroot", r"\bnatural fiber", r"\bnaturals?\b"]),
("Silk", [r"\bsilk"]),
("Linen", [r"\blinen"]),
("Leather", [r"faux leather", r"vegan leather", r"\bleather"]),
("Glass Bead", [r"glass bead", r"beaded"]),
("Flock / Velvet", [r"flock", r"velvet"]),
("Metallic / Foil", [r"metallic", r"\bmica\b", r"mylar", r"\bfoil", r"tedlar"]),
("Wood Veneer", [r"wood veneer", r"\bveneer", r"paulownia"]),
("Vinyl / Type II", [r"vinyl", r"type 2\b", r"type ii\b", r"20 ?oz"]),
]
def match_bucket(text):
"""First-match-wins over BUCKETS precedence. Returns (bucket, matched_kw) or None."""
if not text:
return None
t = text.lower()
for name, pats in BUCKETS:
for p in pats:
m = re.search(p, t)
if m:
return name, m.group(0)
return None
def strip_html(s):
return re.sub(r"<[^>]+>", " ", s or "")
# Tier 1.5: resolve from the FULL body, but only on RELIABLE material statements
# (avoids prose metaphors like "linen canvas" / "delicate weave"). Order: specific
# decorative fibers first, then Non-Woven substrate as the fallback.
FIBER_CTX = [
("Grasscloth", r"grass ?cloth"), # rarely a metaphor
("Cork", r"\bcork\b"),
("Silk", r"\bsilk\b"),
("Linen", r"\blinen\b"),
("Natural Fiber", r"\b(sisal|jute|hemp|abaca|raffia|seagrass|arrowroot)\b"),
]
def body_material(b):
lb = b.lower()
# grasscloth is reliable on a bare mention
if re.search(r"grass ?cloth", lb):
return "Grasscloth", "body:grasscloth"
# other fibers only when stated AS the material: "<fiber> wallcovering/paper"
# or "is a[n] ... <fiber>"
for name, pat in FIBER_CTX[1:]:
if re.search(pat + r"[^.]{0,30}(wall ?cover|wallpaper)", lb) or \
re.search(r"\bis an?\b[^.]{0,40}" + pat, lb):
return name, f"body:{name}"
# Non-Woven substrate fallback (a real filterable material, not in Appendix A yet)
if re.search(r"non.?woven", lb):
return "Non-Woven", "body:non-woven"
return None
def build_sql(scope_all=False):
na = "" if scope_all else "and tags ilike '%New Arrival%'"
return SQL_TMPL.replace("__NA__", na)
SQL_TMPL = r"""
with wc as (
select id, coalesce(shopify_id,'') shopify_id, title, coalesce(vendor,'') vendor,
coalesce(dw_sku,'') dw_sku, coalesce(pattern_name,'') pattern_name,
coalesce(image_url,'') image_url, coalesce(body_html,'') body_html,
lower(regexp_replace(tags,'color:[a-z ]+','','g')) as t
from shopify_products
where product_type='Wallcovering' and status='ACTIVE'
__NA__ and tags is not null
)
select json_agg(json_build_object(
'id',id,'shopify_id',shopify_id,'title',title,'vendor',vendor,'dw_sku',dw_sku,
'pattern_name',pattern_name,'image_url',image_url,'body_html',body_html))
from wc
where not (t ~ '(cork|grasscloth|paperweave|paper weave|sisal|jute|hemp|abaca|raffia|seagrass|arrowroot|silk|linen|leather|glass bead|beaded|flock|metallic|mica|mylar|foil|tedlar|wood veneer|veneer|paulownia|vinyl|type 2|type ii|20 oz|natural wallcovering|naturals|natural fiber|natural textile|natural resource)')
and (t ~ '(woven|textured|texture)');
"""
def load_rows(scope_all=False):
out = subprocess.run(["psql", "host=/tmp dbname=dw_unified", "-tAc", build_sql(scope_all)],
capture_output=True, text=True)
if out.returncode != 0:
sys.exit("psql error: " + out.stderr)
return json.loads(out.stdout.strip() or "[]")
# Extract the explicit "Material:" clause from a description, if present.
MAT_LINE = re.compile(r"material\s*[:\-]\s*([^.<\n]{0,80})", re.I)
def classify(row):
body = strip_html(row["body_html"])
# Tier 1: explicit "Material:" line -> authoritative
m = MAT_LINE.search(body)
if m:
b = match_bucket(m.group(1))
if b:
return "tier1_material_line", b[0], f'Material: {m.group(1).strip()!r}'
# Tier 1.5: reliable material statement anywhere in body
bm = body_material(body)
if bm:
return "tier15_body", bm[0], bm[1]
# Tier 2: fiber word in TITLE/pattern — LOW CONFIDENCE. Fiber words appear in
# colorway/pattern names constantly ("Pop The Cork", "...& Linen" colorway), so this
# is NOT trusted as a resolver. Flagged for review + sent to vision like the rest.
b = match_bucket(row["title"]) or match_bucket(row["pattern_name"])
if b:
src = row["title"] if match_bucket(row["title"]) else row["pattern_name"]
return "needs_vision", None, f'LOWCONF name-match {b[0]} in {src!r} — sent to vision'
# Residual -> vision
return "needs_vision", None, ""
def main():
show_samples = "--samples" in sys.argv
scope_all = "--all" in sys.argv
rows = load_rows(scope_all)
from collections import Counter, defaultdict
tier_counts = Counter()
fiber_split = Counter()
resolved_samples, vision_samples = [], []
for r in rows:
tier, bucket, ev = classify(r)
tier_counts[tier] += 1
if bucket:
fiber_split[bucket] += 1
if len(resolved_samples) < 15:
resolved_samples.append((bucket, r["title"], tier, ev))
else:
lowconf = ev.startswith("LOWCONF")
if lowconf:
tier_counts["lowconf_namematch"] += 1
if len(vision_samples) < 12:
vision_samples.append((r["title"], r["vendor"], ev if lowconf else ""))
total = len(rows)
free = tier_counts["tier1_material_line"] + tier_counts["tier15_body"]
need = tier_counts["needs_vision"]
print("="*72)
print(f"RECLASSIFY — DRY RUN (no writes, no vision calls)")
scope = "ALL collections (full catalog)" if scope_all else "New-Arrivals only"
print(f"Scope: {scope} Textured/Woven catch-all total={total}")
print("="*72)
print(f"Resolved FREE (text): {free:4d} ({free/total*100:.0f}%)")
print(f" - tier1 Material: line (authoritative) {tier_counts['tier1_material_line']:4d}")
print(f" - tier1.5 body material statement {tier_counts['tier15_body']:4d}")
print(f"Needs VISION (residual): {need:4d} ({need/total*100:.0f}%)")
print(f" incl. low-conf name-matches excluded as unreliable: {tier_counts['lowconf_namematch']:4d}")
print(f"Projected vision cost: ${need*VISION_RATE:.2f} ({need} img x ${VISION_RATE}/img)")
print("-"*72)
print("Fiber split of FREE-resolved:")
order = [n for n,_ in BUCKETS] + ["Non-Woven"]
for name in order:
if fiber_split.get(name):
print(f" {name:20s} {fiber_split[name]:4d}")
if show_samples:
print("-"*72); print("Sample resolved (bucket | title | tier | evidence):")
for b,t,tier,ev in resolved_samples:
print(f" [{b}] {t[:42]:42s} <{tier}> {ev[:48]}")
print("-"*72); print("Sample needs-vision (title | vendor | flag):")
for t,v,ev in vision_samples:
print(f" {t[:40]:40s} | {v[:18]:18s} | {ev}")
print("="*72)
print("DRY RUN — nothing written to dw_unified, no API called. Cost this run: $0 (local).")
if __name__ == "__main__":
main()