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

"""
视频处理器模块
"""

import os
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import List

import ffmpeg


@dataclass
class VideoResult:
    """视频结果"""

    output_path: str
    thumbnail_path: str
    duration: float
    width: int
    height: int
    fps: float
    file_size: int


class VideoProcessor:
    """视频处理器"""

    def __init__(self, temp_dir: str = None):
        """
        初始化处理器

        Args:
            temp_dir: 临时目录路径，默认为系统临时目录
        """
        self.temp_dir = temp_dir or tempfile.gettempdir()

    def concatenate_videos(
        self,
        input_paths: List[str],
        output_path: str,
        resolution: tuple[int, int] = (1920, 1080),
        fps: int = 25,
    ) -> VideoResult:
        """
        合并多个视频

        Args:
            input_paths: 输入视频路径列表
            output_path: 输出视频路径
            resolution: 分辨率 (width, height)
            fps: 帧率

        Returns:
            VideoResult: 视频结果
        """
        if not input_paths:
            raise ValueError("input_paths cannot be empty")

        # 创建输出目录
        os.makedirs(os.path.dirname(output_path), exist_ok=True)

        try:
            # P2-6: Use NamedTemporaryFile for automatic cleanup
            concat_file_fd = tempfile.NamedTemporaryFile(
                mode='w', 
                suffix='.txt', 
                delete=True,
                dir=self.temp_dir
            )
            with concat_file_fd as concat_file:
                for path in input_paths:
                    # FFmpeg concat demuxer 要求绝对路径
                    concat_file.write(f"file '{os.path.abspath(path)}'\n")
                # Flush is automatic with context manager

            # 使用 FFmpeg 合并
            width, height = resolution
            (
                ffmpeg.input(concat_file.name, format="concat", safe=0)
                .output(
                    output_path,
                    vcodec="libx264",
                    acodec="aac",
                    s=f"{width}x{height}",
                    r=fps,
                    preset="medium",
                    crf=23,
                )
                .overwrite_output()
                .run(capture_stdout=True, capture_stderr=True)
            )

            # 获取视频信息
            probe = ffmpeg.probe(output_path)
            video_stream = next(s for s in probe["streams"] if s["codec_type"] == "video")

            duration = float(probe["format"]["duration"])
            width = int(video_stream["width"])
            height = int(video_stream["height"])

            # 生成帧率
            fps_str = video_stream.get("r_frame_rate", "25/1")
            fps_parts = fps_str.split("/")
            fps_value = float(fps_parts[0]) / float(fps_parts[1]) if len(fps_parts) == 2 else float(fps_parts[0])

            file_size = os.path.getsize(output_path)

            # 生成缩略图
            thumbnail_path = self.generate_thumbnail(output_path)

            return VideoResult(
                output_path=output_path,
                thumbnail_path=thumbnail_path,
                duration=duration,
                width=width,
                height=height,
                fps=fps_value,
                file_size=file_size,
            )

        except ffmpeg.Error as e:
            stderr = e.stderr.decode() if e.stderr else ""
            raise RuntimeError(f"FFmpeg error: {stderr}") from e

    def generate_thumbnail(
        self,
        video_path: str,
        timestamp: float = 1.0,
        output_path: str = None,
    ) -> str:
        """
        生成视频缩略图

        Args:
            video_path: 视频路径
            timestamp: 截图时间点
            output_path: 输出路径，默认在视频同目录下生成 thumbnail.jpg

        Returns:
            缩略图路径
        """
        if output_path is None:
            output_path = f"{os.path.splitext(video_path)[0]}_thumbnail.jpg"

        try:
            (
                ffmpeg.input(video_path, ss=timestamp)
                .output(output_path, vframes=1, format="image2", vcodec="mjpeg")
                .overwrite_output()
                .run(capture_stdout=True, capture_stderr=True)
            )

            return output_path

        except ffmpeg.Error as e:
            stderr = e.stderr.decode() if e.stderr else ""
            raise RuntimeError(f"FFmpeg thumbnail error: {stderr}") from e

    def get_video_info(self, video_path: str) -> dict:
        """
        获取视频信息

        Args:
            video_path: 视频路径

        Returns:
            视频信息字典
        """
        try:
            probe = ffmpeg.probe(video_path)
            video_info = next(s for s in probe["streams"] if s["codec_type"] == "video")

            duration = float(probe["format"]["duration"])
            width = int(video_info["width"])
            height = int(video_info["height"])

            fps_str = video_info.get("r_frame_rate", "25/1")
            fps_parts = fps_str.split("/")
            fps_value = float(fps_parts[0]) / float(fps_parts[1]) if len(fps_parts) == 2 else float(fps_parts[0])

            return {
                "duration": duration,
                "width": width,
                "height": height,
                "fps": fps_value,
                "codec": video_info.get("codec_name"),
                "bitrate": int(probe["format"].get("bit_rate", 0)),
            }

        except ffmpeg.Error as e:
            stderr = e.stderr.decode() if e.stderr else ""
            raise RuntimeError(f"FFmpeg probe error: {stderr}") from e