xiaoxia-saas/apps/worker/video_processing/thumbnail_generator.py

"""视频封面抽帧工具 — 从已渲染视频中抽取帧作为封面。

统一封面管道：视频渲染时标题已通过 ASS 字幕烧进视频，
渲染完成后直接从此视频抽帧，封面天然带标题，无需额外叠加逻辑。
"""

from __future__ import annotations

import logging
import tempfile
from pathlib import Path

logger = logging.getLogger(__name__)


def extract_first_frame(
    video_path: str,
    output_path: str | None = None,
    *,
    width: int = -1,
    height: int = -1,
    timeout: int = 30,
    seek_ratio: float = 0.15,
    min_seek_seconds: float = 1.0,
) -> str:
    """抽取视频封面帧（默认取视频时长 15% 处的帧，避开片头纯色画面）。

    因为视频渲染时标题已通过 ASS 字幕烧录，抽取的帧天然带标题。

    Args:
        video_path: 视频文件路径
        output_path: 输出图片路径，不传则用临时文件
        width: 输出宽度（默认 -1，保持原始分辨率）
        height: 输出高度（默认 -1，保持原始分辨率）
        timeout: 超时时间（秒）
        seek_ratio: 抽帧位置占视频时长的比例（默认 0.15，即 15% 处）
        min_seek_seconds: 最小抽帧时间（秒），避免极短视频 seek 到 0

    Returns:
        生成的封面帧文件路径

    Raises:
        RuntimeError: ffmpeg 执行失败或输出文件为空
    """
    from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg

    _is_temp_output = False
    if output_path is None:
        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
        tmp.close()
        output_path = tmp.name
        _is_temp_output = True

    try:
        # 计算抽帧时间点：取视频时长 * seek_ratio，最少 min_seek_seconds 秒
        try:
            duration = probe_duration(video_path)
            seek_time = max(min_seek_seconds, duration * seek_ratio)
        except Exception:
            # probe 失败时 fallback 到第1秒
            seek_time = min_seek_seconds

        # 格式化为 HH:MM:SS.xx
        seek_str = _format_seek_time(seek_time)

        # 构建 scale filter：如果指定了宽高则缩放，否则保持原始分辨率。
        # NOTE: scale_filter 在此处通过 if/else 分支赋值，之后不再被覆盖，
        #       后续 cmd / cmd2 均复用同一变量，逻辑无变化。
        if width > 0 or height > 0:
            w_str = str(width) if width > 0 else "-1"
            h_str = str(height) if height > 0 else "-1"
            scale_filter = f"scale={w_str}:{h_str}:force_original_aspect_ratio=decrease,format=yuvj420p"
        else:
            # 保持原始分辨率，只确保格式兼容
            scale_filter = "format=yuvj420p"

        # -ss 放在 -i 前面（input seeking，更快）
        # -vframes 1 只取一帧
        # -q:v 2 jpeg 高质量
        cmd = [
            FFMPEG_BIN,
            "-y",
            "-ss",
            seek_str,
            "-i",
            video_path,
            "-vframes",
            "1",
            "-vf",
            scale_filter,
            "-q:v",
            "2",
            output_path,
        ]

        try:
            run_ffmpeg(cmd, capture_output=True, timeout=timeout)
        except Exception:
            # 失败时退回到第0帧兜底
            cmd2 = [
                FFMPEG_BIN,
                "-y",
                "-i",
                video_path,
                "-ss",
                "00:00:00",
                "-vframes",
                "1",
                "-vf",
                scale_filter,
                "-q:v",
                "2",
                output_path,
            ]
            run_ffmpeg(cmd2, capture_output=True, timeout=timeout)

        if not Path(output_path).exists() or Path(output_path).stat().st_size == 0:
            raise RuntimeError(f"Cover frame extraction failed: {output_path}")

        return output_path
    except Exception:
        # 失败时清理自己创建的临时文件
        if _is_temp_output and output_path:
            try:
                Path(output_path).unlink(missing_ok=True)
            except Exception:
                pass
        raise


def _format_seek_time(seconds: float) -> str:
    """将秒数格式化为 HH:MM:SS.xx 格式。"""
    h = int(seconds // 3600)
    m = int((seconds % 3600) // 60)
    s = seconds % 60
    return f"{h:02d}:{m:02d}:{s:05.2f}"


def generate_and_upload_thumbnail(
    video_path: str,
    storage_key: str,
    *,
    seek_ratio: float = 0.15,
) -> str:
    """从视频中提取一帧缩略图并上传到 OSS。

    Args:
        video_path: 视频文件路径
        storage_key: OSS 存储 key
        seek_ratio: 抽帧位置比例（默认 0.15）

    Returns:
        上传后的 URL 字符串

    Raises:
        RuntimeError: 抽帧或上传失败
    """
    from video_processing.oss_helpers import upload_to_oss

    tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
    tmp.close()
    try:
        frame_path = extract_first_frame(video_path, output_path=tmp.name, seek_ratio=seek_ratio)
        url = upload_to_oss(frame_path, storage_key)
        if not url:
            raise RuntimeError(f"上传缩略图到 OSS 失败: {storage_key}")
        return url
    finally:
        Path(tmp.name).unlink(missing_ok=True)


def extract_and_upload_cover_frames(
    video_path: str,
    plan_id: str,
    *,
    num_frames: int = 3,
    title_text: str = "",
) -> list[dict]:
    """从视频中抽取多帧作为封面候选，上传到 OSS。

    Args:
        video_path: 视频文件路径
        plan_id: 编辑计划 ID（用于生成 storage key）
        num_frames: 抽取帧数（默认 3）
        title_text: 标题文字（当前版本未叠加，预留参数）

    Returns:
        封面候选列表，每项包含 {"url": str, "position": float}
    """
    from video_processing.ffmpeg_utils import probe_duration
    from video_processing.oss_helpers import upload_to_oss

    try:
        duration = probe_duration(video_path)
    except Exception:
        duration = 0.0

    candidates: list[dict] = []
    # 均匀分布抽帧点：从 10% 到 90%
    for i in range(num_frames):
        ratio = 0.1 + 0.8 * i / max(num_frames - 1, 1)
        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
        tmp.close()
        try:
            frame_path = extract_first_frame(
                video_path,
                output_path=tmp.name,
                seek_ratio=ratio,
                min_seek_seconds=0.5,
            )
            storage_key = f"covers/{plan_id}/frame_{i}.jpg"
            url = upload_to_oss(frame_path, storage_key)
            if url:
                seek_time = max(0.5, duration * ratio) if duration > 0 else 0.0
                candidates.append({"url": url, "position": round(seek_time, 2)})
        except Exception as e:
            logger.warning("[thumbnail] 封面候选帧 %d 提取失败: %s", i, e)
        finally:
            Path(tmp.name).unlink(missing_ok=True)

    return candidates