#!/usr/bin/env python3 """PROCITY GODVERSE stock pack — real record sleeves for C's dig (?stock=real), round-7 E2. Build-time only: reads the local recordgod dataset (dealgod Postgres `discogs_release`, the 185k-vinyl set), curates ~200-400 sleeves, **parody-transforms titles/artists deterministically** (the no-real-trademarks law covers metadata), fetches cover images, bakes atlas sheets + a JSON index, publishes atlases to the depot. NOTHING queries the dataset at runtime — C reads the built pack (index + atlases) through the depot / ?localdepot=1. python3 build_stock_pack.py --from-db --count 300 # real pack (needs GODVERSE_DSN + net for covers) python3 build_stock_pack.py --sample 24 # synthetic pipeline test (no DB, solid-colour sleeves) python3 build_stock_pack.py --publish # push built atlases + index to the depot Index schema (proposed to Lane C in LANE_E_NOTES — adjust `INDEX_FIELDS` to their contract): { "version":1, "atlas_px":2048, "cell":256, "items":[ {"id","title","artist","price","price_band","atlas","uv":[u0,v0,u1,v1]} ] } Parody law: real (title,artist) → deterministic seed → parody title/artist recombined from the word banks below. The output shares NO tokens with the source, so zero real-trademark leakage, yet reads as period-plausible AU record-shop stock. Same source row → same parody, every build. """ import json, os, sys, struct, hashlib, urllib.request ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) OUT = os.path.join(ROOT, "pipeline", "_stockpack") STAGE = os.path.join(ROOT, "web", "assets", "models") # ?localdepot mirror os.makedirs(OUT, exist_ok=True) INDEX_FIELDS = ("id", "title", "artist", "price", "price_band", "atlas", "uv") # ── parody word banks (period-plausible AU record stock; no real acts) ────────────────── ADJ = ["Midnight", "Electric", "Velvet", "Golden", "Crimson", "Neon", "Sunburnt", "Hazy", "Concrete", "Paper", "Silver", "Restless", "Quiet", "Broken", "Endless", "Salt"] NOUN = ["Sessions", "Highway", "Ballroom", "Suburbs", "Tide", "Radio", "Static", "Harvest", "Verandah", "Jetty", "Milk Bar", "Cassette", "Reunion", "Departure", "Anthems", "Blue"] BAND_A = ["The", "Big", "Lonesome", "Riptide", "Powderfinger-ish", "Sunset", "Regional", "Mullet"] BAND_B = ["Galahs", "Dropbears", "Hills Hoists", "Milk Crates", "Southerlies", "Panelvans", "Rissoles", "Servos", "Bottlebrush", "Sandgropers", "Cockatoos", "Utes"] FIRST = ["Bazza", "Shazza", "Kylie", "Darryl", "Noeline", "Trev", "Merv", "Dazza", "Coral", "Kev"] LAST = ["Studebaker", "Fitzroy", "Nullarbor", "Grong", "Wollemi", "Barcoo", "Tuckerbox", "Yakka"] def _seed(*parts): return int(hashlib.sha256("|".join(str(p) for p in parts).encode()).hexdigest()[:12], 16) def parody(title, artist, is_group=None): """Deterministic parody of a real (title, artist). Shares no source tokens.""" s = _seed(title or "", artist or "") ptitle = f"{ADJ[s % len(ADJ)]} {NOUN[(s // 7) % len(NOUN)]}" if (s // 13) % 5 == 0: ptitle += f", Vol. {1 + (s // 3) % 3}" # solo vs group name, deterministically if is_group is None: is_group = (s % 2 == 0) if is_group: partist = f"{BAND_A[(s // 5) % len(BAND_A)]} {BAND_B[(s // 11) % len(BAND_B)]}".strip() else: partist = f"{FIRST[(s // 17) % len(FIRST)]} {LAST[(s // 19) % len(LAST)]}" return ptitle, partist def price_band(price): if price is None: return "bargain" if price < 8: return "bargain" if price < 25: return "standard" if price < 60: return "collector" return "grail" # ── data sources ──────────────────────────────────────────────────────────────────────── def rows_from_db(count): """Curate `count` sleeves from dealgod's discogs_release (needs GODVERSE_DSN + psycopg2). Picks rows WITH a cover image + non-empty artist/title, varied by decade for shelf variety.""" import psycopg2 # noqa: build-time only dsn = os.environ.get("GODVERSE_DSN") or os.environ.get("DB_DSN") if not dsn: raise SystemExit("GODVERSE_DSN not set — the DB dsn (creds) is required for --from-db. " "Ask John / read from dealgod's env. (Nothing queries the DB at runtime.)") con = psycopg2.connect(dsn) cur = con.cursor() cur.execute(""" SELECT release_id, artist, title, year, (SELECT current_price FROM prices p WHERE p.release_id=dr.release_id ORDER BY current_price LIMIT 1) AS price, images FROM discogs_release dr WHERE artist <> '' AND title <> '' AND images IS NOT NULL AND images <> '[]' ORDER BY md5(release_id::text) -- deterministic pseudo-random sample LIMIT %s""", (count,)) out = [] for rid, artist, title, year, price, images in cur.fetchall(): imgs = images if isinstance(images, list) else json.loads(images or "[]") url = imgs[0].get("uri") if imgs and isinstance(imgs[0], dict) else (imgs[0] if imgs else None) if url: out.append({"src_id": rid, "artist": artist, "title": title, "price": price, "img": url}) con.close() return out def rows_sample(n): """Synthetic rows for a no-DB pipeline test — solid-colour 'sleeves', fake source metadata.""" from PIL import Image, ImageDraw rows = [] os.makedirs(os.path.join(OUT, "_imgs"), exist_ok=True) for i in range(n): s = _seed("sample", i) col = ((s >> 0) & 255, (s >> 8) & 255, (s >> 16) & 255) im = Image.new("RGB", (256, 256), col) d = ImageDraw.Draw(im) d.rectangle([16, 16, 240, 240], outline=(255, 255, 255), width=3) d.ellipse([88, 88, 168, 168], fill=(20, 20, 20)) # a 'record' centre p = os.path.join(OUT, "_imgs", f"s{i}.png") im.save(p) rows.append({"src_id": f"sample-{i}", "artist": f"Real Artist {i}", "title": f"Real Album {i}", "price": 5 + (s % 80), "img": p}) return rows # ── atlas builder ─────────────────────────────────────────────────────────────────────── def build_pack(rows, cell=256, per_atlas_side=8): from PIL import Image per = per_atlas_side * per_atlas_side items, atlases = [], [] for ai in range((len(rows) + per - 1) // per): chunk = rows[ai * per:(ai + 1) * per] sheet = Image.new("RGB", (cell * per_atlas_side, cell * per_atlas_side), (30, 30, 34)) aname = f"stock_record_atlas_{ai:02d}.webp" for j, r in enumerate(chunk): gx, gy = j % per_atlas_side, j // per_atlas_side try: cov = _load_img(r["img"]).resize((cell, cell)) except Exception: continue sheet.paste(cov, (gx * cell, gy * cell)) ptitle, partist = parody(r["title"], r["artist"]) u0, v0 = gx / per_atlas_side, gy / per_atlas_side item = {"id": f"rec_{len(items):04d}", "title": ptitle, "artist": partist, "price": int(r["price"] or 5), "price_band": price_band(r["price"]), "atlas": aname, "uv": [round(u0, 4), round(v0, 4), round(u0 + 1 / per_atlas_side, 4), round(v0 + 1 / per_atlas_side, 4)]} items.append({k: item[k] for k in INDEX_FIELDS}) sheet.save(os.path.join(OUT, aname), "WEBP", quality=88) atlases.append(aname) index = {"version": 1, "atlas_px": cell * per_atlas_side, "cell": cell, "count": len(items), "atlases": atlases, "items": items} json.dump(index, open(os.path.join(OUT, "stock_record_index.json"), "w"), indent=1) return index def _load_img(src): from PIL import Image import io if src.startswith("http"): req = urllib.request.Request(src, headers={"User-Agent": "procity-stockpack/1"}) with urllib.request.urlopen(req, timeout=20) as r: return Image.open(io.BytesIO(r.read())).convert("RGB") return Image.open(src).convert("RGB") def main(): a = sys.argv if "--publish" in a: # atlases + index publish through the same depot path as GLBs (documented; run when built) for f in os.listdir(OUT): if f.endswith(".webp") or f == "stock_record_index.json": os.makedirs(STAGE, exist_ok=True) import shutil shutil.copy(os.path.join(OUT, f), os.path.join(STAGE, f)) print(f"staged pack → {STAGE} (for ?localdepot). Depot push: adapt publish.py to POST " f"{OUT}/*.webp + index (same /api/upload path).") return if "--sample" in a: n = int(a[a.index("--sample") + 1]) rows = rows_sample(n) elif "--from-db" in a: count = int(a[a.index("--count") + 1]) if "--count" in a else 300 rows = rows_from_db(count) else: print(__doc__); return idx = build_pack(rows) print(f"built pack: {idx['count']} sleeves, {len(idx['atlases'])} atlas(es) → {OUT}") print(f" sample items: {[(i['title'], i['artist'], i['price_band']) for i in idx['items'][:3]]}") if __name__ == "__main__": main()