feat: 素材片段区间持久化去重 + 区间用尽自动轮回
CI/CD Pipeline / Check push changed paths (pull_request) Has been skipped
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 2m39s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 3m43s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 1m8s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 1m15s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 5m30s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 8m33s
AI Code Review / AI Code Review (pull_request) Successful in 8m50s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 9m35s
CI/CD Pipeline / Validate - Code Quality (pull_request) Has been cancelled
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
Preview Deploy / Deploy Preview Environment (pull_request) Has been cancelled

- 新增 asset_segment_tracker 服务:素材 metadata(used_time_ranges) 持久化片段级已用区间
- from-assets 创建片段时读取历史区间,新片段跨任务/跨调用自动避开
- 片段记录与 replace_all_clips_transactional 同事务,失败整体回滚
- MediaKit 异步移动片段起点后同步更新 metadata 区间记录(失败静默)
- _calc_random_start_time 新增 on_exhausted 回调:100次找不到时清空该素材历史区间再重试,实现轮完一圈自动循环

详见 PR body 的 metadata schema 说明
This commit is contained in:
CI Bot
2026-08-29 20:14:28 +08:00
parent 700d6f9130
commit 2174e91c48
5 changed files with 574 additions and 5 deletions
@@ -23,6 +23,12 @@ import re
from app.auth import AuthenticatedUser, get_current_user
from app.core.storage import get_storage_service
from app.dependencies import get_asset_repository, get_db_session
from app.services.asset_segment_tracker import (
get_used_segments,
make_reset_callback,
record_used_segments,
remove_used_segment,
)
from app.services.edit_plan_service import EditPlanService
from app.services.edit_template_service import EditTemplateService
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, status
@@ -600,7 +606,12 @@ def create_clips_from_assets_editor(
asset_durations[asset_id] = float(asset.duration or 0.0)
# 3. 在内存中计算所有片段数据(使用随机起始时间,不调用MediaKit)
used_segments: dict[str, list[tuple[float, float]]] = {}
# 读取素材 metadata 中持久化的历史已用区间(跨任务/跨调用去重),
# 格式与 _calc_random_start_time 的 used_segments 参数一致
used_segments: dict[str, list[tuple[float, float]]] = get_used_segments(
db, unique_asset_ids
)
reset_cb = make_reset_callback(db, used_segments)
clips_data: list[dict] = []
for i, (_seg_order, dur_min, dur_max) in enumerate(segments):
@@ -630,7 +641,11 @@ def create_clips_from_assets_editor(
# 使用随机起始时间(不调用MediaKit,保证接口快速返回)
start_time = _calc_random_start_time(
asset_id, clip_duration, asset_durations, used_segments
asset_id,
clip_duration,
asset_durations,
used_segments,
on_exhausted=reset_cb,
)
if start_time is None:
@@ -639,10 +654,15 @@ def create_clips_from_assets_editor(
detail=f"素材 {asset_id} 时长信息缺失,无法计算起始时间",
)
# 记录已使用时间段
# 记录已使用时间段(内存,供本次后续片段避开)
used_segments.setdefault(asset_id, []).append(
(start_time, start_time + clip_duration)
)
# 同步写入素材 metadata(不 commit,与下方 replace_all_clips_transactional
# 处于同一事务,任一步失败整体回滚,不留脏数据)
record_used_segments(
db, asset_id, start_time, start_time + clip_duration, plan_id
)
clips_data.append(
{
@@ -799,8 +819,35 @@ def _update_mediakit_recommendations_async( # pragma: no cover
# 逐个更新并捕获异常(单点失败不影响其他片段)
try:
old_start = clip.start_time
old_end = old_start + clip_duration
plan_svc.update_clip(clip.id, start_time=recommended_start)
db.commit()
# MediaKit 移动了片段起点 → 同步素材 metadata 的区间记录:
# 删除旧区间记录(按 plan_id + 旧 start 匹配),写入新区间。
# 异步任务,失败静默,不影响已更新的片段。
try:
if remove_used_segment(
db, asset_id, old_start, old_end, plan_id=plan_id
):
record_used_segments(
db,
asset_id,
recommended_start,
recommended_start + clip_duration,
plan_id,
)
db.commit()
except Exception as me:
logger.warning(
"后台任务: 同步素材区间记录失败: clip_id=%s error=%s",
clip.id,
me,
)
try:
db.rollback()
except Exception:
pass
updated_count += 1
updated_clip_ids.add(clip.id)
except Exception as ue:
@@ -0,0 +1,171 @@
"""素材片段级使用记录追踪.
在素材 metadataassets.classification_result JSON)中持久化已使用的片段时间区间,
供 from-assets 创建片段时避开历史区间,实现跨任务/跨调用的片段去重。
metadata 中新增字段 ``used_time_ranges``::
"used_time_ranges": [
{"start": 12.5, "end": 20.3, "plan_id": "plan-xxx", "created_at": "2026-08-29T12:00:00+00:00"},
...
]
注意:本模块所有函数都不自行 commit,由调用方控制事务边界
from-assets 与 replace_all_clips_transactional 同事务;异步任务各自 commit)。
"""
from __future__ import annotations
import json
import logging
from datetime import datetime, timezone
from typing import Callable
from sqlalchemy.orm import Session
from packages.adapters.sqlalchemy_impl.models import AssetModel
logger = logging.getLogger(__name__)
USED_RANGES_KEY = "used_time_ranges"
def _read_ranges(model: AssetModel) -> list[dict]:
"""从 AssetModel 读取 metadata dictclassification_result 列承载的 JSON."""
if not model.classification_result:
return {}
try:
return json.loads(model.classification_result)
except Exception:
return {}
def get_used_segments(db: Session, asset_ids: list[str]) -> dict[str, list[tuple[float, float]]]:
"""聚合多个素材的历史已用片段区间。
Args:
db: SQLAlchemy session
asset_ids: 素材 ID 列表
Returns:
``{asset_id: [(start, end), ...]}`` 格式,与 ``_calc_random_start_time`` 的
``used_segments`` 参数格式一致,可直接传入。
"""
if not asset_ids:
return {}
result: dict[str, list[tuple[float, float]]] = {}
models = db.query(AssetModel).filter(AssetModel.id.in_(list(set(asset_ids)))).all()
for model in models:
meta = _read_ranges(model)
ranges = meta.get(USED_RANGES_KEY) or []
segments: list[tuple[float, float]] = []
for r in ranges:
try:
segments.append((float(r["start"]), float(r["end"])))
except (KeyError, TypeError, ValueError):
continue
if segments:
result[model.id] = segments
return result
def record_used_segments(
db: Session,
asset_id: str,
start: float,
end: float,
plan_id: str,
) -> None:
"""向素材 metadata 追加一条片段使用记录(不 commit)."""
model = db.query(AssetModel).filter(AssetModel.id == asset_id).first()
if model is None:
logger.warning("[片段追踪] 素材不存在,跳过记录: asset_id=%s", asset_id)
return
meta = _read_ranges(model)
ranges = list(meta.get(USED_RANGES_KEY) or [])
ranges.append(
{
"start": round(float(start), 3),
"end": round(float(end), 3),
"plan_id": plan_id,
"created_at": datetime.now(timezone.utc).isoformat(),
}
)
meta[USED_RANGES_KEY] = ranges
model.classification_result = json.dumps(meta, ensure_ascii=False)
model.updated_at = datetime.now(timezone.utc)
def remove_used_segment(
db: Session,
asset_id: str,
start: float,
end: float,
plan_id: str | None = None,
tolerance: float = 0.5,
) -> bool:
"""删除素材 metadata 中匹配的一条使用记录(不 commit).
匹配规则:start/end 与记录值相差不超过 tolerance 秒;plan_id 非空时还需相等。
Returns:
是否找到并删除了记录。
"""
model = db.query(AssetModel).filter(AssetModel.id == asset_id).first()
if model is None:
return False
meta = _read_ranges(model)
ranges = list(meta.get(USED_RANGES_KEY) or [])
remaining: list[dict] = []
removed = False
for r in ranges:
try:
match = (
abs(float(r["start"]) - float(start)) <= tolerance
and abs(float(r["end"]) - float(end)) <= tolerance
)
except (KeyError, TypeError, ValueError):
remaining.append(r)
continue
if plan_id is not None and r.get("plan_id") != plan_id:
match = False
if match and not removed:
removed = True
continue
remaining.append(r)
if removed:
meta[USED_RANGES_KEY] = remaining
model.classification_result = json.dumps(meta, ensure_ascii=False)
model.updated_at = datetime.now(timezone.utc)
return removed
def reset_used_segments(db: Session, asset_id: str) -> None:
"""清空单个素材的历史片段使用记录(不 commit).
单个素材的可用区间被全部占用(轮回一圈)后调用,使后续片段可重新使用整段素材。
"""
model = db.query(AssetModel).filter(AssetModel.id == asset_id).first()
if model is None:
return
meta = _read_ranges(model)
if meta.get(USED_RANGES_KEY):
meta[USED_RANGES_KEY] = []
model.classification_result = json.dumps(meta, ensure_ascii=False)
model.updated_at = datetime.now(timezone.utc)
logger.info("[片段追踪] 素材区间轮回重置: asset_id=%s", asset_id)
def make_reset_callback(db: Session, used_segments: dict) -> Callable[[str], None]:
"""构造给 _calc_random_start_time 用的 reset 回调.
回调同时清空持久化 metadata 和内存中的 used_segments,使重试随机能覆盖全素材。
"""
def _reset(asset_id: str) -> None:
try:
reset_used_segments(db, asset_id)
except Exception:
logger.warning("[片段追踪] reset 持久化记录失败: asset_id=%s", asset_id, exc_info=True)
used_segments.pop(asset_id, None)
return _reset
+14 -1
View File
@@ -12,7 +12,7 @@
from __future__ import annotations
import random
from typing import List
from typing import Callable, List
from packages.domain.edit_plan_clip import EditPlanClip
from packages.domain.editing_mode import EditingMode
@@ -204,6 +204,7 @@ def _calc_random_start_time(
clip_duration: float,
asset_durations: dict[str, float] | None,
used_segments: dict[str, list[tuple[float, float]]] | None = None,
on_exhausted: Callable[[str], None] | None = None,
) -> float | None:
"""计算随机 start_time,避开已使用的时间段.
@@ -216,6 +217,8 @@ def _calc_random_start_time(
clip_duration: 片段时长(秒)
asset_durations: 素材 ID -> 时长映射
used_segments: {asset_id: [(start1, end1), (start2, end2), ...]} 已使用的时间段
on_exhausted: 100 次随机都找不到空闲区间时的回调(入参 asset_id)。
通常用于清空该素材的历史使用记录实现“轮回重置”;回调后会再随机重试一次。
Returns:
随机 start_time 或 None
@@ -255,6 +258,16 @@ def _calc_random_start_time(
if not overlap:
return candidate
# 100 次都找不到空闲区间:触发轮回重置回调(清空历史使用记录)后再随机重试一次
if on_exhausted is not None:
try:
on_exhausted(asset_id)
except Exception:
pass
retry = random.uniform(0.0, max_start)
if not used_segments or asset_id not in used_segments:
return retry
# 如果尝试多次仍找不到,缩短时长使用素材末尾
# 找到最后一个已使用段之后的可用空间
last_used_end = 0.0
+310
View File
@@ -0,0 +1,310 @@
"""素材片段使用记录追踪服务测试(asset_segment_tracker.
覆盖:
- get_used_segments 聚合 metadata 中持久化的区间
- record_used_segments 追加记录(不 commit,保留原有 metadata 字段)
- remove_used_segment 匹配删除(tolerance + plan_id
- reset_used_segments 清空轮回(其他字段不动)
- make_reset_callback 同时清持久化和内存
- _calc_random_start_time 的 on_exhausted 轮回回调
"""
from __future__ import annotations
import json
import os
import sys
from pathlib import Path
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
import pytest
from app.services import asset_segment_tracker as ast
from app.services.asset_segment_tracker import (
get_used_segments,
make_reset_callback,
record_used_segments,
remove_used_segment,
reset_used_segments,
)
from packages.domain.plan_generator_utils import _calc_random_start_time
class FakeModel:
"""模拟 AssetModelid + classification_resultJSON Text+ updated_at。"""
def __init__(self, asset_id: str, meta: dict | None = None):
self.id = asset_id
self.classification_result = json.dumps(meta, ensure_ascii=False) if meta else None
self.updated_at = None
def meta(self) -> dict:
return json.loads(self.classification_result) if self.classification_result else {}
class _InExpr:
def __init__(self, ids, models):
self._ids = ids
self._models = models
def all(self):
return [self._models[i] for i in self._ids if i in self._models]
class _EqExpr:
def __init__(self, target_id, models):
self._target_id = target_id
self._models = models
def first(self):
return self._models.get(self._target_id)
class FakeSession:
"""模拟 dbdb.query(Model).filter(Model.id.in_(ids)).all() / .filter(Model.id == id).first()。
tracker 模块里的 AssetModel 被 monkeypatch 为 FakeModel 类,
这里用挂在类上的伪 column 对象接住 in_ / __eq__。
"""
class _Col:
def __init__(self, models):
self._models = models
def in_(self, ids):
return _InExpr(list(ids), self._models)
def __eq__(self, other):
return _EqExpr(other, self._models)
def __init__(self, models: dict[str, FakeModel]):
self._models = models
self.commits = 0
def query(self, _model):
col = self._Col(self._models)
class _Q:
def filter(self_inner, expr):
return expr
q = _Q()
# 让 tracker 里 AssetModel.id 能取到伪 column
_model.id = col
return q
def commit(self):
self.commits += 1
@pytest.fixture
def patched_model(monkeypatch):
"""把 tracker 模块内的 AssetModel 替换为 FakeModel(供 FakeSession 挂伪 column)。"""
monkeypatch.setattr(ast, "AssetModel", FakeModel)
@pytest.fixture
def models():
return {}
def _db(models):
return FakeSession(models)
# ── get_used_segments ─────────────────────────────────────────────────────────
def test_get_used_segments_aggregates_ranges(patched_model):
models = {
"a1": FakeModel("a1", {"used_time_ranges": [
{"start": 1.0, "end": 5.0, "plan_id": "p1"},
{"start": 9.0, "end": 12.0, "plan_id": "p2"},
]}),
"a2": FakeModel("a2", {"other": 1}), # 无区间记录
"a3": FakeModel("a3"), # metadata 为空
}
db = _db(models)
result = get_used_segments(db, ["a1", "a2", "a3", "missing"])
assert result == {"a1": [(1.0, 5.0), (9.0, 12.0)]}
def test_get_used_segments_empty_input(patched_model):
assert get_used_segments(_db({}), []) == {}
# ── record_used_segments ──────────────────────────────────────────────────────
def test_record_appends_and_no_commit(patched_model):
models = {"a1": FakeModel("a1", {"generation_use_count": 3})}
db = _db(models)
record_used_segments(db, "a1", 2.0, 6.5, "plan-x")
meta = models["a1"].meta()
assert meta["generation_use_count"] == 3 # 原有字段保留
ranges = meta["used_time_ranges"]
assert len(ranges) == 1
assert ranges[0]["start"] == 2.0
assert ranges[0]["end"] == 6.5
assert ranges[0]["plan_id"] == "plan-x"
assert "created_at" in ranges[0]
assert db.commits == 0 # 不自行 commit(事务由调用方控制)
def test_record_multiple_appends_in_order(patched_model):
models = {"a1": FakeModel("a1")}
db = _db(models)
record_used_segments(db, "a1", 0.0, 4.0, "p1")
record_used_segments(db, "a1", 10.0, 14.0, "p1")
ranges = models["a1"].meta()["used_time_ranges"]
assert [r["start"] for r in ranges] == [0.0, 10.0]
def test_record_missing_asset_is_noop(patched_model):
db = _db({})
record_used_segments(db, "ghost", 0.0, 1.0, "p1") # 不抛异常
# ── remove_used_segment ───────────────────────────────────────────────────────
def test_remove_matching_segment(patched_model):
models = {"a1": FakeModel("a1")}
db = _db(models)
record_used_segments(db, "a1", 0.0, 4.0, "p1")
record_used_segments(db, "a1", 10.0, 14.0, "p1")
removed = remove_used_segment(db, "a1", 0.0, 4.0, plan_id="p1")
assert removed is True
ranges = models["a1"].meta()["used_time_ranges"]
assert len(ranges) == 1
assert ranges[0]["start"] == 10.0
def test_remove_not_found_returns_false(patched_model):
models = {"a1": FakeModel("a1")}
db = _db(models)
record_used_segments(db, "a1", 0.0, 4.0, "p1")
assert remove_used_segment(db, "a1", 99.0, 100.0, plan_id="p1") is False
def test_remove_respects_tolerance(patched_model):
models = {
"a1": FakeModel("a1", {"used_time_ranges": [
{"start": 5.0, "end": 9.0, "plan_id": "p1"}
]}),
"a2": FakeModel("a2", {"used_time_ranges": [
{"start": 5.0, "end": 9.0, "plan_id": "p1"}
]}),
}
db = _db(models)
# 偏差 0.3 秒,在 tolerance=0.5 内 → 删除成功
assert remove_used_segment(db, "a1", 5.3, 8.7, plan_id="p1") is True
# 偏差 2 秒,超出 tolerance → 删除失败
assert remove_used_segment(db, "a2", 7.0, 11.0, plan_id="p1") is False
def test_remove_plan_id_must_match(patched_model):
models = {"a1": FakeModel("a1", {"used_time_ranges": [
{"start": 5.0, "end": 9.0, "plan_id": "plan-A"}
]})}
db = _db(models)
# 时间匹配但 plan_id 不同 → 不删除
assert remove_used_segment(db, "a1", 5.0, 9.0, plan_id="plan-B") is False
assert len(models["a1"].meta()["used_time_ranges"]) == 1
# ── reset_used_segments ───────────────────────────────────────────────────────
def test_reset_clears_ranges_keeps_other_fields(patched_model):
models = {"a1": FakeModel("a1", {
"generation_use_count": 9,
"used_time_ranges": [{"start": 1, "end": 2}],
})}
db = _db(models)
reset_used_segments(db, "a1")
meta = models["a1"].meta()
assert meta["used_time_ranges"] == []
assert meta["generation_use_count"] == 9
assert db.commits == 0
# ── make_reset_callback ───────────────────────────────────────────────────────
def test_reset_callback_clears_persisted_and_memory(patched_model):
models = {"a1": FakeModel("a1", {"used_time_ranges": [
{"start": 0, "end": 30}
]})}
db = _db(models)
used_segments = {"a1": [(0.0, 30.0)], "a2": [(1.0, 2.0)]}
cb = make_reset_callback(db, used_segments)
cb("a1")
assert "a1" not in used_segments # 内存清空
assert "a2" in used_segments # 其他素材不受影响
assert models["a1"].meta()["used_time_ranges"] == []
# ── _calc_random_start_time 轮回回调 ──────────────────────────────────────────
def test_calc_random_start_invokes_reset_when_exhausted():
"""素材区间被占满(100 次随机必重叠)→ 触发 on_exhausted,重置后重试成功。"""
durations = {"a1": 30.0}
used = {"a1": [(0.0, 10.0), (10.0, 20.0), (20.0, 30.0)]}
reset_called = []
def _on_exhausted(asset_id):
reset_called.append(asset_id)
used.pop(asset_id, None) # 模拟轮回清空
result = _calc_random_start_time(
"a1", 10.0, durations, used, on_exhausted=_on_exhausted
)
assert reset_called == ["a1"]
assert result is not None
assert 0.0 <= result <= 20.0 # max_start = 30 - 10
def test_calc_random_start_no_callback_keeps_legacy_fallback():
"""不传 on_exhausted 时保持旧降级行为,不报错。"""
durations = {"a1": 30.0}
used = {"a1": [(0.0, 10.0), (10.0, 20.0), (20.0, 30.0)]}
result = _calc_random_start_time("a1", 10.0, durations, used)
assert result is not None
def test_calc_random_start_with_space_does_not_reset():
"""有充足空闲区间时不触发 reset。"""
durations = {"a1": 100.0}
used = {"a1": [(0.0, 50.0)]}
reset_called = []
result = _calc_random_start_time(
"a1", 5.0, durations, used, on_exhausted=lambda aid: reset_called.append(aid)
)
assert reset_called == []
assert result is not None
+29 -1
View File
@@ -81,6 +81,34 @@ def _get_clips_data_from_call(mock_plan_svc):
return call_args.kwargs.get("clips_data", [])
@pytest.fixture(autouse=True)
def _mock_segment_tracker():
"""from-assets 现在会读/写素材 metadata 的片段区间记录,测试中 mock 掉避免依赖真实 DB。
get_used_segments 返回空 dict(等价历史行为:无历史区间);
record/remove/reset 回调均无副作用。
"""
with (
patch(
"app.api.routes.templates_editor.clips.get_used_segments",
return_value={},
),
patch(
"app.api.routes.templates_editor.clips.record_used_segments",
return_value=None,
),
patch(
"app.api.routes.templates_editor.clips.make_reset_callback",
return_value=lambda asset_id: None,
),
patch(
"app.api.routes.templates_editor.clips.remove_used_segment",
return_value=False,
),
):
yield
class TestEditorClipsBySegments:
"""测试按 segment 数量创建片段 + 素材轮询。"""
@@ -387,7 +415,7 @@ class TestEditorClipsDurationAndStartTime:
captured_used_segments = []
def fake_calc(asset_id, clip_duration, asset_durations, used_segments):
def fake_calc(asset_id, clip_duration, asset_durations, used_segments, on_exhausted=None):
captured_used_segments.append({aid: list(segs) for aid, segs in (used_segments or {}).items()})
return (len(captured_used_segments) - 1) * 5.0