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

"""智能降重微变换纯逻辑模块 — #1970 PR2.

所有函数均为纯函数：不调用 FFmpeg、不读写文件，只负责按可复现种子
生成每个片段 / 整片的微变换参数与 filter_complex 片段。

6 个维度：
1. hflip 水平翻转（每片段 50%，有字幕/文字的片段不翻转）
2. 播放速度 0.97~1.03x（视频 setpts + 音频 atempo）
3. 亮度 ±2%（eq=brightness）
4. 对比度 ±2%（eq=contrast）
5. 饱和度 ±2%（eq=saturation）
6. BGM 起始偏移 2~8 秒（音频 atrim 起点）

随机种子 = hash(task_id + video_index) % 10000，保证同一任务同一视频
可复现；dedup_enabled=False 时不生成本模块任何输出。
"""

from __future__ import annotations

import random
from dataclasses import dataclass, field

# ── 常量（与需求文档 §2 对齐）──────────────────────────────────────────────────

SPEED_MIN = 0.97
SPEED_MAX = 1.03
COLOR_DELTA = 0.02
HFLIP_PROBABILITY = 0.5
BGM_OFFSET_MIN = 2.0
BGM_OFFSET_MAX = 8.0
SEED_MODULO = 10000


def make_video_seed(task_id: str, video_index: int) -> int:
    """生成视频级可复现种子：hash(task_id+video_index) % 10000。

    用 sha256 而非内置 hash()：内置 hash 对字符串带进程级随机盐（PYTHONHASHSEED），
    跨进程不可复现。结果映射到 0~9999。
    """
    import hashlib

    raw = f"{task_id or ''}:{int(video_index)}"
    digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()
    return int(digest[:8], 16) % SEED_MODULO


@dataclass(slots=True)
class ClipMicroTransform:
    """单个片段的微变换参数。"""

    clip_index: int
    hflip: bool = False
    speed: float = 1.0
    brightness: float = 0.0
    contrast: float = 1.0
    saturation: float = 1.0
    has_text: bool = False

    def video_filter_suffix(self) -> str:
        """返回追加在片段视频处理链上的 filter 后缀（无末尾标签）。

        顺序：trim/setpts（已有）→ 调速 setpts → hflip → eq → format。
        调速的 setpts 必须位于 trim 之后；hflip/eq 在缩放之后即可，
        concat_engine 按「调速 → hflip → eq」顺序拼接到 scale/fps 之前的
        trim 之后、scale 之后均可，这里只产出独立步骤、由引擎决定插入点。
        """
        parts: list[str] = []
        # 速度：setpts=PTS/speed（speed>1 时画面加速，时间戳变小）
        if abs(self.speed - 1.0) > 1e-4:
            parts.append(f"setpts=PTS/{self.speed:.5f}")
        # 水平翻转：有文字/字幕片段不翻转
        if self.hflip and not self.has_text:
            parts.append("hflip")
        # 色彩微调：brightness 取值 -1~1（±0.02），contrast/saturation 围绕 1.0
        if abs(self.brightness) > 1e-4 or abs(self.contrast - 1.0) > 1e-4 or abs(self.saturation - 1.0) > 1e-4:
            parts.append(
                f"eq=brightness={self.brightness:+.4f}:"
                f"contrast={self.contrast:.4f}:saturation={self.saturation:.4f}"
            )
        return ",".join(parts)

    def audio_filter_suffix(self) -> str:
        """返回片段音频链上的调速 filter（atempo），无调速时返回空串。"""
        if abs(self.speed - 1.0) <= 1e-4:
            return ""
        return f"atempo={self.speed:.5f}"


@dataclass(slots=True)
class VideoMicroTransformPlan:
    """一个成片视频的全部微变换参数。"""

    task_id: str
    video_index: int
    seed: int
    clips: list[ClipMicroTransform] = field(default_factory=list)
    bgm_start_offset: float = 0.0

    def clip(self, index: int) -> ClipMicroTransform | None:
        for c in self.clips:
            if c.clip_index == index:
                return c
        return None


def _draw_speed(rng: random.Random) -> float:
    return round(rng.uniform(SPEED_MIN, SPEED_MAX), 5)


def _draw_signed_delta(rng: random.Random) -> float:
    return round(rng.uniform(-COLOR_DELTA, COLOR_DELTA), 4)


def build_micro_transform_plan(
    task_id: str,
    video_index: int,
    clip_count: int,
    *,
    clip_has_text: list[bool] | None = None,
    enable_bgm_offset: bool = True,
) -> VideoMicroTransformPlan:
    """按可复现种子生成整片的微变换计划。

    Args:
        task_id: 生成任务 ID（种子输入）
        video_index: 视频在批次中的序号（0 起）
        clip_count: 片段数量
        clip_has_text: 每个片段是否有字幕/文字轨道（True 的片段不翻转）；
                       None 时按 P1 约定视为无可靠文字检测——保守起见 hflip 一律关闭
        enable_bgm_offset: 是否生成 BGM 起始偏移（无 BGM 时调用方可忽略该值）

    Returns:
        VideoMicroTransformPlan
    """
    seed = make_video_seed(task_id, video_index)
    rng = random.Random(seed)

    # P1 字幕检测约定：无法判断片段是否有文字时，一律不翻转（宁可少一个维度也不误翻字幕）
    safe_has_text = clip_has_text if clip_has_text is not None else [True] * max(clip_count, 0)

    clips: list[ClipMicroTransform] = []
    for i in range(max(clip_count, 0)):
        has_text = bool(safe_has_text[i]) if i < len(safe_has_text) else True
        do_hflip = (not has_text) and rng.random() < HFLIP_PROBABILITY
        clips.append(
            ClipMicroTransform(
                clip_index=i,
                hflip=do_hflip,
                speed=_draw_speed(rng),
                brightness=_draw_signed_delta(rng),
                contrast=round(1.0 + _draw_signed_delta(rng), 4),
                saturation=round(1.0 + _draw_signed_delta(rng), 4),
                has_text=has_text,
            )
        )

    bgm_offset = rng.uniform(BGM_OFFSET_MIN, BGM_OFFSET_MAX) if enable_bgm_offset else 0.0
    return VideoMicroTransformPlan(
        task_id=task_id,
        video_index=video_index,
        seed=seed,
        clips=clips,
        bgm_start_offset=round(bgm_offset, 3),
    )


def build_bgm_offset_trim(start_offset: float, bgm_duration: float) -> str:
    """生成 BGM 起始偏移的 atrim 片段。

    偏移超出 BGM 长度时回退为 0（从头播放），避免空输入。
    返回的字符串形如 "atrim=start=3.200,"，可拼到 BGM filter chain 最前面；
    无需偏移时返回空串。
    """
    if start_offset <= 0 or bgm_duration <= 0 or start_offset >= bgm_duration - 0.5:
        return ""
    return f"atrim=start={start_offset:.3f},"