Spaces:
Running
Running
| """Persistent, append-only storage for community gallery results. | |
| Each published result is committed as one immutable directory:: | |
| <gallery_dir>/<32-hex item id>/ | |
| item.json | |
| reference.png | |
| model.bundle.js | |
| ... | |
| Writers first build a hidden staging directory and atomically rename it to | |
| the public item id. Readers ignore staging directories and rebuild their view | |
| from validated metadata on every request, so a process restart requires no | |
| in-memory index and a concurrent publish is never observed half-written. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import math | |
| import os | |
| import re | |
| import shutil | |
| import threading | |
| import time | |
| import uuid | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| from typing import Any | |
| ITEM_ID_RE = re.compile(r"^[0-9a-f]{32}$") | |
| METADATA_NAME = "item.json" | |
| METADATA_SCHEMA_VERSION = 1 | |
| MAX_METADATA_BYTES = 64 * 1024 | |
| # This is deliberately narrower than the per-job artifact set. In particular, | |
| # events.jsonl is operational history rather than a gallery artifact and is | |
| # never made permanent or public. | |
| GALLERY_ARTIFACT_NAMES = { | |
| "reference.png": "image/png", | |
| "probe.json": "application/json", | |
| "spec.json": "application/json", | |
| "compile-spec.json": "application/json", | |
| "validation.json": "application/json", | |
| "factory.ts": "text/plain; charset=utf-8", | |
| "model.bundle.js": "text/javascript; charset=utf-8", | |
| "standalone.html": "text/html; charset=utf-8", | |
| } | |
| REQUIRED_GALLERY_ARTIFACTS = frozenset({ | |
| "reference.png", | |
| "spec.json", | |
| "factory.ts", | |
| "model.bundle.js", | |
| "standalone.html", | |
| }) | |
| class GalleryError(RuntimeError): | |
| """A gallery item could not be committed or storage is unavailable.""" | |
| def valid_item_id(item_id: str) -> bool: | |
| return bool(ITEM_ID_RE.fullmatch(item_id or "")) | |
| def _iso_utc(epoch: float) -> str: | |
| return ( | |
| datetime.fromtimestamp(epoch, timezone.utc) | |
| .isoformat(timespec="milliseconds") | |
| .replace("+00:00", "Z") | |
| ) | |
| def _safe_int(value: Any, default: int = 0) -> int: | |
| if isinstance(value, bool): | |
| return default | |
| try: | |
| parsed = int(value) | |
| except (TypeError, ValueError, OverflowError): | |
| return default | |
| return max(0, parsed) | |
| def _safe_float(value: Any, default: float = 0.0) -> float: | |
| if isinstance(value, bool): | |
| return default | |
| try: | |
| parsed = float(value) | |
| except (TypeError, ValueError, OverflowError): | |
| return default | |
| return max(0.0, parsed) if math.isfinite(parsed) else default | |
| class GalleryStore: | |
| """Concurrency-safe persistent gallery backed by immutable directories.""" | |
| def __init__(self, root: Path) -> None: | |
| self.root = Path(root) | |
| self._lock = threading.RLock() | |
| def ensure_ready(self) -> None: | |
| try: | |
| self.root.mkdir(parents=True, exist_ok=True) | |
| if not self.root.is_dir(): | |
| raise OSError("gallery path is not a directory") | |
| except OSError as exc: | |
| raise GalleryError( | |
| f"Community gallery storage is unavailable: {type(exc).__name__}." | |
| ) from exc | |
| def publish( | |
| self, | |
| *, | |
| item_id: str, | |
| job_dir: Path, | |
| result: dict[str, Any], | |
| created_at: float | None = None, | |
| elapsed_started_at: float | None = None, | |
| staging_token: str | None = None, | |
| ) -> dict[str, Any]: | |
| """Atomically copy one successful job into permanent gallery storage. | |
| The item id is normally the job id. Publishing the same completed item | |
| twice is idempotent; a corrupt pre-existing destination is refused. | |
| """ | |
| if not valid_item_id(item_id): | |
| raise GalleryError("Gallery item id must be 32 lowercase hex characters.") | |
| self.ensure_ready() | |
| root = self.root.resolve() | |
| source_root = Path(job_dir).resolve() | |
| if not source_root.is_dir(): | |
| raise GalleryError("Completed job artifacts are unavailable.") | |
| epoch = time.time() if created_at is None else _safe_float(created_at, time.time()) | |
| metadata = self._metadata_from_result( | |
| item_id=item_id, | |
| result=result, | |
| created_at=epoch, | |
| ) | |
| final_dir = root / item_id | |
| if staging_token is None: | |
| staging_token = uuid.uuid4().hex | |
| if not re.fullmatch(r"[0-9a-f]{32}", staging_token): | |
| raise GalleryError("Gallery staging token must be 32 lowercase hex characters.") | |
| staging = root / f".publishing-{item_id}-{staging_token}" | |
| with self._lock: | |
| existing = self.get(item_id) | |
| if existing is not None: | |
| return existing | |
| if final_dir.exists(): | |
| raise GalleryError("A corrupt gallery item already uses this id.") | |
| copied_names: list[str] = [] | |
| committed = False | |
| try: | |
| staging.mkdir(mode=0o750) | |
| for name in GALLERY_ARTIFACT_NAMES: | |
| source = (source_root / name).resolve() | |
| if not source.is_file() or source.parent != source_root: | |
| continue | |
| destination = staging / name | |
| with source.open("rb") as reader, destination.open("xb") as writer: | |
| shutil.copyfileobj(reader, writer, length=1024 * 1024) | |
| writer.flush() | |
| try: | |
| os.fsync(writer.fileno()) | |
| except OSError: | |
| # Some mounted object stores do not implement fsync. | |
| pass | |
| copied_names.append(name) | |
| missing = REQUIRED_GALLERY_ARTIFACTS.difference(copied_names) | |
| if missing: | |
| raise GalleryError( | |
| "Completed result is missing required gallery artifacts: " | |
| + ", ".join(sorted(missing)) | |
| ) | |
| metadata["artifactNames"] = copied_names | |
| if elapsed_started_at is not None: | |
| elapsed_origin = _safe_float( | |
| elapsed_started_at, default=time.time() | |
| ) | |
| metadata["elapsedSeconds"] = round( | |
| max(0.0, time.time() - elapsed_origin), | |
| 1, | |
| ) | |
| # Build the response before the filesystem commit. Once the | |
| # rename succeeds, no fallible transformation remains that | |
| # could make the pipeline report an error for a visible item. | |
| public_item = self._public_item(metadata) | |
| metadata_path = staging / METADATA_NAME | |
| with metadata_path.open("x", encoding="utf-8") as writer: | |
| json.dump(metadata, writer, ensure_ascii=False, separators=(",", ":")) | |
| writer.write("\n") | |
| writer.flush() | |
| try: | |
| os.fsync(writer.fileno()) | |
| except OSError: | |
| pass | |
| # The rename is the commit point. A reader sees either no item | |
| # directory or the complete immutable directory. | |
| staging.rename(final_dir) | |
| committed = True | |
| return public_item | |
| except GalleryError: | |
| raise | |
| except OSError as exc: | |
| # A second process may have won an extremely unlikely | |
| # same-id race. Treat a valid committed item as idempotent. | |
| existing = self.get(item_id) | |
| if existing is not None: | |
| return existing | |
| raise GalleryError( | |
| f"Could not publish the community gallery item: {type(exc).__name__}." | |
| ) from exc | |
| finally: | |
| if not committed and staging.exists(): | |
| shutil.rmtree(staging, ignore_errors=True) | |
| def list_items(self, *, offset: int = 0, limit: int = 24) -> dict[str, Any]: | |
| """Return validated items newest first with offset pagination.""" | |
| self.ensure_ready() | |
| offset = max(0, int(offset)) | |
| limit = max(1, min(100, int(limit))) | |
| items: list[dict[str, Any]] = [] | |
| try: | |
| children = list(self.root.iterdir()) | |
| except OSError as exc: | |
| raise GalleryError( | |
| f"Could not read community gallery storage: {type(exc).__name__}." | |
| ) from exc | |
| for child in children: | |
| if not child.is_dir() or not valid_item_id(child.name): | |
| continue | |
| item = self.get(child.name) | |
| if item is not None: | |
| items.append(item) | |
| items.sort(key=lambda item: (item["createdAt"], item["id"]), reverse=True) | |
| total = len(items) | |
| page = items[offset:offset + limit] | |
| return { | |
| "items": page, | |
| "offset": offset, | |
| "limit": limit, | |
| "total": total, | |
| "hasMore": offset + len(page) < total, | |
| } | |
| def get(self, item_id: str) -> dict[str, Any] | None: | |
| """Read and validate one complete item directly from persistent disk.""" | |
| if not valid_item_id(item_id): | |
| return None | |
| try: | |
| root = self.root.resolve() | |
| item_dir = (root / item_id).resolve() | |
| if item_dir.parent != root or not item_dir.is_dir(): | |
| return None | |
| metadata_path = (item_dir / METADATA_NAME).resolve() | |
| if ( | |
| metadata_path.parent != item_dir | |
| or not metadata_path.is_file() | |
| or metadata_path.stat().st_size > MAX_METADATA_BYTES | |
| ): | |
| return None | |
| raw = json.loads(metadata_path.read_text(encoding="utf-8")) | |
| except (OSError, UnicodeError, json.JSONDecodeError, TypeError, ValueError): | |
| return None | |
| metadata = self._validated_metadata(raw, expected_id=item_id, item_dir=item_dir) | |
| return self._public_item(metadata) if metadata is not None else None | |
| def artifact_path(self, item_id: str, name: str) -> Path | None: | |
| """Resolve one allowlisted artifact with metadata and path containment.""" | |
| if name not in GALLERY_ARTIFACT_NAMES: | |
| return None | |
| item = self.get(item_id) | |
| if item is None or name not in item["artifacts"]: | |
| return None | |
| root = self.root.resolve() | |
| item_dir = (root / item_id).resolve() | |
| path = (item_dir / name).resolve() | |
| if path.parent != item_dir or not path.is_file(): | |
| return None | |
| return path | |
| def _metadata_from_result( | |
| *, | |
| item_id: str, | |
| result: dict[str, Any], | |
| created_at: float, | |
| ) -> dict[str, Any]: | |
| target_name = result.get("targetName") | |
| if not isinstance(target_name, str) or not target_name.strip(): | |
| target_name = "Object" | |
| return { | |
| "schemaVersion": METADATA_SCHEMA_VERSION, | |
| "id": item_id, | |
| "jobId": item_id, | |
| "createdAtEpoch": created_at, | |
| "targetName": target_name.strip()[:80], | |
| "components": _safe_int(result.get("components")), | |
| "materials": _safe_int(result.get("materials")), | |
| "elapsedSeconds": _safe_float(result.get("elapsedSeconds")), | |
| "generationMode": str(result.get("generationMode") or "unknown")[:80], | |
| "generatedPass": str(result.get("generatedPass") or "unknown")[:80], | |
| "reviewStatus": str(result.get("reviewStatus") or "unreviewed")[:80], | |
| "artifactNames": [], | |
| } | |
| def _validated_metadata( | |
| raw: Any, | |
| *, | |
| expected_id: str, | |
| item_dir: Path, | |
| ) -> dict[str, Any] | None: | |
| if not isinstance(raw, dict): | |
| return None | |
| if raw.get("schemaVersion") != METADATA_SCHEMA_VERSION: | |
| return None | |
| if raw.get("id") != expected_id or raw.get("jobId") != expected_id: | |
| return None | |
| target_name = raw.get("targetName") | |
| if not isinstance(target_name, str) or not target_name.strip() or len(target_name) > 80: | |
| return None | |
| epoch = raw.get("createdAtEpoch") | |
| if isinstance(epoch, bool) or not isinstance(epoch, (int, float)): | |
| return None | |
| epoch = float(epoch) | |
| if not math.isfinite(epoch) or epoch < 0: | |
| return None | |
| names = raw.get("artifactNames") | |
| if not isinstance(names, list) or any(not isinstance(name, str) for name in names): | |
| return None | |
| artifact_names: list[str] = [] | |
| for name in GALLERY_ARTIFACT_NAMES: | |
| if name not in names: | |
| continue | |
| path = (item_dir / name).resolve() | |
| if path.parent != item_dir or not path.is_file(): | |
| return None | |
| artifact_names.append(name) | |
| if not REQUIRED_GALLERY_ARTIFACTS.issubset(artifact_names): | |
| return None | |
| return { | |
| "schemaVersion": METADATA_SCHEMA_VERSION, | |
| "id": expected_id, | |
| "jobId": expected_id, | |
| "createdAtEpoch": epoch, | |
| "targetName": target_name.strip(), | |
| "components": _safe_int(raw.get("components")), | |
| "materials": _safe_int(raw.get("materials")), | |
| "elapsedSeconds": _safe_float(raw.get("elapsedSeconds")), | |
| "generationMode": str(raw.get("generationMode") or "unknown")[:80], | |
| "generatedPass": str(raw.get("generatedPass") or "unknown")[:80], | |
| "reviewStatus": str(raw.get("reviewStatus") or "unreviewed")[:80], | |
| "artifactNames": artifact_names, | |
| } | |
| def _public_item(metadata: dict[str, Any]) -> dict[str, Any]: | |
| item_id = metadata["id"] | |
| artifacts = { | |
| name: f"/api/gallery/{item_id}/artifacts/{name}" | |
| for name in GALLERY_ARTIFACT_NAMES | |
| if name in metadata["artifactNames"] | |
| } | |
| return { | |
| "id": item_id, | |
| "jobId": metadata["jobId"], | |
| "targetName": metadata["targetName"], | |
| "createdAt": _iso_utc(metadata["createdAtEpoch"]), | |
| "thumbnailUrl": artifacts["reference.png"], | |
| "detailUrl": f"/api/gallery/{item_id}", | |
| "artifacts": artifacts, | |
| "stats": { | |
| "components": metadata["components"], | |
| "materials": metadata["materials"], | |
| "elapsedSeconds": metadata["elapsedSeconds"], | |
| }, | |
| "generation": { | |
| "mode": metadata["generationMode"], | |
| "generatedPass": metadata["generatedPass"], | |
| "reviewStatus": metadata["reviewStatus"], | |
| }, | |
| } | |