tl/scenes/thumbnails.py

import json
import logging
import math
import shutil
import subprocess
import threading
from pathlib import Path

from django.conf import settings

from .models import Project

logger = logging.getLogger(__name__)

THUMB_INTERVAL_SECONDS = 1.0
THUMB_MAX_HEIGHT = 360
THUMB_JPEG_QUALITY = 9
CLIP_SELECT_CHUNK_SIZE = 30


def _otio_frames(rational_time):
    return rational_time.get("value", 0) if rational_time else 0


def _clip_sample_frame(start, duration):
    end = start + duration
    first = math.ceil(start)
    last = math.ceil(end) - 1
    if last < first:
        return max(0, math.floor(start))
    midpoint = math.floor(start + (duration / 2))
    return max(0, min(max(midpoint, first), last))


def _clip_tiles(start, duration, fps):
    frames_per_tile = max(1, round(fps * THUMB_INTERVAL_SECONDS))
    tiles = []
    offset = 0
    while offset < duration:
        end = min(duration, offset + frames_per_tile)
        sample = _clip_sample_frame(start + offset, end - offset)
        tiles.append({"start": offset, "end": end, "frame": sample})
        offset = end
    return tiles


def _otio_clip_items(project):
    if not project.otio:
        return []

    with open(project.otio.path, encoding="utf-8") as fh:
        otio = json.load(fh)

    tracks = otio.get("tracks", {}).get("children", [])
    source_starts = [
        _otio_frames(item.get("source_range", {}).get("start_time"))
        for track in tracks
        for item in track.get("children", [])
        if str(item.get("OTIO_SCHEMA", "")).startswith("Clip")
    ]
    media_offset = min(source_starts) if source_starts else 0

    clips = []
    for track_index, track in enumerate(tracks):
        clip_index = 0
        for item in track.get("children", []):
            if not str(item.get("OTIO_SCHEMA", "")).startswith("Clip"):
                continue
            source_range = item.get("source_range", {})
            start = _otio_frames(source_range.get("start_time")) - media_offset
            duration = _otio_frames(source_range.get("duration"))
            clips.append({
                "id": f"t{track_index}-c{clip_index}",
                "tiles": _clip_tiles(start, duration, project.fps),
            })
            clip_index += 1
    return clips


def thumbnail_root(project_id):
    return Path(settings.MEDIA_ROOT) / "projects" / str(project_id) / "thumbs" / "v1"


def thumbnail_url_path(project_id, filename):
    media_url = "/" + settings.MEDIA_URL.strip("/")
    return f"{media_url}/projects/{project_id}/thumbs/v1/{filename}"


def _chunks(items, size):
    for i in range(0, len(items), size):
        yield i // size, items[i:i + size]


def enqueue_thumbnail_generation(project_id):
    thread = threading.Thread(
        target=generate_project_thumbnails,
        args=(project_id,),
        name=f"thumbs-project-{project_id}",
        daemon=True,
    )
    thread.start()


def generate_project_thumbnails(project_id):
    try:
        project = Project.objects.get(pk=project_id)
    except Project.DoesNotExist:
        return

    if not project.clip:
        Project.objects.filter(pk=project_id).update(
            thumbnail_status="failed",
            thumbnail_error="project has no clip",
        )
        return

    Project.objects.filter(pk=project_id).update(
        thumbnail_status="running",
        thumbnail_error="",
        thumbnail_manifest={},
    )

    out_dir = thumbnail_root(project_id)
    tmp_dir = out_dir.with_name("v1.tmp")
    try:
        if tmp_dir.exists():
            shutil.rmtree(tmp_dir)
        tmp_dir.mkdir(parents=True, exist_ok=True)

        input_path = Path(project.clip.path)
        pattern = tmp_dir / "frame_%06d.jpg"
        vf = (
            f"fps=1/{THUMB_INTERVAL_SECONDS},"
            f"scale=w=-2:h={THUMB_MAX_HEIGHT}:force_original_aspect_ratio=decrease"
        )
        scale = f"scale=w=-2:h={THUMB_MAX_HEIGHT}:force_original_aspect_ratio=decrease"
        cmd = [
            "ffmpeg",
            "-hide_banner",
            "-loglevel",
            "error",
            "-y",
            "-i",
            str(input_path),
            "-vf",
            vf,
            "-q:v",
            str(THUMB_JPEG_QUALITY),
            str(pattern),
        ]
        subprocess.run(cmd, check=True, capture_output=True, text=True)

        files = sorted(tmp_dir.glob("frame_*.jpg"))
        if not files:
            raise RuntimeError("ffmpeg produced no thumbnails")

        clip_items = _otio_clip_items(project)
        unique_clip_frames = sorted({
            max(0, tile["frame"])
            for clip in clip_items
            for tile in clip["tiles"]
        })
        frame_filenames = {}
        for chunk_index, frames in _chunks(unique_clip_frames, CLIP_SELECT_CHUNK_SIZE):
            select_expr = "+".join(f"eq(n\\,{frame})" for frame in frames)
            chunk_pattern = f"clipframe_{chunk_index:03d}_%06d.jpg"
            subprocess.run([
                "ffmpeg",
                "-hide_banner",
                "-loglevel",
                "error",
                "-y",
                "-i",
                str(input_path),
                "-vf",
                f"select='{select_expr}',{scale}",
                "-vsync",
                "0",
                "-q:v",
                str(THUMB_JPEG_QUALITY),
                str(tmp_dir / chunk_pattern),
            ], check=True, capture_output=True, text=True)
            for idx, frame in enumerate(frames, start=1):
                frame_filenames[frame] = f"clipframe_{chunk_index:03d}_{idx:06d}.jpg"

        clip_thumbnails = {}
        for clip in clip_items:
            tiles = []
            for tile in clip["tiles"]:
                if tile["frame"] not in frame_filenames:
                    continue
                tiles.append({
                    "start": tile["start"],
                    "end": tile["end"],
                    "url": thumbnail_url_path(project_id, frame_filenames[tile["frame"]]),
                })
            if tiles:
                clip_thumbnails[clip["id"]] = tiles

        manifest = {
            "version": 1,
            "interval": THUMB_INTERVAL_SECONDS,
            "max_height": THUMB_MAX_HEIGHT,
            "quality": THUMB_JPEG_QUALITY,
            "count": len(files),
            "template": thumbnail_url_path(project_id, "frame_%06d.jpg"),
            "clips": clip_thumbnails,
        }
        (tmp_dir / "manifest.json").write_text(json.dumps(manifest, indent=2), encoding="utf-8")

        if out_dir.exists():
            shutil.rmtree(out_dir)
        tmp_dir.rename(out_dir)

        Project.objects.filter(pk=project_id).update(
            thumbnail_status="ready",
            thumbnail_manifest=manifest,
            thumbnail_error="",
        )
    except Exception as exc:
        logger.exception("thumbnail generation failed for project %s", project_id)
        if tmp_dir.exists():
            shutil.rmtree(tmp_dir, ignore_errors=True)
        Project.objects.filter(pk=project_id).update(
            thumbnail_status="failed",
            thumbnail_error=str(exc),
            thumbnail_manifest={},
        )