PROCITY/pipeline/godverse_stock.py

#!/usr/bin/env python3
"""Lane G — G2a: per-shop REAL stock atlases (tier 1) for `?stock=real`.

THE FIRST REAL CRATE. One real shop's real crate — real sleeves, real titles, real prices —
baked to a static atlas keyed by godverse shop id. No server, no runtime query: tier 1 is files.

Two stages, split at THE SNAPSHOT BOUNDARY (the charter's determinism rule, and the same
discipline build_towns.py applies to Overpass raw):

    snapshot   POS DB + image map ->  pipeline/_godverse/<shop>_<crate>.snapshot.json
    pack       snapshot + images  ->  web/assets/stock_godverse/<shop>/stock_record_*.{json,webp}

The snapshot is the artifact of record: it runs where the databases are, pins every source row
AND every cover's sha256, and is committed. The pack stage is pure — same snapshot + same image
bytes => byte-identical atlas, on any machine, with no DB. That split is what lets Lane E's
validate_atlas gate G's output on m3 without the dealgod DB.

    python3 pipeline/godverse_stock.py snapshot --crate 550 --image-map <tsv>
    ~/Documents/MODELBEAST/venvs/mflux/bin/python pipeline/godverse_stock.py pack \
        pipeline/_godverse/monsterrobot_550.snapshot.json      # needs PIL (house convention)

THE SHOP (John's ruling, R23): `monsterrobot` — a real, trading record shop at 147 Musgrave Road,
Red Hill, Brisbane QLD. It is the same shop in three datasets, which is what makes tier 1 honest:
  recordgod   its POS: real crates, real slot numbers, real prices, real grading
  monsterrobot.party  its own website — the covers are the SHOP'S OWN product shots
  dealgod store 3962749  the scrape of that site; the cover cache + the godverse shop id
It is NOT in thriftgod's OSM census (see G-progress R23), so its godverse id is its dealgod store
id 3962749 — real, stable, and clear of the census id space (thriftgod max shop id = 2992).

DETERMINISM: ordered queries, no RNG, no sampling, no clock in the output. Prices/titles/artists
are FACTS copied from the POS, not minted (thriftgod's mint() is TABLESAMPLE + unseeded random —
it cannot produce tier 1; see G3_ECONOMY_DESIGN.md §2).

LICENCE (charter law #3): covers are the shop's own product shots from the shop's own domain,
cached by dealgod. In-house 🟢. FLAG BEFORE ANY PUBLIC/COMMERCIAL RELEASE. Real titles/artists
ship at tier 1 per John's R23 amendment to the no-real-trademarks law (in-house green, same
flag). Provenance rides in every index. PII FENCE: recordgod holds real customer/staff/sales
tables — this pipeline reads ONLY inventory/crate/disc_cache and must never read the others.
"""
import argparse, hashlib, json, os, subprocess, sys

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
SNAP_DIR = os.path.join(ROOT, "pipeline", "_godverse")
STOCK_DIR = os.path.join(ROOT, "web", "assets", "stock_godverse")
IMG_CACHE = os.environ.get("DEALGOD_IMG_CACHE", "/Volumes/johnking/dealgod_backup/images")

# The shop, per John's R23 ruling. `godverse_id` = dealgod store id (see module docstring).
SHOP = {
    "godverse_id": 3962749, "slug": "monsterrobot", "name": "Monster Robot Party",
    "address": "147 Musgrave Road", "suburb": "Red Hill", "city": "Brisbane",
    "state": "QLD", "country": "AU", "type": "record",
}
OWN_PHOTO_HOST = "monsterrobot.party"   # the licence test: the shop's own domain, nothing else

CELL_W, CELL_H, GRID, ATLAS_PX = 256, 256, 8, 2048   # Lane E's record contract (build_stock_pack.py)
PER_ATLAS = GRID * (ATLAS_PX // CELL_H)              # 64


def price_band(price):
    """Lane E's record bands, verbatim — C's dig UI reads this (grail => `rare`)."""
    p = price if price is not None else 4
    return "bargain" if p < 8 else "standard" if p < 25 else "collector" if p < 60 else "grail"


def psql(db, sql):
    """Read-only query -> list of field-lists. psql CLI, not psycopg2: the m3 has no psycopg2, and
    a snapshot stage that shells out has one less dep between the DB and the frozen file."""
    out = subprocess.run(["psql", "-d", db, "-tAF\t", "-c", sql],
                         capture_output=True, text=True, check=True).stdout
    return [ln.split("\t") for ln in out.splitlines() if ln.strip()]


def _n(v):
    return None if v in ("", "\\N") else v


# ── stage 1: snapshot ────────────────────────────────────────────────────────────────────
def load_image_map(path):
    """release_id -> local_image, restricted to the SHOP'S OWN photos (the licence test).

    On ultra (dealgod present) this is a query. On the m3 dealgod is a 2 GB dump and never
    restored, so the map is streamed out of it once — the command is recorded in the snapshot's
    provenance so the extraction is reproducible:

      gzcat dealgod-<date>.sql.gz | awk -F'\\t' '/^COPY public.products \\(/{c=1;next}
        c && /^\\\\.$/{c=0;next} c && $50!="\\\\N"{print $50"\\t"$5"\\t"$34"\\t"$52"\\t"$56}'
      (fields: 50=release_id 5=image_url 34=local_image 52=current_price 56=discogs_artist_name)
    """
    m = {}
    with open(path) as fh:
        for ln in fh:
            f = ln.rstrip("\n").split("\t")
            if len(f) < 3:
                continue
            rel, url, local = f[0], f[1], f[2]
            if OWN_PHOTO_HOST not in url or not local.startswith("/img/"):
                continue
            m.setdefault(rel, local)          # first wins; ids are sorted upstream => stable
    return m


def snapshot(args):
    crate = psql("recordgod", f"SELECT id,name,COALESCE(label_text,'') FROM crate WHERE id={args.crate}")
    if not crate:
        sys.exit(f"crate {args.crate} not found in recordgod")
    cid, cname, clabel = crate[0][0], crate[0][1], crate[0][2]

    # ONLY inventory/crate/disc_cache — the PII fence (see module docstring).
    rows = psql("recordgod", f"""
        SELECT i.sku, COALESCE(i.slot_number::text,''), i.release_id,
               COALESCE(d.artist,''), COALESCE(d.title, COALESCE(i.title,'')),
               i.price, COALESCE(i.condition,''), COALESCE(i.sleeve_cond,'')
        FROM inventory i
        JOIN crate c ON c.id = i.crate_id
        LEFT JOIN disc_cache d USING (release_id)
        WHERE i.in_stock AND c.id = {args.crate} AND i.release_id IS NOT NULL
        ORDER BY i.slot_number NULLS LAST, i.sku""")

    imgs = load_image_map(args.image_map)
    items, dropped = [], {"no_own_photo": 0, "file_missing": 0, "no_price": 0}
    for sku, slot, rel, artist, title, price, cond, sleeve in rows:
        local = imgs.get(rel)
        if not local:
            dropped["no_own_photo"] += 1
            continue
        path = os.path.join(IMG_CACHE, local[len("/img/"):])
        if not os.path.isfile(path):
            dropped["file_missing"] += 1
            continue
        if not _n(price):
            dropped["no_price"] += 1
            continue
        with open(path, "rb") as fh:
            sha = hashlib.sha256(fh.read()).hexdigest()
        items.append({"sku": sku, "slot": int(slot) if slot else None, "release_id": int(rel),
                      "artist": artist, "title": title, "price": float(price),
                      "condition": cond, "sleeve_cond": sleeve,
                      "cover": local, "cover_sha256": sha})

    if args.max_items:
        items = items[:args.max_items]

    snap = {
        "version": 1, "kind": "record", "tier": 1,
        "shop": SHOP,
        "crate": {"id": int(cid), "name": cname, "label": clabel},
        "counts": {"in_crate": len(rows), "baked": len(items), "dropped": dropped},
        "source": {
            "pos_db": "recordgod", "pos_dump": args.pos_dump,
            "image_map_dump": args.image_map_dump, "image_cache": IMG_CACHE,
            "captured": args.captured,
            "extraction": "see load_image_map() docstring — the products projection command",
        },
        # US `license` throughout (R24: British `licence` retired per C's contract §7.3; E's
        # validator reads `provenance.licence or provenance.license`, so US passes the gate).
        "license": {
            "covers": "the shop's own product shots (monsterrobot.party), cached by dealgod",
            "zone": "in-house-green", "flag": "FLAG BEFORE ANY PUBLIC/COMMERCIAL RELEASE",
            "metadata": "real titles/artists — John's R23 amendment to the no-real-trademarks law",
            "pii": "none — inventory/crate/disc_cache only; customer/staff/sales never read",
        },
        "items": items,
    }
    os.makedirs(SNAP_DIR, exist_ok=True)
    dst = os.path.join(SNAP_DIR, f"{SHOP['slug']}_{cid}.snapshot.json")
    with open(dst, "w") as fh:
        json.dump(snap, fh, indent=1, sort_keys=False)
        fh.write("\n")
    print(f"snapshot: crate {cid} '{clabel or cname}' — {len(items)} real records "
          f"(of {len(rows)} in crate; dropped {dropped}) -> {dst}")


# ── stage 2: pack (pure: snapshot + image bytes -> atlas) ─────────────────────────────────
def pack(args):
    from PIL import Image
    snap = json.load(open(args.snapshot))
    shop, items = snap["shop"], snap["items"]
    out_dir = os.path.join(STOCK_DIR, str(shop["godverse_id"]))
    os.makedirs(out_dir, exist_ok=True)

    idx_items, atlases = [], []
    for ai in range((len(items) + PER_ATLAS - 1) // PER_ATLAS):
        chunk = items[ai * PER_ATLAS:(ai + 1) * PER_ATLAS]
        W, H = GRID * CELL_W, (ATLAS_PX // CELL_H) * CELL_H
        sheet = Image.new("RGB", (W, H), (28, 28, 32))
        aname = f"stock_record_atlas_{ai:02d}.webp"
        for j, it in enumerate(chunk):
            gx, gy = j % GRID, j // GRID
            path = os.path.join(IMG_CACHE, it["cover"][len("/img/"):])
            with open(path, "rb") as fh:
                raw = fh.read()
            if hashlib.sha256(raw).hexdigest() != it["cover_sha256"]:
                sys.exit(f"cover drift: {it['sku']} {it['cover']} — snapshot sha mismatch, re-snapshot")
            import io
            cov = Image.open(io.BytesIO(raw)).convert("RGB").resize((CELL_W, CELL_H), Image.LANCZOS)
            sheet.paste(cov, (gx * CELL_W, gy * CELL_H))
            u0, v0 = gx * CELL_W / W, gy * CELL_H / H
            idx_items.append({
                # C §7.2a (R25): sku-derived, never positional. `rec_NNNN` is the generic packs'
                # namespace and a real-stock pack must not emit into it — positionally, all 120 of
                # our ids WERE generic-pack ids, which is what made an id-equality gate vacuous.
                # And position isn't identity: sell one record out of the crate and `rec_0050`
                # silently becomes a different record on the next bake, which is exactly what
                # tier-2's sold-means-gone must never do. The POS sku is the shop's own identity
                # for that copy (release_id identifies the pressing; the sku identifies the copy).
                "id": f"sku_{it['sku']}", "title": it["title"], "artist": it["artist"],
                "price": int(round(it["price"])), "price_band": price_band(it["price"]),
                "atlas": aname, "uv": [round(u0, 4), round(v0, 4),
                                       round(u0 + CELL_W / W, 4), round(v0 + CELL_H / H, 4)],
            })
        sheet.save(os.path.join(out_dir, aname), "WEBP", quality=88, method=6)
        atlases.append(aname)

    with open(args.snapshot, "rb") as fh:
        snap_sha = hashlib.sha256(fh.read()).hexdigest()
    lic = snap["license"]
    generator, snapshot_name = "pipeline/godverse_stock.py", os.path.basename(args.snapshot)
    attribution = (f"Stock, prices + covers: {shop['name']} ({OWN_PHOTO_HOST}) — the shop's own "
                   f"product shots · cover cache: GODVERSE/dealgod · POS: recordgod")
    index = {
        "version": 1, "kind": "record", "tier": 1,
        "atlas_px": ATLAS_PX, "cell_w": CELL_W, "cell_h": CELL_H,
        "count": len(idx_items), "atlases": atlases,
        # TOP LEVEL, per C's contract §7.3 + the house convention of the shipped town caches
        # (`license` is a short US-spelled string there, `attribution` its companion).
        "license": f"{lic['zone']} — {lic['covers']}; {lic['flag']}",
        "attribution": attribution,
        "generator": generator,
        "snapshot": snapshot_name,
        "shop": {k: shop[k] for k in ("godverse_id", "slug", "name", "suburb", "state")},
        "crate": snap["crate"],
        # NESTED, per E's validate_atlas.py — which is the authority C's §7.3 defers to for these
        # fields, and which reads provenance.{generator,snapshot,licence|license{zone,flag,covers}}.
        # C's doc asserts top-level is "exactly what E's validator reads"; measured, it is not (see
        # G-progress R24). Emitting both keeps C's contract and E's gate green off ONE source of
        # truth (`lic`/`generator`/`snapshot` above) — flagged for F/Fable to collapse to one.
        "provenance": {
            "generator": generator,
            "source": snap["source"], "license": lic,
            "snapshot": snapshot_name, "snapshot_sha256": snap_sha,
            "determinism": "pure bake — same snapshot + same cover bytes => byte-identical atlas; "
                           "covers pinned by sha256, no RNG, no clock",
        },
        "items": idx_items,
    }
    dst = os.path.join(out_dir, "stock_record_index.json")
    with open(dst, "w") as fh:
        json.dump(index, fh, indent=1)
        fh.write("\n")
    print(f"pack: {len(idx_items)} real sleeves -> {len(atlases)} atlas(es) in {out_dir}")
    print(f"      shop {shop['godverse_id']} ({shop['slug']}) · crate "
          f"{snap['crate']['id']} '{snap['crate']['label'] or snap['crate']['name']}'")


ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
sub = ap.add_subparsers(dest="cmd", required=True)
s1 = sub.add_parser("snapshot", help="POS + image map -> frozen snapshot (runs where the DBs are)")
s1.add_argument("--crate", type=int, required=True)
s1.add_argument("--image-map", required=True, help="TSV: release_id, image_url, local_image, ...")
s1.add_argument("--max-items", type=int, default=0, help="cap (atlas budget — charter risk #2)")
s1.add_argument("--pos-dump", default="", help="recordgod dump filename, for provenance")
s1.add_argument("--image-map-dump", default="", help="dealgod dump filename, for provenance")
s1.add_argument("--captured", default="", help="snapshot date (YYYY-MM-DD), for provenance")
s1.set_defaults(func=snapshot)
s2 = sub.add_parser("pack", help="snapshot -> atlas + index (pure; needs PIL, no DB)")
s2.add_argument("snapshot")
s2.set_defaults(func=pack)
args = ap.parse_args()
args.func(args)