wip(#1749): voice_duration_planner + variant_voice_resolver 纯函数（checkpoint）

2026-09-07 00:58:35 +08:00
parent bcbcb41750
commit c4abc71a2b
2 changed files with 190 additions and 0 deletions
@@ -0,0 +1,70 @@
+"""批量变体配音严格守卫与解析（#1749 问题 A）。
+
+事故：批量独立配音视频3 错绑视频1 配音——根因是 voice_library_ids
+缺值/长度不符时静默 fallback 到 voice_library_id（单值），导致多个变体
+共用同一条配音。
+
+定稿规则：
+- 独立配音（voice_library_ids 非空）：长度必须 == count，逐项非空，
+  否则 400；**禁止静默 fallback** 到单值 voice_library_id；
+- 统一配音（voice_library_id 非空、voice_library_ids 空）：所有变体共用；
+- 两者都空：返回 [""] * count（无配音，渲染走原 BGM/原声路径）。
+
+纯函数，不碰 DB；调用方（路由）负责把 VariantVoiceError 转 400。
+"""
+
+from __future__ import annotations
+
+from typing import List, Optional
+
+
+class VariantVoiceError(ValueError):
+    """配音参数错误（路由层捕获后返回 400）。"""
+
+
+def resolve_variant_voice_ids(
+    *,
+    count: int,
+    voice_library_id: Optional[str] = None,
+    voice_library_ids: Optional[List[str]] = None,
+) -> List[str]:
+    """解析每个变体使用的配音 voice_library_id。
+
+    Args:
+        count: 变体数量（必须 >= 1）。
+        voice_library_id: 统一配音 ID（所有变体共用）。
+        voice_library_ids: 独立配音 ID 列表，长度必须 == count。
+
+    Returns:
+        长度 == count 的列表；元素为 "" 表示该变体无配音。
+
+    Raises:
+        VariantVoiceError: count 非法 / 独立配音长度不符 / 存在缺值。
+    """
+    if not count or count < 1:
+        raise VariantVoiceError("变体数量 count 必须 >= 1")
+
+    single = (voice_library_id or "").strip()
+    multi = [str(v).strip() for v in (voice_library_ids or []) if str(v or "").strip()]
+
+    if voice_library_ids:
+        # 显式传了独立配音列表：严格校验，禁止静默 fallback
+        raw = [str(v or "").strip() for v in voice_library_ids]
+        if len(raw) != count:
+            raise VariantVoiceError(
+                f"独立配音数量({len(raw)})与视频数量({count})不一致；"
+                f"批量独立配音必须为每个视频分别指定配音，或改用统一配音 voice_library_id"
+            )
+        missing = [i for i, v in enumerate(raw) if not v]
+        if missing:
+            raise VariantVoiceError(
+                f"第 {[i + 1 for i in missing]} 个视频缺少独立配音 voice_library_id；"
+                f"独立配音不允许缺值，也不会回退为统一配音"
+            )
+        return raw
+
+    if single:
+        return [single] * count
+
+    # 无配音
+    return [""] * count
@@ -0,0 +1,120 @@
+"""配音时长 → 片段时长分配纯函数（#1749）。
+
+定稿规则（工单 #1749）：
+1. 片段数 = 模板片段数，定死，不因素材增减；
+2. 成片总时长 = 配音时长：逐段分配段长，含转场重叠扣减
+   （cut 零重叠；xfade 等转场按转场时长重叠），误差 < 0.5s；
+3. 素材短于段长 → 末帧冻结（tpad）/ 音频补静音（apad）铺满，
+   禁止慢放、禁止截断配音；
+4. 任何情况下不得因素材时长/数量报错打断用户。
+
+本模块为纯函数：输入片段骨架（每段转场效果/时长）与配音总时长，
+输出每段目标时长（target duration）与成片总时长。不碰 DB、不碰素材。
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+#: 单段最小时长（秒）：低于此值播放器/渲染链路易出问题
+MIN_CLIP_DURATION = 1.0
+
+#: 成片总时长与配音时长的可接受误差（秒）
+TOTAL_DURATION_TOLERANCE = 0.5
+
+
+def transition_overlap_seconds(transition_effect: Optional[str], transition_duration: float) -> float:
+    """转场导致的相邻片段重叠时长。
+
+    cut（或空/None）无重叠；其余转场（xfade/fade/slide 等）按转场时长重叠。
+    """
+    effect = (transition_effect or "cut").strip().lower()
+    if effect in ("cut", "", "none"):
+        return 0.0
+    dur = float(transition_duration or 0.0)
+    return max(0.0, dur)
+
+
+def plan_clip_durations(
+    clip_count: int,
+    voice_duration: float,
+    transition_effects: Optional[list[Optional[str]]] = None,
+    transition_durations: Optional[list[float]] = None,
+) -> list[float]:
+    """把配音总时长分配到 clip_count 段，返回每段目标时长（秒）。
+
+    分配口径：Σ段长 − Σ转场重叠 = 配音时长（成片净时长 = 配音）。
+    转场重叠发生在相邻片段之间，共 clip_count-1 处；第 i 处重叠取
+    **后一段（i+1）** 的转场设置（与 xfade 构建口径一致：转场挂在后段）。
+
+    舍入误差全部由最后一段吸收，保证 total_output_duration(结果) ≈ 配音。
+
+    Args:
+        clip_count: 片段数（模板定死，必须 > 0）。
+        voice_duration: 配音总时长（秒）；<=0 返回空列表表示"无配音"。
+        transition_effects: 每段转场效果（长度 clip_count，index 0 的转场无效）。
+        transition_durations: 每段转场时长（长度 clip_count）。
+
+    Returns:
+        每段目标时长列表（长度 clip_count）；无配音/非法输入返回 []。
+    """
+    if not clip_count or clip_count <= 0:
+        return []
+    try:
+        voice = float(voice_duration)
+    except (TypeError, ValueError):
+        return []
+    if voice <= 0:
+        return []
+
+    effects = transition_effects or [None] * clip_count
+    durations = transition_durations or [0.0] * clip_count
+
+    # 相邻片段间的转场重叠总和（转场挂在后段，取 i=1..clip_count-1）
+    total_overlap = 0.0
+    for i in range(1, clip_count):
+        effect = effects[i] if i < len(effects) else None
+        tdur = durations[i] if i < len(durations) else 0.0
+        total_overlap += transition_overlap_seconds(effect, tdur)
+
+    # 需要的段长总和 = 配音 + 重叠（重叠部分被算了两次，扣回一次）
+    gross = voice + total_overlap
+    if gross < clip_count * MIN_CLIP_DURATION:
+        # 配音极短：保底每段 MIN_CLIP_DURATION（成片略长于配音，末段可冻结）
+        gross = clip_count * MIN_CLIP_DURATION
+        logger.info(
+            "配音时长 %.2fs 过短，%d 段按最小段长 %.1fs 保底（成片将略长于配音）",
+            voice,
+            clip_count,
+            MIN_CLIP_DURATION,
+        )
+
+    per_clip = gross / clip_count
+    result = [round(per_clip, 3) for _ in range(clip_count)]
+    # 末段吸收舍入误差：直接用 gross - 前段之和
+    result[-1] = round(gross - sum(result[:-1]), 3)
+    if result[-1] < MIN_CLIP_DURATION:
+        # 极端情况下末段被舍入压得过小，摊平
+        result[-1] = MIN_CLIP_DURATION
+    return result
+
+
+def total_output_duration(
+    clip_durations: list[float],
+    transition_effects: Optional[list[Optional[str]]] = None,
+    transition_durations: Optional[list[float]] = None,
+) -> float:
+    """按分配后的段长与转场计算成片净时长 = Σ段长 − Σ转场重叠。"""
+    if not clip_durations:
+        return 0.0
+    effects = transition_effects or [None] * len(clip_durations)
+    durations = transition_durations or [0.0] * len(clip_durations)
+    total = sum(float(d) for d in clip_durations)
+    for i in range(1, len(clip_durations)):
+        effect = effects[i] if i < len(effects) else None
+        tdur = durations[i] if i < len(durations) else 0.0
+        total -= transition_overlap_seconds(effect, tdur)
+    return round(max(0.0, total), 3)