feat(fc2a): add Thumbnailer service

Generates 400px-max thumbnails: PNG with alpha preserved when source has transparency, JPEG (q=85, progressive) otherwise. Storage layout is /images/thumbs/{sha[:3]}/{sha}.{jpg|png} so each prefix bucket holds at most ~4k files (16^3 buckets across all possible SHAs). Video path uses ffmpeg's first-frame snapshot at ~5% offset; we have a 1-second floor for short clips. ffmpeg call has a 60s timeout. Video tests are deferred to FC-2e's integration suite where transcoding is also wired up. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-14 12:05:19 -04:00
parent f6c7117231
commit 1a2a120da1
2 changed files with 168 additions and 0 deletions
@@ -0,0 +1,102 @@
+"""Image + video thumbnail generation.
+
+Images go through Pillow; videos go through ffmpeg's first-frame snapshot
+at ~5% into the runtime (avoids fade-ins/title cards while still being a
+representative frame).
+
+Thumbnails are stored at /images/thumbs/{sha[:3]}/{sha}.jpg or .png; the
+3-char prefix bucket avoids hot single-directory inode pressure.
+"""
+
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+
+from PIL import Image
+
+THUMB_MAX_PX = 400
+JPEG_QUALITY = 85
+FFMPEG_TIMEOUT_SECONDS = 60
+
+
+@dataclass(frozen=True)
+class ThumbnailResult:
+    path: Path
+    width: int
+    height: int
+    mime: str
+
+
+def thumbnail_path_for(sha256_hex: str, *, alpha: bool, root: Path) -> Path:
+    ext = ".png" if alpha else ".jpg"
+    bucket = sha256_hex[:3]
+    return root / "thumbs" / bucket / f"{sha256_hex}{ext}"
+
+
+class Thumbnailer:
+    def __init__(self, images_root: Path):
+        self.images_root = images_root
+
+    def generate_image_thumbnail(
+        self, source: Path, sha256_hex: str
+    ) -> ThumbnailResult:
+        """Reads source, fits inside THUMB_MAX_PX × THUMB_MAX_PX preserving
+        aspect ratio. Writes PNG if source has alpha, else JPEG.
+        """
+        with Image.open(source) as im:
+            im.load()
+            has_alpha = im.mode in ("RGBA", "LA") or (
+                im.mode == "P" and "transparency" in im.info
+            )
+            if has_alpha:
+                im = im.convert("RGBA")
+            else:
+                im = im.convert("RGB")
+            im.thumbnail((THUMB_MAX_PX, THUMB_MAX_PX), Image.Resampling.LANCZOS)
+
+            dest = thumbnail_path_for(sha256_hex, alpha=has_alpha, root=self.images_root)
+            dest.parent.mkdir(parents=True, exist_ok=True)
+            if has_alpha:
+                im.save(dest, "PNG", optimize=True)
+                mime = "image/png"
+            else:
+                im.save(dest, "JPEG", quality=JPEG_QUALITY, optimize=True, progressive=True)
+                mime = "image/jpeg"
+            return ThumbnailResult(path=dest, width=im.width, height=im.height, mime=mime)
+
+    def generate_video_thumbnail(
+        self, source: Path, sha256_hex: str, duration_seconds: float | None = None
+    ) -> ThumbnailResult:
+        """Snapshots a frame ~5% into the video via ffmpeg, then runs that
+        frame through the image-thumbnail pipeline.
+
+        duration_seconds is optional; if absent, we snapshot at the 1-second
+        mark which is a reasonable fallback for short clips.
+        """
+        offset = max(1.0, (duration_seconds or 20.0) * 0.05)
+        tmp = self.images_root / "thumbs" / "_tmp" / f"{sha256_hex}.png"
+        tmp.parent.mkdir(parents=True, exist_ok=True)
+        try:
+            cmd = [
+                "ffmpeg",
+                "-ss",
+                f"{offset:.2f}",
+                "-i",
+                str(source),
+                "-frames:v",
+                "1",
+                "-vf",
+                f"scale='if(gt(iw,ih),{THUMB_MAX_PX},-2)':'if(gt(iw,ih),-2,{THUMB_MAX_PX})'",
+                "-y",
+                str(tmp),
+            ]
+            subprocess.run(
+                cmd,
+                check=True,
+                capture_output=True,
+                timeout=FFMPEG_TIMEOUT_SECONDS,
+            )
+            return self.generate_image_thumbnail(tmp, sha256_hex)
+        finally:
+            if tmp.exists():
+                tmp.unlink()