feat: 跨视频片段避让 — 生成前注入已用区间 #1670 #1671

Merged
xiaoxia merged 2 commits from feat/cross-video-avoidance-1670 into develop 2026-09-03 22:42:11 +08:00
4 changed files with 431 additions and 9 deletions
@@ -131,6 +131,7 @@ class PlanGeneratorService:
editing_mode,
random_selection=random_preview,
asset_durations=asset_durations,
user_id=created_by_user_id,
)
# 5. 持久化所有 clips 并计算总时长
@@ -218,6 +219,7 @@ class PlanGeneratorService:
*,
random_selection: bool = False,
asset_durations: dict[str, float] | None = None,
user_id: str = "",
) -> None:
"""按 editing_mode 将素材分配到 clips(就地修改,未持久化).
@@ -239,6 +241,14 @@ class PlanGeneratorService:
asset_ids = list(asset_ids) # 复制避免修改调用方原列表
random.shuffle(asset_ids)
# 查询已有视频的已用区间(跨视频避让)
external_used_segments = None
if user_id and self._clip_repo:
try:
external_used_segments = self._clip_repo.list_used_segments_by_user(user_id, limit_recent=50)
except Exception:
logger.warning("跨视频避让查询失败,回退到纯随机", exc_info=True)
distribute_assets(
clips,
asset_ids,
@@ -246,6 +256,7 @@ class PlanGeneratorService:
random_selection=random_selection,
asset_durations=asset_durations,
asset_scene_points=asset_scene_points,
external_used_segments=external_used_segments,
)
def _fetch_asset_scene_points(self, asset_ids: List[str]) -> dict[str, list[float]]:
@@ -131,3 +131,65 @@ class SQLAlchemyEditPlanClipRepository:
created_at=model.created_at,
updated_at=model.updated_at,
)
def list_used_segments_by_user(
self,
user_id: str,
*,
limit_recent: int = 50,
) -> dict[str, list[tuple[float, float]]]:
"""查询用户已有视频中已使用的素材区间(跨视频避让).
JOIN edit_plans 表,按 created_by_user_id 过滤,只查 status='completed'
的 plan 下 status='rendered' 且 asset_id 非空的 clips。按 plan 的
created_at DESC 取最近 limit_recent 个 plan。
Returns:
{asset_id: [(start_time, start_time + duration), ...]}
空结果返回空 dict。
"""
from packages.adapters.sqlalchemy_impl.models import EditPlanModel
if not user_id:
return {}
# 1. 查出最近 limit_recent 个已完成 plan 的 ID
recent_plan_ids = [
row[0]
for row in self.session.query(EditPlanModel.id)
.filter(
EditPlanModel.created_by_user_id == user_id,
EditPlanModel.status == "completed",
)
.order_by(EditPlanModel.created_at.desc())
.limit(limit_recent)
.all()
]
if not recent_plan_ids:
return {}
# 2. 查这些 plan 下已渲染、有素材的 clips
clips = (
self.session.query(
EditPlanClipModel.asset_id,
EditPlanClipModel.start_time,
EditPlanClipModel.duration,
)
.filter(
EditPlanClipModel.plan_id.in_(recent_plan_ids),
EditPlanClipModel.status == "rendered",
EditPlanClipModel.asset_id != "",
EditPlanClipModel.asset_id.isnot(None),
)
.all()
)
# 3. 聚合为 {asset_id: [(start, start+duration), ...]}
result: dict[str, list[tuple[float, float]]] = {}
for asset_id, start_time, duration in clips:
if asset_id not in result:
result[asset_id] = []
result[asset_id].append((start_time or 0.0, (start_time or 0.0) + (duration or 0.0)))
return result
+23 -9
View File
@@ -169,6 +169,7 @@ def distribute_assets(
random_selection: bool = False,
asset_durations: dict[str, float] | None = None,
asset_scene_points: dict[str, list[float]] | None = None,
external_used_segments: dict[str, list[tuple[float, float]]] | None = None,
) -> None:
"""按 editing_mode 将素材分配到 clips(就地修改).
@@ -188,6 +189,7 @@ def distribute_assets(
random_selection: 是否随机选择素材(用于预览生成)
asset_durations: 素材 ID -> 时长(秒)映射,用于设置 start_time
asset_scene_points: 素材 ID -> 场景切换点列表(metadata 缓存)
external_used_segments: 跨视频已用区间(来自其他视频的 clips),注入到分配逻辑中避让
"""
if not asset_ids or not clips:
return
@@ -198,16 +200,16 @@ def distribute_assets(
random.shuffle(asset_ids)
if editing_mode == EditingMode.ONE_TAKE.value:
_distribute_one_take(clips, asset_ids, asset_durations, asset_scene_points)
_distribute_one_take(clips, asset_ids, asset_durations, asset_scene_points, external_used_segments)
elif editing_mode == EditingMode.PIP.value:
_distribute_pip(clips, asset_ids, asset_durations, asset_scene_points)
_distribute_pip(clips, asset_ids, asset_durations, asset_scene_points, external_used_segments)
elif editing_mode == EditingMode.VOICE_OVER.value:
_distribute_voice_over(clips, asset_ids, asset_durations, asset_scene_points)
_distribute_voice_over(clips, asset_ids, asset_durations, asset_scene_points, external_used_segments)
elif editing_mode == EditingMode.VOICE_PIP.value:
_distribute_voice_pip(clips, asset_ids, asset_durations, asset_scene_points)
_distribute_voice_pip(clips, asset_ids, asset_durations, asset_scene_points, external_used_segments)
else:
# 未知模式,退化为 one_take
_distribute_one_take(clips, asset_ids, asset_durations, asset_scene_points)
_distribute_one_take(clips, asset_ids, asset_durations, asset_scene_points, external_used_segments)
def _resolve_start_time(
@@ -248,9 +250,12 @@ def _distribute_one_take(
asset_ids: List[str],
asset_durations: dict[str, float] | None = None,
asset_scene_points: dict[str, list[float]] | None = None,
external_used_segments: dict[str, list[tuple[float, float]]] | None = None,
) -> None:
"""ONE_TAKE: 素材按顺序依次分配给 main 类型 clips."""
used_segments: dict[str, list[tuple[float, float]]] = {}
used_segments: dict[str, list[tuple[float, float]]] = (
{k: list(v) for k, v in external_used_segments.items()} if external_used_segments else {}
)
main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
for i, clip in enumerate(main_clips):
if i < len(asset_ids):
@@ -271,9 +276,12 @@ def _distribute_pip(
asset_ids: List[str],
asset_durations: dict[str, float] | None = None,
asset_scene_points: dict[str, list[float]] | None = None,
external_used_segments: dict[str, list[tuple[float, float]]] | None = None,
) -> None:
"""PIP: 第1个素材→main(全屏背景),其余→overlay clips."""
used_segments: dict[str, list[tuple[float, float]]] = {}
used_segments: dict[str, list[tuple[float, float]]] = (
{k: list(v) for k, v in external_used_segments.items()} if external_used_segments else {}
)
# 第1个素材 → main clip
main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
if main_clips and asset_ids:
@@ -310,9 +318,12 @@ def _distribute_voice_over(
asset_ids: List[str],
asset_durations: dict[str, float] | None = None,
asset_scene_points: dict[str, list[float]] | None = None,
external_used_segments: dict[str, list[tuple[float, float]]] | None = None,
) -> None:
"""VOICE_OVER: 素材→main clips (B-roll)."""
used_segments: dict[str, list[tuple[float, float]]] = {}
used_segments: dict[str, list[tuple[float, float]]] = (
{k: list(v) for k, v in external_used_segments.items()} if external_used_segments else {}
)
main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
for i, clip in enumerate(main_clips):
if i < len(asset_ids):
@@ -333,9 +344,12 @@ def _distribute_voice_pip(
asset_ids: List[str],
asset_durations: dict[str, float] | None = None,
asset_scene_points: dict[str, list[float]] | None = None,
external_used_segments: dict[str, list[tuple[float, float]]] | None = None,
) -> None:
"""VOICE_PIP: 第1个→background, 第2个→corner_voice, 其余→b_roll."""
used_segments: dict[str, list[tuple[float, float]]] = {}
used_segments: dict[str, list[tuple[float, float]]] = (
{k: list(v) for k, v in external_used_segments.items()} if external_used_segments else {}
)
bg_clips = [c for c in clips if c.clip_type == "background"]
voice_clips = [c for c in clips if c.clip_type == "corner_voice"]
broll_clips = [c for c in clips if c.clip_type == "b_roll"]
+335
View File
@@ -0,0 +1,335 @@
"""Tests for Issue #1670 — 跨视频片段避让(生成前注入已用区间)."""
from __future__ import annotations
from datetime import datetime, timezone
from unittest.mock import MagicMock, patch
import pytest
from packages.adapters.sqlalchemy_impl.edit_plan_clip_repository import (
SQLAlchemyEditPlanClipRepository,
)
from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus
from packages.domain.plan_generator_utils import (
_distribute_one_take,
distribute_assets,
)
# ── Repository 层测试 ─────────────────────────────────────────────────────────
class TestListUsedSegmentsByUser:
"""测试 list_used_segments_by_user 方法."""
def _make_repo(self, session_mock):
return SQLAlchemyEditPlanClipRepository(session_mock)
def test_empty_user_id_returns_empty_dict(self):
"""空 user_id 直接返回空 dict,不查 DB."""
session = MagicMock()
repo = self._make_repo(session)
result = repo.list_used_segments_by_user("")
assert result == {}
session.query.assert_not_called()
def test_no_completed_plans_returns_empty_dict(self):
"""用户没有已完成的 plan 时返回空 dict."""
session = MagicMock()
# Mock plan query returns empty
plan_query = MagicMock()
plan_query.filter.return_value = plan_query
plan_query.order_by.return_value = plan_query
plan_query.limit.return_value = plan_query
plan_query.all.return_value = []
session.query.return_value = plan_query
repo = self._make_repo(session)
result = repo.list_used_segments_by_user("user_123")
assert result == {}
def test_aggregates_clips_from_multiple_plans(self):
"""从多个已完成 plan 的 clips 聚合已用区间."""
session = MagicMock()
# Mock plan query: 2 completed plans
plan_query = MagicMock()
plan_query.filter.return_value = plan_query
plan_query.order_by.return_value = plan_query
plan_query.limit.return_value = plan_query
plan_query.all.return_value = [("plan_1",), ("plan_2",)]
session.query.return_value = plan_query
# Mock clip query: clips from both plans
clip_query = MagicMock()
clip_query.filter.return_value = clip_query
clip_query.all.return_value = [
("asset_A", 0.0, 5.0), # plan_1, asset A: 0~5s
("asset_A", 10.0, 3.0), # plan_1, asset A: 10~13s
("asset_B", 2.0, 4.0), # plan_2, asset B: 2~6s
]
# Second session.query call is for clips
session.query.side_effect = [plan_query, clip_query]
repo = self._make_repo(session)
result = repo.list_used_segments_by_user("user_123")
assert "asset_A" in result
assert len(result["asset_A"]) == 2
assert result["asset_A"][0] == (0.0, 5.0)
assert result["asset_A"][1] == (10.0, 13.0)
assert "asset_B" in result
assert result["asset_B"][0] == (2.0, 6.0)
def test_respects_limit_recent_parameter(self):
"""limit_recent 参数限制查询的 plan 数量."""
session = MagicMock()
plan_query = MagicMock()
plan_query.filter.return_value = plan_query
plan_query.order_by.return_value = plan_query
plan_query.limit.return_value = plan_query
plan_query.all.return_value = [("plan_1",)]
session.query.return_value = plan_query
clip_query = MagicMock()
clip_query.filter.return_value = clip_query
clip_query.all.return_value = [("asset_X", 1.0, 2.0)]
session.query.side_effect = [plan_query, clip_query]
repo = self._make_repo(session)
result = repo.list_used_segments_by_user("user_123", limit_recent=10)
# Verify limit was called with the parameter
plan_query.limit.assert_called_once_with(10)
assert "asset_X" in result
# ── Domain 层测试 ─────────────────────────────────────────────────────────────
class TestDistributeAssetsWithExternalSegments:
"""测试 distribute_assets 传入 external_used_segments 的行为."""
def _make_clips(self, count: int, duration: float = 3.0) -> list[EditPlanClip]:
"""创建指定数量的 MAIN 类型 clips."""
return [
EditPlanClip(
id=f"clip_{i}",
plan_id="plan_1",
clip_type="main",
order=i,
template_clip_config_id="",
asset_id="",
text_content="",
start_time=0.0,
duration=duration,
status=EditPlanClipStatus.PENDING,
)
for i in range(count)
]
def test_external_used_segments_none_backward_compatible(self):
"""external_used_segments=None 时行为不变(向后兼容)."""
clips = self._make_clips(3)
asset_ids = ["asset_1", "asset_2", "asset_3"]
asset_durations = {aid: 30.0 for aid in asset_ids}
# Should not raise
distribute_assets(
clips,
asset_ids,
"one_take",
asset_durations=asset_durations,
external_used_segments=None,
)
# All clips should have assets assigned
for clip in clips:
assert clip.asset_id != ""
def test_external_used_segments_avoids_existing_ranges(self):
"""传入 external_used_segments 后,新分配的 start_time 避开已有区间."""
clips = self._make_clips(2, duration=3.0)
asset_ids = ["asset_1"]
asset_durations = {"asset_1": 30.0}
# Pretend asset_1 0~10s is already used by another video
external = {"asset_1": [(0.0, 10.0)]}
# Run multiple times to check that start_time always avoids 0~10s
# (with some randomness, but the avoidance should be consistent)
for _ in range(10):
test_clips = self._make_clips(1, duration=3.0)
distribute_assets(
test_clips,
asset_ids,
"one_take",
asset_durations=asset_durations,
external_used_segments=external,
)
start = test_clips[0].start_time
# Start time + duration (3s) should not overlap with 0~10
# i.e., start >= 10.0 or start + 3 <= 0.0 (impossible since start >= 0)
assert (
start >= 10.0 or start + 3.0 <= 0.0 or start >= 10.0
), f"start_time {start} overlaps with existing segment 0~10"
def test_external_used_segments_deep_copy(self):
"""external_used_segments 会被深拷贝,不会修改外部数据."""
external = {"asset_1": [(0.0, 5.0)]}
original = {"asset_1": [(0.0, 5.0)]}
clips = self._make_clips(1, duration=2.0)
asset_ids = ["asset_1"]
asset_durations = {"asset_1": 20.0}
distribute_assets(
clips,
asset_ids,
"one_take",
asset_durations=asset_durations,
external_used_segments=external,
)
# External dict should be unchanged
assert external == original
def test_empty_external_used_segments_same_as_none(self):
"""空 dict 的 external_used_segments 行为与 None 相同."""
clips = self._make_clips(2, duration=3.0)
asset_ids = ["asset_1", "asset_2"]
asset_durations = {aid: 30.0 for aid in asset_ids}
# Should not raise and should assign assets normally
distribute_assets(
clips,
asset_ids,
"one_take",
asset_durations=asset_durations,
external_used_segments={},
)
for clip in clips:
assert clip.asset_id != ""
# ── Service 层测试 ────────────────────────────────────────────────────────────
class TestServiceLayerIntegration:
"""测试 _distribute_assets 在 service 层的查询逻辑."""
def _make_service(self, clip_repo_mock, asset_repo_mock=None):
"""创建 PlanGeneratorService 并注入 mock repos."""
from unittest.mock import MagicMock, patch
from apps.api.app.services.plan_generator_service import PlanGeneratorService
with (
patch("apps.api.app.services.plan_generator_service.SQLAlchemyEditPlanRepository"),
patch(
"apps.api.app.services.plan_generator_service.SQLAlchemyEditPlanClipRepository",
return_value=clip_repo_mock,
),
):
db = MagicMock()
svc = PlanGeneratorService(db, asset_repo=asset_repo_mock)
svc._clip_repo = clip_repo_mock
return svc
def _make_clip(self):
return EditPlanClip(
id="clip_1",
plan_id="plan_1",
clip_type="main",
order=0,
template_clip_config_id="",
asset_id="",
text_content="",
start_time=0.0,
duration=3.0,
status=EditPlanClipStatus.PENDING,
)
def test_query_called_with_user_id(self):
"""有 user_id 时调用 list_used_segments_by_user."""
clip_repo = MagicMock()
clip_repo.list_used_segments_by_user.return_value = {"asset_A": [(0.0, 5.0)]}
asset_repo = MagicMock()
asset_repo.get.return_value = None # smart_match fallback
svc = self._make_service(clip_repo, asset_repo)
clips = [self._make_clip()]
svc._distribute_assets(
clips,
["asset_A"],
"one_take",
asset_durations={"asset_A": 30.0},
user_id="user_123",
)
clip_repo.list_used_segments_by_user.assert_called_once_with("user_123", limit_recent=50)
def test_query_not_called_without_user_id(self):
"""无 user_id 时不调用查询."""
clip_repo = MagicMock()
asset_repo = MagicMock()
asset_repo.get.return_value = None
svc = self._make_service(clip_repo, asset_repo)
clips = [self._make_clip()]
svc._distribute_assets(
clips,
["asset_A"],
"one_take",
asset_durations={"asset_A": 30.0},
user_id="",
)
clip_repo.list_used_segments_by_user.assert_not_called()
def test_query_failure_does_not_block_generation(self):
"""查询失败时不阻塞生成,回退到纯随机."""
clip_repo = MagicMock()
clip_repo.list_used_segments_by_user.side_effect = Exception("DB error")
asset_repo = MagicMock()
asset_repo.get.return_value = None
svc = self._make_service(clip_repo, asset_repo)
clips = [self._make_clip()]
# Should not raise
svc._distribute_assets(
clips,
["asset_A"],
"one_take",
asset_durations={"asset_A": 30.0},
user_id="user_123",
)
# Clip should still get an asset assigned (fallback to random)
assert clips[0].asset_id == "asset_A"
def test_preview_and_final_both_query(self):
"""预览和正式生成都触发查询."""
for random_selection in [True, False]:
clip_repo = MagicMock()
clip_repo.list_used_segments_by_user.return_value = {}
asset_repo = MagicMock()
asset_repo.get.return_value = None
svc = self._make_service(clip_repo, asset_repo)
clips = [self._make_clip()]
svc._distribute_assets(
clips,
["asset_A"],
"one_take",
random_selection=random_selection,
asset_durations={"asset_A": 30.0},
user_id="user_123",
)
clip_repo.list_used_segments_by_user.assert_called_once()