PROCITY/pipeline/validate_atlas.py

#!/usr/bin/env python3
"""Validate per-shop real-stock atlases — the G2a format gate (v5.0-alpha, ROUND23 ledger #4).

Lane G bakes one atlas + index per real shop from dealgod inventory (tier 1 of the charter's stock
ladder). This gate is the FORMAT side: it proves G's output is well-formed, inside C's ceilings, and
carries its provenance — **on m3, from the files alone, with no dealgod DB and no network**. That's the
point: the gate must run anywhere, on the committed artefacts.

    python3 pipeline/validate_atlas.py        # exit 0 = every atlas valid (or none yet), 1 = a bad one

Contract (published R22 → LANE_E_NOTES E's half · LANE_C_PUB §7 C's ceilings · charter law #3 licence):
  index  `web/assets/stock_godverse/**/stock_shop_<godverseShopId>_index.json`
         {version, atlas_px, cell, atlases[], items:[{id,title,artist,price,price_band,atlas,uv}]}
         + provenance: {license, attribution, generator, snapshot}
  C's ceilings — 1 atlas/shop (hard cap 2); ≤1024² typical, 2048² only for a 60+ item shop.
  charter law #3 — every atlas carries provenance; no private-individual data, ever.
  determinism — `snapshot` pins the inventory query result (G snapshots it like E snapshots Overpass raw).

Fail-soft is NOT this gate's business: a missing atlas is legal at runtime (parody canvas, R7 law). This
gate only judges atlases that EXIST — a malformed one must fail loudly rather than render garbled sleeves.
"""
import json, os, sys, glob, struct

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
STOCK_G = os.path.join(ROOT, "web", "assets", "stock_godverse")      # Lane G's namespace (charter law #5)
BANDS = {"bargain", "standard", "collector", "grail"}
REQUIRED_ITEM = ("id", "title", "artist", "price", "price_band", "atlas", "uv")
REQUIRED_PROV = ("license", "attribution", "generator", "snapshot")   # charter law #3 + determinism
MAX_ATLASES = 2          # C: 1 atlas/shop, hard cap 2 (each extra = +1 draw)
PX_TYPICAL, PX_MAX = 1024, 2048
BIG_SHOP_ITEMS = 60      # C: 2048² only for a 60+ item shop (else spill to the type-pack)
# charter law #3: "No private-individual data, ever" — a real shop's stock row must never carry a person.
FORBIDDEN_ITEM_KEYS = {"seller", "seller_name", "owner", "owner_name", "customer", "user", "username",
                       "email", "phone", "mobile", "address", "member", "account", "consignor"}


def image_size(path):
    """(w, h) for PNG/WebP from the header alone — dependency-free, house style (cf. glb_stat.py)."""
    with open(path, "rb") as f:
        head = f.read(32)
    if head[:8] == b"\x89PNG\r\n\x1a\n":
        w, h = struct.unpack(">II", head[16:24])
        return w, h
    if head[:4] == b"RIFF" and head[8:12] == b"WEBP":
        fmt = head[12:16]
        if fmt == b"VP8X":                                  # extended: 24-bit minus-one
            b = head[24:30]
            return (b[0] | b[1] << 8 | b[2] << 16) + 1, (b[3] | b[4] << 8 | b[5] << 16) + 1
        if fmt == b"VP8 ":                                  # lossy: 14-bit each
            w, h = struct.unpack("<HH", head[26:30])
            return w & 0x3FFF, h & 0x3FFF
        if fmt == b"VP8L":                                  # lossless: packed 14-bit
            b = struct.unpack("<I", head[21:25])[0]
            return (b & 0x3FFF) + 1, ((b >> 14) & 0x3FFF) + 1
    raise ValueError("unrecognised image header (want PNG or WebP)")


def validate_index(path):
    errs, warns = [], []
    name = os.path.basename(path)
    try:
        idx = json.load(open(path))
    except Exception as e:
        return [f"{name}: does not parse: {e}"], []
    d = os.path.dirname(path)

    # ── provenance (charter law #3) — every atlas carries it, and the gate PRINTS the licence ──
    miss = [f for f in REQUIRED_PROV if not str(idx.get(f, "")).strip()]
    if miss:
        errs.append(f"{name}: missing/empty provenance {miss} (charter law #3 — every atlas carries provenance)")

    # ── C's draw ceiling (LANE_C_PUB §7.1) ──
    atlases = idx.get("atlases", [])
    if not isinstance(atlases, list) or not atlases:
        errs.append(f"{name}: no atlases[] listed")
        atlases = []
    if len(atlases) > MAX_ATLASES:
        errs.append(f"{name}: {len(atlases)} atlases — C's hard cap is {MAX_ATLASES} "
                    f"(1/shop; a 60+ item shop spills to the shared type-pack instead of fragmenting)")
    elif len(atlases) == 2:
        warns.append(f"{name}: 2 atlases (+1 draw) — C's rule is 1/shop; 2 is the cap, not the target")

    items = idx.get("items", [])
    if not isinstance(items, list) or not items:
        errs.append(f"{name}: no items")
        items = []

    # ── C's texture/VRAM ceiling (§7.2) + index↔atlas consistency (the declared size must be REAL) ──
    declared = idx.get("atlas_px")
    for a in atlases:
        ap = os.path.join(d, a)
        if not os.path.isfile(ap):
            errs.append(f"{name}: atlas {a!r} listed but missing on disk")
            continue
        try:
            w, h = image_size(ap)
        except Exception as e:
            errs.append(f"{name}: atlas {a!r} unreadable: {e}")
            continue
        if declared and (w != declared or h != declared):
            errs.append(f"{name}: atlas {a!r} is {w}×{h} but index declares atlas_px={declared}")
        big = len(items) >= BIG_SHOP_ITEMS
        cap = PX_MAX if big else PX_TYPICAL
        if max(w, h) > cap:
            errs.append(f"{name}: atlas {a!r} {w}×{h} exceeds C's ceiling {cap}² "
                        f"({'60+ item shop' if big else f'{len(items)}-item shop → ≤{PX_TYPICAL}²'})")
        if max(w, h) > PX_TYPICAL and not big:
            warns.append(f"{name}: atlas {a!r} is {w}×{h} for {len(items)} items — prefer ≤{PX_TYPICAL}²")

    # ── the index itself (schema confirmed verbatim in LANE_C_PUB §7.3) ──
    ids = set()
    for i, it in enumerate(items):
        if not isinstance(it, dict):
            errs.append(f"{name}[{i}]: item is not an object"); continue
        m = [f for f in REQUIRED_ITEM if f not in it]
        if m:
            errs.append(f"{name}[{i}]: missing {m}"); continue
        priv = sorted(set(k.lower() for k in it) & FORBIDDEN_ITEM_KEYS)
        if priv:
            errs.append(f"{name}[{it['id']}]: private-individual field(s) {priv} — charter law #3 forbids it")
        if not str(it["title"]).strip() or not str(it["artist"]).strip():
            errs.append(f"{name}[{it['id']}]: empty title/artist")
        if it["id"] in ids:
            errs.append(f"{name}: duplicate id {it['id']!r}")
        ids.add(it["id"])
        if it["price_band"] not in BANDS:
            errs.append(f"{name}[{it['id']}]: bad price_band {it['price_band']!r}")
        if not isinstance(it["price"], (int, float)) or it["price"] < 0:
            errs.append(f"{name}[{it['id']}]: bad price {it['price']!r}")
        if it["atlas"] not in atlases:
            errs.append(f"{name}[{it['id']}]: atlas {it['atlas']!r} not in atlases[]")
        uv = it["uv"]
        if not (isinstance(uv, list) and len(uv) == 4 and all(isinstance(v, (int, float)) for v in uv)):
            errs.append(f"{name}[{it['id']}]: uv must be [u0,v0,u1,v1] numbers"); continue
        u0, v0, u1, v1 = uv
        if not (0 <= u0 < u1 <= 1 and 0 <= v0 < v1 <= 1):     # UV origin top-left; C flips V (R7 law)
            errs.append(f"{name}[{it['id']}]: uv out of range / inverted: {uv}")
    return errs, warns


def main():
    idxs = sorted(glob.glob(os.path.join(STOCK_G, "**", "stock_shop_*_index.json"), recursive=True))
    if not idxs:
        print("atlas-QA: no per-shop atlases yet (Lane G's G2a) — nothing to validate")
        return 0
    all_errs, all_warns = [], []
    for p in idxs:
        e, w = validate_index(p)
        all_errs += e; all_warns += w
        idx = {}
        try:
            idx = json.load(open(p))
        except Exception:
            pass
        print(f"  {os.path.basename(p):40} items={len(idx.get('items', [])):>3} "
              f"atlases={len(idx.get('atlases', []))} px={idx.get('atlas_px')} "
              f"snapshot={str(idx.get('snapshot', '—'))[:24]} {'OK' if not e else 'ERR(' + str(len(e)) + ')'}")
        if idx.get("license"):                    # licence law: the gate SHOWS it, a human clears it
            print(f"      licence: {idx['license']}")
    for w in all_warns:
        print(f"  WARN {w}")
    for e in all_errs:
        print(f"  ERR  {e}")
    if all_errs:
        print(f"atlas-QA FAIL — {len(all_errs)} error(s), {len(all_warns)} warning(s)")
        return 1
    print(f"atlas-QA OK — {len(idxs)} per-shop atlas(es) valid, {len(all_warns)} warning(s)")
    return 0


if __name__ == "__main__":
    sys.exit(main())