From c4abc71a2b621752906d6a737d7d02db8ab7d51a Mon Sep 17 00:00:00 2001 From: saas-backend-agent Date: Mon, 7 Sep 2026 00:58:35 +0800 Subject: [PATCH] =?UTF-8?q?wip(#1749):=20voice=5Fduration=5Fplanner=20+=20?= =?UTF-8?q?variant=5Fvoice=5Fresolver=20=E7=BA=AF=E5=87=BD=E6=95=B0?= =?UTF-8?q?=EF=BC=88checkpoint=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/domain/variant_voice_resolver.py | 70 +++++++++++++ packages/domain/voice_duration_planner.py | 120 ++++++++++++++++++++++ 2 files changed, 190 insertions(+) create mode 100644 packages/domain/variant_voice_resolver.py create mode 100644 packages/domain/voice_duration_planner.py diff --git a/packages/domain/variant_voice_resolver.py b/packages/domain/variant_voice_resolver.py new file mode 100644 index 000000000..a13a97aef --- /dev/null +++ b/packages/domain/variant_voice_resolver.py @@ -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 diff --git a/packages/domain/voice_duration_planner.py b/packages/domain/voice_duration_planner.py new file mode 100644 index 000000000..69054aa7d --- /dev/null +++ b/packages/domain/voice_duration_planner.py @@ -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)