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

"""视频输出校验 — 渲染后验证输出文件完整性.

在 FFmpeg 渲染完成后，验证输出文件是否有效（非损坏/截断）：
  1. 文件存在且大小 > 0
  2. moov atom 存在（MP4 容器完整性标志）
  3. ffprobe 可正常读取视频流信息

用途：
  - RenderAdapter 渲染后调用，避免将损坏文件上传到 OSS
  - 提前发现 FFmpeg 异常退出但未抛异常的情况（如 exit=0 但输出截断）
"""

from __future__ import annotations

import logging
import subprocess
from dataclasses import dataclass
from pathlib import Path

from shared.ffmpeg_utils import FFPROBE_BIN

logger = logging.getLogger(__name__)


# ── 数据结构 ──────────────────────────────────────────────────────────────────


@dataclass
class VideoValidationResult:
    """视频校验结果。"""

    valid: bool
    file_exists: bool = False
    file_size: int = 0
    moov_atom_found: bool = False
    has_video_stream: bool = False
    duration: float = 0.0
    width: int = 0
    height: int = 0
    error_message: str = ""

    @property
    def is_valid(self) -> bool:
        """是否通过所有校验。"""
        return self.valid


# ── FFmpeg 退出码映射 ──────────────────────────────────────────────────────────

# 常见 FFmpeg 退出码及其含义
FFMPEG_EXIT_CODES: dict[int, tuple[str, str]] = {
    0: ("成功", "渲染正常完成"),
    1: ("通用错误", "FFmpeg 执行出错，请检查输入参数和素材"),
    69: ("权限错误", "无权限写入输出文件或访问输入文件"),
    126: ("权限不足", "命令不可执行"),
    127: ("命令不存在", "FFmpeg 二进制文件未找到"),
    134: ("Abnormal termination", "FFmpeg 异常终止（可能内存不足）"),
    137: ("OOM Killed", "FFmpeg 被系统 OOM Killer 终止（内存不足）"),
    139: ("段错误", "FFmpeg Segmentation Fault（可能是编解码器 Bug）"),
    141: ("管道断裂", "FFmpeg 输出管道断裂"),
    143: ("SIGTERM", "FFmpeg 收到终止信号"),
    183: ("滤镜错误", "FFmpeg 滤镜链配置错误"),
    234: ("素材异常", "输入素材格式不兼容或已损坏"),
    255: ("严重错误", "FFmpeg 执行严重错误"),
}


def get_exit_code_message(exit_code: int) -> str:
    """获取 FFmpeg 退出码的中文描述。

    Args:
        exit_code: FFmpeg 进程退出码

    Returns:
        人类可读的错误描述
    """
    if exit_code in FFMPEG_EXIT_CODES:
        name, desc = FFMPEG_EXIT_CODES[exit_code]
        return f"exit={exit_code} ({name}): {desc}"
    if exit_code > 128:
        signal_num = exit_code - 128
        return f"exit={exit_code}: 被信号 {signal_num} 终止"
    return f"exit={exit_code}: 未知错误"


# ── 校验函数 ──────────────────────────────────────────────────────────────────


def validate_video_output(video_path: str | Path, *, min_duration: float = 0.1) -> VideoValidationResult:
    """校验渲染输出视频文件的完整性。

    校验项：
      1. 文件存在且大小 > 0
      2. 包含 moov atom（MP4 容器完整性）
      3. ffprobe 可读取至少一个视频流

    Args:
        video_path: 输出视频文件路径
        min_duration: 最小有效时长（秒），低于此值视为无效，默认 0.1s

    Returns:
        VideoValidationResult
    """
    path = Path(video_path)
    result = VideoValidationResult(valid=False)

    # 1. 文件存在性检查
    if not path.exists():
        result.error_message = f"输出文件不存在: {path}"
        logger.error("[video-validation] %s", result.error_message)
        return result

    result.file_exists = True

    # 2. 文件大小检查
    try:
        result.file_size = path.stat().st_size
    except OSError as e:
        result.error_message = f"无法读取文件大小: {e}"
        logger.error("[video-validation] %s", result.error_message)
        return result

    if result.file_size == 0:
        result.error_message = "输出文件大小为 0（FFmpeg 未写入任何数据）"
        logger.error("[video-validation] %s", result.error_message)
        return result

    # 极小文件（< 1KB）几乎不可能是有效视频
    if result.file_size < 1024:
        result.error_message = f"输出文件过小 ({result.file_size} bytes)，可能渲染未完成"
        logger.error("[video-validation] %s", result.error_message)
        return result

    # 3. moov atom 检查（MP4 容器完整性标志）
    result.moov_atom_found = _check_moov_atom(path)
    if not result.moov_atom_found:
        result.error_message = (
            "输出文件缺少 moov atom（MP4 容器不完整）。" "可能原因：FFmpeg 被强制终止、磁盘空间不足、或渲染过程中断。"
        )
        logger.error("[video-validation] %s — file_size=%d", result.error_message, result.file_size)
        return result

    # 4. ffprobe 验证视频流
    try:
        probe_result = subprocess.run(  # nosec B603
            [
                FFPROBE_BIN,
                "-v",
                "error",
                "-select_streams",
                "v:0",
                "-show_entries",
                "stream=width,height,duration,codec_name",
                "-show_entries",
                "format=duration",
                "-of",
                "json",
                str(path),
            ],
            check=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            timeout=15,
        )

        import json

        probe_data = json.loads(probe_result.stdout)
        streams = probe_data.get("streams", [])
        fmt = probe_data.get("format", {})

        if not streams:
            result.error_message = "输出文件无视频流（ffprobe 未检测到 video stream）"
            logger.error("[video-validation] %s", result.error_message)
            return result

        result.has_video_stream = True

        video_stream = streams[0]
        result.width = int(video_stream.get("width", 0) or 0)
        result.height = int(video_stream.get("height", 0) or 0)
        result.duration = float(fmt.get("duration", 0) or video_stream.get("duration", 0) or 0)

        # 时长校验
        if result.duration > 0 and result.duration < min_duration:
            result.error_message = (
                f"输出视频时长过短 ({result.duration:.2f}s < {min_duration}s)，" "可能渲染只处理了极少帧"
            )
            logger.warning("[video-validation] %s", result.error_message)
            # 不标记为失败，只是警告（某些预览场景确实很短）
            # 但如果时长为 0 且文件大小较大，说明 moov 有问题
        elif result.duration == 0 and result.file_size > 0:
            logger.warning(
                "[video-validation] ffprobe 无法读取时长，但文件存在且大小=%d，标记为可疑",
                result.file_size,
            )

    except subprocess.TimeoutExpired:
        result.error_message = "ffprobe 超时（15s），输出文件可能已损坏"
        logger.error("[video-validation] %s", result.error_message)
        return result
    except subprocess.CalledProcessError as e:
        stderr_text = (e.stderr or "").strip()
        result.error_message = f"ffprobe 校验失败: exit={e.returncode}, stderr={stderr_text[:200]}"
        logger.error("[video-validation] %s", result.error_message)
        return result
    except (json.JSONDecodeError, KeyError, ValueError) as e:
        result.error_message = f"ffprobe 输出解析失败: {e}"
        logger.error("[video-validation] %s", result.error_message)
        return result

    # 全部通过
    result.valid = True
    logger.info(
        "[video-validation] 校验通过: path=%s size=%d duration=%.2fs resolution=%dx%d",
        path,
        result.file_size,
        result.duration,
        result.width,
        result.height,
    )
    return result


def _check_moov_atom(video_path: Path) -> bool:
    """检查 MP4 文件是否包含 moov atom。

    moov atom 是 MP4 容器的元数据容器，包含视频时长、编解码器信息等。
    FFmpeg 使用 -movflags +faststart 时 moov 在文件头部；否则在尾部。
    如果 FFmpeg 被强制终止，moov 可能完全不存在。

    方法：读取文件前 64KB + 尾部 64KB，搜索 "moov" 字节标记。
    对于 -movflags +faststart 的快启文件，moov 在头部。

    Args:
        video_path: 视频文件路径

    Returns:
        True 表示找到 moov atom
    """
    try:
        file_size = video_path.stat().st_size
        if file_size < 8:
            return False

        # 搜索范围：头部 64KB + 尾部 64KB（覆盖 faststart 和普通 MP4）
        search_size = min(64 * 1024, file_size)

        with open(video_path, "rb") as f:
            # 读取头部
            head_data = f.read(search_size)
            if b"moov" in head_data:
                return True

            # 读取尾部
            if file_size > search_size:
                f.seek(file_size - search_size)
                tail_data = f.read(search_size)
                if b"moov" in tail_data:
                    return True

        return False

    except OSError as e:
        logger.warning("[video-validation] 检查 moov atom 失败: %s", e)
        return False