saas-backend-agent
317 lines
15 KiB
Python
317 lines
15 KiB
Python
"""批量变体独立选片核心（#1743 起，#1749 强化素材级去重）。
|
||
|
||
总原则：多视频 = 单视频逻辑 × N。批量正式生成/批量预览/variant-plans 时，
|
||
每个变体**完整重跑单视频的选片流程**：
|
||
|
||
1. 源 plan 片段骨架（clip_type/order/text/transition）保持不变；
|
||
2. 素材池选片（#1749 定稿三轮策略）：
|
||
- **第一轮 fresh 素材优先**：本批次尚未被任何变体使用过的素材优先分配，
|
||
从根上避免跨视频素材重复；
|
||
- **第二轮受控复用**：fresh 素材不足时才允许复用已用素材，但必须通过
|
||
起点扫描（_best_start_for_asset，0.25s 窗口）使与批次内已有区间的
|
||
overlap ≤ 20%（BATCH_CLIP_OVERLAP_LIMIT），且不得完全重叠；
|
||
- **短素材数学上无法错开**（素材时长 < 段长 ×(1−0.20)，任何起点
|
||
重叠都 >20%）→ **禁止跨变体复用**，跳过该素材继续找；
|
||
- **第三轮兜底尽力而为**：池子耗尽时取最优（重叠最小）起点，不报错、
|
||
不打断生成（#1749 铁律：任何情况下不得因素材时长/数量报错打断）；
|
||
3. main 片段之间洗牌顺序；起点走场景镜头洗牌 + 随机起点 + 历史已用区间
|
||
避让（pick_scene_aware_start / _resolve_start_time，与单视频同一入口）；
|
||
4. target_durations：#1749 配音时长分配后每段目标段长（voice_duration_planner），
|
||
落库到片段 duration；素材短于段长由渲染侧末帧冻结（tpad/apad）铺满。
|
||
|
||
本模块只产出 clips_data（dict 列表），不碰 DB 事务边界；素材时长/场景点/
|
||
已用区间由调用方注入，便于单测。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import random
|
||
|
||
from packages.domain.plan_generator_utils import _resolve_start_time
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ── 阈值常量 ────────────────────────────────────────────────────────────────
|
||
BATCH_CLIP_OVERLAP_LIMIT = 0.20
|
||
"""批次内同一素材片段时间区间重叠占比上限（20%）。超过则重选起点/换素材。"""
|
||
|
||
VARIANT_RESELECT_MAX_ATTEMPTS = 6
|
||
"""单片段重叠避让/起点重选的最大尝试次数。"""
|
||
|
||
START_SCAN_STEP = 0.25
|
||
"""复用素材时起点扫描窗口步长（秒）。"""
|
||
|
||
MAIN_CLIP_TYPES = {"main"}
|
||
"""参与素材洗牌重选的片段类型（intro/outro/overlay 等固定角色片段保持源 plan）。"""
|
||
|
||
|
||
def _clip_overlap_ratio(
|
||
asset_id: str,
|
||
start: float,
|
||
duration: float,
|
||
batch_segments: dict[str, list[tuple[float, float]]],
|
||
) -> float:
|
||
"""计算新区间 [start, start+duration) 与批次内同素材已选区间的重叠占比。
|
||
|
||
返回重叠总时长 / 片段时长。
|
||
"""
|
||
if not asset_id or duration <= 0:
|
||
return 0.0
|
||
end = start + duration
|
||
overlap = 0.0
|
||
for seg_start, seg_end in batch_segments.get(asset_id, []):
|
||
ov = max(0.0, min(end, seg_end) - max(start, seg_start))
|
||
overlap += ov
|
||
return min(1.0, overlap / duration)
|
||
|
||
|
||
def _best_start_for_asset(
|
||
asset_id: str,
|
||
clip_duration: float,
|
||
asset_durations: dict[str, float],
|
||
batch_segments: dict[str, list[tuple[float, float]]],
|
||
) -> tuple[float, float] | None:
|
||
"""在素材可用范围内扫描起点，找重叠最小的 (start, ratio)。
|
||
|
||
扫描步长 START_SCAN_STEP；返回 (best_start, best_ratio)。
|
||
短素材（max_start<=0）直接返回 (0.0, ratio)——由调用方判断 ratio 是否可接受。
|
||
"""
|
||
total = asset_durations.get(asset_id, 0.0)
|
||
if total <= 0:
|
||
return None
|
||
max_start = max(0.0, total - clip_duration)
|
||
if max_start <= 0.0:
|
||
return 0.0, _clip_overlap_ratio(asset_id, 0.0, clip_duration, batch_segments)
|
||
best_start, best_ratio = 0.0, 1.0
|
||
steps = int(max_start / START_SCAN_STEP) + 1
|
||
for i in range(steps + 1):
|
||
s = min(max_start, i * START_SCAN_STEP)
|
||
r = _clip_overlap_ratio(asset_id, s, clip_duration, batch_segments)
|
||
if r < best_ratio:
|
||
best_start, best_ratio = s, r
|
||
if r <= BATCH_CLIP_OVERLAP_LIMIT:
|
||
return s, r
|
||
return best_start, best_ratio
|
||
|
||
|
||
def reselect_clips_for_variant(
|
||
source_clips: list[dict],
|
||
candidate_asset_ids: list[str],
|
||
*,
|
||
asset_durations: dict[str, float],
|
||
asset_scene_points: dict[str, list[float]] | None = None,
|
||
historical_used_segments: dict[str, list[tuple[float, float]]] | None = None,
|
||
batch_segments: dict[str, list[tuple[float, float]]] | None = None,
|
||
target_durations: list[float] | dict[int, float] | None = None,
|
||
rng: random.Random | None = None,
|
||
) -> list[dict]:
|
||
"""为一个变体基于源片段骨架重新独立选片。
|
||
|
||
Args:
|
||
source_clips: 源 plan 片段（dict 列表，每项至少含
|
||
order/asset_id/start_time/duration/clip_type，可含
|
||
playback_speed/transition_effect/transition_duration/text_content）。
|
||
candidate_asset_ids: 素材池（源 plan 素材 ∪ 批次任务素材）。
|
||
asset_durations: {asset_id: 时长秒}，起点避让/区间计算必需。
|
||
asset_scene_points: {asset_id: 场景切换点}，有则走镜头洗牌选起点。
|
||
historical_used_segments: 素材 metadata 中持久化的历史已用区间
|
||
（跨任务/跨变体避让），函数内会就地追加本变体选中的区间。
|
||
batch_segments: 本批次已选片段区间（变体间素材级去重 + 20% 重叠检查），
|
||
函数内会就地追加本变体选中的区间。
|
||
target_durations: #1749 配音分配后的每段目标时长（按 order 对齐的 list，
|
||
或 {order: duration} dict）；落库到片段 duration，素材不足由渲染冻结铺满。
|
||
rng: 可选随机数生成器（测试可注入固定种子）。
|
||
|
||
Returns:
|
||
clips_data: 与源片段等长、order 对齐的新片段 dict 列表。
|
||
|
||
Raises:
|
||
ValueError: 源片段为空 / 素材池为空 / 素材时长全为 0（无法差异化选片）。
|
||
"""
|
||
rng = rng or random.Random()
|
||
if not source_clips:
|
||
raise ValueError("源 plan 无片段，无法为变体重新选片")
|
||
if not candidate_asset_ids:
|
||
raise ValueError("素材池为空，无法为变体独立选片（不允许退回同源成片）")
|
||
|
||
# 仅保留时长可知（>0）的素材；时长未知无法做区间避让/重叠计算
|
||
usable_assets = [a for a in dict.fromkeys(candidate_asset_ids) if asset_durations.get(a, 0.0) > 0]
|
||
if not usable_assets:
|
||
raise ValueError("素材池时长全部未知（0），无法为变体独立选片")
|
||
|
||
# 历史已用区间：复制一份，本变体选中的区间就地追加（随 clip record 持久化由调用方负责）
|
||
used_segments: dict[str, list[tuple[float, float]]] = (
|
||
{k: list(v) for k, v in (historical_used_segments or {}).items()} if historical_used_segments else {}
|
||
)
|
||
batch_segments = batch_segments if batch_segments is not None else {}
|
||
|
||
# 按 order 排序源片段，保持骨架顺序
|
||
ordered = sorted(source_clips, key=lambda c: c.get("order", 0))
|
||
|
||
def _target_dur(idx: int, src: dict) -> float:
|
||
"""配音分配的目标段长（优先），否则用源片段段长。"""
|
||
if target_durations is not None:
|
||
if isinstance(target_durations, dict):
|
||
v = target_durations.get(idx, target_durations.get(src.get("order", 0)))
|
||
else:
|
||
v = target_durations[idx] if idx < len(target_durations) else None
|
||
if v is not None and float(v) > 0:
|
||
return float(v)
|
||
return float(src.get("duration", 0.0) or 0.0)
|
||
|
||
# ── 1. 素材池洗牌（组合随机），分 fresh / 已用两轮 ──────────────────────
|
||
shuffled_pool = list(usable_assets)
|
||
rng.shuffle(shuffled_pool)
|
||
|
||
# ── 2. main 片段之间洗牌顺序（顺序随机） ────────────────────────────────
|
||
main_indexes = [i for i, c in enumerate(ordered) if c.get("clip_type", "main") in MAIN_CLIP_TYPES]
|
||
rng.shuffle(main_indexes)
|
||
|
||
result: list[dict | None] = [None] * len(ordered)
|
||
|
||
for idx in main_indexes:
|
||
src = ordered[idx]
|
||
target_dur = _target_dur(idx, src)
|
||
if target_dur <= 0:
|
||
# 异常片段：原样保留
|
||
result[idx] = _base_clip_data(
|
||
src, asset_id=src.get("asset_id", ""), start=float(src.get("start_time", 0.0)), duration=target_dur
|
||
)
|
||
continue
|
||
|
||
asset_id, start, eff_dur = _pick_asset_and_start(
|
||
clip_duration=target_dur,
|
||
shuffled_pool=shuffled_pool,
|
||
asset_durations=asset_durations,
|
||
asset_scene_points=asset_scene_points,
|
||
used_segments=used_segments,
|
||
batch_segments=batch_segments,
|
||
rng=rng,
|
||
)
|
||
|
||
interval = (start, start + eff_dur)
|
||
used_segments.setdefault(asset_id, []).append(interval)
|
||
batch_segments.setdefault(asset_id, []).append(interval)
|
||
|
||
result[idx] = _base_clip_data(src, asset_id=asset_id, start=start, duration=target_dur)
|
||
|
||
# ── 3. 非 main 片段（intro/outro/overlay 等固定角色）：保留源素材，仅重算起点 ──
|
||
for idx, c in enumerate(ordered):
|
||
if result[idx] is not None:
|
||
continue
|
||
src = c
|
||
aid = src.get("asset_id", "")
|
||
target_dur = _target_dur(idx, src)
|
||
start = float(src.get("start_time", 0.0))
|
||
total = asset_durations.get(aid, 0.0)
|
||
if aid and target_dur > 0 and total > 0:
|
||
# 固定角色片段也走批次避让（但不换素材）
|
||
eff_dur = min(target_dur, total)
|
||
scan = _best_start_for_asset(aid, eff_dur, asset_durations, batch_segments)
|
||
if scan is not None and scan[1] <= BATCH_CLIP_OVERLAP_LIMIT:
|
||
start = scan[0]
|
||
else:
|
||
cand = _resolve_start_time(aid, eff_dur, asset_durations, used_segments, asset_scene_points)
|
||
if cand is not None:
|
||
start = cand
|
||
elif scan is not None:
|
||
start = scan[0]
|
||
interval = (start, start + eff_dur)
|
||
used_segments.setdefault(aid, []).append(interval)
|
||
batch_segments.setdefault(aid, []).append(interval)
|
||
result[idx] = _base_clip_data(src, asset_id=aid, start=start, duration=target_dur)
|
||
|
||
return [c for c in result if c is not None]
|
||
|
||
|
||
def _pick_asset_and_start(
|
||
*,
|
||
clip_duration: float,
|
||
shuffled_pool: list[str],
|
||
asset_durations: dict[str, float],
|
||
asset_scene_points: dict[str, list[float]] | None,
|
||
used_segments: dict[str, list[tuple[float, float]]],
|
||
batch_segments: dict[str, list[tuple[float, float]]],
|
||
rng: random.Random,
|
||
) -> tuple[str, float, float]:
|
||
"""三轮选片：fresh 优先 → 受控复用（重叠≤20%，短素材禁复用）→ 兜底尽力而为。
|
||
|
||
Returns:
|
||
(asset_id, start, eff_dur)：eff_dur = min(段长, 素材时长)，
|
||
段长超出素材时长的部分由渲染侧末帧冻结铺满。
|
||
"""
|
||
# ── 第一轮：fresh 素材（本批次未用过）──────────────────────────────────
|
||
fresh = [a for a in shuffled_pool if a not in batch_segments]
|
||
rng.shuffle(fresh)
|
||
for asset_id in fresh:
|
||
total = asset_durations.get(asset_id, 0.0)
|
||
if total <= 0:
|
||
continue
|
||
eff_dur = min(clip_duration, total)
|
||
cand = _resolve_start_time(asset_id, eff_dur, asset_durations, used_segments, asset_scene_points)
|
||
if cand is None:
|
||
max_start = max(0.0, total - eff_dur)
|
||
cand = rng.uniform(0.0, max_start) if max_start > 0 else 0.0
|
||
# fresh 素材批次内无区间，重叠必然为 0，直接采用
|
||
return asset_id, cand, eff_dur
|
||
|
||
# ── 第二轮：受控复用 —— 扫描起点使重叠 ≤20%；短素材数学无法错开则跳过 ──
|
||
reused = [a for a in shuffled_pool if a in batch_segments]
|
||
rng.shuffle(reused)
|
||
fallback: tuple[str, float, float, float] | None = None # (asset, start, eff, ratio)
|
||
for asset_id in reused:
|
||
total = asset_durations.get(asset_id, 0.0)
|
||
if total <= 0:
|
||
continue
|
||
eff_dur = min(clip_duration, total)
|
||
# 短素材判定：素材时长 < 段长 ×(1−0.20) → 任何起点重叠都 >20%，禁跨变体复用
|
||
if total < clip_duration * (1.0 - BATCH_CLIP_OVERLAP_LIMIT) - 1e-6:
|
||
logger.info(
|
||
"素材 %s 时长 %.2fs 短于段长 %.2fs 的 80%，数学上无法错开，禁止跨变体复用",
|
||
asset_id,
|
||
total,
|
||
clip_duration,
|
||
)
|
||
continue
|
||
scan = _best_start_for_asset(asset_id, eff_dur, asset_durations, batch_segments)
|
||
if scan is None:
|
||
continue
|
||
start, ratio = scan
|
||
if ratio <= BATCH_CLIP_OVERLAP_LIMIT:
|
||
return asset_id, start, eff_dur
|
||
if fallback is None or ratio < fallback[3]:
|
||
fallback = (asset_id, start, eff_dur, ratio)
|
||
|
||
# ── 第三轮：兜底尽力而为（池耗尽/全部超阈值）——不报错，取最优 ──────────
|
||
if fallback is not None:
|
||
asset_id, start, eff_dur, ratio = fallback
|
||
logger.info(
|
||
"变体选片素材池不足，受控复用重叠 %.0f%%（>20%% 阈值，尽力而为不打断）: asset=%s",
|
||
ratio * 100,
|
||
asset_id,
|
||
)
|
||
return asset_id, start, eff_dur
|
||
|
||
# 理论不可达（usable_assets 非空）；保底取池首
|
||
asset_id = shuffled_pool[0]
|
||
total = asset_durations.get(asset_id, 0.0)
|
||
eff_dur = min(clip_duration, total) if total > 0 else clip_duration
|
||
return asset_id, 0.0, eff_dur
|
||
|
||
|
||
def _base_clip_data(src: dict, *, asset_id: str, start: float, duration: float | None = None) -> dict:
|
||
"""从源片段构造落库 dict（保留骨架/转场/文案/速度，替换素材与起点）。"""
|
||
return {
|
||
"order": src.get("order", 0),
|
||
"asset_id": asset_id,
|
||
"start_time": round(float(start), 3),
|
||
"duration": float(duration if duration is not None else src.get("duration", 0.0) or 0.0),
|
||
"clip_type": src.get("clip_type", "main"),
|
||
"playback_speed": float(src.get("playback_speed", 1.0) or 1.0),
|
||
"transition_effect": src.get("transition_effect", "cut"),
|
||
"transition_duration": float(src.get("transition_duration", 0.0) or 0.0),
|
||
"text_content": src.get("text_content", ""),
|
||
"config": src.get("config") or {},
|
||
}
|