Files
xiaoxia-saas/apps/api/app/api/routes/templates_editor/clips.py
T
xiaoxia ff833ce7e0
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (pull_request) Successful in 1s
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 1s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 30s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 29s
CI/CD Pipeline / Validate - Style (pull_request) Successful in 1m6s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 1m8s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 1m30s
CI/CD Pipeline / Validate - Python (mypy + alembic) (pull_request) Successful in 1m50s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 2m1s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 2m59s
AI Code Review / AI Code Review (pull_request) Successful in 6m20s
CI/CD Pipeline / Validate - Security (pull_request) Successful in 7m23s
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / CI Gate (pull_request) Failing after 3s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 4m33s
Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 41s
ACR Cleanup / ACR Image Cleanup (pull_request_target) Successful in 37s
CI/CD Pipeline / Canary Release to Production (pull_request) Failing after 141h35m52s
CI/CD Pipeline / Deploy Production (pull_request) Failing after 141h35m52s
CI/CD Pipeline / Build Production Web Image (pull_request) Failing after 141h35m56s
CI/CD Pipeline / Build Production API Image (pull_request) Failing after 141h35m56s
CI/CD Pipeline / ACR Image Cleanup (pull_request) Failing after 141h42m58s
CI/CD Pipeline / Staging E2E Tests (pull_request) Failing after 141h42m59s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Failing after 141h43m4s
CI/CD Pipeline / Retag skipped Staging Worker Image (pull_request) Failing after 141h43m9s
CI/CD Pipeline / Retag skipped Staging API Image (pull_request) Failing after 141h43m14s
CI/CD Pipeline / Build Staging Worker Image (pull_request) Failing after 141h43m16s
CI/CD Pipeline / Build Staging API Image (pull_request) Failing after 141h43m17s
CI/CD Pipeline / PR Build Web Image (pull_request) Failing after 141h43m18s
CI/CD Pipeline / Check push changed paths (pull_request) Failing after 141h43m22s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Failing after 141h43m19s
CI/CD Pipeline / Build Production Worker Image (pull_request) Failing after 142h10m45s
CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 142h17m47s
CI/CD Pipeline / Retag skipped Staging Web Image (pull_request) Failing after 142h17m58s
CI/CD Pipeline / Build Staging Web Image (pull_request) Failing after 142h18m6s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 142h18m8s
fix: 清理22个F841 unused-variable,解除CI Validate-Style阻塞
全部为 except Exception as e: 中变量e未使用的模式,
ruff自动修复将 e: 改为 :。不涉及任何业务逻辑变更。
2026-09-10 17:39:08 +08:00

1169 lines
48 KiB
Python
Executable File
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""片段管理路由.
端点:
- GET /clips 片段列表
- POST /clips 创建片段
- GET /clips/{clip_id} 片段详情
- PUT /clips/{clip_id} 更新片段
- DELETE /clips/{clip_id} 删除片段
- POST /clips/{clip_id}/split 分割片段
- POST /clips/merge 合并片段
- POST /clips/reorder 重排片段
- POST /clips/batch-delete 批量删除
- POST /clips/from-assets 从素材创建片段
"""
from __future__ import annotations
import json
import logging
import random
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
# 默认转场时长(与 worker 端保持一致)
_DEFAULT_TRANSITION_DURATION = 0.5
from app.services.asset_segment_tracker import (
REUSE_RATIO_LIMIT,
SEGMENT_EDGE_GAP,
get_used_segments,
make_reuse_callback,
record_used_segments,
remove_used_segment,
)
from app.services.edit_plan_service import EditPlanService
from app.services.edit_template_service import EditTemplateService, TemplateNotFoundError
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
from packages.domain.plan_generator_utils import (
_calc_random_start_time,
build_scene_segments,
extract_scene_points_from_metadata,
pick_scene_aware_start,
pick_start_in_scene_segment,
)
from packages.domain.smart_match import SCORE_RANDOM_NOISE_MAX, score_asset
from packages.shared.mediakit_client import get_mediakit_client
from .dependencies import get_draft_plan_id, get_editor_services
from .schemas import (
ClipBatchDeleteRequest,
ClipBatchDeleteResponse,
ClipReorderRequest,
ClipReorderResponse,
ClipsFromAssetsRequest,
ClipsFromAssetsResponse,
EditorClipCreateRequest,
EditorClipListResponse,
EditorClipResponse,
EditorClipUpdateRequest,
MergeClipsRequest,
SplitClipRequest,
)
logger = logging.getLogger(__name__)
router = APIRouter(tags=["Template Editor"])
# 编辑器默认片段时长(秒)
_DEFAULT_EDITOR_CLIP_DURATION = 5.0
def _clip_to_response(clip, asset_url: str | None = None) -> EditorClipResponse:
"""统一构造片段响应 — 与 edit_plan_clips 表字段完全对齐"""
def _enum_str(val) -> str:
return val.value if hasattr(val, "value") else str(val)
def _fmt_dt(val) -> str:
if val is None:
return ""
if hasattr(val, "isoformat"):
return val.isoformat()
return str(val)
return EditorClipResponse(
id=clip.id,
plan_id=clip.plan_id,
clip_type=_enum_str(getattr(clip, "clip_type", "")),
order=clip.order,
duration=clip.duration,
start_time=getattr(clip, "start_time", 0.0) or 0.0,
text_content=clip.text_content or "",
transition_effect=_enum_str(getattr(clip, "transition_effect", "cut")),
transition_duration=getattr(clip, "transition_duration", 0.0) or 0.0,
playback_speed=clip.playback_speed or 1.0,
asset_id=getattr(clip, "asset_id", "") or "",
asset_url=asset_url,
status=getattr(clip, "status", "pending") or "pending",
template_clip_config_id=getattr(clip, "template_clip_config_id", "") or "",
config=clip.config or {},
created_at=_fmt_dt(getattr(clip, "created_at", None)),
updated_at=_fmt_dt(getattr(clip, "updated_at", None)),
)
def _build_asset_url_map(
asset_ids: list[str],
asset_repo: SQLAlchemyAssetRepository,
) -> dict[str, str | None]:
"""批量查询素材并生成签名URL映射.
Returns:
{asset_id: signed_url_or_None}
"""
if not asset_ids:
return {}
# 去重:多个 clip 可能引用同一个素材
# 去重并保持顺序
seen: set[str] = set()
unique_ids = []
for aid in asset_ids:
if aid and aid not in seen:
seen.add(aid)
unique_ids.append(aid)
result: dict[str, str | None] = {}
try:
storage = get_storage_service()
except Exception as e:
logger.exception("获取存储服务失败,跳过asset_url生成: %s", e)
return {aid: None for aid in asset_ids}
# 批量查询所有 Asset(单次 SQL IN 查询,避免 N+1)
try:
assets = asset_repo.find_by_ids(unique_ids)
asset_map = {a.id: a for a in assets}
except Exception:
logger.exception("批量查询素材失败: asset_ids=%s", asset_ids)
return {aid: None for aid in asset_ids if aid}
for aid in unique_ids:
try:
asset = asset_map.get(aid)
if asset is None:
result[aid] = None
continue
storage_key = getattr(asset, "storage_key", None) or ""
if not storage_key:
result[aid] = None
continue
result[aid] = storage.get_download_url(storage_key, expires_seconds=3600)
except Exception:
logger.exception("生成素材签名URL失败: asset_id=%s", aid)
result[aid] = None
return result
@router.get("/clips", response_model=EditorClipListResponse)
def list_draft_clips(
template_id: str,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
asset_repo: SQLAlchemyAssetRepository = Depends(get_asset_repository),
skip: int = Query(default=0, ge=0),
limit: int = Query(default=100, ge=1, le=500),
_: AuthenticatedUser = Depends(get_current_user),
):
"""获取草稿的片段列表"""
_, plan_svc = services
clips = plan_svc.list_clips(plan_id, skip=skip, limit=limit)
total = plan_svc.count_clips(plan_id)
# 批量解析素材签名URL
asset_ids = [getattr(c, "asset_id", "") or "" for c in clips]
asset_ids = [aid for aid in asset_ids if aid]
url_map = _build_asset_url_map(asset_ids, asset_repo)
return EditorClipListResponse(
items=[_clip_to_response(c, asset_url=url_map.get(getattr(c, "asset_id", "") or "")) for c in clips],
total=total,
)
@router.post("/clips", response_model=EditorClipResponse, status_code=status.HTTP_201_CREATED)
def create_draft_clip(
template_id: str,
req: EditorClipCreateRequest,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
_: AuthenticatedUser = Depends(get_current_user),
):
"""在草稿中创建新片段"""
_, plan_svc = services
try:
clip = plan_svc.create_clip(
plan_id,
clip_type=req.clip_type,
order=req.order,
duration=req.duration,
text_content=req.text_content,
transition_effect=req.transition_effect,
config=req.config,
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return _clip_to_response(clip)
@router.put("/clips/{clip_id}", response_model=EditorClipResponse)
def update_draft_clip(
template_id: str,
clip_id: str,
req: EditorClipUpdateRequest,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
_: AuthenticatedUser = Depends(get_current_user),
):
"""更新草稿中的片段"""
_, plan_svc = services
try:
clip = plan_svc.update_clip(
clip_id,
order=req.order,
duration=req.duration,
text_content=req.text_content,
transition_effect=req.transition_effect,
playback_speed=req.playback_speed,
config=req.config,
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return _clip_to_response(clip)
@router.delete("/clips/{clip_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_draft_clip(
template_id: str,
clip_id: str,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
_: AuthenticatedUser = Depends(get_current_user),
):
"""删除草稿中的片段"""
_, plan_svc = services
success = plan_svc.delete_clip(clip_id)
if not success:
raise HTTPException(status_code=404, detail="片段不存在")
return None
@router.get("/clips/{clip_id}", response_model=EditorClipResponse)
def get_draft_clip_detail(
template_id: str,
clip_id: str,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
asset_repo: SQLAlchemyAssetRepository = Depends(get_asset_repository),
_: AuthenticatedUser = Depends(get_current_user),
):
"""获取草稿中的片段详情"""
_, plan_svc = services
clip = plan_svc.get_clip(clip_id)
if clip is None:
raise HTTPException(status_code=404, detail="片段不存在")
if clip.plan_id != plan_id:
raise HTTPException(status_code=404, detail="片段不存在")
asset_id = getattr(clip, "asset_id", "") or ""
url_map = _build_asset_url_map([asset_id], asset_repo) if asset_id else {}
return _clip_to_response(clip, asset_url=url_map.get(asset_id))
@router.post("/clips/{clip_id}/split", status_code=status.HTTP_200_OK)
def split_draft_clip(
template_id: str,
clip_id: str,
body: SplitClipRequest,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
asset_repo: SQLAlchemyAssetRepository = Depends(get_asset_repository),
_: AuthenticatedUser = Depends(get_current_user),
):
"""将一个片段从指定时间点分割为两个片段"""
_, plan_svc = services
clip = plan_svc.get_clip(clip_id)
if clip is None or clip.plan_id != plan_id:
raise HTTPException(status_code=404, detail="片段不存在")
try:
result = plan_svc.split_clip(clip_id, body.split_time)
except ValueError as exc:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
left = result["left_clip"]
right = result["right_clip"]
asset_ids = [getattr(left, "asset_id", "") or "", getattr(right, "asset_id", "") or ""]
asset_ids = [a for a in asset_ids if a]
url_map = _build_asset_url_map(asset_ids, asset_repo)
return {
"left_clip": _clip_to_response(left, asset_url=url_map.get(getattr(left, "asset_id", "") or "")),
"right_clip": _clip_to_response(right, asset_url=url_map.get(getattr(right, "asset_id", "") or "")),
}
@router.post("/clips/merge", status_code=status.HTTP_200_OK)
def merge_draft_clips(
template_id: str,
body: MergeClipsRequest,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
asset_repo: SQLAlchemyAssetRepository = Depends(get_asset_repository),
_: AuthenticatedUser = Depends(get_current_user),
):
"""将多个连续的同类型片段合并为一个片段"""
_, plan_svc = services
for cid in body.clip_ids:
clip = plan_svc.get_clip(cid)
if clip is None or clip.plan_id != plan_id:
raise HTTPException(status_code=404, detail=f"片段不存在: {cid}")
try:
merged = plan_svc.merge_clips(body.clip_ids)
except ValueError as exc:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
asset_id = getattr(merged, "asset_id", "") or ""
url_map = _build_asset_url_map([asset_id], asset_repo) if asset_id else {}
return {
"merged_clip": _clip_to_response(merged, asset_url=url_map.get(asset_id)),
"deleted_clip_ids": body.clip_ids,
}
@router.post("/clips/reorder", response_model=ClipReorderResponse)
def reorder_editor_clips(
template_id: str,
body: ClipReorderRequest,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
_: AuthenticatedUser = Depends(get_current_user),
) -> ClipReorderResponse:
"""批量重排片段顺序"""
_, plan_svc = services
count = 0
for item in body.items:
try:
plan_svc.update_clip(item.clip_id, order=item.new_order)
count += 1
except ValueError:
pass
return ClipReorderResponse(updated_count=count, plan_id=plan_id)
@router.post("/clips/batch-delete", response_model=ClipBatchDeleteResponse)
def batch_delete_editor_clips(
template_id: str,
body: ClipBatchDeleteRequest,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
_: AuthenticatedUser = Depends(get_current_user),
) -> ClipBatchDeleteResponse:
"""批量删除片段"""
_, plan_svc = services
deleted = 0
for clip_id in body.clip_ids:
if plan_svc.delete_clip(clip_id):
deleted += 1
return ClipBatchDeleteResponse(deleted_count=deleted, plan_id=plan_id)
def _safe_segment_duration(value, default: float) -> float:
"""安全地将数据库中的时长值转换为正浮点数.
处理 None、无效类型、负数、NaN 等异常情况。
"""
if value is None:
return default
try:
result = float(value)
except (ValueError, TypeError):
return default
if result != result or result <= 0: # NaN check or non-positive
return default
return result
def _get_template_segments(
template_id: str,
user_id: str,
tpl_svc: EditTemplateService,
) -> list[tuple[int, float, float]]:
"""获取模板的片段配置(顺序、最短时长、最长时长).
单一数据源:模板主表为 ``templates``(用户自建,归属 user_id)/
``edit_templates``(全局模板库),片段配置主表为 ``template_clip_configs``
(由 ``EditTemplateService.list_clip_configs_for_editor`` 统一读取)。
不再使用"新表抛异常 → 降级直查配置表 → 再降级查 segments"的异常控制流,
也不在正常请求中打印 ``ValueError: 模板不存在`` 堆栈。
Args:
template_id: 模板 ID
user_id: 当前登录用户 ID(用于归属校验)
tpl_svc: 模板编辑器服务
Returns:
[(segment_order, duration_min, duration_max), ...] 按 order 排序;
模板存在但未配置片段时返回空列表。
Raises:
TemplateNotFoundError: 模板不存在、已删除或不归属于当前用户。
"""
clip_configs = tpl_svc.list_clip_configs_for_editor(template_id, user_id)
result = []
for cc in clip_configs:
dur_min = _safe_segment_duration(cc.min_duration, _DEFAULT_EDITOR_CLIP_DURATION)
dur_max = _safe_segment_duration(
cc.max_duration or cc.min_duration,
_DEFAULT_EDITOR_CLIP_DURATION,
)
dur_min, dur_max = min(dur_min, dur_max), max(dur_min, dur_max)
result.append((cc.order, dur_min, dur_max))
return sorted(result, key=lambda x: x[0])
def _recommended_time_conflicts(
start: float,
duration: float,
used: list[tuple[float, float]],
edge_gap: float = SEGMENT_EDGE_GAP,
) -> bool:
"""检查推荐起始时间是否与已使用时间段冲突.
冲突检测统一加 ``edge_gap`` 秒边缘间隙:已用区间按 [s-gap, e+gap] 扩边后判定,
避免推荐片段与已用片段首尾紧贴导致画面观感重复。
"""
end = start + duration
for used_start, used_end in used:
if start < used_end + edge_gap and end > used_start - edge_gap:
return True
return False
# 向后兼容别名:镜头段构建/段内取点逻辑已下沉到 packages.domain.plan_generator_utils,
# 旧测试与历史代码仍按 clips._build_scene_segments / _pick_start_in_scene_segment 导入
_build_scene_segments = build_scene_segments
_pick_start_in_scene_segment = pick_start_in_scene_segment
def _get_mediakit_recommendations(
asset_ids: list[str],
asset_repo,
) -> dict[str, float]:
"""调用 MediaKit 视频理解,获取智能选片推荐起始时间.
尝试让 MediaKit 分析视频内容,返回每个素材的推荐起始时间。
任何异常都优雅降级,返回空字典(调用方降级到随机选择)。
"""
try:
client = get_mediakit_client()
if not client.is_available:
logger.info("MediaKit 未配置,使用随机起始时间")
return {}
storage = get_storage_service()
video_urls: list[str] = []
valid_asset_ids: list[str] = []
for asset_id in asset_ids[:10]:
asset = asset_repo.get(asset_id)
if not asset or not getattr(asset, "storage_key", None):
continue
mime = getattr(asset, "mime_type", "")
if not mime.startswith("video/"):
continue
try:
url = storage.get_download_url(asset.storage_key)
if url:
video_urls.append(url)
valid_asset_ids.append(asset_id)
except Exception:
logger.exception("获取素材URL失败: asset_id=%s", asset_id)
if not video_urls:
return {}
prompt = (
"请分析每段视频,找出最精彩的5秒片段应该从哪个时间点开始。"
"考虑因素:画面清晰度、主体是否明确、是否有明显的动作或场景变化。"
"请严格以JSON数组格式返回,不要包含其他文字:"
'[{"asset_id": "素材ID", "recommended_start_time": 12.5, "reason": "原因"}]'
)
contents = client.analyze_videos(
video_urls=video_urls,
prompt=prompt,
level="Economy",
poll_interval=2.0,
max_poll_attempts=15,
)
if not contents:
logger.info("MediaKit 分析无结果,降级为随机选择")
return {}
# 按索引映射结果:contents[i] 对应 valid_asset_ids[i]
recommendations: dict[str, float] = {}
for idx, content_text in enumerate(contents):
if idx >= len(valid_asset_ids):
break
asset_id = valid_asset_ids[idx]
if not content_text:
continue
# 尝试从文本中提取 JSON
parsed = False
# 尝试直接解析
try:
data = json.loads(content_text.strip())
if isinstance(data, list) and data:
for item in data:
if isinstance(item, dict) and "recommended_start_time" in item:
recommendations[asset_id] = float(item["recommended_start_time"])
parsed = True
break
except (json.JSONDecodeError, ValueError, TypeError):
pass
# 尝试从 markdown 代码块中提取 JSON
if not parsed:
json_match = re.search(r"\[\s*(\{.*?\})\s*\]", content_text, re.DOTALL)
if json_match:
try:
item = json.loads(json_match.group(1))
if isinstance(item, dict) and "recommended_start_time" in item:
recommendations[asset_id] = float(item["recommended_start_time"])
parsed = True
except (json.JSONDecodeError, ValueError, TypeError):
pass
# 尝试正则提取
if not parsed:
time_match = re.search(r'recommended_start_time["\s:]+([\d.]+)', content_text)
if time_match:
try:
recommendations[asset_id] = float(time_match.group(1))
except (ValueError, TypeError):
pass
if recommendations:
logger.info("MediaKit 智能选片推荐: %s", recommendations)
else:
logger.info("MediaKit 结果解析失败,降级为随机选择")
return recommendations
except Exception as e:
logger.exception("MediaKit 智能选片异常,降级为随机选择: %s", e)
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,
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配置创建,MediaKit异步更新).
逻辑:
1. 从模板读取 segments,片段数量 = segment 数量(忽略前端传的 required_clips_count)
2. 每个片段时长在 segment 的 duration_min ~ duration_max 之间随机取值(保留一位小数)
3. 素材按片段顺序轮询分配,素材不够时同一素材切多个片段
4. 使用 replace_all_clips_transactional 原子性地清空旧片段并创建新的(随机起始时间)
5. 立即返回响应(目标 <1秒)
6. 后台异步任务:调用 MediaKit 智能选片并更新片段的 start_time
7. 素材时长为 0 或缺失时报 400,不创建无效片段
"""
tpl_svc, plan_svc = services
user_id = str(current_user.user.id)
# 1. 查询模板片段配置。模板不存在/已删除/无权限 → 404;
# 模板存在但确实未配置片段 → 422(配置错误,与 404 区分)。
try:
segments = _get_template_segments(template_id, user_id, tpl_svc)
except TemplateNotFoundError as exc:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="模板不存在或无权访问",
) from exc
if not segments:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="模板未配置片段",
)
# 防御:schema validator 已过滤 null/空串,这里再归一化一次,
# 避免异常入参(undefined → null)导致后续 /assets/{id} 404 / 422
asset_ids = [str(aid).strip() for aid in (body.asset_ids or []) if isinstance(aid, str) and aid.strip()]
if not asset_ids:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="素材列表为空,无法创建片段",
)
# 2. 获取素材实际时长(去重查询)
unique_asset_ids = list(dict.fromkeys(asset_ids))
asset_durations: dict[str, float] = {}
asset_smart_scores: dict[str, float] = {}
# 素材 metadata 中缓存的场景切换点(由后台 MediaKit SceneChange 检测写入):
# 有缓存时片段起点从随机镜头段中选取(不同片段来自不同镜头),无缓存回退随机起点
asset_scene_points: dict[str, list[float]] = {}
for asset_id in unique_asset_ids:
asset = asset_repo.get(asset_id)
if asset and hasattr(asset, "duration"):
asset_durations[asset_id] = float(asset.duration or 0.0)
# 计算 smart_match 综合评分,用于候选排序
smart_score, _ = score_asset(asset)
asset_smart_scores[asset_id] = smart_score
# 读取场景切换点缓存(新素材未检测过时为 None,走随机起点兜底)
cached_points = extract_scene_points_from_metadata(getattr(asset, "metadata", None))
if cached_points:
asset_scene_points[asset_id] = cached_points
logger.info(
"from-assets 场景缓存命中: %d/%d 个素材有场景切换点",
len(asset_scene_points),
len(unique_asset_ids),
)
# 3. 在内存中计算所有片段数据(使用随机起始时间,不调用MediaKit)
# 读取素材 metadata 中持久化的历史已用区间(跨任务/跨调用去重),
# 格式与 _calc_random_start_time 的 used_segments 参数一致
used_segments: dict[str, list[tuple[float, float]]] = get_used_segments(db, unique_asset_ids)
# 受控复用回调:可用区间耗尽时复用最久未用且未达复用上限(3次)的历史区间,
# 复用片段时长累加到 reused_durations 供 15% 占比控制
reused_durations: dict[str, float] = {}
# 本条成片中每个素材被分配的片段总时长(复用占比分母)
asset_assigned_durations: dict[str, float] = {}
# 受控复用回调:区间耗尽时复用最久未用且 use_count<3 的历史区间;
# 回调内部预判复用后占比是否超 15%,超限拒绝复用(返回 None)
reuse_cb = make_reuse_callback(
db,
asset_durations,
reused_durations,
assigned_tracker=asset_assigned_durations,
)
clips_data: list[dict] = []
def _reuse_ratio_exceeded(aid: str, extra: float = 0.0) -> bool:
"""该素材在本条成片中「已复用片段时长 / 已分配片段总时长」是否已超 15%。
在为下一片段选素材时调用:本片段尚未分配,复用状态只在分配后的回调里
更新,因此直接检查当前占比——一旦已超 15%,该素材不再参与后续分配。
assigned=0(首个片段)放行;reused=0(尚未发生复用)时不误拦正常分配。
"""
assigned = asset_assigned_durations.get(aid, 0.0)
if assigned <= 0:
return False
return reused_durations.get(aid, 0.0) / assigned > REUSE_RATIO_LIMIT
# 素材耗尽标志:某轮循环中所有素材均被跳过时为 True
all_assets_exhausted = False
# 计算转场重叠补偿:每个 clip 需要额外增加的时长
# 目标:渲染后视频总时长 = 模板设定的各片段时长之和
# 公式:每 clip 增加 (n_segments - 1) * td / n_segments
n_segments = len(segments)
if n_segments > 1:
transition_compensation = (n_segments - 1) * _DEFAULT_TRANSITION_DURATION / n_segments
else:
transition_compensation = 0.0
# 打乱 segments 的处理顺序(分配素材的顺序随机化),但最终 clips_data 按原始 order 排序
shuffled_indices = list(range(len(segments)))
random.shuffle(shuffled_indices)
for idx in shuffled_indices:
_seg_order, dur_min, dur_max = segments[idx]
# 在 segment 的 duration_min ~ duration_max 之间随机取值(保留一位小数)
raw_duration = random.uniform(dur_min, dur_max)
# 加上转场补偿,确保最终输出时长 = 模板设定总时长
raw_duration += transition_compensation
# 贪心分配素材:按"已使用次数"升序排列候选素材(使用最少的优先),
# 同次数随机打散,避免"A-B-C-D"的固定组合反复出现。
# 跳过时长缺失、复用占比已超 10% 阈值的素材;
# 选中后计算起点,若该素材可用区间耗尽且复用被闸门拒绝(calc 返回 None),
# 继续尝试下一个素材
asset_id = ""
clip_duration = 0.0
start_time: float | None = None
# 动态按使用次数排序:优先选使用最少的素材,同次数随机打散
asset_use_counts = {aid: len(used_segments.get(aid, [])) for aid in asset_ids}
# 排序键:smart_match 评分(注入随机噪声)→ 使用次数 → 纯随机。
# 噪声让得分接近的素材排名每次浮动,避免同一批素材反复选出相同组合,
# 从素材组合层面降低成片查重率;分差 > SCORE_RANDOM_NOISE_MAX 时排名稳定,
# 质量差距显著的素材仍保持优先级。
sorted_candidates = sorted(
asset_ids,
key=lambda aid: (
-(asset_smart_scores.get(aid, 0.0) + random.uniform(0.0, SCORE_RANDOM_NOISE_MAX)),
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
candidate_duration = min(round(raw_duration, 1), candidate_total)
if candidate_duration <= 0:
continue
if _reuse_ratio_exceeded(candidate, candidate_duration):
logger.info(
"from-assets 素材复用占比超 %.0f%% 阈值,跳过分配: asset_id=%s",
REUSE_RATIO_LIMIT * 100,
candidate,
)
continue
# 起始时间选取(不调用 MediaKit,保证接口快速返回):
# 1) 素材有场景切换点缓存时,优先从随机镜头段中选起点(不同片段来自不同镜头,
# 画面内容本质不同),与 used_segments 做冲突避让(含 1.5s 边缘间隙)
# 2) 无缓存 / 镜头段全冲突 → _calc_random_start_time 随机起点兜底;
# 100 次避不开历史区间时走受控复用回调(复用片段累加 reused_durations,
# 回调内部预判复用后占比超 10% 则拒绝并返回 None)
candidate_start = None
if candidate in asset_scene_points:
candidate_start = pick_scene_aware_start(
candidate,
candidate_duration,
asset_durations,
asset_scene_points,
used_segments,
edge_gap=SEGMENT_EDGE_GAP,
)
if candidate_start is None:
candidate_start = _calc_random_start_time(
candidate,
candidate_duration,
asset_durations,
used_segments,
on_exhausted=reuse_cb,
)
if candidate_start is None:
# 该素材可用区间耗尽且复用被闸门/use_count 上限拒绝 → 尝试下一素材
logger.info(
"from-assets 素材无可用可切区间(复用被拒),轮询下一素材: asset_id=%s",
candidate,
)
continue
asset_id = candidate
clip_duration = candidate_duration
start_time = candidate_start
break
if not asset_id or start_time is None:
# 所有素材时长缺失、复用占比超阈值,或区间耗尽且复用被拒 → 素材可切区间不足
all_assets_exhausted = True
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="素材可切区间不足,请补充新素材",
)
# 记录已使用时间段(内存,供本次后续片段避开)
used_segments.setdefault(asset_id, []).append((start_time, start_time + clip_duration))
asset_assigned_durations[asset_id] = asset_assigned_durations.get(asset_id, 0.0) + clip_duration
# 同步写入素材 metadata(不 commit,与下方 replace_all_clips_transactional
# 处于同一事务,任一步失败整体回滚,不留脏数据);
# 复用区间与历史记录高度重叠时 record 内部自动累加 use_count
record_used_segments(db, asset_id, start_time, start_time + clip_duration, plan_id)
clips_data.append(
{
"order": _seg_order,
"asset_id": asset_id,
"start_time": start_time,
"duration": clip_duration,
"clip_type": body.clip_type or "main",
}
)
# 按原始 segment order 排序,确保 clips_data 的 order 字段有序(0,1,2,3...)
clips_data.sort(key=lambda c: c["order"])
# 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",
template_id,
plan_id,
len(segments),
created_count,
current_user.user.id,
)
# 5. 触发后台任务:异步调用 MediaKit 并更新片段起始时间
background_tasks.add_task(
_update_mediakit_recommendations_async,
plan_id,
unique_asset_ids,
)
# 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.exception(
"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,
)
def _update_mediakit_recommendations_async( # pragma: no cover
plan_id: str,
asset_ids: list[str],
) -> None:
"""后台任务:使用 SceneChange 智能选帧并更新片段的起始时间.
优先使用 SceneChange 策略检测视频镜头切换点,将每个素材按镜头段拆分,
各片段优先从不同镜头段中选取起始时间,实现「不同片段展示不同场景」的效果。
降级策略:
1. SceneChange 优先 → detect_scene_changes 内部已含 TimeInterval 降级
2. 若 detect_scene_changes 仍返回 None → 回退到旧的 analyze_videos 方式
3. 所有方式都失败 → 保持现有随机 start_time,不影响视频生成
此函数在后台异步执行,不影响接口响应时间。
失败时静默处理,不影响已创建的片段。
"""
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)
# 查询该 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 预分组片段对象(按 order 排序,保证按模板顺序分配镜头段)
clips_by_asset: dict[str, list] = defaultdict(list)
for clip in clips:
aid = getattr(clip, "asset_id", "") or ""
if aid:
clips_by_asset[aid].append(clip)
for aid in clips_by_asset:
clips_by_asset[aid].sort(key=lambda c: c.order)
# 读取素材全部历史已用区间(跨任务/跨 plan 持久化记录)
historical_segments = get_used_segments(db, unique_asset_ids)
# 已更新的片段ID(用于排除已移动的旧时间段)
updated_clip_ids: set[str] = set()
# 已更新的时间段
updated_segments: dict[str, list[tuple[float, float]]] = {}
updated_count = 0
# 尝试获取存储服务(用于生成视频 URL)
try:
storage = get_storage_service()
except Exception as e:
logger.exception("后台任务: 获取存储服务失败,跳过 SceneChange 更新: %s", e)
return
# 获取 MediaKit 客户端
client = get_mediakit_client()
# 对每个素材,检测场景切换点并分配镜头段
for asset_id in unique_asset_ids:
asset_clips = clips_by_asset.get(asset_id, [])
if not asset_clips:
continue
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
# 获取素材视频 URL
video_url: str | None = None
storage_key = getattr(asset, "storage_key", None) or ""
mime = getattr(asset, "mime_type", "") or ""
if storage_key and mime.startswith("video/"):
try:
video_url = storage.get_download_url(storage_key)
except Exception:
logger.exception("后台任务: 获取素材URL失败: asset_id=%s", asset_id)
# 构建该素材的占用区间列表(排除已更新片段)
def _get_other_segments(asset_id_inner, clip_id_inner):
segs: list[tuple[float, float]] = []
for c in clips_by_asset.get(asset_id_inner, []):
cid = c.id
if cid != clip_id_inner and cid not in updated_clip_ids:
segs.append((c.start_time, c.start_time + c.duration))
segs.extend(updated_segments.get(asset_id_inner, []))
# 并入历史已用区间
def _norm(segs_in):
return {(round(float(a), 3), round(float(b), 3)) for a, b in segs_in}
return list(_norm(segs) | _norm(historical_segments.get(asset_id_inner, [])))
# 优先使用 SceneChange 策略
scene_segments: list[tuple[float, float]] = []
# 先查素材 metadata 中的场景点缓存:命中则直接复用,跳过 MediaKit 检测
# (缓存由本任务首次检测后写入,跨任务/跨 plan 复用)
cached_points = extract_scene_points_from_metadata(getattr(asset, "metadata", None))
if cached_points:
scene_segments = build_scene_segments(cached_points, asset_total)
logger.info(
"后台任务: 命中场景点缓存: asset_id=%s scenes=%d",
asset_id,
len(scene_segments),
)
if not scene_segments and client.is_available and video_url:
scene_changes = client.detect_scene_changes(video_url)
if scene_changes is not None:
scene_segments = build_scene_segments(scene_changes, asset_total)
logger.info(
"后台任务: 素材场景检测完成: asset_id=%s scenes=%d",
asset_id,
len(scene_segments),
)
# 检测结果写入素材 metadata 缓存:首次生成用随机起点,
# 检测完成后后续生成的渲染前同步路径即可读缓存选镜头段
try:
existing_meta = dict(getattr(asset, "metadata", None) or {})
existing_meta["scene_change_points"] = scene_changes
asset.metadata = existing_meta # type: ignore[attr-defined]
asset_repo.update(asset) # type: ignore[arg-type]
logger.info(
"后台任务: 场景点已写入素材缓存: asset_id=%s points=%d",
asset_id,
len(scene_changes),
)
except Exception:
# 缓存写入失败不影响本次片段更新
logger.exception(
"后台任务: 场景点缓存写入失败: asset_id=%s",
asset_id,
)
# SceneChange 未获得有效结果 → 尝试 analyze_videos 作为 fallback
if not scene_segments and video_url:
fallback_recs = _get_mediakit_recommendations([asset_id], asset_repo)
if fallback_recs and asset_id in fallback_recs:
# analyze_videos 只返回单个推荐点,转为单镜头段
rec_start = fallback_recs[asset_id]
scene_segments = [(rec_start, asset_total)]
logger.info(
"后台任务: 使用 analyze_videos fallback: asset_id=%s start=%.2f",
asset_id,
rec_start,
)
if not scene_segments:
# 所有方式都失败 → 保持现有随机 start_time
logger.info(
"后台任务: SceneChange 与 analyze_videos 均无结果,保持随机起点: asset_id=%s",
asset_id,
)
continue
# 为每个片段分配不同的镜头段
scene_segments_pool = list(scene_segments) # 可消费的镜头段池
for clip in asset_clips:
clip_duration = clip.duration
recommended_start: float | None = None
# 从镜头段池中依次尝试,选一个不冲突的
for seg_idx, (seg_start, seg_end) in enumerate(scene_segments_pool):
candidate_start = pick_start_in_scene_segment(seg_start, seg_end, clip_duration)
if candidate_start is None:
continue # 镜头段太短,跳过
# 检查越界
if candidate_start + clip_duration > asset_total:
continue
# 检查与已用区间冲突
other_segs = _get_other_segments(asset_id, clip.id)
if _recommended_time_conflicts(candidate_start, clip_duration, other_segs):
continue
recommended_start = candidate_start
# 消费该镜头段(从池中移除,下一个片段用不同镜头段)
scene_segments_pool.pop(seg_idx)
break
if recommended_start is None:
# 镜头段用完或都冲突 → 尝试 _calc_random_start_time 兜底
used_segs_for_calc: dict[str, list[tuple[float, float]]] = {
asset_id: _get_other_segments(asset_id, clip.id)
}
fallback_start = _calc_random_start_time(
asset_id,
clip_duration,
{asset_id: asset_total},
used_segs_for_calc,
)
if fallback_start is None:
continue # 完全无法分配,保持原起点
recommended_start = fallback_start
# 更新片段起始时间
try:
old_start = clip.start_time
old_end = old_start + clip_duration
plan_svc.update_clip(clip.id, start_time=recommended_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,
)
except Exception:
logger.exception(
"后台任务: 同步素材区间记录失败,回滚本次片段更新: clip_id=%s",
clip.id,
)
db.rollback()
continue
db.commit()
updated_count += 1
updated_clip_ids.add(clip.id)
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,
)
except Exception:
logger.exception("后台任务: 单个片段更新失败: clip_id=%s", clip.id)
try:
db.rollback()
except Exception:
pass
continue
logger.info("后台任务完成: plan_id=%s 成功更新 %d 个片段", plan_id, updated_count)
except Exception:
# 后台任务失败不影响已创建的片段,静默处理
logger.exception("后台任务异常: plan_id=%s", plan_id)
if db:
try:
db.rollback()
except Exception:
pass
finally:
if db:
try:
db.close()
except Exception:
pass