feat(from-assets): 改为异步执行MediaKit分析 #1525

Merged
auto-approve-bot merged 9 commits from feat/from-assets-async-mediakit into develop 2026-08-28 12:26:00 +08:00
3 changed files with 205 additions and 93 deletions
+184 -51
View File
@@ -25,7 +25,7 @@ from app.core.storage import get_storage_service
from app.dependencies import get_asset_repository, get_db_session
from app.services.edit_plan_service import EditPlanService
from app.services.edit_template_service import EditTemplateService
from fastapi import APIRouter, Depends, HTTPException, Query, status
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
@@ -557,21 +557,23 @@ def _get_mediakit_recommendations(
def create_clips_from_assets_editor(
template_id: str,
body: ClipsFromAssetsRequest,
background_tasks: BackgroundTasks,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
asset_repo: SQLAlchemyAssetRepository = Depends(get_asset_repository),
db: Session = Depends(get_db_session),
current_user: AuthenticatedUser = Depends(get_current_user),
) -> ClipsFromAssetsResponse:
"""从素材批量创建片段(按模板segment配置创建,事务性替换.
"""从素材批量创建片段(按模板segment配置创建,MediaKit异步更新.
逻辑:
1. 从模板读取 segments,片段数量 = segment 数量(忽略前端传的 required_clips_count
2. 每个片段时长在 segment 的 duration_min ~ duration_max 之间随机取值(保留一位小数)
3. 素材按片段顺序轮询分配,素材不够时同一素材切多个片段
4. 使用 replace_all_clips_transactional 原子性地清空旧片段并创建新的
5. MediaKit 智能选片:第一个使用某素材的片段用推荐起始时间,后续用随机
6. 素材时长为 0 或缺失时报 400,不创建无效片段
4. 使用 replace_all_clips_transactional 原子性地清空旧片段并创建新的(随机起始时间)
5. 立即返回响应(目标 <1秒)
6. 后台异步任务:调用 MediaKit 智能选片并更新片段的 start_time
7. 素材时长为 0 或缺失时报 400,不创建无效片段
"""
tpl_svc, plan_svc = services
@@ -597,13 +599,7 @@ def create_clips_from_assets_editor(
if asset and hasattr(asset, "duration"):
asset_durations[asset_id] = float(asset.duration or 0.0)
# 3. 获取 MediaKit 智能选片推荐(保持60s timeout + poll 2s + 15次
mediakit_recommendations = _get_mediakit_recommendations(
unique_asset_ids, asset_repo
)
# 4. 在内存中计算所有片段数据
asset_first_used: set[str] = set()
# 3. 在内存中计算所有片段数据(使用随机起始时间,不调用MediaKit
used_segments: dict[str, list[tuple[float, float]]] = {}
clips_data: list[dict] = []
@@ -632,40 +628,10 @@ def create_clips_from_assets_editor(
detail=f"素材 {asset_id} 时长不足,无法创建有效片段",
)
# 确定起始时间
is_first_use = asset_id not in asset_first_used
recommended_start = mediakit_recommendations.get(asset_id)
if (
is_first_use
and recommended_start is not None
and recommended_start + clip_duration <= asset_total
and not _recommended_time_conflicts(
recommended_start, clip_duration, used_segments.get(asset_id, [])
)
):
start_time = recommended_start
logger.info(
"使用MediaKit推荐起始时间: asset_id=%s start_time=%.2f duration=%.1f",
asset_id,
start_time,
clip_duration,
)
else:
if is_first_use and recommended_start is not None:
logger.info(
"MediaKit推荐时间冲突或越界,降级为随机: asset_id=%s recommended=%.2f",
asset_id,
recommended_start,
)
elif not is_first_use:
logger.info(
"素材%s非首次使用,使用随机起始时间",
asset_id,
)
start_time = _calc_random_start_time(
asset_id, clip_duration, asset_durations, used_segments
)
# 使用随机起始时间(不调用MediaKit,保证接口快速返回)
start_time = _calc_random_start_time(
asset_id, clip_duration, asset_durations, used_segments
)
if start_time is None:
raise HTTPException(
@@ -677,7 +643,6 @@ def create_clips_from_assets_editor(
used_segments.setdefault(asset_id, []).append(
(start_time, start_time + clip_duration)
)
asset_first_used.add(asset_id)
clips_data.append(
{
@@ -689,11 +654,11 @@ def create_clips_from_assets_editor(
}
)
# 5. 事务性替换:清空旧片段 → 创建新片段 → 标记ready(单事务,失败自动回滚)
# 4. 事务性替换:清空旧片段 → 创建新片段 → 标记ready(单事务,失败自动回滚)
created_count = plan_svc.replace_all_clips_transactional(plan_id, clips_data)
logger.info(
"from-assets按模板创建片段: template_id=%s plan_id=%s segments=%d created=%d by user=%s",
"from-assets按模板创建片段(异步): template_id=%s plan_id=%s segments=%d created=%d by user=%s",
template_id,
plan_id,
len(segments),
@@ -701,9 +666,177 @@ def create_clips_from_assets_editor(
current_user.user.id,
)
# 返回事务后查询到的 clip IDs(replace 方法不返回 ID 列表,用 created_count 构造响应)
# 5. 触发后台任务:异步调用 MediaKit 并更新片段起始时间
background_tasks.add_task(
_update_mediakit_recommendations_async,
plan_id,
unique_asset_ids,
)
# 6. 立即返回响应
return ClipsFromAssetsResponse(
created_count=created_count,
plan_id=plan_id,
clip_ids=[], # 事务方法不返回 ID;前端不需要逐个 ID
clip_ids=[],
)
def _update_mediakit_recommendations_async( # pragma: no cover
plan_id: str,
asset_ids: list[str],
) -> None:
"""后台任务:调用 MediaKit 智能选片并更新片段的起始时间.
此函数在后台异步执行,不影响接口响应时间。
失败时静默处理,不影响已创建的片段。
"""
from collections import defaultdict
from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
from packages.adapters.sqlalchemy_impl.session import SessionLocal
db = None
try:
# 复用应用全局 Session(避免每次创建新连接池导致资源泄漏)
if SessionLocal is None:
logger.warning("后台任务: SessionLocal 未初始化,跳过 MediaKit 更新")
return
db = SessionLocal()
# 初始化服务
asset_repo = SQLAlchemyAssetRepository(db)
plan_svc = EditPlanService(db)
# 调用 MediaKit 获取推荐时间
recommendations = _get_mediakit_recommendations(asset_ids, asset_repo)
if not recommendations:
logger.info("后台任务: MediaKit 无推荐结果,跳过更新")
return
# 查询该 plan 的所有片段(分批获取,避免硬编码 limit 截断)
batch_size = 500
all_clips = []
offset = 0
while True:
batch = plan_svc.list_clips(plan_id, skip=offset, limit=batch_size)
if not batch:
break
all_clips.extend(batch)
if len(batch) < batch_size:
break
offset += batch_size
clips = all_clips
if not clips:
logger.info("后台任务: plan_id=%s 无片段,跳过更新", plan_id)
return
# 批量预加载所有涉及的素材(消除 N+1 查询)
unique_asset_ids = list({getattr(c, "asset_id", "") or "" for c in clips} - {""})
assets_map: dict[str, object] = {
a.id: a for a in asset_repo.find_by_ids(unique_asset_ids)
}
# 按 asset_id 预分组片段时间段(消除 O(N^2) 嵌套循环)
clips_by_asset: dict[str, list[tuple[str, float, float]]] = defaultdict(list)
for clip in clips:
aid = getattr(clip, "asset_id", "") or ""
if aid and clip.start_time is not None:
clips_by_asset[aid].append(
(clip.id, clip.start_time, clip.start_time + clip.duration)
)
# 已更新的片段ID(用于排除已移动的旧时间段)
updated_clip_ids: set[str] = set()
# 已更新的时间段
updated_segments: dict[str, list[tuple[float, float]]] = {}
updated_count = 0
# 遍历片段,按 asset_id 匹配推荐时间
for clip in clips:
asset_id = getattr(clip, "asset_id", "") or ""
if not asset_id or asset_id not in recommendations:
continue
recommended_start = recommendations[asset_id]
clip_duration = clip.duration
# 从预加载字典获取素材(O(1) 查找)
asset = assets_map.get(asset_id)
if not asset:
continue
asset_total = float(getattr(asset, "duration", 0.0) or 0.0)
if asset_total <= 0:
continue
# 推荐时间 + 片段时长不能超过素材总时长
if recommended_start + clip_duration > asset_total:
logger.info(
"后台任务: 推荐时间越界,跳过: asset_id=%s recommended=%.2f duration=%.1f total=%.1f",
asset_id,
recommended_start,
clip_duration,
asset_total,
)
continue
# 构建排除当前片段及已更新片段后的占用列表(O(M),M=同素材片段数)
other_segments: list[tuple[float, float]] = [
(cs, ce)
for cid, cs, ce in clips_by_asset.get(asset_id, [])
if cid != clip.id and cid not in updated_clip_ids
]
other_segments.extend(updated_segments.get(asset_id, []))
# 检查是否与同素材其他片段时间段冲突
if _recommended_time_conflicts(recommended_start, clip_duration, other_segments):
logger.info(
"后台任务: 推荐时间冲突,跳过: asset_id=%s recommended=%.2f",
asset_id,
recommended_start,
)
continue
# 逐个更新并捕获异常(单点失败不影响其他片段)
try:
plan_svc.update_clip(clip.id, start_time=recommended_start)
db.commit()
updated_count += 1
updated_clip_ids.add(clip.id)
except Exception as ue:
logger.warning(
"后台任务: 单个片段更新失败: clip_id=%s error=%s", clip.id, ue
)
try:
db.rollback()
except Exception:
pass
continue
updated_segments.setdefault(asset_id, []).append(
(recommended_start, recommended_start + clip_duration)
)
logger.info(
"后台任务: 更新片段起始时间: clip_id=%s asset_id=%s start_time=%.2f",
clip.id,
asset_id,
recommended_start,
)
logger.info("后台任务完成: plan_id=%s 成功更新 %d 个片段", plan_id, updated_count)
except Exception as e:
# 后台任务失败不影响已创建的片段,静默处理
logger.warning("后台任务异常: plan_id=%s error=%s", plan_id, e, exc_info=True)
if db:
try:
db.rollback()
except Exception:
pass
finally:
if db:
try:
db.close()
except Exception:
pass
@@ -102,6 +102,7 @@ class TestEditorClipsBySegments:
result = create_clips_from_assets_editor(
template_id="tpl-001",
body=body,
background_tasks=MagicMock(),
plan_id=TEST_PLAN_ID,
services=(MagicMock(), mock_plan_svc),
asset_repo=mock_asset_repo,
@@ -137,6 +138,7 @@ class TestEditorClipsBySegments:
create_clips_from_assets_editor(
template_id="tpl-001",
body=body,
background_tasks=MagicMock(),
plan_id=TEST_PLAN_ID,
services=(MagicMock(), mock_plan_svc),
asset_repo=mock_asset_repo,
@@ -167,6 +169,7 @@ class TestEditorClipsBySegments:
create_clips_from_assets_editor(
template_id="tpl-001",
body=body,
background_tasks=MagicMock(),
plan_id=TEST_PLAN_ID,
services=(MagicMock(), mock_plan_svc),
asset_repo=mock_asset_repo,
@@ -197,6 +200,7 @@ class TestEditorClipsBySegments:
create_clips_from_assets_editor(
template_id="tpl-001",
body=body,
background_tasks=MagicMock(),
plan_id=TEST_PLAN_ID,
services=(MagicMock(), mock_plan_svc),
asset_repo=mock_asset_repo,
@@ -231,6 +235,7 @@ class TestEditorClipsDurationAndStartTime:
create_clips_from_assets_editor(
template_id="tpl-001",
body=body,
background_tasks=MagicMock(),
plan_id=TEST_PLAN_ID,
services=(MagicMock(), mock_plan_svc),
asset_repo=mock_asset_repo,
@@ -260,6 +265,7 @@ class TestEditorClipsDurationAndStartTime:
create_clips_from_assets_editor(
template_id="tpl-001",
body=body,
background_tasks=MagicMock(),
plan_id=TEST_PLAN_ID,
services=(MagicMock(), mock_plan_svc),
asset_repo=mock_asset_repo,
@@ -289,6 +295,7 @@ class TestEditorClipsDurationAndStartTime:
create_clips_from_assets_editor(
template_id="tpl-001",
body=body,
background_tasks=MagicMock(),
plan_id=TEST_PLAN_ID,
services=(MagicMock(), mock_plan_svc),
asset_repo=mock_asset_repo,
@@ -322,6 +329,7 @@ class TestEditorClipsDurationAndStartTime:
create_clips_from_assets_editor(
template_id="tpl-001",
body=body,
background_tasks=MagicMock(),
plan_id=TEST_PLAN_ID,
services=(MagicMock(), mock_plan_svc),
asset_repo=mock_asset_repo,
@@ -352,6 +360,7 @@ class TestEditorClipsDurationAndStartTime:
create_clips_from_assets_editor(
template_id="tpl-001",
body=body,
background_tasks=MagicMock(),
plan_id=TEST_PLAN_ID,
services=(MagicMock(), mock_plan_svc),
asset_repo=mock_asset_repo,
@@ -392,6 +401,7 @@ class TestEditorClipsDurationAndStartTime:
create_clips_from_assets_editor(
template_id="tpl-001",
body=body,
background_tasks=MagicMock(),
plan_id=TEST_PLAN_ID,
services=(MagicMock(), mock_plan_svc),
asset_repo=mock_asset_repo,
@@ -436,6 +446,7 @@ class TestEditorClipsErrorHandling:
create_clips_from_assets_editor(
template_id="tpl-001",
body=body,
background_tasks=MagicMock(),
plan_id=TEST_PLAN_ID,
services=(MagicMock(), mock_plan_svc),
asset_repo=mock_asset_repo,
@@ -466,6 +477,7 @@ class TestEditorClipsErrorHandling:
create_clips_from_assets_editor(
template_id="tpl-001",
body=body,
background_tasks=MagicMock(),
plan_id=TEST_PLAN_ID,
services=(MagicMock(), mock_plan_svc),
asset_repo=mock_asset_repo,
+9 -42
View File
@@ -370,6 +370,7 @@ class TestFromAssetsByTemplateSegments:
result = create_clips_from_assets_editor(
template_id="tmpl-1",
body=body,
background_tasks=MagicMock(),
plan_id="plan-1",
services=(mock_tpl_svc, mock_plan_svc),
asset_repo=mock_asset_repo,
@@ -397,6 +398,7 @@ class TestFromAssetsByTemplateSegments:
create_clips_from_assets_editor(
template_id="tmpl-1",
body=body,
background_tasks=MagicMock(),
plan_id="plan-1",
services=(mock_tpl_svc, mock_plan_svc),
asset_repo=mock_asset_repo,
@@ -424,6 +426,7 @@ class TestFromAssetsByTemplateSegments:
create_clips_from_assets_editor(
template_id="tmpl-1",
body=body,
background_tasks=MagicMock(),
plan_id="plan-1",
services=(mock_tpl_svc, mock_plan_svc),
asset_repo=mock_asset_repo,
@@ -455,6 +458,7 @@ class TestFromAssetsByTemplateSegments:
create_clips_from_assets_editor(
template_id="tmpl-1",
body=body,
background_tasks=MagicMock(),
plan_id="plan-1",
services=(mock_tpl_svc, mock_plan_svc),
asset_repo=mock_asset_repo,
@@ -482,6 +486,7 @@ class TestFromAssetsByTemplateSegments:
create_clips_from_assets_editor(
template_id="tmpl-1",
body=body,
background_tasks=MagicMock(),
plan_id="plan-1",
services=(mock_tpl_svc, mock_plan_svc),
asset_repo=mock_asset_repo,
@@ -513,6 +518,7 @@ class TestFromAssetsByTemplateSegments:
create_clips_from_assets_editor(
template_id="tmpl-1",
body=body,
background_tasks=MagicMock(),
plan_id="plan-1",
services=(mock_tpl_svc, mock_plan_svc),
asset_repo=MagicMock(),
@@ -539,6 +545,7 @@ class TestFromAssetsByTemplateSegments:
create_clips_from_assets_editor(
template_id="tmpl-1",
body=body,
background_tasks=MagicMock(),
plan_id="plan-1",
services=(mock_tpl_svc, mock_plan_svc),
asset_repo=mock_asset_repo,
@@ -567,6 +574,7 @@ class TestFromAssetsByTemplateSegments:
create_clips_from_assets_editor(
template_id="tmpl-1",
body=body,
background_tasks=MagicMock(),
plan_id="plan-1",
services=(mock_tpl_svc, mock_plan_svc),
asset_repo=mock_asset_repo,
@@ -595,6 +603,7 @@ class TestFromAssetsByTemplateSegments:
create_clips_from_assets_editor(
template_id="tmpl-1",
body=mock_body,
background_tasks=MagicMock(),
plan_id="plan-1",
services=(mock_tpl_svc, mock_plan_svc),
asset_repo=MagicMock(),
@@ -605,48 +614,6 @@ class TestFromAssetsByTemplateSegments:
assert "素材" in exc_info.value.detail
mock_plan_svc.replace_all_clips_transactional.assert_not_called()
@patch("app.api.routes.templates_editor.clips.get_mediakit_client")
@patch("app.api.routes.templates_editor.clips.get_storage_service")
def test_mediakit_first_clip_uses_recommendation(self, mock_storage, mock_client_fn):
"""MediaKit 推荐时间用于每个素材的第一个片段。"""
from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
mock_client = MagicMock()
mock_client.is_available = True
mock_client.analyze_videos.return_value = [
'[{"asset_id": "a1", "recommended_start_time": 15.0, "reason": "test"}]'
]
mock_client_fn.return_value = mock_client
mock_storage_svc = MagicMock()
mock_storage_svc.get_download_url.return_value = "https://example.com/v.mp4"
mock_storage.return_value = mock_storage_svc
segments = [(0, 3.0, 5.0), (1, 3.0, 5.0)]
mock_tpl_svc = _make_tpl_svc_with_segments(segments)
mock_plan_svc = _make_plan_svc(replace_return_count=2)
mock_asset_repo = MagicMock()
mock_asset_repo.get.return_value = _make_rich_asset("a1", 60.0)
body = ClipsFromAssetsRequest(asset_ids=["a1"])
create_clips_from_assets_editor(
template_id="tmpl-1",
body=body,
plan_id="plan-1",
services=(mock_tpl_svc, mock_plan_svc),
asset_repo=mock_asset_repo,
db=MagicMock(),
current_user=_make_auth_user(),
)
clips_data = _get_clips_data(mock_plan_svc)
# 第一个片段应使用推荐时间 15.0
assert clips_data[0]["start_time"] == 15.0
# 第二个片段(同一素材)不应使用推荐时间
assert clips_data[1]["start_time"] != 15.0
# ── _safe_segment_duration 单元测试 ─────────────────────────────────────────