← back to Fentucci Naturals
scripts/enrich.py
183 lines
#!/usr/bin/env python3
"""Fentucci Naturals (Tokiwa PL) enrichment — increment 2.
Per-PATTERN luxury copy via Mac1 Ollama qwen3:14b (think:false, $0 local),
cached in data/pattern-copy.json (resumable). Per-ROW description = pattern
copy + deterministic colorway sentence. Tags/specs deterministic. Writes back
to dw_unified.tokiwa_catalog (description, tags, specs, enriched_at,
settlement_status).
Fallback: if Ollama unreachable, deterministic material-copy template blocks.
HARD RULES: never the word "wallpaper"; never "Tokiwa"/mill identifiers.
"Washi" / "Japanese silk" are generic MATERIAL terms and allowed.
"""
import json, os, re, subprocess, sys, time, colorsys, urllib.request
HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
CACHE = os.path.join(ROOT, "data", "pattern-copy.json")
CONN = "host=/tmp dbname=dw_unified"
OLLAMA = "http://192.168.1.133:11434/api/generate"
MODEL = "qwen3:14b"
BANNED = re.compile(r"wallpaper|tokiwa|japan(?!ese silk)|mill\b|imported", re.I)
SETTLE = re.compile(r"\b(leaf|leaves|foliage|palm|frond|botanic\w*|floral|flower|bird|butterfl\w*|banana|grape|vine|tropic\w*)\b", re.I)
# Gilding terminology — "metal/gold/silver leaf" is a MATERIAL (applied leafing),
# not a foliage motif; strip before the settlement scan to avoid false flags.
GILDING = re.compile(r"\b(?:metal|gold|silver|copper|aluminum|champagne)[\s-]*leaf(?:ing|ed)?\b|\bleafing\b", re.I)
def psql(sql):
return subprocess.run(["psql", CONN, "-Atc", sql], capture_output=True, text=True, check=True).stdout.strip()
# ---------- deterministic template blocks (fallback + colorway sentences) ----------
MATERIAL_COPY = {
"Abaca": "Hand-woven abaca fibers lend this wallcovering an artisanal, sun-washed texture with the relaxed elegance of the Italian coast.",
"Sisal": "Natural sisal is woven strand by strand into a crisp, tailored texture that brings quiet organic warmth to refined interiors.",
"Raffia": "Supple raffia is hand-loomed into a softly irregular weave, carrying the breezy sophistication of a seaside villa.",
"Paperweave": "An intricately hand-knotted paperweave delivers dimensional texture with a light, architectural rhythm.",
"Washi Paper": "Delicate washi paper is layered into a serene, luminous surface with a whisper-soft tactility.",
"Gilding Washi": "Gilded washi paper catches the light with a burnished, jewel-box radiance over a serene handcrafted ground.",
"Japanese Silk": "Lustrous Japanese silk threads are woven into a shimmering, couture-grade surface that shifts with the light.",
"Metal Leaf": "Hand-applied metal leaf creates a gleaming, faceted surface with the opulence of an artisan atelier.",
"Glass Beads": "Thousands of glass beads are set across the surface, scattering light into a subtle, champagne-like sparkle.",
"Wood Veneer": "Fine wood veneer brings the warmth of natural grain to the wall in a sleek, tailored format.",
"Water Hyacinth": "Hand-braided water hyacinth lends a robust, basket-woven texture with grounded organic character.",
"Fine Arrowroot": "Finely woven arrowroot creates a refined, linear texture with an understated natural sheen.",
"Mica": "Crushed mica minerals give the surface a stony, shimmering depth that reads as quiet luxury.",
"Bamboo": "Slender bamboo reeds are aligned in a rhythmic, architectural weave with a warm natural finish.",
"Heavy Jute": "Heavyweight jute is woven into a bold, tactile texture with rustic-luxe presence.",
"Denim": "Woven denim cloth brings an unexpected, tailored softness and casual-luxe depth to the wall.",
"Textile": "A finely woven textile surface adds tailored softness and dimensional warmth to the wall.",
}
STYLE_TAGS = {
"Abaca": ["Grasscloth", "Organic", "Coastal"],
"Sisal": ["Grasscloth", "Organic", "Tailored"],
"Raffia": ["Grasscloth", "Organic", "Coastal"],
"Paperweave": ["Grasscloth", "Organic", "Textured"],
"Washi Paper": ["Textured", "Serene", "Modern"],
"Gilding Washi": ["Metallic", "Luxe", "Glamour"],
"Japanese Silk": ["Silk", "Elegant", "Luxe"],
"Metal Leaf": ["Metallic", "Luxe", "Glamour"],
"Glass Beads": ["Beaded", "Luxe", "Glamour"],
"Wood Veneer": ["Wood", "Warm Modern", "Organic"],
"Water Hyacinth": ["Woven", "Organic", "Coastal"],
"Fine Arrowroot": ["Grasscloth", "Organic", "Refined"],
"Mica": ["Mineral", "Shimmer", "Modern"],
"Bamboo": ["Woven", "Organic", "Zen"],
"Heavy Jute": ["Woven", "Organic", "Rustic Luxe"],
"Denim": ["Woven", "Casual Luxe", "Modern"],
"Textile": ["Woven", "Tailored", "Modern"],
}
def color_family(hexstr, name):
n = (name or "").lower()
try:
h = hexstr.lstrip("#"); r, g, b = (int(h[i:i+2], 16)/255 for i in (0, 2, 4))
hue, lig, sat = colorsys.rgb_to_hls(r, g, b)
except Exception:
return "Neutral"
if sat < 0.12:
return "Charcoal & Black" if lig < 0.3 else ("White & Ivory" if lig > 0.8 else "Gray")
deg = hue * 360
if deg < 45 or deg >= 330: fam = "Warm Neutral" if sat < 0.35 else "Terracotta & Red"
elif deg < 70: fam = "Gold & Champagne"
elif deg < 160: fam = "Green"
elif deg < 260: fam = "Blue"
else: fam = "Plum & Purple"
if fam in ("Warm Neutral", "Gold & Champagne") and lig > 0.75: fam = "Cream & Beige"
return fam
def colorway_sentence(color, fam):
return f" Shown in {color}, a {fam.lower()} colorway that settles beautifully into layered, light-filled interiors."
# ---------- LLM per-pattern copy ----------
def ollama_up():
try:
urllib.request.urlopen("http://192.168.1.133:11434/api/tags", timeout=5)
return True
except Exception:
return False
def llm_pattern_copy(pattern, material):
prompt = (
f"You are the copywriter for Designer Wallcoverings, a luxury brand. Write EXACTLY 2 sentences of "
f"elegant product copy for a wallcovering pattern named \"{pattern}\" from the Fentucci Naturals "
f"collection, made of {material}. Tell the material story (how it is crafted/woven and how it feels "
f"in a room). Rules: NEVER use the word 'wallpaper' (say 'wallcovering'). Never mention any "
f"manufacturer, country of origin, or the words Tokiwa or Japan (the material name itself is fine). "
f"No color names. No quotes, no markdown, just the 2 sentences."
)
body = json.dumps({"model": MODEL, "prompt": prompt, "stream": False, "think": False,
"options": {"temperature": 0.7, "num_predict": 160}}).encode()
req = urllib.request.Request(OLLAMA, data=body, headers={"Content-Type": "application/json"})
with urllib.request.urlopen(req, timeout=120) as r:
out = json.loads(r.read())["response"].strip()
out = re.sub(r"\s+", " ", out).strip().strip('"')
return out
def get_pattern_copy(rows):
cache = {}
if os.path.exists(CACHE):
cache = json.load(open(CACHE))
use_llm = ollama_up()
patterns = sorted({(r["pattern_name"], r["material"]) for r in rows})
t0 = time.time(); done = 0
for pat, mat in patterns:
if pat in cache and not BANNED.search(cache[pat]):
continue
copy = None
if use_llm:
for attempt in range(3):
try:
c = llm_pattern_copy(pat, mat)
if c and not BANNED.search(c) and c.count(".") >= 1 and len(c) > 80:
copy = c; break
except Exception as e:
print(f" llm fail {pat} attempt {attempt+1}: {e}", flush=True)
if not copy:
copy = f"{pat} is a study in natural craft. " + MATERIAL_COPY.get(mat, MATERIAL_COPY["Textile"])
print(f" template fallback: {pat}", flush=True)
cache[pat] = copy; done += 1
json.dump(cache, open(CACHE, "w"), indent=1)
if done % 8 == 0:
el = time.time() - t0
print(f" {done}/{len(patterns)} patterns, {el/done:.1f}s/pattern", flush=True)
return cache, use_llm
def main():
rows = json.loads(psql(
"select coalesce(json_agg(row_to_json(t))::text,'[]') from (select id, dw_sku, mfr_sku, pattern_name, "
"color_name, color_hex, material, source_category from tokiwa_catalog where enriched_at is null order by dw_sku) t"))
if not rows:
print("nothing to enrich (all rows have enriched_at)"); return
cache, use_llm = get_pattern_copy(rows)
print(f"pattern copy ready ({'LLM qwen3:14b' if use_llm else 'TEMPLATE fallback'}), enriching {len(rows)} rows", flush=True)
esc = lambda s: s.replace("'", "''")
t0 = time.time()
for i, r in enumerate(rows):
fam = color_family(r["color_hex"], r["color_name"])
desc = cache[r["pattern_name"]] + colorway_sentence(r["color_name"], fam)
mat = r["material"]
tags = ["Fentucci Naturals", "Fentucci Naturals Collection", "Natural Wallcovering",
"Natural Fiber", "Wallcovering", mat, r["pattern_name"], r["color_name"], fam,
r["dw_sku"], "display_variant", "quotes", "Needs-Price", "Needs-Width"] + STYLE_TAGS.get(mat, [])
tags = list(dict.fromkeys(tags))
specs = {"material": mat, "type": "Sidewall", "product_type": "Wallcovering",
"collection": "Fentucci Naturals", "pattern": r["pattern_name"],
"color": r["color_name"], "color_family": fam, "sold_by": "quote_only",
"sample_price": 4.25}
hay = " ".join([desc, r["pattern_name"], r["source_category"] or "", mat])
settle = "review" if SETTLE.search(GILDING.sub(" ", hay)) else "clear"
arr = "ARRAY[" + ",".join(f"'{esc(t)}'" for t in tags) + "]"
psql(f"update tokiwa_catalog set description='{esc(desc)}', tags={arr}, "
f"specs='{esc(json.dumps(specs))}'::jsonb, settlement_status='{settle}', "
f"enriched_at=now() where id={r['id']}")
if (i + 1) % 100 == 0:
print(f" rows {i+1}/{len(rows)} ({(time.time()-t0)/(i+1):.2f}s/row)", flush=True)
print(f"done: {len(rows)} rows in {time.time()-t0:.0f}s (path={'llm' if use_llm else 'template'})")
if __name__ == "__main__":
main()