Deploy relay source

src/small_cuts/hf_relay.py

@@ -0,0 +1,245 @@
+"""Hugging Face bucket relay for finished Small Cuts scenes.
+
+The Space uses this as a read-only scene source. The private engine or a local
+publisher writes finished scene manifests + media into an HF bucket; the Space
+downloads those files into a temp cache and serves them through Gradio.
+"""
+
+from __future__ import annotations
+
+import copy
+import json
+import tempfile
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Protocol
+from urllib.parse import quote, urlparse
+
+import httpx
+
+RELAY_BUCKET_ENV = "SMALL_CUTS_RELAY_BUCKET"
+RELAY_PREFIX_ENV = "SMALL_CUTS_RELAY_PREFIX"
+DEFAULT_RELAY_PREFIX = "relay"
+RELAY_MANIFEST = "manifest.json"
+RELAY_CACHE_DIR = Path(tempfile.gettempdir()) / "small-cuts-hf-relay"
+GRADIO_FILE_ROUTE = "/gradio_api/file="
+DEFAULT_SCENE_LIMIT = 60
+MEDIA_KEYS = ("frame_url", "card_url", "audio_url", "clip_url")
+PUBLISH_VISIBILITIES = frozenset({"shared", "public"})
+HTTP_TIMEOUT_S = 20.0
+
+
+class BucketFileSystem(Protocol):
+    def cat(self, path: str) -> bytes: ...
+
+
+class BucketRelayError(RuntimeError):
+    """Raised when the bucket relay cannot read or hydrate its manifest."""
+
+
+@dataclass(frozen=True)
+class RelaySnapshot:
+    path: Path
+    scene_count: int
+    manifest_path: Path
+
+
+def gradio_file_url(path: str | Path) -> str:
+    return f"{GRADIO_FILE_ROUTE}{quote(str(path))}"
+
+
+def _normalize_prefix(prefix: str) -> str:
+    return prefix.strip().strip("/")
+
+
+def _safe_bucket_slug(bucket_id: str) -> str:
+    return bucket_id.replace("/", "__")
+
+
+class BucketSceneClient:
+    """Read finished NarratedScene payloads from a Hugging Face bucket manifest."""
+
+    base_url = ""
+    readonly = True
+
+    def __init__(
+        self,
+        bucket_id: str,
+        *,
+        prefix: str = DEFAULT_RELAY_PREFIX,
+        fs: BucketFileSystem | None = None,
+        cache_dir: str | Path | None = None,
+        register_static_paths: Any | None = None,
+    ) -> None:
+        self.bucket_id = bucket_id.strip()
+        if not self.bucket_id:
+            raise ValueError("bucket_id is required")
+        self.prefix = _normalize_prefix(prefix)
+        self.root = f"hf://buckets/{self.bucket_id}"
+        if self.prefix:
+            self.root = f"{self.root}/{self.prefix}"
+        self._fs = fs
+        self.cache_dir = (
+            Path(cache_dir)
+            if cache_dir is not None
+            else (RELAY_CACHE_DIR / _safe_bucket_slug(self.bucket_id))
+        )
+        if cache_dir is None and self.prefix:
+            self.cache_dir = self.cache_dir / self.prefix
+        self.cache_dir.mkdir(parents=True, exist_ok=True)
+        if register_static_paths is not None:
+            register_static_paths([self.cache_dir])
+
+    @property
+    def fs(self) -> BucketFileSystem:
+        if self._fs is None:
+            from huggingface_hub import HfFileSystem
+
+            self._fs = HfFileSystem()
+        return self._fs
+
+    def list_scenes(self, limit: int = DEFAULT_SCENE_LIMIT) -> list[dict[str, Any]]:
+        try:
+            raw = self.fs.cat(f"{self.root}/{RELAY_MANIFEST}")
+            manifest = json.loads(raw.decode("utf-8"))
+            scenes = manifest.get("scenes", [])
+            if not isinstance(scenes, list):
+                raise ValueError("relay manifest scenes must be a list")
+            return [self._hydrate_scene(scene) for scene in scenes[-limit:]]
+        except FileNotFoundError:
+            return []
+        except (OSError, json.JSONDecodeError, KeyError, TypeError, ValueError) as exc:
+            raise BucketRelayError(f"could not read relay bucket {self.bucket_id}: {exc}") from exc
+
+    def media_url(self, path: str | None) -> str | None:
+        if not path:
+            return None
+        if path.startswith(("http://", "https://", "data:", GRADIO_FILE_ROUTE)):
+            return path
+        relative = self._relative_media_path(path)
+        target = self.cache_dir / relative
+        if not target.exists():
+            target.parent.mkdir(parents=True, exist_ok=True)
+            target.write_bytes(self.fs.cat(f"{self.root}/{relative.as_posix()}"))
+        return gradio_file_url(target)
+
+    def _hydrate_scene(self, scene: dict[str, Any]) -> dict[str, Any]:
+        hydrated = copy.deepcopy(scene)
+        media = hydrated.get("media")
+        if not isinstance(media, dict):
+            hydrated["media"] = {}
+            return hydrated
+        for key in MEDIA_KEYS:
+            media[key] = self.media_url(media.get(key))
+        return hydrated
+
+    def _relative_media_path(self, path: str) -> Path:
+        value = path.strip().lstrip("/")
+        if self.prefix and value.startswith(f"{self.prefix}/"):
+            value = value[len(self.prefix) + 1 :]
+        relative = Path(value)
+        if relative.is_absolute() or ".." in relative.parts:
+            raise ValueError(f"unsafe bucket media path: {path}")
+        return relative
+
+
+def prepare_relay_snapshot(
+    engine_url: str,
+    output_dir: str | Path,
+    *,
+    limit: int = DEFAULT_SCENE_LIMIT,
+    include_private: bool = False,
+    client: httpx.Client | None = None,
+) -> RelaySnapshot:
+    """Stage a bucket-ready manifest + media snapshot from the private engine."""
+    base_url = engine_url.rstrip("/")
+    output = Path(output_dir)
+    media_root = output / "media"
+    output.mkdir(parents=True, exist_ok=True)
+    media_root.mkdir(parents=True, exist_ok=True)
+    close_client = client is None
+    http = client or httpx.Client(timeout=HTTP_TIMEOUT_S)
+    try:
+        response = http.get(f"{base_url}/v1/scenes")
+        response.raise_for_status()
+        scenes = response.json().get("scenes", [])[-limit:]
+        published = [
+            _stage_scene_media(base_url, output, scene, http)
+            for scene in scenes
+            if _should_publish_scene(scene, include_private=include_private)
+        ]
+    finally:
+        if close_client:
+            http.close()
+    manifest = {
+        "contract_version": "1.1.0",
+        "published_at": datetime.now(timezone.utc).isoformat(),
+        "source_engine": base_url,
+        "scenes": published,
+    }
+    manifest_path = output / RELAY_MANIFEST
+    manifest_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n")
+    return RelaySnapshot(output, len(published), manifest_path)
+
+
+def _should_publish_scene(scene: dict[str, Any], *, include_private: bool) -> bool:
+    if include_private:
+        return True
+    return scene.get("visibility") in PUBLISH_VISIBILITIES
+
+
+def _stage_scene_media(
+    engine_url: str,
+    output_dir: Path,
+    scene: dict[str, Any],
+    client: httpx.Client,
+) -> dict[str, Any]:
+    staged = copy.deepcopy(scene)
+    media = staged.get("media")
+    if not isinstance(media, dict):
+        staged["media"] = {}
+        return staged
+    scene_dir = _safe_path_segment(str(staged.get("scene_id") or "scene"))
+    for key in MEDIA_KEYS:
+        media[key] = _download_media(engine_url, output_dir, scene_dir, media.get(key), client)
+    return staged
+
+
+def _download_media(
+    engine_url: str,
+    output_dir: Path,
+    scene_dir: str,
+    url: str | None,
+    client: httpx.Client,
+) -> str | None:
+    if not url:
+        return None
+    if url.startswith(("data:", GRADIO_FILE_ROUTE)):
+        return None
+    absolute = url if url.startswith(("http://", "https://")) else f"{engine_url}/{url.lstrip('/')}"
+    relative = _relay_media_path(url, scene_dir)
+    target = output_dir / relative
+    target.parent.mkdir(parents=True, exist_ok=True)
+    response = client.get(absolute)
+    response.raise_for_status()
+    target.write_bytes(response.content)
+    return relative.as_posix()
+
+
+def _relay_media_path(url: str, scene_dir: str) -> Path:
+    parsed = urlparse(url)
+    source_path = (parsed.path if parsed.scheme else url.split("?", 1)[0]).lstrip("/")
+    if source_path.startswith("media/"):
+        relative = Path(source_path)
+    else:
+        filename = _safe_path_segment(Path(source_path).name or "media.bin")
+        relative = Path("media") / scene_dir / filename
+    if relative.is_absolute() or ".." in relative.parts:
+        raise ValueError(f"unsafe relay media path: {url}")
+    return relative
+
+
+def _safe_path_segment(value: str) -> str:
+    cleaned = "".join(ch if ch.isalnum() or ch in ("-", "_", ".") else "-" for ch in value)
+    return cleaned.strip(".-") or "item"

src/small_cuts/observability.py

@@ -0,0 +1,74 @@
+"""Optional Sentry observability for demo/runtime failures.
+
+No Sentry traffic is sent unless ``SENTRY_DSN`` is configured. Payload scrubbing
+keeps frames, audio, cookies, and auth headers out of events.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+
+SENTRY_DSN_ENV = "SENTRY_DSN"
+SENTRY_ENV_ENV = "SENTRY_ENVIRONMENT"
+SENTRY_RELEASE_ENV = "SENTRY_RELEASE"
+SENSITIVE_HEADERS = {"authorization", "cookie", "x-api-key", "x-forwarded-for"}
+
+_INITIALIZED = False
+
+
+def init_sentry(dsn: str | None = None, *, sdk: Any | None = None) -> bool:
+    global _INITIALIZED
+    dsn = dsn if dsn is not None else os.environ.get(SENTRY_DSN_ENV, "").strip()
+    if not dsn:
+        return False
+    if _INITIALIZED:
+        return True
+    sdk = sdk or _import_sentry_sdk()
+    if sdk is None:
+        return False
+    sdk.init(
+        dsn=dsn,
+        environment=os.environ.get(SENTRY_ENV_ENV) or os.environ.get("SPACE_ID") or "local",
+        release=os.environ.get(SENTRY_RELEASE_ENV) or os.environ.get("SPACE_COMMIT_SHA"),
+        send_default_pii=False,
+        attach_stacktrace=True,
+        traces_sample_rate=0.0,
+        before_send=_scrub_event,
+    )
+    _INITIALIZED = True
+    return True
+
+
+def capture_exception(exc: BaseException, *, sdk: Any | None = None) -> None:
+    if not _INITIALIZED and not init_sentry(sdk=sdk):
+        return
+    sdk = sdk or _import_sentry_sdk()
+    if sdk is not None:
+        sdk.capture_exception(exc)
+
+
+def _import_sentry_sdk() -> Any | None:
+    try:
+        import sentry_sdk
+    except ImportError:
+        return None
+    return sentry_sdk
+
+
+def _scrub_event(event: dict[str, Any], _hint: dict[str, Any]) -> dict[str, Any]:
+    request = event.get("request")
+    if isinstance(request, dict):
+        request.pop("data", None)
+        request.pop("cookies", None)
+        headers = request.get("headers")
+        if isinstance(headers, dict):
+            request["headers"] = {
+                key: value for key, value in headers.items() if key.lower() not in SENSITIVE_HEADERS
+            }
+    return event
+
+
+def reset_for_tests() -> None:
+    global _INITIALIZED
+    _INITIALIZED = False

src/small_cuts/viewer.py

@@ -37,6 +37,16 @@ from PIL import Image
 from . import demo_seed
 from ._icons import ICON_CSS
 from .frames import pick_key_frame, sample_frames
+from .hf_relay import (
+    DEFAULT_RELAY_PREFIX,
+    RELAY_BUCKET_ENV,
+    RELAY_PREFIX_ENV,
+    BucketRelayError,
+)
+from .hf_relay import (
+    BucketSceneClient as _BucketSceneClient,
+)
+from .observability import capture_exception
 from .styles import DEFAULT_STYLE_KEY, STYLES
 from .title_card import derive_title
 from .tts import speak
@@ -496,7 +506,17 @@ class EngineClient:
         return path if path.startswith("http") else f"{self.base_url}{path}"
 
 
-def shelf_items(scenes: list[dict[str, Any]], client: EngineClient) -> list[tuple[str, str]]:
+class BucketSceneClient(_BucketSceneClient):
+    """Bucket relay client wired to Gradio's static file serving."""
+
+    def __init__(self, *args, **kwargs) -> None:
+        kwargs.setdefault("register_static_paths", gr.set_static_paths)
+        super().__init__(*args, **kwargs)
+
+
+def shelf_items(
+    scenes: list[dict[str, Any]], client: EngineClient | BucketSceneClient
+) -> list[tuple[str, str]]:
     """Gallery payload: POV frame thumbnails captioned with the generated scene title."""
     items = []
     for scene in scenes:
@@ -508,7 +528,7 @@ def shelf_items(scenes: list[dict[str, Any]], client: EngineClient) -> list[tupl
 
 
 def poll_engine(
-    client: EngineClient,
+    client: EngineClient | BucketSceneClient,
     scenes_prev: list[dict[str, Any]],
     pinned_id: str | None,
     playing_id: str | None,
@@ -523,7 +543,8 @@ def poll_engine(
     """
     try:
         scenes = client.list_scenes(limit=SHELF_LIMIT)
-    except (httpx.HTTPError, KeyError, ValueError) as exc:
+    except (httpx.HTTPError, BucketRelayError, KeyError, ValueError) as exc:
+        capture_exception(exc)
         kind = type(exc).__name__
         print(
             f"small_cuts.viewer: engine poll failed for {client.base_url}: {kind}: {exc!r}",
@@ -644,7 +665,8 @@ def _go_live_handler(
         speech = speak(narration)
         scene["duration"] = len(speech.audio) / speech.sample_rate if speech.sample_rate else None
         scene["audio_src"] = _write_voice(speech.audio, speech.sample_rate, scene["scene_id"])
-    except Exception:
+    except Exception as exc:
+        capture_exception(exc)
         scene["audio_src"] = None
     scenes = [*(scenes or []), scene][-SHELF_LIMIT:]
     payload = format_stage(scene)
@@ -994,7 +1016,14 @@ PLAYBACK_SYNC_JS = """
 def build_viewer_app() -> gr.Blocks:
     """The P1 viewer page. Mode is decided once, at build time, from the env."""
     engine_url = os.environ.get(ENGINE_URL_ENV, "").strip()
-    client = EngineClient(engine_url) if engine_url else None
+    relay_bucket = os.environ.get(RELAY_BUCKET_ENV, "").strip()
+    relay_prefix = os.environ.get(RELAY_PREFIX_ENV, DEFAULT_RELAY_PREFIX).strip()
+    if engine_url:
+        client: EngineClient | BucketSceneClient | None = EngineClient(engine_url)
+    elif relay_bucket:
+        client = BucketSceneClient(relay_bucket, prefix=relay_prefix or DEFAULT_RELAY_PREFIX)
+    else:
+        client = None
     seed = _seed_scenes() if client is None else []
 
     if client: