"""routes_assets.py — product image service (wave 18, contract C2-ASSET). Serves the catalog's product imagery. Two sources, checked in order: 1. `AIOS_ASSET_DIR` — a local directory of master images whose FILENAME STEM is the product `code` (Odoo default_code, case-insensitive). Dev mode: the owner's master folder on this machine. Derivatives are generated on demand into a disk cache. 2. `AIOS_ASSET_REPO` — a private HF dataset (default `royal-imports/product-assets`) laid out by `platform/ingest_product_assets.py`: assets/products/orig/ originals, archived verbatim (R9) assets/products/web/.png ~800px alpha-preserving derivative manifest.json {"codes": {"": {"ext": ".png"}}, ...} Files download once per container into the HF cache; responses carry immutable cache headers so the browser re-fetches nothing. QUALITY (R9): `q=web` = the low-res derivative the app displays; `q=print` and `q=orig` BOTH serve the ORIGINAL bytes — v1's "high quality" IS the master file (dated amendment in the wave doc). The gate is the PRODUCT grant: images decorate the product surface, so whoever may open `product_data` may fetch them; nobody else may. Codes are validated against a closed charset before they touch a filesystem or a repo path — a path separator can never reach the lookup, so traversal is refused at the door (400), and an unknown code is 404 `asset_not_found`, never an empty 200. """ import base64 import io import os import re import tempfile import time import uuid from pathlib import Path from fastapi import Body, Depends from fastapi import APIRouter from fastapi.responses import FileResponse from deps import Session, err, module_gate, require_session from routes_admin import admin_gate router = APIRouter(prefix="/api/v1") #: Same grant as the product surface these images decorate (routes_products.MODULE, restated #: here to avoid importing that module for one string). MODULE = "product_data" _WEB_EDGE = 800 # max edge (px) of the in-app derivative _QUALITIES = ("web", "print", "orig") _EXT_MEDIA = {".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg"} #: closed charset — no slashes, no backslashes, no colons, so a code can never be a path. _CODE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9 ._+&()\-]{0,63}$") _IMMUTABLE = "public, max-age=31536000, immutable" #: per-process index of the local master dir: rebuilt when the dir path or its mtime changes. _DIR_INDEX = {"path": None, "mtime": None, "map": {}} def _norm(code): return str(code or "").strip().upper() def _local_dir(): d = os.environ.get("AIOS_ASSET_DIR", "").strip() if not d: return None p = Path(d) return p if p.is_dir() else None def _cache_root(): root = Path(os.environ.get("AIOS_ASSET_CACHE", "").strip() or Path(tempfile.gettempdir()) / "aios_asset_cache") root.mkdir(parents=True, exist_ok=True) return root def _dir_index(d): """{CODE: Path} over the master dir. Two files sharing an upper-cased stem keep the lexicographically first — the same rule the ingest script applies, so dir mode and repo mode resolve identically.""" try: mtime = d.stat().st_mtime_ns except OSError: return {} if _DIR_INDEX["path"] == str(d) and _DIR_INDEX["mtime"] == mtime: return _DIR_INDEX["map"] idx = {} for p in sorted(d.iterdir()): if not p.is_file() or p.name.startswith("."): continue if p.suffix.lower() not in _EXT_MEDIA: continue idx.setdefault(p.stem.upper(), p) _DIR_INDEX.update(path=str(d), mtime=mtime, map=idx) return idx def _web_derivative(src, code): """The ~800px alpha-preserving PNG for one master file, built lazily into the disk cache. Rebuilt when the master is newer than the cached copy.""" out = _cache_root() / "web" out.mkdir(parents=True, exist_ok=True) dst = out / f"{code}.png" try: if dst.is_file() and dst.stat().st_mtime >= src.stat().st_mtime: return dst except OSError: pass try: from PIL import Image except ImportError: raise err(503, "pillow_missing", "image derivatives need Pillow — `pip install pillow` on this host") with Image.open(src) as im: im.load() if im.mode not in ("RGBA", "RGB", "LA", "L"): im = im.convert("RGBA") im.thumbnail((_WEB_EDGE, _WEB_EDGE)) im.save(dst, format="PNG", optimize=True) return dst def _assets(): """The asset SOURCE lives in `platform/core/assets.py`, not here. ⛔ NOT TIDINESS — a CONTRACT. `ops/verify_portability.py` B1/B2 assert the HuggingFace SDK appears only in its declared homes and NEVER under `aios-web/api/`, because this process is meant to lift to Hetzner/EC2 without a hub client. The first version of this file imported `huggingface_hub` directly and the gate caught it.""" import core.assets as assets return assets @router.get("/assets/products") def asset_manifest(session: Session = Depends(module_gate(MODULE))): """The codes that HAVE an image — so the catalog designer can mark gaps instead of rendering broken tiles. `source` names which backend answered.""" d = _local_dir() if d is not None: return {"codes": sorted(_dir_index(d).keys()), "source": "dir"} codes = _assets().manifest() if codes: return {"codes": sorted(codes.keys()), "source": "repo"} return {"codes": [], "source": "none"} @router.get("/assets/products/{code}") def product_asset(code: str, q: str = "web", session: Session = Depends(module_gate(MODULE))): if q not in _QUALITIES: raise err(400, "bad_quality", f"q must be one of {', '.join(_QUALITIES)}") norm = _norm(code) if not _CODE_RE.match(norm): raise err(400, "bad_code", "that is not a product code") path = None d = _local_dir() if d is not None: src = _dir_index(d).get(norm) if src is not None: path = _web_derivative(src, norm) if q == "web" else src if path is None: path = _assets().fetch(norm, q) if path is None or not Path(path).is_file(): raise err(404, "asset_not_found", f"no image for {norm}") media = _EXT_MEDIA.get(Path(path).suffix.lower(), "application/octet-stream") return FileResponse(str(path), media_type=media, headers={"Cache-Control": _IMMUTABLE}) # ------------------------------------------------------------------ editorial (DEBT-5) # Non-SKU imagery for catalog GALLERY pages — the 2027 book's full-bleed lifestyle # photography, which has no product code to live under. Second prefix in the same repo: # assets/editorial/orig/ masters, verbatim # assets/editorial/web/.png ~1600px derivative (full-page display) # manifest.json gains {"editorial": {"": {"ext": ...}}} # The designer addresses these as `ed:` in a page's imageCode — a prefix the product # charset can never produce (':' is outside _CODE_RE), so the two namespaces cannot collide. # # MODAL, not layered: with AIOS_ASSET_DIR set, `/editorial/` is AUTHORITATIVE (dev and # the hermetic gate); otherwise the repo is. Falling through dir→repo here would let a gate # with a temp dir silently read the production repo and go green on prod state. _ED_WEB_EDGE = 1600 # full-page display derivative; print still serves the master _SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,63}$") _ED_MAX_BYTES = 25 * 1024 * 1024 def _ed_dir(create=False): d = _local_dir() if d is None: return None p = d / "editorial" if create: p.mkdir(parents=True, exist_ok=True) return p def _ed_dir_index(ed): idx = {} try: for p in sorted(ed.iterdir()): if p.is_file() and p.suffix.lower() in _EXT_MEDIA and not p.name.startswith("."): idx.setdefault(p.stem.lower(), p) except OSError: pass return idx @router.get("/assets/editorial") def editorial_manifest(session: Session = Depends(module_gate(MODULE))): """The slugs that HAVE an editorial image — the designer's picker reads this.""" if _local_dir() is not None: ed = _ed_dir() return {"slugs": sorted(_ed_dir_index(ed).keys()) if ed else [], "source": "dir"} slugs = _assets().editorial() return {"slugs": sorted(slugs.keys()), "source": "repo" if slugs else "none"} @router.get("/assets/editorial/{slug}") def editorial_asset(slug: str, q: str = "web", session: Session = Depends(module_gate(MODULE))): if q not in _QUALITIES: raise err(400, "bad_quality", f"q must be one of {', '.join(_QUALITIES)}") slug = str(slug or "").strip().lower() if not _SLUG_RE.match(slug): raise err(400, "bad_slug", "that is not an editorial slug") path = None if _local_dir() is not None: ed = _ed_dir() src = _ed_dir_index(ed).get(slug) if ed else None if src is not None: path = _web_derivative(src, f"ed_{slug}") if q == "web" else src else: path = _assets().fetch_editorial(slug, q) if path is None or not Path(path).is_file(): raise err(404, "asset_not_found", f"no editorial image named {slug}") media = _EXT_MEDIA.get(Path(path).suffix.lower(), "application/octet-stream") return FileResponse(str(path), media_type=media, headers={"Cache-Control": _IMMUTABLE}) @router.post("/assets/editorial", status_code=201) def editorial_upload(body: dict = Body(default=None), session: Session = Depends(admin_gate)): """Store one editorial image: `{slug, data}` with `data` base64 (JSON keeps the API free of a multipart dependency; an admin upload is rare enough to wear the +33%). The bytes must OPEN as an image — a corrupt file is refused here, not discovered on a printed page.""" body = body or {} slug = str(body.get("slug") or "").strip().lower() if not _SLUG_RE.match(slug): raise err(400, "bad_slug", "slugs are lowercase letters, digits and hyphens (max 64)") raw = str(body.get("data") or "") raw = raw.split(",", 1)[1] if raw.startswith("data:") and "," in raw else raw try: blob = base64.b64decode(raw, validate=True) except Exception: raise err(400, "bad_image", "data must be base64 image bytes") if not blob: raise err(400, "bad_image", "the upload is empty") if len(blob) > _ED_MAX_BYTES: raise err(400, "image_too_large", f"editorial masters cap at {_ED_MAX_BYTES // (1024 * 1024)} MB") try: from PIL import Image except ImportError: raise err(503, "pillow_missing", "image validation needs Pillow — `pip install pillow` on this host") try: with Image.open(io.BytesIO(blob)) as im: im.load() fmt = (im.format or "").upper() if fmt not in ("PNG", "JPEG"): raise err(400, "bad_image", "PNG or JPEG only") ext = ".png" if fmt == "PNG" else ".jpg" web = im.convert("RGB") if im.mode not in ("RGB", "RGBA", "L", "LA") else im.copy() web.thumbnail((_ED_WEB_EDGE, _ED_WEB_EDGE)) buf = io.BytesIO() web.save(buf, format="PNG", optimize=True) except Exception as e: if getattr(e, "status_code", None): raise raise err(400, "bad_image", "that file does not open as an image") if _local_dir() is not None: ed = _ed_dir(create=True) (ed / f"{slug}{ext}").write_bytes(blob) return {"slug": slug, "source": "dir"} try: _assets().put_editorial(slug, blob, ext, buf.getvalue()) except Exception: raise err(503, "asset_repo_unavailable", "the asset repo did not accept the upload — try again") return {"slug": slug, "source": "repo"} # ───────────────────────────────── RECORD IMAGES (wave 19, owner item 5 / R7 / contract C5) # # The third namespace, and the first one ORDINARY USERS write. An `image` field's cell holds a # string reference; `rec:` names a picture uploaded here. # # ⛔ THREE WALLS, and each exists because of a specific way this could go wrong: # # 1. SESSION-GATED, NOT ADMIN-GATED. R7 is explicit — a non-admin attaches a picture to their # own record. The editorial upload above IS admin-only because it publishes into Royal's # shared catalogue library; this writes a per-record attachment, which is the same class of # act as typing a note. # 2. TENANT FROM THE SESSION, NEVER FROM THE STRING. The stored address is `/` and # both halves of it are composed server-side. A `rec:` reference pasted from another # tenant's cell is looked up under THIS tenant's prefix, finds nothing, and 404s. There is # no code path in which a client-supplied string chooses a tenant. # 3. IT MUST OPEN AS AN IMAGE. A 2 MB cap on the DECODED bytes (the client checks too, but a # client check only spares the user a pointless upload) and Pillow must parse it. Without # the parse, "upload an image" is "upload 2 MB of anything to the tenant's dataset". # # The id is server-minted hex, so a client never names a path component at all — traversal is # not refused so much as unreachable. The read route re-validates it anyway (defence in depth, # and the same closed-charset posture `_CODE_RE` takes for product codes). _REC_MAX_BYTES = 2 * 1024 * 1024 # C5's cap, enforced on the DECODED bytes _REC_WEB_EDGE = 1200 # bigger than a SKU thumbnail; smaller than editorial _REC_ID_RE = re.compile(r"^[a-f0-9]{8,32}$") def _rec_dir(tenant, create=False): """Dir-mode home for one tenant's record images: `/records//`. Dir mode is MODAL exactly as editorial is (see that section's note): with `AIOS_ASSET_DIR` set it is authoritative and the repo is never consulted, so a hermetic gate cannot silently read production state and go green on it. """ d = _local_dir() if d is None: return None p = d / "records" / str(tenant or "").strip().lower() if create: p.mkdir(parents=True, exist_ok=True) return p @router.post("/assets/records", status_code=201) def record_upload(body: dict = Body(default=None), session: Session = Depends(require_session)): """Store one record image and return the REFERENCE to put in the cell. `{data: ""}` → `{"ref": "rec:"}`. The caller PATCHes that ref onto the row through the ordinary overlay wall; this route never writes a cell. That separation is deliberate: the per-key `permissions.edit` wall, the pid wall and the hidden-field wall all live in `core.grid_events`, and an asset route that wrote cells directly would be a second write path with none of them. """ body = body or {} raw = str(body.get("data") or "") raw = raw.split(",", 1)[1] if raw.startswith("data:") and "," in raw else raw try: blob = base64.b64decode(raw, validate=True) except Exception: raise err(400, "bad_image", "data must be base64 image bytes") if not blob: raise err(400, "bad_image", "the upload is empty") if len(blob) > _REC_MAX_BYTES: raise err(400, "image_too_large", f"record images cap at {_REC_MAX_BYTES // (1024 * 1024)} MB") try: from PIL import Image except ImportError: raise err(503, "pillow_missing", "image validation needs Pillow — `pip install pillow` on this host") try: with Image.open(io.BytesIO(blob)) as im: im.load() fmt = (im.format or "").upper() if fmt not in ("PNG", "JPEG"): raise err(400, "bad_image", "PNG or JPEG only") ext = ".png" if fmt == "PNG" else ".jpg" web = im.convert("RGB") if im.mode not in ("RGB", "RGBA", "L", "LA") else im.copy() web.thumbnail((_REC_WEB_EDGE, _REC_WEB_EDGE)) buf = io.BytesIO() web.save(buf, format="PNG", optimize=True) except Exception as e: if getattr(e, "status_code", None): raise raise err(400, "bad_image", "that file does not open as an image") asset_id = uuid.uuid4().hex[:24] tenant = session.tenant d = _rec_dir(tenant, create=True) if d is not None: (d / f"{asset_id}{ext}").write_bytes(blob) (d / f"{asset_id}.web.png").write_bytes(buf.getvalue()) return {"ref": f"rec:{asset_id}", "source": "dir"} try: _assets().put_record(tenant, asset_id, blob, ext, buf.getvalue()) except Exception: raise err(503, "asset_repo_unavailable", "the asset repo did not accept the upload — try again") return {"ref": f"rec:{asset_id}", "source": "repo"} @router.get("/assets/records/{asset_id}") def record_asset(asset_id: str, q: str = "web", session: Session = Depends(require_session)): """Serve one record image for THIS session's tenant. Any signed-in user of the tenant may read it — a picture on a row is visible to whoever can open the row, and the row walls already decided that.""" if q not in _QUALITIES: raise err(400, "bad_quality", f"q must be one of {', '.join(_QUALITIES)}") asset_id = str(asset_id or "").strip().lower() if not _REC_ID_RE.match(asset_id): raise err(400, "bad_asset_id", "that is not a record image id") tenant = session.tenant d = _rec_dir(tenant) path = None if d is not None: # Dir mode is MODAL: this tenant's folder, or nothing. Never a fall-through to the repo. cand = (d / f"{asset_id}.web.png") if q == "web" else None if cand is None or not cand.is_file(): cand = next((d / f"{asset_id}{e}" for e in _EXT_MEDIA if (d / f"{asset_id}{e}").is_file()), None) path = cand if cand is not None and cand.is_file() else None else: path = _assets().fetch_record(tenant, asset_id, q) if path is None or not Path(path).is_file(): # 404 — and it is the SAME answer for "no such picture" and "that picture belongs to # another tenant", which is the cross-tenant wall stated as a status code. raise err(404, "asset_not_found", "no image for that reference") media = _EXT_MEDIA.get(Path(path).suffix.lower(), "application/octet-stream") return FileResponse(str(path), media_type=media, headers={"Cache-Control": _IMMUTABLE})