feat(asset-dedup): 素材去重机制6项优化 — 调参+警告+贪心分配 #1611

Merged
auto-approve-bot merged 2 commits from feat/asset-dedup-optimization into develop 2026-09-01 17:40:34 +08:00
6 changed files with 119 additions and 38 deletions
@@ -582,6 +582,47 @@ def _get_mediakit_recommendations(
return {}
def _calc_plan_internal_duplicate_rate(clips_data: list[dict]) -> float:
"""估算单条成片内部重复率(%.
检查本条成片中同一素材是否有重叠的时间区间。
重叠时长 / 成片总时长 * 100 = 内部重复率。
这是一个轻量估算,不依赖视频指纹;完整查重由 worker 异步完成。
"""
if not clips_data:
return 0.0
# 按素材分组
by_asset: dict[str, list[tuple[float, float]]] = {}
total_duration = 0.0
for c in clips_data:
aid = c.get("asset_id", "")
if not aid:
continue
start = c.get("start_time", 0.0)
end = start + c.get("duration", 0.0)
by_asset.setdefault(aid, []).append((start, end))
total_duration += c.get("duration", 0.0)
if total_duration <= 0:
return 0.0
# 检查同素材内的区间重叠
overlap_duration = 0.0
for segments in by_asset.values():
if len(segments) < 2:
continue
segments_sorted = sorted(segments, key=lambda s: s[0])
last_end = segments_sorted[0][1]
for start, end in segments_sorted[1:]:
overlap = max(0.0, min(end, last_end) - start)
if overlap > 0:
overlap_duration += overlap
last_end = max(last_end, end)
return round(overlap_duration / total_duration * 100, 1)
@router.post("/clips/from-assets", response_model=ClipsFromAssetsResponse)
def create_clips_from_assets_editor(
template_id: str,
@@ -662,19 +703,31 @@ def create_clips_from_assets_editor(
return False
return reused_durations.get(aid, 0.0) / assigned > REUSE_RATIO_LIMIT
# 素材耗尽标志:某轮循环中所有素材均被跳过时为 True
all_assets_exhausted = False
for i, (_seg_order, dur_min, dur_max) in enumerate(segments):
# 在 segment 的 duration_min ~ duration_max 之间随机取值(保留一位小数)
raw_duration = random.uniform(dur_min, dur_max)
# 轮询分配素材:跳过时长缺失、复用占比已超 15% 阈值的素材;
# 贪心分配素材:按"已使用次数"升序排列候选素材(使用最少的优先),
# 同次数随机打散,避免"A-B-C-D"的固定组合反复出现。
# 跳过时长缺失、复用占比已超 10% 阈值的素材;
# 选中后计算起点,若该素材可用区间耗尽且复用被闸门拒绝(calc 返回 None),
# 继续轮询下一个素材
# 继续尝试下一个素材
asset_id = ""
clip_duration = 0.0
start_time: float | None = None
n_assets = len(asset_ids)
for offset in range(n_assets):
candidate = asset_ids[(i + offset) % n_assets]
# 动态按使用次数排序:优先选使用最少的素材,同次数随机打散
asset_use_counts = {
aid: len(used_segments.get(aid, []))
for aid in asset_ids
}
sorted_candidates = sorted(
asset_ids,
key=lambda aid: (asset_use_counts.get(aid, 0), random.random()),
)
for candidate in sorted_candidates:
candidate_total = asset_durations.get(candidate, 0.0)
if candidate_total <= 0:
continue
@@ -690,7 +743,7 @@ def create_clips_from_assets_editor(
continue
# 随机起始时间(不调用 MediaKit,保证接口快速返回);100 次避不开
# 历史区间时走受控复用回调(复用片段累加 reused_durations,回调内部
# 预判复用后占比超 15% 则拒绝并返回 None
# 预判复用后占比超 10% 则拒绝并返回 None
candidate_start = _calc_random_start_time(
candidate,
candidate_duration,
@@ -712,6 +765,7 @@ def create_clips_from_assets_editor(
if not asset_id or start_time is None:
# 所有素材时长缺失、复用占比超阈值,或区间耗尽且复用被拒 → 素材可切区间不足
all_assets_exhausted = True
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="素材可切区间不足,请补充新素材",
@@ -754,11 +808,31 @@ def create_clips_from_assets_editor(
unique_asset_ids,
)
# 6. 立即返回响应
# 6. 估算成片内部重复率(本条成片中同一素材的重叠片段时长占比)
dup_rate = _calc_plan_internal_duplicate_rate(clips_data)
duplicate_warning = None
if dup_rate > 50:
duplicate_warning = f"查重率 {dup_rate:.1f}% 超过50%,建议更换素材或模板"
logger.warning(
"from-assets 成片查重率超标: plan_id=%s dup_rate=%.1f%%",
plan_id, dup_rate,
)
# 7. 素材耗尽提示
exhaustion_warning = None
if all_assets_exhausted and created_count < len(segments):
exhaustion_warning = (
"素材可切区间不足,部分片段使用了复用素材。"
"建议:1) 补充更多素材到素材库 2) 使用不同的素材组合生成"
)
# 8. 立即返回响应
return ClipsFromAssetsResponse(
created_count=created_count,
plan_id=plan_id,
clip_ids=[],
duplicate_warning=duplicate_warning,
exhaustion_warning=exhaustion_warning,
)
@@ -189,6 +189,8 @@ class ClipsFromAssetsResponse(BaseModel):
plan_id: str = ""
message: str = ""
clip_ids: List[str] = Field(default_factory=list, description="创建的片段ID列表")
duplicate_warning: Optional[str] = Field(default=None, description="查重率超标警告")
exhaustion_warning: Optional[str] = Field(default=None, description="素材耗尽警告")
# ── 封面配置 ────────────────────────────────────────────────────────────────
@@ -3,7 +3,7 @@
在素材 metadataassets.classification_result JSON)中持久化已使用的片段时间区间,
供 from-assets 创建片段时避开历史区间,实现跨任务/跨调用的片段去重;
素材可用区间耗尽后进入受控复用:允许有限次数(MAX_RANGE_USE_COUNT)复用最久未用
的历史区间,配合调用方的成片复用占比控制(MAX_REUSE_RATIO = 15%),把任意两条
的历史区间,配合调用方的成片复用占比控制(MAX_REUSE_RATIO = 10%),把任意两条
成片的画面重复率控制在阈值内。
metadata 中的记录字段 ``used_time_ranges``::
@@ -40,14 +40,14 @@ logger = logging.getLogger(__name__)
USED_RANGES_KEY = "used_time_ranges"
# ── 受控复用配置常量 ─────────────────────────────────────────────────────────
MAX_RANGE_USE_COUNT = 3
MAX_RANGE_USE_COUNT = 2
"""单条历史区间最多被使用次数(含首次),达到后不再参与复用。"""
REUSE_RATIO_LIMIT = 0.15
"""单条成片中,单个素材的复用片段累计时长 / 该素材在成片中的总时长上限(15%)。
REUSE_RATIO_LIMIT = 0.10
"""单条成片中,单个素材的复用片段累计时长 / 该素材在成片中的总时长上限(10%)。
超过则该素材不再分配新片段(调用方在轮询分配时跳过)。"""
SEGMENT_EDGE_GAP = 0.3
SEGMENT_EDGE_GAP = 1.5
"""冲突判定边缘间隙(秒):历史区间按 [start-gap, end+gap] 扩边后参与冲突检测,
避免两条片段首尾紧贴导致画面观感重复;记录仍存实际值。"""
@@ -397,12 +397,12 @@ def make_reuse_callback(
db: SQLAlchemy session
asset_durations: 素材 ID -> 总时长(回调需要素材总时长做边界约束)
reused_tracker: 可选的 ``{asset_id: 累计复用时长}``,回调成功返回复用区间时
会把本次片段时长累加进去,供调用方统计成片复用占比(15% 阈值)。
会把本次片段时长累加进去,供调用方统计成片复用占比(10% 阈值)。
assigned_tracker: 可选的 ``{asset_id: 已分配片段总时长}``,配合 ratio_limit
在复用前预判:若复用本片段后占比 (reused + clip_duration) /
(assigned + clip_duration) 超过 ratio_limit,则拒绝复用、返回 None
(保证成片复用占比不超阈值)。
ratio_limit: 单条成片复用时长占比上限,默认 15%
ratio_limit: 单条成片复用时长占比上限,默认 10%
Returns:
回调函数 ``(asset_id, clip_duration) -> (start, end) | None``。
+10 -10
View File
@@ -124,9 +124,9 @@ def _ranges(db, aid="a1"):
def test_config_constants():
assert MAX_RANGE_USE_COUNT == 3
assert REUSE_RATIO_LIMIT == 0.15
assert SEGMENT_EDGE_GAP == 0.3
assert MAX_RANGE_USE_COUNT == 2
assert REUSE_RATIO_LIMIT == 0.10
assert SEGMENT_EDGE_GAP == 1.5
# ── get_used_segments ─────────────────────────────────────────────────────────
@@ -320,13 +320,13 @@ def test_find_reusable_prefers_oldest_unused(patched_model):
def test_find_reusable_excludes_max_use_count(patched_model):
"""use_count 达到上限(3)的区间不再参与复用;全部达上限返回 None。"""
"""use_count 达到上限(2)的区间不再参与复用;全部达上限返回 None。"""
models = {
"a1": FakeModel(
"a1",
{
"used_time_ranges": [
{"start": 0.0, "end": 10.0, "use_count": 3, "last_used_at": "2026-01-01T00:00:00"},
{"start": 0.0, "end": 10.0, "use_count": 2, "last_used_at": "2026-01-01T00:00:00"},
]
},
)
@@ -335,22 +335,22 @@ def test_find_reusable_excludes_max_use_count(patched_model):
assert find_reusable_range(db, "a1", 5.0, 30.0) is None
def test_find_reusable_fourth_use_rejected(patched_model):
"""同区间复用第 4 次被拒绝:use_count=2 的可复用,use_count=3 的不可复用。"""
def test_find_reusable_third_use_rejected(patched_model):
"""同区间复用第 3 次被拒绝:use_count=1 的可复用,use_count=2 的不可复用。"""
models = {
"a1": FakeModel(
"a1",
{
"used_time_ranges": [
{"start": 0.0, "end": 10.0, "use_count": 2, "last_used_at": "2026-03-01T00:00:00"},
{"start": 10.0, "end": 20.0, "use_count": 3, "last_used_at": "2026-01-01T00:00:00"},
{"start": 0.0, "end": 10.0, "use_count": 1, "last_used_at": "2026-03-01T00:00:00"},
{"start": 10.0, "end": 20.0, "use_count": 2, "last_used_at": "2026-01-01T00:00:00"},
]
},
)
}
db = _db(models)
result = find_reusable_range(db, "a1", 5.0, 30.0)
# 只能选 use_count=2 的区间(start=0),不能选 use_count=3 的(虽然它更老)
# 只能选 use_count=1 的区间(start=0),不能选 use_count=2 的(虽然它更老)
assert result is not None and result[0] == 0.0
+6 -5
View File
@@ -142,11 +142,12 @@ class TestEditorClipsBySegments:
clips_data = _get_clips_data_from_call(mock_plan_svc)
assert len(clips_data) == 4
# 验证轮询分配:a1, a2, a1, a2
assert clips_data[0]["asset_id"] == "a1"
assert clips_data[1]["asset_id"] == "a2"
assert clips_data[2]["asset_id"] == "a1"
assert clips_data[3]["asset_id"] == "a2"
# 验证均衡分配(贪心策略保证):2个素材分4个片段,每个素材恰好使用2次
from collections import Counter
asset_ids = [c["asset_id"] for c in clips_data]
counts = Counter(asset_ids)
assert counts["a1"] == 2 and counts["a2"] == 2
@patch("app.api.routes.templates_editor.clips.get_storage_service")
def test_orders_start_at_zero(self, mock_storage):
+13 -9
View File
@@ -69,13 +69,13 @@ class TestRecommendedTimeConflicts:
def test_conflict_exact_boundary_no_overlap(self):
from app.api.routes.templates_editor.clips import _recommended_time_conflicts
# 新语义:默认 0.3s 边缘间隙扩边,推荐 [10, 15] 与已用 [0, 10] 首尾相接
# 新语义:默认 1.5s 边缘间隙扩边,推荐 [10, 15] 与已用 [0, 10] 首尾相接
# 落在扩边范围内 → 判为冲突(避免观感重复)
assert _recommended_time_conflicts(10.0, 5.0, [(0.0, 10.0)]) is True
# 显式 edge_gap=0 时退回纯区间重叠判定:相接不算重叠
assert _recommended_time_conflicts(10.0, 5.0, [(0.0, 10.0)], edge_gap=0.0) is False
# 间隙大于边缘间隙(0.5 > 0.3)→ 不冲突
assert _recommended_time_conflicts(10.5, 5.0, [(0.0, 10.0)]) is False
# 间隙大于边缘间隙(2.0 > 1.5)→ 不冲突
assert _recommended_time_conflicts(12.0, 5.0, [(0.0, 10.0)]) is False
def test_conflict_multiple_used(self):
from app.api.routes.templates_editor.clips import _recommended_time_conflicts
@@ -83,10 +83,10 @@ class TestRecommendedTimeConflicts:
used = [(0.0, 5.0), (10.0, 15.0), (20.0, 25.0)]
# 推荐 [6, 11] 与 [10, 15] 冲突
assert _recommended_time_conflicts(6.0, 5.0, used) is True
# 推荐 [15, 20] 与 [10, 15] 首尾相接:0.3s 扩边内 → 冲突
# 推荐 [15, 20] 与 [10, 15] 首尾相接:1.5s 扩边内 → 冲突
assert _recommended_time_conflicts(15.0, 5.0, used) is True
# 空闲段 [5.3, 9.7] 长 4.4s:推荐 [5.5, 9.5]dur=4)与三区间扩边均不接触
assert _recommended_time_conflicts(5.5, 4.0, used) is False
# 空闲段 (6.5, 8.5) 长 2.0s:推荐 [6.6, 8.4]dur=1.8)与三区间扩边均不接触
assert _recommended_time_conflicts(6.6, 1.8, used) is False
# ── _get_mediakit_recommendations 单元测试 ──────────────────────────────────
@@ -445,8 +445,8 @@ class TestFromAssetsByTemplateSegments:
assert 3.0 <= clips_data[0]["duration"] <= 5.0
assert 4.0 <= clips_data[1]["duration"] <= 8.0
def test_assets_round_robin_assignment(self):
"""素材按片段顺序轮询分配"""
def test_assets_balanced_assignment(self):
"""素材按使用次数贪心分配(使用少的优先),保证均衡使用"""
from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
@@ -475,7 +475,11 @@ class TestFromAssetsByTemplateSegments:
clips_data = _get_clips_data(mock_plan_svc)
asset_ids = [c["asset_id"] for c in clips_data]
assert asset_ids == ["a1", "a2", "a1", "a2"]
# 贪心分配保证均衡:2个素材分4个片段,每个素材恰好使用2次
from collections import Counter
counts = Counter(asset_ids)
assert counts["a1"] == 2 and counts["a2"] == 2
def test_orders_start_from_zero(self):
"""片段 order 从 0 开始递增。"""