"""智能降重微变换纯逻辑模块 — #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},"