feat: 统一渲染引擎 + 打通一键生成全链路 #202

Merged
xiaoxia merged 6 commits from feature/unified-rendering-and-pipeline into develop 2026-07-09 22:59:19 +08:00
26 changed files with 3717 additions and 504 deletions
@@ -0,0 +1,26 @@
"""Add editing_mode to edit_templates
Revision ID: 035_editing_mode
Revises: 034_cms_enhance
Create Date: 2026-07-09
"""
import sqlalchemy as sa
from alembic import op
revision = "035_editing_mode"
down_revision = "034_cms_enhance"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.add_column(
"edit_templates",
sa.Column("editing_mode", sa.String(20), nullable=False, server_default="one_take"),
)
def downgrade() -> None:
op.drop_column("edit_templates", "editing_mode")
+126 -1
View File
@@ -10,6 +10,8 @@ RESTful CRUD for EditPlan:
- GET /api/v1/edit-plans/{id}/generation-status 查询生成进度(任务 2.05)
- POST /api/v1/edit-plans/{id}/ai-recommend AI 推荐片段方案(任务 3.09)
- POST /api/v1/edit-plans/{id}/generate-cover AI 生成封面(任务 3.09)
- GET /api/v1/edit-plans/{id}/timeline 时间线场景数据
- POST /api/v1/edit-plans/generate-from-template 基于模板+素材自动生成剪辑计划
业务逻辑委托给 EditPlanService 服务层。
"""
@@ -24,7 +26,7 @@ from app.auth import AuthenticatedUser, get_current_user
from app.core.celery_app import celery_app
from app.dependencies import get_asset_library_repository, get_asset_repository, get_db_session, get_project_repository
from app.schemas.generation_task import GenerationTaskResponse
from app.services import EditPlanService
from app.services import EditPlanService, PlanGeneratorService
from fastapi import APIRouter, Depends, HTTPException, Query, status
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
@@ -203,6 +205,44 @@ class GenerateCoverResponse(BaseModel):
cover: dict[str, Any] = Field(..., description="封面数据(type / image_url / frame_time 等)")
# ── 基于模板生成剪辑计划 Schemas ─────────────────────────────────────────────
class GenerateFromTemplateRequest(BaseModel):
"""基于模板生成剪辑计划请求体"""
template_id: str = Field(..., description="剪辑模板 ID")
asset_ids: List[str] = Field(default_factory=list, description="素材 ID 列表")
project_id: str = Field(default="", description="所属项目 ID")
name: str = Field(default="", description="计划名称(为空则自动取模板名)")
class _PlanClipItem(BaseModel):
"""片段响应体"""
id: str
clip_type: str
order: int
asset_id: str
text_content: str
start_time: float
duration: float
transition_effect: str
status: str
config: Optional[dict[str, Any]] = None
created_at: datetime
updated_at: datetime
model_config = {"from_attributes": True}
class GenerateFromTemplateResponse(BaseModel):
"""基于模板生成剪辑计划响应体"""
plan: EditPlanResponse
clips: List[_PlanClipItem]
# ── Helpers ───────────────────────────────────────────────────────────────────
@@ -1078,3 +1118,88 @@ def get_plan_timeline(
total_duration=total_duration,
scenes=scenes,
)
# ── 基于模板生成剪辑计划 ─────────────────────────────────────────────────────
@router.post(
"/generate-from-template",
response_model=GenerateFromTemplateResponse,
status_code=status.HTTP_201_CREATED,
)
def generate_from_template(
body: GenerateFromTemplateRequest,
db: Session = Depends(get_db_session),
current_user: AuthenticatedUser = Depends(get_current_user),
project_repository: Any = Depends(get_project_repository),
) -> GenerateFromTemplateResponse:
"""基于模板 + 素材自动生成剪辑计划
流程:
1. 获取模板及其片段配置
2. 调用 PlanGeneratorService 生成 EditPlan + EditPlanClips
3. 返回完整的计划和片段列表
"""
from app.services import EditTemplateService
# 项目鉴权
if body.project_id:
_check_project_access(body.project_id, current_user.user.id, project_repository)
template_svc = EditTemplateService(db)
# 获取模板
try:
template = template_svc.get_template_or_raise(body.template_id)
except ValueError as exc:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=str(exc),
)
# 获取模板片段配置
clip_configs = template_svc.list_clip_configs(body.template_id, skip=0, limit=200)
# 调用 PlanGeneratorService 生成计划
generator = PlanGeneratorService(db)
result = generator.generate_from_template(
template=template,
clip_configs=clip_configs,
asset_ids=body.asset_ids,
project_id=body.project_id,
created_by_user_id=current_user.user.id,
name=body.name,
)
plan = result["plan"]
clips = result["clips"]
logger.info(
"基于模板生成剪辑计划: plan_id=%s template_id=%s clips=%d by user=%s",
plan.id,
body.template_id,
len(clips),
current_user.user.id,
)
return GenerateFromTemplateResponse(
plan=_to_response(plan),
clips=[
_PlanClipItem(
id=c.id,
clip_type=c.clip_type,
order=c.order,
asset_id=c.asset_id,
text_content=c.text_content,
start_time=c.start_time,
duration=c.duration,
transition_effect=c.transition_effect,
status=c.status.value if hasattr(c.status, "value") else c.status,
config=c.config,
created_at=c.created_at,
updated_at=c.updated_at,
)
for c in clips
],
)
+10
View File
@@ -41,6 +41,9 @@ class EditTemplateCreateRequest(BaseModel):
name: str = Field(..., min_length=1, max_length=200, description="模板名称")
description: str = Field(default="", max_length=2000, description="模板描述")
template_type: str = Field(default="default", max_length=50, description="模板类型")
editing_mode: str = Field(
default="one_take", max_length=20, description="剪辑模式: one_take/pip/voice_over/voice_pip"
)
config: dict[str, Any] = Field(default_factory=dict, description="模板配置 (JSON)")
preview_url: str = Field(default="", max_length=500, description="预览地址")
sort_weight: int = Field(default=0, ge=0, le=9999, description="排序权重")
@@ -52,6 +55,9 @@ class EditTemplateUpdateRequest(BaseModel):
name: Optional[str] = Field(default=None, min_length=1, max_length=200, description="模板名称")
description: Optional[str] = Field(default=None, max_length=2000, description="模板描述")
template_type: Optional[str] = Field(default=None, max_length=50, description="模板类型")
editing_mode: Optional[str] = Field(
default=None, max_length=20, description="剪辑模式: one_take/pip/voice_over/voice_pip"
)
config: Optional[dict[str, Any]] = Field(default=None, description="模板配置 (JSON)")
preview_url: Optional[str] = Field(default=None, max_length=500, description="预览地址")
sort_weight: Optional[int] = Field(default=None, ge=0, le=9999, description="排序权重")
@@ -65,6 +71,7 @@ class EditTemplateResponse(BaseModel):
name: str
description: str
template_type: str
editing_mode: str
config: dict[str, Any]
preview_url: str
sort_weight: int
@@ -102,6 +109,7 @@ def _to_response(t: EditTemplate) -> EditTemplateResponse:
name=t.name,
description=t.description,
template_type=t.template_type,
editing_mode=t.editing_mode,
config=t.config,
preview_url=t.preview_url,
sort_weight=t.sort_weight,
@@ -195,6 +203,7 @@ def create_template(
name=body.name,
description=body.description,
template_type=body.template_type,
editing_mode=body.editing_mode,
config=normalized_config,
preview_url=body.preview_url,
sort_weight=body.sort_weight,
@@ -239,6 +248,7 @@ def update_template(
name=body.name,
description=body.description,
template_type=body.template_type,
editing_mode=body.editing_mode,
config=config_to_update,
preview_url=body.preview_url,
sort_weight=body.sort_weight,
+2
View File
@@ -4,6 +4,7 @@ from .auto_clip_service import AutoClipService
from .edit_plan_service import EditPlanService
from .edit_template_service import EditTemplateService
from .job_service import JobService
from .plan_generator_service import PlanGeneratorService
from .video_compose_service import VideoComposeService
__all__ = [
@@ -11,5 +12,6 @@ __all__ = [
"EditPlanService",
"EditTemplateService",
"JobService",
"PlanGeneratorService",
"VideoComposeService",
]
@@ -100,6 +100,7 @@ class EditTemplateService:
*,
description: str = "",
template_type: str = "default",
editing_mode: str = "one_take",
config: Optional[dict[str, Any]] = None,
preview_url: str = "",
sort_weight: int = 0,
@@ -124,6 +125,7 @@ class EditTemplateService:
name=clean_name,
description=description,
template_type=template_type,
editing_mode=editing_mode,
config=config,
preview_url=preview_url,
sort_weight=sort_weight,
@@ -139,6 +141,7 @@ class EditTemplateService:
name: Optional[str] = None,
description: Optional[str] = None,
template_type: Optional[str] = None,
editing_mode: Optional[str] = None,
config: Optional[dict[str, Any]] = None,
preview_url: Optional[str] = None,
sort_weight: Optional[int] = None,
@@ -165,6 +168,7 @@ class EditTemplateService:
name=new_name,
description=description.strip() if description is not None else existing.description,
template_type=template_type.strip() if template_type is not None else existing.template_type,
editing_mode=editing_mode.strip() if editing_mode is not None else existing.editing_mode,
config=config if config is not None else existing.config,
preview_url=preview_url.strip() if preview_url is not None else existing.preview_url,
sort_weight=sort_weight if sort_weight is not None else existing.sort_weight,
@@ -0,0 +1,391 @@
"""PlanGeneratorService — 基于模板+素材自动生成剪辑计划.
核心职责:
- 根据 EditTemplate 的 editing_mode 和 TemplateClipConfig 列表,
自动生成 EditPlan + EditPlanClip 列表
- 四种模式素材分配策略:
- ONE_TAKE: 素材顺序分配给 main 类型 clips
- PIP: 第1个素材→main(全屏背景),其余→overlay clips
- VOICE_OVER: 素材→main clips (B-roll),标记需要配音叠加
- VOICE_PIP: 第1个→background, 第2个→corner_voice, 其余→b_roll
"""
from __future__ import annotations
import logging
from typing import Any, List, Optional
from sqlalchemy.orm import Session
from packages.adapters.sqlalchemy_impl import (
SQLAlchemyEditPlanClipRepository,
SQLAlchemyEditPlanRepository,
)
from packages.domain.config_schemas import normalize_plan_config
from packages.domain.edit_plan import EditPlan
from packages.domain.edit_plan_clip import EditPlanClip
from packages.domain.edit_template import EditTemplate
from packages.domain.editing_mode import EditingMode
from packages.domain.template_clip_config import ClipType, TemplateClipConfig
logger = logging.getLogger(__name__)
# ── 默认片段时长(秒) ────────────────────────────────────────────────────────
_DEFAULT_CLIP_DURATION = 5.0
_DEFAULT_INTRO_DURATION = 3.0
_DEFAULT_OUTRO_DURATION = 3.0
class PlanGeneratorService:
"""剪辑计划生成器
基于模板 + 素材,自动生成 EditPlan 及 EditPlanClip 列表。
"""
def __init__(self, db: Session) -> None:
self._plan_repo = SQLAlchemyEditPlanRepository(db)
self._clip_repo = SQLAlchemyEditPlanClipRepository(db)
# ── 公开接口 ─────────────────────────────────────────────────────────────
def generate_from_template(
self,
template: EditTemplate,
clip_configs: List[TemplateClipConfig],
asset_ids: List[str],
*,
project_id: str = "",
created_by_user_id: str = "",
name: str = "",
) -> dict[str, Any]:
"""基于模板+素材生成剪辑计划
Args:
template: 剪辑模板实体
clip_configs: 模板片段配置列表(可为空,自动生成默认结构)
asset_ids: 素材 ID 列表
project_id: 所属项目 ID
created_by_user_id: 创建者用户 ID
name: 计划名称(为空则自动取模板名)
Returns:
dict: {"plan": EditPlan, "clips": List[EditPlanClip]}
"""
editing_mode = template.editing_mode or EditingMode.ONE_TAKE.value
plan_name = name.strip() or f"{template.name} - 剪辑计划"
# 1. 构建 plan config(继承模板的 title/subtitle/bgm,记录 editing_mode
plan_config = self._build_plan_config(template, editing_mode)
# 2. 创建 EditPlan
plan = EditPlan.create(
template_id=template.id,
name=plan_name,
config=plan_config,
total_duration=0.0,
project_id=project_id,
created_by_user_id=created_by_user_id,
)
plan = self._plan_repo.create(plan)
logger.info(
"生成剪辑计划: plan_id=%s template=%s mode=%s assets=%d",
plan.id,
template.id,
editing_mode,
len(asset_ids),
)
# 3. 生成片段列表
if clip_configs:
clips = self._create_clips_from_configs(plan.id, clip_configs)
else:
clips = self._generate_default_clips(plan.id, editing_mode, len(asset_ids))
# 4. 按 editing_mode 分配素材
if asset_ids:
self._distribute_assets(clips, asset_ids, editing_mode)
# 5. 持久化所有 clips 并计算总时长
created_clips: List[EditPlanClip] = []
total_duration = 0.0
for clip in clips:
saved = self._clip_repo.create(clip)
created_clips.append(saved)
total_duration += saved.duration
# 6. 更新 plan 的 total_duration
plan.total_duration = total_duration
plan = self._plan_repo.update(plan)
# 7. 流转到 editing 状态
try:
plan.start_editing()
plan = self._plan_repo.update(plan)
except ValueError as exc:
logger.warning("计划状态流转失败: plan_id=%s error=%s", plan.id, exc)
logger.info(
"剪辑计划生成完成: plan_id=%s clips=%d duration=%.1f",
plan.id,
len(created_clips),
total_duration,
)
return {"plan": plan, "clips": created_clips}
# ── 内部方法 ─────────────────────────────────────────────────────────────
def _build_plan_config(
self,
template: EditTemplate,
editing_mode: str,
) -> dict[str, Any]:
"""从模板配置构建 plan config"""
template_config = template.config or {}
plan_config: dict[str, Any] = {
"editing_mode": editing_mode,
}
# 继承模板的 cover/title/subtitle/bgm 配置
for key in ("cover", "title", "subtitle", "bgm"):
if key in template_config:
plan_config[key] = template_config[key]
return normalize_plan_config(plan_config)
def _create_clips_from_configs(
self,
plan_id: str,
clip_configs: List[TemplateClipConfig],
) -> List[EditPlanClip]:
"""从 TemplateClipConfig 列表创建 EditPlanClip 列表(未持久化)"""
clips: List[EditPlanClip] = []
# 按 order 排序
sorted_configs = sorted(clip_configs, key=lambda c: c.order)
for cfg in sorted_configs:
# 计算时长:取 min_duration 和 max_duration 的中间值
if cfg.min_duration > 0 and cfg.max_duration > 0:
duration = (cfg.min_duration + cfg.max_duration) / 2
elif cfg.min_duration > 0:
duration = cfg.min_duration
elif cfg.max_duration > 0:
duration = cfg.max_duration
else:
duration = _DEFAULT_CLIP_DURATION
# clip_type 可能是枚举或字符串
clip_type = cfg.clip_type.value if hasattr(cfg.clip_type, "value") else cfg.clip_type
# transition_effect 可能是枚举或字符串
transition = (
cfg.transition_effect.value if hasattr(cfg.transition_effect, "value") else cfg.transition_effect
)
clip = EditPlanClip.create(
plan_id=plan_id,
clip_type=clip_type,
order=cfg.order,
template_clip_config_id=cfg.id,
text_content=getattr(cfg, "text_template", "") or "",
duration=duration,
transition_effect=transition or "cut",
)
clips.append(clip)
return clips
def _generate_default_clips(
self,
plan_id: str,
editing_mode: str,
asset_count: int,
) -> List[EditPlanClip]:
"""无 clip_configs 时,根据 editing_mode 生成默认 clip 结构
- ONE_TAKE: N 个 main clipsN = asset_count,至少1个)
- PIP: 1 个 main + (N-1) 个 overlayN = asset_count
- VOICE_OVER: N 个 main clips + 标记需要配音
- VOICE_PIP: 1 个 background + 1 个 corner_voice + (N-2) 个 b_roll
"""
n = max(asset_count, 1)
clips: List[EditPlanClip] = []
order = 0
if editing_mode == EditingMode.PIP.value:
# 1 个 main(全屏背景)
clips.append(
EditPlanClip.create(
plan_id=plan_id,
clip_type=ClipType.MAIN.value,
order=order,
duration=_DEFAULT_CLIP_DURATION,
)
)
order += 1
# 剩余为 overlay
for i in range(1, n):
clips.append(
EditPlanClip.create(
plan_id=plan_id,
clip_type="overlay",
order=order,
duration=_DEFAULT_CLIP_DURATION,
)
)
order += 1
elif editing_mode == EditingMode.VOICE_OVER.value:
# N 个 main clipsB-roll
for i in range(n):
clips.append(
EditPlanClip.create(
plan_id=plan_id,
clip_type=ClipType.MAIN.value,
order=order,
duration=_DEFAULT_CLIP_DURATION,
config={"role": "b_roll"},
)
)
order += 1
elif editing_mode == EditingMode.VOICE_PIP.value:
# 1 个 background
clips.append(
EditPlanClip.create(
plan_id=plan_id,
clip_type="background",
order=order,
duration=_DEFAULT_CLIP_DURATION,
)
)
order += 1
# 1 个 corner_voice
clips.append(
EditPlanClip.create(
plan_id=plan_id,
clip_type="corner_voice",
order=order,
duration=_DEFAULT_CLIP_DURATION,
)
)
order += 1
# 剩余为 b_roll
for i in range(2, n):
clips.append(
EditPlanClip.create(
plan_id=plan_id,
clip_type="b_roll",
order=order,
duration=_DEFAULT_CLIP_DURATION,
)
)
order += 1
else:
# ONE_TAKE: N 个 main clips
for i in range(n):
clips.append(
EditPlanClip.create(
plan_id=plan_id,
clip_type=ClipType.MAIN.value,
order=order,
duration=_DEFAULT_CLIP_DURATION,
)
)
order += 1
return clips
def _distribute_assets(
self,
clips: List[EditPlanClip],
asset_ids: List[str],
editing_mode: str,
) -> None:
"""按 editing_mode 将素材分配到 clips(就地修改,未持久化)
分配策略:
- ONE_TAKE: 素材按顺序依次分配给 main 类型 clips
- PIP: 第1个素材→main(全屏背景),其余→交替分配给 overlay clips
- VOICE_OVER: 素材→main clips (B-roll)
- VOICE_PIP: 第1个→background, 第2个→corner_voice, 其余→b_roll
"""
if not asset_ids or not clips:
return
if editing_mode == EditingMode.ONE_TAKE.value:
self._distribute_one_take(clips, asset_ids)
elif editing_mode == EditingMode.PIP.value:
self._distribute_pip(clips, asset_ids)
elif editing_mode == EditingMode.VOICE_OVER.value:
self._distribute_voice_over(clips, asset_ids)
elif editing_mode == EditingMode.VOICE_PIP.value:
self._distribute_voice_pip(clips, asset_ids)
else:
# 未知模式,退化为 one_take
self._distribute_one_take(clips, asset_ids)
def _distribute_one_take(
self,
clips: List[EditPlanClip],
asset_ids: List[str],
) -> None:
"""ONE_TAKE: 素材按顺序依次分配给 main 类型 clips"""
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):
clip.assign_asset(asset_ids[i])
def _distribute_pip(
self,
clips: List[EditPlanClip],
asset_ids: List[str],
) -> None:
"""PIP: 第1个素材→main(全屏背景),其余→overlay clips"""
# 第1个素材 → main clip
main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
if main_clips and asset_ids:
main_clips[0].assign_asset(asset_ids[0])
# 其余素材 → overlay clips
overlay_clips = [c for c in clips if c.clip_type == "overlay"]
remaining = asset_ids[1:]
for i, clip in enumerate(overlay_clips):
if i < len(remaining):
clip.assign_asset(remaining[i])
def _distribute_voice_over(
self,
clips: List[EditPlanClip],
asset_ids: List[str],
) -> None:
"""VOICE_OVER: 素材→main clips (B-roll)"""
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):
clip.assign_asset(asset_ids[i])
def _distribute_voice_pip(
self,
clips: List[EditPlanClip],
asset_ids: List[str],
) -> None:
"""VOICE_PIP: 第1个→background, 第2个→corner_voice, 其余→b_roll"""
bg_clips = [c for c in clips if c.clip_type == "background"]
corner_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"]
# 第1个素材 → background
if bg_clips and len(asset_ids) > 0:
bg_clips[0].assign_asset(asset_ids[0])
# 第2个素材 → corner_voice
if corner_clips and len(asset_ids) > 1:
corner_clips[0].assign_asset(asset_ids[1])
# 其余素材 → b_roll
remaining = asset_ids[2:]
for i, clip in enumerate(broll_clips):
if i < len(remaining):
clip.assign_asset(remaining[i])
+12 -1
View File
@@ -2,6 +2,17 @@
视频处理模块
"""
# 共享工具模块(供 editing_modes / generation / edit_plan_generation 等复用)
from . import dedup_helpers, ffmpeg_utils, oss_helpers
from .processor import VideoProcessor, VideoResult
from .unified_render_service import RenderResult, UnifiedRenderService
__all__ = ["VideoProcessor", "VideoResult"]
__all__ = [
"VideoProcessor",
"VideoResult",
"ffmpeg_utils",
"oss_helpers",
"dedup_helpers",
"UnifiedRenderService",
"RenderResult",
]
@@ -0,0 +1,126 @@
"""查重辅助函数 — 从 generation.py 提取的 GeneratedVideo 记录 + 查重逻辑.
供 render_edit_plan 和 generate_video 共同复用,
创建 GeneratedVideo 记录后计算指纹并执行项目级 + 批次内查重。
"""
from __future__ import annotations
import logging
from uuid import uuid4
from sqlalchemy.orm import Session
logger = logging.getLogger(__name__)
def create_video_record_and_dedup(
*,
generation_task_id: str,
project_id: str,
batch_id: str,
file_url: str,
file_size: int,
duration: float,
video_path: str,
mode: str,
session: Session,
width: int = 1280,
height: int = 720,
fps: float = 25.0,
) -> int:
"""创建 GeneratedVideo 记录,计算指纹并执行查重(历史 + 批次)。
Args:
generation_task_id: 生成任务 ID
project_id: 项目 ID
batch_id: 批次 ID(可为空字符串)
file_url: 视频文件 URL
file_size: 文件大小(字节)
duration: 视频时长(秒)
video_path: 视频本地路径(用于计算指纹)
mode: 剪辑模式名称
session: 数据库会话
width: 视频宽度
height: 视频高度
fps: 视频帧率
Returns:
创建的视频记录数量(1 表示成功,0 表示失败)
"""
from video_processing.dedup import VideoDeduplicator
from packages.adapters.sqlalchemy_impl.generated_video_repository import (
SQLAlchemyGeneratedVideoRepository,
)
from packages.domain import GeneratedVideo
try:
video_id = uuid4().hex
generated_video = GeneratedVideo(
id=video_id,
project_id=project_id,
generation_task_id=generation_task_id,
name=f"generated-{generation_task_id[:8]}.mp4",
file_url=file_url,
file_size=file_size,
duration=duration,
width=width,
height=height,
fps=fps,
status="completed",
generation_params={"mode": mode},
)
video_repo = SQLAlchemyGeneratedVideoRepository(session)
video_repo.create(generated_video)
# 计算视频指纹
deduplicator = VideoDeduplicator()
try:
fingerprint = deduplicator.compute_fingerprint(video_path)
except Exception as fp_err:
logger.warning("Fingerprint computation failed for %s: %s", video_id, fp_err)
session.commit()
return 1
generated_video.video_fingerprint = fingerprint.to_dict()
# (a) 历史成片查重
duplicate_result = deduplicator.check_duplicate(fingerprint, project_id, session)
# (b) 批次内查重(仅当有 batch_id 时)
if not duplicate_result and batch_id:
duplicate_result = deduplicator.check_batch_duplicate(fingerprint, batch_id, video_id, session)
if duplicate_result:
generated_video.is_duplicate = True
generated_video.duplicate_of = duplicate_result["duplicate_of"]
logger.info(
"Duplicate detected: %s -> %s (reason=%s, similarity=%.3f)",
video_id,
duplicate_result["duplicate_of"],
duplicate_result["reason"],
duplicate_result["similarity"],
)
else:
generated_video.is_duplicate = False
generated_video.duplicate_of = None
video_repo.update(generated_video)
session.commit()
logger.info(
"GeneratedVideo record created: %s (task=%s, dup=%s)",
video_id,
generation_task_id,
generated_video.is_duplicate,
)
return 1
except Exception as e:
logger.error(
"Failed to create video record / dedup for task %s: %s",
generation_task_id,
e,
)
session.rollback()
return 0
+41 -84
View File
@@ -5,7 +5,6 @@
import logging
import os
import subprocess
import sys
import tempfile
from dataclasses import dataclass
@@ -22,6 +21,8 @@ else:
from pathlib import Path
from typing import Optional
from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_video_info, run_ffmpeg
logger = logging.getLogger(__name__)
@@ -67,8 +68,6 @@ class EditingModeProcessor:
"""
self.config = config
self.work_dir = work_dir or tempfile.gettempdir()
self._ffmpeg_bin = "ffmpeg"
self._ffprobe_bin = "ffprobe"
def process(
self,
@@ -129,62 +128,20 @@ class EditingModeProcessor:
return os.path.join(self.work_dir, f"output_{self.config.mode}_{os.getpid()}.mp4")
def _run_ffmpeg(self, command: list[str], capture_output: bool = True) -> tuple:
"""执行 FFmpeg 命令"""
logger.debug(f"Running FFmpeg: {' '.join(command)}")
"""执行 FFmpeg 命令 — 委托给共享 ffmpeg_utils.run_ffmpeg"""
try:
result = subprocess.run(
command,
check=True,
stdout=subprocess.PIPE if capture_output else None,
stderr=subprocess.PIPE if capture_output else None,
text=capture_output,
)
return result.stdout or "", result.stderr or ""
except subprocess.CalledProcessError as e:
stderr = e.stderr.decode() if e.stderr else str(e)
logger.error(f"FFmpeg error: {stderr}")
raise RuntimeError(f"FFmpeg execution failed: {stderr}") from e
return run_ffmpeg(command, capture_output=capture_output)
except RuntimeError as e:
logger.error(f"FFmpeg error: {e}")
raise
def _get_video_info(self, video_path: str) -> dict:
"""获取视频信息"""
"""获取视频信息 — 委托给共享 ffmpeg_utils.probe_video_info,补充 codec/size 字段"""
try:
result = subprocess.run(
[
self._ffprobe_bin,
"-v",
"error",
"-show_entries",
"stream=width,height,r_frame_rate,duration,codec_name",
"-show_entries",
"format=duration,size",
"-of",
"json",
video_path,
],
check=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
import json
data = json.loads(result.stdout)
streams = data.get("streams", [{}])
video_stream = next((s for s in streams if s.get("codec_type") == "video"), streams[0] if streams else {})
fmt = data.get("format", {})
fps_str = video_stream.get("r_frame_rate", "25/1")
fps_parts = fps_str.split("/")
fps = float(fps_parts[0]) / float(fps_parts[1]) if len(fps_parts) == 2 else float(fps_parts[0])
return {
"width": int(video_stream.get("width", 0)),
"height": int(video_stream.get("height", 0)),
"fps": fps,
"duration": float(fmt.get("duration", 0)),
"codec": video_stream.get("codec_name", "unknown"),
"size": int(fmt.get("size", 0)),
}
info = probe_video_info(video_path)
info["codec"] = "unknown"
info["size"] = os.path.getsize(video_path) if os.path.exists(video_path) else 0
return info
except Exception as e:
logger.warning(f"Failed to get video info for {video_path}: {e}")
return {"width": 0, "height": 0, "fps": 25, "duration": 0, "codec": "unknown", "size": 0}
@@ -205,7 +162,7 @@ class EditingModeProcessor:
def _normalize_video(self, input_path: str, output_path: str) -> dict:
"""标准化视频格式:先统一帧率,再缩放/填充"""
command = [
self._ffmpeg_bin,
FFMPEG_BIN,
"-y",
"-i",
input_path,
@@ -228,7 +185,7 @@ class EditingModeProcessor:
"-an",
output_path,
]
self._run_ffmpeg(command)
run_ffmpeg(command)
return self._get_video_info(output_path)
def _one_take(self, video_paths: list[str], output_path: str) -> str:
@@ -265,7 +222,7 @@ class EditingModeProcessor:
offset1 = durations[0] - transition / 2
command = [
self._ffmpeg_bin,
FFMPEG_BIN,
"-y",
"-i",
normalized_paths[0],
@@ -285,7 +242,7 @@ class EditingModeProcessor:
"yuv420p",
output_path,
]
self._run_ffmpeg(command)
run_ffmpeg(command)
return output_path
else:
return self._one_take_simple_concat(normalized_paths, output_path)
@@ -298,7 +255,7 @@ class EditingModeProcessor:
f.write(f"file '{os.path.abspath(path)}'\n")
command = [
self._ffmpeg_bin,
FFMPEG_BIN,
"-y",
"-f",
"concat",
@@ -310,7 +267,7 @@ class EditingModeProcessor:
"copy",
output_path,
]
self._run_ffmpeg(command)
run_ffmpeg(command)
try:
os.remove(concat_file)
@@ -344,7 +301,7 @@ class EditingModeProcessor:
if pip_info["duration"] > main_info["duration"]:
temp_pip = os.path.join(self.work_dir, f"pip_temp_{os.getpid()}.mp4")
command = [
self._ffmpeg_bin,
FFMPEG_BIN,
"-y",
"-i",
video_paths[1],
@@ -362,11 +319,11 @@ class EditingModeProcessor:
"yuv420p",
temp_pip,
]
self._run_ffmpeg(command)
run_ffmpeg(command)
pip_normalized_input = temp_pip
else:
command = [
self._ffmpeg_bin,
FFMPEG_BIN,
"-y",
"-i",
video_paths[1],
@@ -382,13 +339,13 @@ class EditingModeProcessor:
"yuv420p",
pip_normalized,
]
self._run_ffmpeg(command)
run_ffmpeg(command)
pip_normalized_input = pip_normalized
if main_info["duration"] > pip_info["duration"]:
looped_pip = os.path.join(self.work_dir, f"pip_looped_{os.getpid()}.mp4")
command = [
self._ffmpeg_bin,
FFMPEG_BIN,
"-y",
"-stream_loop",
"-1",
@@ -408,11 +365,11 @@ class EditingModeProcessor:
"yuv420p",
looped_pip,
]
self._run_ffmpeg(command)
run_ffmpeg(command)
pip_normalized_input = looped_pip
command = [
self._ffmpeg_bin,
FFMPEG_BIN,
"-y",
"-i",
main_normalized,
@@ -432,7 +389,7 @@ class EditingModeProcessor:
"yuv420p",
output_path,
]
self._run_ffmpeg(command)
run_ffmpeg(command)
for temp_file in [main_normalized, pip_normalized]:
if temp_file and temp_file != output_path:
@@ -462,7 +419,7 @@ class EditingModeProcessor:
if bg_info["duration"] < audio_duration:
looped_bg = os.path.join(self.work_dir, f"bg_looped_{os.getpid()}.mp4")
command = [
self._ffmpeg_bin,
FFMPEG_BIN,
"-y",
"-stream_loop",
"-1",
@@ -482,12 +439,12 @@ class EditingModeProcessor:
"yuv420p",
looped_bg,
]
self._run_ffmpeg(command)
run_ffmpeg(command)
bg_normalized = looped_bg
elif bg_info["duration"] > audio_duration:
temp_bg = os.path.join(self.work_dir, f"bg_trimmed_{os.getpid()}.mp4")
command = [
self._ffmpeg_bin,
FFMPEG_BIN,
"-y",
"-i",
bg_normalized,
@@ -497,12 +454,12 @@ class EditingModeProcessor:
"copy",
temp_bg,
]
self._run_ffmpeg(command)
run_ffmpeg(command)
bg_normalized = temp_bg
blurred_bg = os.path.join(self.work_dir, f"bg_blurred_{os.getpid()}.mp4")
command = [
self._ffmpeg_bin,
FFMPEG_BIN,
"-y",
"-i",
bg_normalized,
@@ -518,10 +475,10 @@ class EditingModeProcessor:
"yuv420p",
blurred_bg,
]
self._run_ffmpeg(command)
run_ffmpeg(command)
command = [
self._ffmpeg_bin,
FFMPEG_BIN,
"-y",
"-i",
blurred_bg,
@@ -544,7 +501,7 @@ class EditingModeProcessor:
"-shortest",
output_path,
]
self._run_ffmpeg(command)
run_ffmpeg(command)
for temp_file in [bg_normalized, blurred_bg]:
try:
@@ -582,7 +539,7 @@ class EditingModeProcessor:
voice_adjusted = os.path.join(self.work_dir, f"voice_adj_{os.getpid()}.mp4")
command = [
self._ffmpeg_bin,
FFMPEG_BIN,
"-y",
"-i",
voice_normalized,
@@ -600,11 +557,11 @@ class EditingModeProcessor:
"yuv420p",
voice_adjusted,
]
self._run_ffmpeg(command)
run_ffmpeg(command)
bg_adjusted = os.path.join(self.work_dir, f"bg_adj_{os.getpid()}.mp4")
command = [
self._ffmpeg_bin,
FFMPEG_BIN,
"-y",
"-i",
bg_normalized,
@@ -614,11 +571,11 @@ class EditingModeProcessor:
"copy",
bg_adjusted,
]
self._run_ffmpeg(command)
run_ffmpeg(command)
if audio_path:
command = [
self._ffmpeg_bin,
FFMPEG_BIN,
"-y",
"-i",
bg_adjusted,
@@ -645,7 +602,7 @@ class EditingModeProcessor:
]
else:
command = [
self._ffmpeg_bin,
FFMPEG_BIN,
"-y",
"-i",
bg_adjusted,
@@ -668,7 +625,7 @@ class EditingModeProcessor:
"yuv420p",
output_path,
]
self._run_ffmpeg(command)
run_ffmpeg(command)
for temp_file in [voice_normalized, voice_adjusted, bg_normalized, bg_adjusted]:
try:
@@ -0,0 +1,289 @@
"""FFmpeg 工具函数 — 从 editing_modes.py / video_compose_service.py 提取的共享原语.
提供 FFmpeg / FFprobe 调用、视频信息探测、视频标准化、xfade 转场滤镜构建
等底层能力,供 EditingModeProcessor、VideoComposeService、UnifiedRenderService
共同复用。
"""
from __future__ import annotations
import logging
import shutil
import subprocess # nosec B404
from pathlib import Path
from typing import Any
logger = logging.getLogger(__name__)
# ── 常量 ──────────────────────────────────────────────────────────────────────
FFMPEG_BIN: str = shutil.which("ffmpeg") or "ffmpeg"
FFPROBE_BIN: str = shutil.which("ffprobe") or "ffprobe"
DEFAULT_OUTPUT_WIDTH = 1280
DEFAULT_OUTPUT_HEIGHT = 720
DEFAULT_FPS = 25
# xfade 转场映射:transition_effect 名称 → FFmpeg xfade transition 名称
# 键同时支持 TransitionEffect 枚举值和字符串名称(向后兼容)
XFADE_TRANSITION_MAP: dict[str, str] = {
"fade": "fade",
"slideleft": "slideleft",
"slide_left": "slideleft",
"slideright": "slideright",
"slide_right": "slideright",
"dissolve": "dissolve",
"wipe": "wipeleft",
"wipeleft": "wipeleft",
}
DEFAULT_TRANSITION_DURATION = 0.5
# ── FFmpeg 执行 ───────────────────────────────────────────────────────────────
def run_ffmpeg(
command: list[str],
*,
capture_output: bool = True,
) -> tuple[str, str]:
"""执行 FFmpeg 命令。
Args:
command: 完整的 ffmpeg 命令列表(含 "ffmpeg" 本身)
capture_output: 是否捕获 stdout/stderr
Returns:
(stdout, stderr) 元组
Raises:
subprocess.CalledProcessError: 命令执行失败时抛出
"""
result = subprocess.run( # nosec B603
command,
check=True,
stdout=subprocess.PIPE if capture_output else None,
stderr=subprocess.PIPE if capture_output else None,
text=True,
)
return (result.stdout or "", result.stderr or "")
def probe_duration(local_path: str | Path) -> float:
"""用 ffprobe 获取视频时长(秒)。
失败时返回默认值 5.0 秒。
"""
try:
result = subprocess.run( # nosec B603
[
FFPROBE_BIN,
"-v",
"error",
"-show_entries",
"format=duration",
"-of",
"default=noprint_wrappers=1:nokey=1",
str(local_path),
],
check=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
return round(float(result.stdout.strip()), 3)
except Exception:
return 5.0
def probe_video_info(video_path: str) -> dict[str, Any]:
"""获取视频信息(宽、高、时长、fps)。
Returns:
{"width": int, "height": int, "duration": float, "fps": float}
失败时返回默认值。
"""
try:
result = subprocess.run( # nosec B603
[
FFPROBE_BIN,
"-v",
"error",
"-select_streams",
"v:0",
"-show_entries",
"stream=width,height,r_frame_rate,duration",
"-show_entries",
"format=duration",
"-of",
"json",
video_path,
],
check=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
import json
info = json.loads(result.stdout)
stream = info.get("streams", [{}])[0]
fmt = info.get("format", {})
width = int(stream.get("width", DEFAULT_OUTPUT_WIDTH))
height = int(stream.get("height", DEFAULT_OUTPUT_HEIGHT))
# 解析帧率
fps_str = stream.get("r_frame_rate", "25/1")
if "/" in fps_str:
num, den = fps_str.split("/")
fps = float(num) / float(den) if float(den) > 0 else DEFAULT_FPS
else:
fps = float(fps_str) if fps_str else DEFAULT_FPS
# 时长
duration = float(fmt.get("duration", 0)) or float(stream.get("duration", 0))
return {
"width": width,
"height": height,
"duration": duration,
"fps": round(fps, 2),
}
except Exception as e:
logger.warning("获取视频信息失败: %s, error: %s", video_path, e)
return {
"width": DEFAULT_OUTPUT_WIDTH,
"height": DEFAULT_OUTPUT_HEIGHT,
"duration": 0.0,
"fps": DEFAULT_FPS,
}
def normalize_video(
input_path: str,
output_path: str,
*,
width: int = DEFAULT_OUTPUT_WIDTH,
height: int = DEFAULT_OUTPUT_HEIGHT,
fps: int = DEFAULT_FPS,
) -> dict[str, Any]:
"""标准化视频(缩放 + 恒定帧率)。
使用 scale + pad 保持宽高比,黑边填充到目标分辨率。
Returns:
{"width": int, "height": int, "path": str}
"""
command = [
FFMPEG_BIN,
"-y",
"-i",
input_path,
"-vf",
f"scale={width}:{height}:force_original_aspect_ratio=decrease,"
f"pad={width}:{height}:(ow-iw)/2:(oh-ih)/2:black,"
f"fps={fps}",
"-c:v",
"libx264",
"-crf",
"23",
"-preset",
"medium",
"-c:a",
"aac",
"-b:a",
"128k",
"-movflags",
"+faststart",
output_path,
]
run_ffmpeg(command)
return {"width": width, "height": height, "path": output_path}
# ── xfade / concat 滤镜构建 ──────────────────────────────────────────────────
def chain_filters(filters: list[str], output_label: str, *, input_label: str = "0:v") -> str:
"""将滤镜列表串联为 FFmpeg 滤镜字符串。
例:chain_filters(["scale=1280:720", "fps=25"], "v0")
"[0:v]scale=1280:720,fps=25[v0]"
"""
filter_body = ",".join(filters)
return f"[{input_label}]{filter_body}[{output_label}]"
def resolve_xfade_transition(transition_name: str) -> str:
"""将转场效果名称映射为 FFmpeg xfade transition 名称。
支持 TransitionEffect 枚举值和字符串名称,未知值回退到 "fade"
"""
# 兼容 TransitionEffect 枚举(有 .value 属性)
if hasattr(transition_name, "value"):
transition_name = transition_name.value
return XFADE_TRANSITION_MAP.get(transition_name, "fade")
def build_xfade_filter_chain(
clip_durations: list[float],
clip_video_labels: list[str],
transitions: list[str],
*,
transition_duration: float = DEFAULT_TRANSITION_DURATION,
output_label: str = "outv",
) -> tuple[str, float]:
"""构建 xfade 转场滤镜链。
Args:
clip_durations: 每个片段的时长
clip_video_labels: 每个片段的视频流标签(如 "v0", "v1"
transitions: 每个片段对应的转场效果(第一个片段的转场被忽略)
transition_duration: 转场时长(秒)
output_label: 最终输出标签
Returns:
(filter_string, estimated_total_duration)
"""
n = len(clip_durations)
parts: list[str] = []
total_duration = sum(clip_durations)
if n == 0:
return "", 0.0
if n == 1:
parts.append(f"[{clip_video_labels[0]}]copy[{output_label}]")
return ";".join(parts), total_duration
# xfade 链
cumulative = 0.0
prev_label = clip_video_labels[0]
for i in range(1, n):
cumulative += clip_durations[i - 1]
offset = max(0.0, cumulative - transition_duration * i)
transition = transitions[i] if i < len(transitions) else "cut"
xfade_transition = resolve_xfade_transition(transition)
if i == n - 1:
out_label = output_label
else:
out_label = f"xf{i}"
parts.append(
f"[{prev_label}][{clip_video_labels[i]}]"
f"xfade=transition={xfade_transition}"
f":duration={transition_duration}"
f":offset={offset:.3f}"
f"[{out_label}]"
)
prev_label = out_label
# 总时长减去转场重叠部分
total_duration -= transition_duration * (n - 1)
return ";".join(parts), max(0.0, total_duration)
+164
View File
@@ -0,0 +1,164 @@
"""OSS 工具函数 — 从 generation.py / edit_plan_generation.py 提取的共享 OSS 操作.
提供 OSS 配置读取、Bucket 创建、素材上传/下载、asset_id → 本地路径解析
等能力,供 render_edit_plan 和 generate_video 共同复用。
"""
from __future__ import annotations
import hashlib
import logging
import os
from pathlib import Path
from typing import Optional
from urllib.parse import urlparse
import oss2
logger = logging.getLogger(__name__)
# ── OSS 配置 ──────────────────────────────────────────────────────────────────
def oss_settings() -> tuple[str, str, str, str] | None:
"""获取 OSS 配置。
Returns:
(access_key_id, access_key_secret, endpoint, bucket_name) 元组,
配置缺失时返回 None。
"""
access_key_id = os.getenv("OSS_ACCESS_KEY_ID")
access_key_secret = os.getenv("OSS_ACCESS_KEY_SECRET")
endpoint = os.getenv("OSS_ENDPOINT")
bucket_name = os.getenv("OSS_BUCKET_NAME")
if not all([access_key_id, access_key_secret, endpoint, bucket_name]):
return None
return access_key_id, access_key_secret, endpoint, bucket_name
def oss_bucket() -> oss2.Bucket | None:
"""获取 OSS Bucket 实例。
Returns:
oss2.Bucket 实例,配置缺失时返回 None。
"""
settings = oss_settings()
if settings is None:
return None
access_key_id, access_key_secret, endpoint, bucket_name = settings
return oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)
def normalize_storage_key(storage_key_or_url: str) -> str:
"""标准化存储键 — 如果是完整 URL 则提取 path 部分。
Examples:
"https://bucket.oss-cn-hangzhou.aliyuncs.com/path/to/file.mp4"
"path/to/file.mp4"
"path/to/file.mp4""path/to/file.mp4"
"""
if storage_key_or_url.startswith(("http://", "https://")):
return urlparse(storage_key_or_url).path.lstrip("/")
return storage_key_or_url.lstrip("/")
# ── 上传 / 下载 ───────────────────────────────────────────────────────────────
def download_asset(asset_storage_key: str, local_path: Path) -> bool:
"""从 OSS 下载素材文件到本地路径。
Args:
asset_storage_key: 素材的存储键(或完整 URL)
local_path: 本地保存路径
Returns:
True 表示下载成功,False 表示失败。
"""
bucket = oss_bucket()
if bucket is None:
return False
try:
bucket.get_object_to_file(normalize_storage_key(asset_storage_key), str(local_path))
return local_path.exists() and local_path.stat().st_size > 0
except Exception:
logger.exception("下载素材失败: %s", asset_storage_key)
return False
def upload_to_oss(local_path: Path, storage_key: str) -> str | None:
"""上传文件到 OSS,返回公开 URL。
Args:
local_path: 本地文件路径
storage_key: 目标存储键
Returns:
公开访问 URL,上传失败或 OSS 未配置时返回 None。
"""
bucket = oss_bucket()
if bucket is None:
return None
try:
bucket.put_object_from_file(storage_key, str(local_path))
settings = oss_settings()
if settings:
_, _, endpoint, bucket_name = settings
return f"https://{bucket_name}.{endpoint.replace('https://', '').replace('http://', '')}/{storage_key}"
return None
except Exception:
logger.exception("上传 OSS 失败: %s", storage_key)
return None
# ── Asset 解析 ────────────────────────────────────────────────────────────────
def resolve_asset_path(asset_id: str, work_dir: Path) -> Path | None:
"""从 asset_id 解析到本地文件路径。
策略(按优先级):
1. 如果 asset_id 是本地绝对路径(/var/storage/...)→ 直接返回
2. 如果 work_dir 下已有缓存文件 → 返回缓存路径
3. 从 OSS 下载到 work_dir/{hash}.mp4 → 返回下载路径
4. 下载失败 → 返回 None
缓存策略:以 asset_id 的 SHA256 前 16 位为文件名,避免重复下载。
"""
# 1. 本地绝对路径
if asset_id.startswith("/") and os.path.exists(asset_id):
return Path(asset_id)
# 2. 缓存命中
cache_hash = hashlib.sha256(asset_id.encode()).hexdigest()[:16]
cached_path = work_dir / f"{cache_hash}.mp4"
if cached_path.exists() and cached_path.stat().st_size > 0:
return cached_path
# 3. 从 OSS 下载
if download_asset(asset_id, cached_path):
return cached_path
return None
def resolve_asset_ids_to_paths(
asset_ids: list[str],
work_dir: Path,
) -> dict[str, Path]:
"""批量解析 asset_id → 本地路径。
Args:
asset_ids: 素材 ID 列表
work_dir: 工作目录
Returns:
{asset_id: local_path} 映射,仅包含成功解析的条目。
"""
result: dict[str, Path] = {}
for aid in asset_ids:
local_path = resolve_asset_path(aid, work_dir)
if local_path:
result[aid] = local_path
return result
@@ -0,0 +1,463 @@
"""统一渲染引擎 — 输入 EditPlan + EditPlanClips,按时间线+图层渲染视频.
核心原则(灵应):渲染引擎是统一的,不判断模式,只按 clip_type/config.role
分组为图层再合成。
图层分组:
main (无 config.role) → main (z=0)
main + config.role=b_roll → broll (z=0,与 main 同层替换)
overlay → overlay (z=1,画中画叠加)
background → background (z=0,全屏底图)
corner_voice → corner_voice (z=1,右上角小窗)
b_roll → broll (z=0)
intro / outro → main (z=0,按 order 排在首/尾)
合成流程:
1. 每个 clip 先 trim + scale + setpts 预处理
2. 同层 clips 按 order 用 xfade 串联
3. overlay/corner_voice 层 overlay 到主层
4. 如有独立音频轨,amix 混入
"""
from __future__ import annotations
import logging
import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from video_processing.ffmpeg_utils import (
DEFAULT_FPS,
DEFAULT_OUTPUT_HEIGHT,
DEFAULT_OUTPUT_WIDTH,
DEFAULT_TRANSITION_DURATION,
FFMPEG_BIN,
build_xfade_filter_chain,
probe_duration,
probe_video_info,
run_ffmpeg,
)
logger = logging.getLogger(__name__)
# ── 数据结构 ──────────────────────────────────────────────────────────────────
@dataclass
class ResolvedClip:
"""已解析到本地路径的片段。"""
clip_id: str
asset_id: str
local_path: Path
clip_type: str
order: int
start_time: float = 0.0
duration: float = 0.0 # 0 表示使用素材完整时长
transition_effect: str = "cut"
config: dict[str, Any] = field(default_factory=dict)
# 运行时填充
actual_duration: float = 0.0 # 素材实际时长(probe 后填充)
@dataclass
class RenderLayer:
"""渲染图层。"""
role: str # "main" | "overlay" | "pip" | "background" | "corner_voice" | "broll" | "audio"
clips: list[ResolvedClip] = field(default_factory=list)
z_index: int = 0
opacity: float = 1.0
position: tuple[int, int] | None = None # (x, y) 偏移,None 表示全屏
@dataclass
class RenderResult:
"""渲染结果。"""
output_path: Path
duration: float
file_size: int
width: int
height: int
# ── clip_type → layer role 映射 ──────────────────────────────────────────────
def _resolve_layer_role(clip_type: str, config: dict[str, Any]) -> str:
"""根据 clip_type 和 config.role 确定图层角色。
映射规则:
intro / outro → "main"(按 order 排在首/尾)
overlay → "overlay"(画中画叠加,z=1
corner_voice → "corner_voice"(右上角小窗,z=1
background → "background"(全屏底图,z=0
b_roll → "broll"z=0
main + config.role=b_roll → "broll"
main (default) → "main"
"""
role = config.get("role", "")
if clip_type in ("intro", "outro"):
return "main"
if clip_type == "overlay":
return "overlay"
if clip_type == "corner_voice":
return "corner_voice"
if clip_type == "background":
return "background"
if clip_type == "b_roll":
return "broll"
# main type
if role == "b_roll":
return "broll"
return "main"
# ── 图层默认 z_index ─────────────────────────────────────────────────────────
_LAYER_Z_INDEX: dict[str, int] = {
"background": -1,
"broll": 0,
"main": 0,
"overlay": 1,
"corner_voice": 1,
"audio": 2,
}
# 图层默认 PiP 位置(相对输出画布的偏移)
_PIP_SCALE = 0.25 # PiP 占主画面的比例
# ── 统一渲染引擎 ─────────────────────────────────────────────────────────────
class UnifiedRenderService:
"""统一渲染引擎。
输入 EditPlan + EditPlanClips + 素材路径映射,按时间线+图层执行渲染。
"""
def __init__(
self,
plan: Any, # EditPlan
clips: list[Any], # list[EditPlanClip]
asset_path_map: dict[str, Path], # asset_id → local_path
work_dir: Path,
*,
output_width: int = DEFAULT_OUTPUT_WIDTH,
output_height: int = DEFAULT_OUTPUT_HEIGHT,
output_fps: int = DEFAULT_FPS,
transition_duration: float = DEFAULT_TRANSITION_DURATION,
):
self.plan = plan
self.clips = clips
self.asset_path_map = asset_path_map
self.work_dir = work_dir
self.output_width = output_width
self.output_height = output_height
self.output_fps = output_fps
self.transition_duration = transition_duration
def render(self) -> RenderResult:
"""执行渲染,返回 RenderResult。
Raises:
ValueError: 没有可渲染的片段时抛出
"""
# 1. 解析 clips → ResolvedClips(跳过无素材的 clip
resolved = self._resolve_clips()
if not resolved:
raise ValueError("没有可渲染的片段(所有片段素材缺失或下载失败)")
# 2. 分组为 RenderLayers
layers = self._group_clips_into_layers(resolved)
# 3. 构建 filter_complex
output_path = self.work_dir / f"rendered_{self.plan.id}.mp4"
filter_complex, input_args = self._build_filter_complex(layers)
# 4. 执行 FFmpeg
self._execute_ffmpeg(filter_complex, input_args, output_path)
# 5. 探测输出
duration, file_size, width, height = self._probe_output(output_path)
return RenderResult(
output_path=output_path,
duration=duration,
file_size=file_size,
width=width,
height=height,
)
# ── 内部方法 ──────────────────────────────────────────────────────────────
def _resolve_clips(self) -> list[ResolvedClip]:
"""将 EditPlanClip 列表解析为 ResolvedClip 列表。
跳过 asset_id 为空或在 asset_path_map 中找不到的片段。
"""
resolved: list[ResolvedClip] = []
for clip in self.clips:
asset_id = clip.asset_id
if not asset_id:
logger.warning("片段无素材: clip_id=%s", clip.id)
continue
local_path = self.asset_path_map.get(asset_id)
if local_path is None or not local_path.exists():
logger.warning("素材不存在: clip_id=%s asset_id=%s", clip.id, asset_id)
continue
# 探测实际时长
try:
actual_duration = probe_duration(local_path)
except Exception:
actual_duration = clip.duration or 5.0
rc = ResolvedClip(
clip_id=clip.id,
asset_id=asset_id,
local_path=local_path,
clip_type=clip.clip_type,
order=clip.order,
start_time=clip.start_time,
duration=clip.duration,
transition_effect=clip.transition_effect or "cut",
config=clip.config or {},
actual_duration=actual_duration,
)
resolved.append(rc)
# 按 order 排序
resolved.sort(key=lambda c: c.order)
return resolved
def _group_clips_into_layers(self, resolved_clips: list[ResolvedClip]) -> list[RenderLayer]:
"""将 ResolvedClips 分组为 RenderLayers。
分组规则见 _resolve_layer_role 函数文档。
"""
layer_map: dict[str, RenderLayer] = {}
for clip in resolved_clips:
role = _resolve_layer_role(clip.clip_type, clip.config)
if role not in layer_map:
z = _LAYER_Z_INDEX.get(role, 0)
layer_map[role] = RenderLayer(role=role, z_index=z)
layer_map[role].clips.append(clip)
# 每个 layer 内的 clips 按 order 排序
for layer in layer_map.values():
layer.clips.sort(key=lambda c: c.order)
# 计算 PiP 位置
pip_width = int(self.output_width * _PIP_SCALE)
pip_height = int(self.output_height * _PIP_SCALE)
margin = 20 # 边距
if "overlay" in layer_map:
layer_map["overlay"].position = (
self.output_width - pip_width - margin,
margin,
)
if "corner_voice" in layer_map:
layer_map["corner_voice"].position = (
self.output_width - pip_width - margin,
margin,
)
# 按 z_index 排序返回
layers = sorted(layer_map.values(), key=lambda lyr: lyr.z_index)
return layers
def _build_filter_complex(self, layers: list[RenderLayer]) -> tuple[str, list[str]]:
"""构建 FFmpeg filter_complex 字符串和输入参数列表。
Returns:
(filter_complex_str, input_args_list)
input_args_list 是 ["-i", path1, "-i", path2, ...] 格式
"""
if not layers:
raise ValueError("没有可渲染的图层")
# 收集所有 clips(按图层顺序,同层按 order)
all_clips: list[ResolvedClip] = []
for layer in layers:
all_clips.extend(layer.clips)
# 构建输入参数
input_args: list[str] = []
clip_to_input_idx: dict[str, int] = {}
for i, clip in enumerate(all_clips):
input_args.extend(["-i", str(clip.local_path)])
clip_to_input_idx[clip.clip_id] = i
filter_parts: list[str] = []
# Step 1: 预处理每个 clip — scale + setpts
# 为每个 clip 生成预处理后的标签 [v0], [v1], ...
preprocessed_labels: list[str] = []
for i, clip in enumerate(all_clips):
label = f"v{i}"
role = _resolve_layer_role(clip.clip_type, clip.config)
filters: list[str] = []
# trim(如果指定了 duration
if clip.duration > 0 and clip.duration < clip.actual_duration:
filters.append(f"trim=duration={clip.duration}")
filters.append("setpts=PTS-STARTPTS")
# scale
if role in ("overlay", "corner_voice"):
pip_w = int(self.output_width * _PIP_SCALE)
pip_h = int(self.output_height * _PIP_SCALE)
filters.append(f"scale={pip_w}:{pip_h}")
elif role == "background":
filters.append(
f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=increase"
)
filters.append(f"crop={self.output_width}:{self.output_height}")
else:
# main / broll: scale + pad 保持宽高比
filters.append(
f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=decrease"
)
filters.append(f"pad={self.output_width}:{self.output_height}" ":(ow-iw)/2:(oh-ih)/2:black")
filters.append(f"fps={self.output_fps}")
filters.append("setpts=PTS-STARTPTS")
filter_str = f"[{i}:v]{','.join(filters)}[{label}]"
filter_parts.append(filter_str)
preprocessed_labels.append(label)
# Step 2: 同层 clips 用 xfade 串联
layer_output_labels: dict[str, str] = {}
for layer in layers:
layer_clip_indices = [all_clips.index(c) for c in layer.clips]
layer_labels = [preprocessed_labels[i] for i in layer_clip_indices]
layer_durations = [
all_clips[i].duration if all_clips[i].duration > 0 else all_clips[i].actual_duration
for i in layer_clip_indices
]
layer_transitions = [all_clips[i].transition_effect for i in layer_clip_indices]
if len(layer_labels) == 1:
# 单 clip 层,直接使用预处理标签
layer_output_labels[layer.role] = layer_labels[0]
else:
# 多 clip 层,用 xfade 串联
out_label = f"{layer.role}_merged"
xfade_filter, _ = build_xfade_filter_chain(
clip_durations=layer_durations,
clip_video_labels=layer_labels,
transitions=layer_transitions,
transition_duration=self.transition_duration,
output_label=out_label,
)
if xfade_filter:
filter_parts.append(xfade_filter)
layer_output_labels[layer.role] = out_label
# Step 3: 合成各层
# 找到主层 — background 优先作为底图,其次 broll / main
final_video_label = None
if "background" in layer_output_labels:
final_video_label = layer_output_labels["background"]
# b_roll / main 叠加到 background 上
for role in ("broll", "main"):
if role in layer_output_labels:
base_label = layer_output_labels[role]
combined_label = f"combined_{role}"
filter_parts.append(
f"[{final_video_label}][{base_label}]" f"overlay=(W-w)/2:(H-h)/2[{combined_label}]"
)
final_video_label = combined_label
else:
# 无 background 时,取 broll 或 main 作为基础
for role in ("broll", "main"):
if role in layer_output_labels:
final_video_label = layer_output_labels[role]
break
if final_video_label is None:
# 没有任何主层,使用第一个层
final_video_label = layer_output_labels[layers[0].role]
# 叠加 overlay 层
for layer in layers:
if layer.role in ("overlay", "corner_voice"):
if layer.role not in layer_output_labels:
continue
overlay_label = layer_output_labels[layer.role]
x, y = layer.position or (
self.output_width - int(self.output_width * _PIP_SCALE) - 20,
20,
)
combined_label = f"combined_{layer.role}"
filter_parts.append(f"[{final_video_label}][{overlay_label}]" f"overlay={x}:{y}[{combined_label}]")
final_video_label = combined_label
filter_parts.append(f"[{final_video_label}]format=yuv420p[final_video]")
filter_complex = ";".join(filter_parts)
return filter_complex, input_args
def _execute_ffmpeg(
self,
filter_complex: str,
input_args: list[str],
output_path: Path,
) -> None:
"""执行 FFmpeg 渲染命令。"""
command = [
FFMPEG_BIN,
"-y",
*input_args,
"-filter_complex",
filter_complex,
"-map",
"[final_video]",
"-c:v",
"libx264",
"-crf",
"23",
"-preset",
"medium",
"-pix_fmt",
"yuv420p",
"-movflags",
"+faststart",
str(output_path),
]
logger.info(
"执行渲染: plan_id=%s inputs=%d output=%s",
self.plan.id,
input_args.count("-i"),
output_path,
)
run_ffmpeg(command)
def _probe_output(self, output_path: Path) -> tuple[float, int, int, int]:
"""探测输出文件的时长、大小、宽高。
Returns:
(duration, file_size, width, height)
"""
info = probe_video_info(str(output_path))
file_size = output_path.stat().st_size if output_path.exists() else 0
return (
info["duration"],
file_size,
info["width"],
info["height"],
)
@@ -3,177 +3,39 @@
Celery 任务 worker.render_edit_plan:
1. 加载 EditPlan + EditPlanClips
2. 下载各片段素材
3. 按 order 顺序拼接片段
3. 使用 UnifiedRenderService 按时间线+图层渲染
4. 上传渲染结果到 OSS
5. 更新 EditPlan / EditPlanClip 状态
6. 更新 GenerationTask 进度
5. 创建 GeneratedVideo 记录 + 查重
6. 更新 EditPlan / EditPlanClip 状态
7. 更新 GenerationTask 进度
"""
from __future__ import annotations
import logging
import os
import shutil
import subprocess # nosec B404
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
from urllib.parse import urlparse
import oss2
from worker_app.celery_app import celery_app
from worker_app.db import SessionLocal
logger = logging.getLogger(__name__)
FFMPEG_BIN = shutil.which("ffmpeg") or "ffmpeg"
FFPROBE_BIN = shutil.which("ffprobe") or "ffprobe"
OUTPUT_WIDTH = 1280
OUTPUT_HEIGHT = 720
OUTPUT_FPS = 25.0
# ── OSS helpers ───────────────────────────────────────────────────────────────
def _oss_settings() -> tuple[str, str, str, str] | None:
"""获取 OSS 配置"""
access_key_id = os.getenv("OSS_ACCESS_KEY_ID")
access_key_secret = os.getenv("OSS_ACCESS_KEY_SECRET")
endpoint = os.getenv("OSS_ENDPOINT")
bucket_name = os.getenv("OSS_BUCKET_NAME")
if not all([access_key_id, access_key_secret, endpoint, bucket_name]):
return None
return access_key_id, access_key_secret, endpoint, bucket_name
def _oss_bucket() -> oss2.Bucket | None:
"""获取 OSS Bucket"""
settings = _oss_settings()
if settings is None:
return None
access_key_id, access_key_secret, endpoint, bucket_name = settings
return oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)
def _normalize_storage_key(storage_key_or_url: str) -> str:
"""标准化存储键"""
if storage_key_or_url.startswith(("http://", "https://")):
return urlparse(storage_key_or_url).path.lstrip("/")
return storage_key_or_url.lstrip("/")
def _download_asset(asset_storage_key: str, local_path: Path) -> bool:
"""下载素材文件到本地"""
bucket = _oss_bucket()
if bucket is None:
return False
try:
bucket.get_object_to_file(_normalize_storage_key(asset_storage_key), str(local_path))
return local_path.exists() and local_path.stat().st_size > 0
except Exception:
logger.exception("下载素材失败: %s", asset_storage_key)
return False
def _upload_to_oss(local_path: Path, storage_key: str) -> str | None:
"""上传文件到 OSS,返回公开 URL"""
bucket = _oss_bucket()
if bucket is None:
return None
try:
bucket.put_object_from_file(storage_key, str(local_path))
settings = _oss_settings()
if settings:
_, _, endpoint, bucket_name = settings
return f"https://{bucket_name}.{endpoint.replace('https://', '').replace('http://', '')}/{storage_key}"
return None
except Exception:
logger.exception("上传 OSS 失败: %s", storage_key)
return None
# ── FFmpeg helpers ────────────────────────────────────────────────────────────
def _run_ffmpeg(command: list[str]) -> None:
"""执行 FFmpeg 命令"""
subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) # nosec B603
def _probe_duration(local_path: Path) -> float:
"""获取视频/音频时长"""
try:
result = subprocess.run( # nosec B603
[
FFPROBE_BIN,
"-v",
"error",
"-show_entries",
"format=duration",
"-of",
"default=noprint_wrappers=1:nokey=1",
str(local_path),
],
check=True,
capture_output=True,
text=True,
)
return float(result.stdout.strip())
except Exception:
return 0.0
def _concatenate_clips(
clip_paths: list[Path],
output_path: Path,
transition_effects: list[str] | None = None,
) -> bool:
"""将多个片段拼接为最终视频
使用 FFmpeg concat demuxer 实现。
"""
if not clip_paths:
return False
if len(clip_paths) == 1:
# 单片段直接复制
try:
shutil.copy2(str(clip_paths[0]), str(output_path))
return True
except Exception:
return False
# 多片段:使用 concat demuxer
concat_file = output_path.parent / "concat_list.txt"
try:
with open(concat_file, "w") as f:
for p in clip_paths:
f.write(f"file '{p}'\n")
command = [
FFMPEG_BIN,
"-y",
"-f",
"concat",
"-safe",
"0",
"-i",
str(concat_file),
"-c",
"copy",
str(output_path),
]
_run_ffmpeg(command)
return output_path.exists() and output_path.stat().st_size > 0
except Exception:
logger.exception("拼接片段失败")
return False
finally:
if concat_file.exists():
concat_file.unlink()
# ── 共享工具模块导入 ──────────────────────────────────────────────────────────
from video_processing.dedup_helpers import create_video_record_and_dedup
from video_processing.oss_helpers import (
download_asset,
upload_to_oss,
)
from video_processing.unified_render_service import UnifiedRenderService
# ── Repository imports (延迟导入避免循环依赖) ─────────────────────────────────
@@ -207,11 +69,12 @@ def render_edit_plan(self, plan_id: str) -> dict:
流程:
1. 加载 EditPlan + EditPlanClips
2. 下载各片段素材到临时目录
3. 按 order 顺序拼接片段
2. 下载各片段素材到临时目录,构建 asset_path_map
3. 使用 UnifiedRenderService 按时间线+图层渲染
4. 上传渲染结果到 OSS
5. 更新 EditPlan → completed, EditPlanClips → rendered
6. 更新 GenerationTask 进度
5. 创建 GeneratedVideo 记录 + 查重
6. 更新 EditPlan → completed, EditPlanClips → rendered
7. 更新 GenerationTask 进度
"""
logger.info("开始渲染剪辑计划: plan_id=%s", plan_id)
@@ -246,10 +109,10 @@ def render_edit_plan(self, plan_id: str) -> dict:
gen_task.started_at = datetime.now(timezone.utc)
gen_task_repo.update(gen_task)
# 3. 下载素材并拼接
# 3. 下载素材并构建 asset_path_map
with tempfile.TemporaryDirectory(prefix="edit_plan_") as tmpdir:
tmpdir_path = Path(tmpdir)
clip_paths: list[Path] = []
asset_path_map: dict[str, Path] = {}
rendered_clip_ids: list[str] = []
failed_clip_ids: list[str] = []
@@ -261,18 +124,23 @@ def render_edit_plan(self, plan_id: str) -> dict:
failed_clip_ids.append(clip.id)
continue
if clip.asset_id in asset_path_map:
# 同一素材已下载(多个 clip 共享同一素材)
rendered_clip_ids.append(clip.id)
continue
# 下载素材
ext = Path(clip.asset_id).suffix or ".mp4"
local_path = tmpdir_path / f"clip_{clip.order:04d}{ext}"
if _download_asset(clip.asset_id, local_path):
clip_paths.append(local_path)
if download_asset(clip.asset_id, local_path):
asset_path_map[clip.asset_id] = local_path
rendered_clip_ids.append(clip.id)
else:
clip.mark_failed()
clip_repo.update(clip)
failed_clip_ids.append(clip.id)
if not clip_paths:
if not asset_path_map:
logger.error("所有片段素材下载失败: %s", plan_id)
plan.mark_failed()
plan_repo.update(plan)
@@ -285,42 +153,75 @@ def render_edit_plan(self, plan_id: str) -> dict:
gen_task_repo.update(gen_task)
return {"status": "error", "message": "所有片段素材下载失败"}
# 4. 拼接片段
output_path = tmpdir_path / f"rendered_{plan_id}.mp4"
transition_effects = [c.transition_effect for c in clips if c.asset_id]
success = _concatenate_clips(clip_paths, output_path, transition_effects)
# 4. 使用 UnifiedRenderService 渲染
render_service = UnifiedRenderService(
plan=plan,
clips=clips,
asset_path_map=asset_path_map,
work_dir=tmpdir_path,
output_width=OUTPUT_WIDTH,
output_height=OUTPUT_HEIGHT,
output_fps=int(OUTPUT_FPS),
)
if not success:
logger.error("片段拼接失败: %s", plan_id)
try:
render_result = render_service.render()
except Exception as render_err:
logger.error("渲染失败: %s%s", plan_id, render_err)
plan.mark_failed()
plan_repo.update(plan)
if generation_task_id:
gen_task = gen_task_repo.get(generation_task_id)
if gen_task:
gen_task.status = "failed"
gen_task.error_message = "片段拼接失败"
gen_task.error_message = f"渲染失败: {render_err}"
gen_task.completed_at = datetime.now(timezone.utc)
gen_task_repo.update(gen_task)
return {"status": "error", "message": "片段拼接失败"}
return {"status": "error", "message": f"渲染失败: {render_err}"}
output_path = render_result.output_path
# 5. 上传到 OSS
storage_key = f"rendered/{plan_id}/output.mp4"
output_url = _upload_to_oss(output_path, storage_key)
output_url = upload_to_oss(output_path, storage_key)
# 6. 更新片段状态为 rendered
# 6. 创建 GeneratedVideo 记录 + 查重
project_id = plan.project_id or ""
batch_id = plan.config.get("batch_id", "")
mode = plan.config.get("mode", "edit_plan")
if generation_task_id and project_id:
try:
create_video_record_and_dedup(
generation_task_id=generation_task_id,
project_id=project_id,
batch_id=batch_id,
file_url=output_url or "",
file_size=render_result.file_size,
duration=render_result.duration,
video_path=str(output_path),
mode=mode,
session=db,
width=render_result.width,
height=render_result.height,
fps=OUTPUT_FPS,
)
except Exception as dedup_err:
logger.warning("查重失败(不影响渲染结果): %s", dedup_err)
# 7. 更新片段状态为 rendered
for clip_id in rendered_clip_ids:
clip = clip_repo.get(clip_id)
if clip and clip.status.value == "ready":
clip.mark_rendered()
clip_repo.update(clip)
# 7. 更新 EditPlan 状态为 completed
# 8. 更新 EditPlan 状态为 completed
plan.config["rendered_url"] = output_url or ""
plan.config["rendered_storage_key"] = storage_key
plan.mark_completed()
plan_repo.update(plan)
# 8. 更新 GenerationTask 状态为 completed
# 9. 更新 GenerationTask 状态为 completed
if generation_task_id:
gen_task = gen_task_repo.get(generation_task_id)
if gen_task:
@@ -331,10 +232,11 @@ def render_edit_plan(self, plan_id: str) -> dict:
gen_task_repo.update(gen_task)
logger.info(
"剪辑计划渲染完成: plan_id=%s rendered=%d failed=%d",
"剪辑计划渲染完成: plan_id=%s rendered=%d failed=%d duration=%.1fs",
plan_id,
len(rendered_clip_ids),
len(failed_clip_ids),
render_result.duration,
)
return {
@@ -343,6 +245,7 @@ def render_edit_plan(self, plan_id: str) -> dict:
"rendered_count": len(rendered_clip_ids),
"failed_count": len(failed_clip_ids),
"output_url": output_url,
"duration": render_result.duration,
}
except Exception as exc:
+240 -238
View File
@@ -1,18 +1,25 @@
"""
视频生成任务
支持四种剪辑模式:一镜到底、画中画、口播、口播+画中画
视频生成任务 — 使用 UnifiedRenderService 统一渲染引擎.
支持四种剪辑模式:一镜到底、画中画、口播、口播+画中画。
模式差异体现在虚拟剪辑计划的 clip_type 分布上,渲染引擎不判断模式。
模式 → clip_type 映射:
ONE_TAKE: N 个 main clips
PIP: 1 main + N-1 overlay
VOICE_OVER: N 个 main(config.role=b_roll)
VOICE_PIP: 1 background + 1 corner_voice + N-2 b_roll
"""
from __future__ import annotations
import logging
import os
import shutil
import subprocess # nosec B404
import tempfile
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
from urllib.parse import urlparse
from typing import Any, Optional
import oss2
from worker_app.celery_app import celery_app
from worker_app.db import SessionLocal
@@ -20,8 +27,6 @@ OUTPUT_WIDTH = 1280
OUTPUT_HEIGHT = 720
OUTPUT_FPS = 25.0
OUTPUT_DURATION_SECONDS = 5.0
FFMPEG_BIN = shutil.which("ffmpeg") or "ffmpeg"
FFPROBE_BIN = shutil.which("ffprobe") or "ffprobe"
GENERATED_FILES_DIR = Path(os.getenv("GENERATED_FILES_DIR", "/app/generated"))
GENERATED_FILES_URL_PREFIX = os.getenv("GENERATED_FILES_URL_PREFIX", "/generated-files")
PUBLIC_API_BASE_URL = os.getenv("PUBLIC_API_BASE_URL", "https://api.xiaoxiajianji.com").rstrip("/")
@@ -78,81 +83,138 @@ def _update_task_status(task_id: str, status_action: str, **kwargs) -> bool:
return False
# ── FFmpeg / OSS helpers ─────────────────────────────────────────────────────
# ── 共享工具模块导入 ──────────────────────────────────────────────────────────
from video_processing.dedup_helpers import create_video_record_and_dedup
from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg
from video_processing.oss_helpers import (
download_asset,
oss_bucket,
upload_to_oss,
)
from video_processing.unified_render_service import UnifiedRenderService
# ── 虚拟 Plan / Clip(内存中构建,不写数据库) ────────────────────────────────
def _run_ffmpeg(command: list[str]) -> None:
"""执行 FFmpeg 命令"""
subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) # nosec B603
@dataclass
class _VirtualPlan:
"""内存中的虚拟剪辑计划,供 UnifiedRenderService 使用。"""
id: str
name: str = ""
def _oss_settings() -> tuple[str, str, str, str] | None:
"""获取 OSS 配置"""
access_key_id = os.getenv("OSS_ACCESS_KEY_ID")
access_key_secret = os.getenv("OSS_ACCESS_KEY_SECRET")
endpoint = os.getenv("OSS_ENDPOINT")
bucket_name = os.getenv("OSS_BUCKET_NAME")
if not all([access_key_id, access_key_secret, endpoint, bucket_name]):
return None
return access_key_id, access_key_secret, endpoint, bucket_name
@dataclass
class _VirtualClip:
"""内存中的虚拟剪辑片段,供 UnifiedRenderService 使用。"""
id: str
plan_id: str = ""
clip_type: str = "main"
order: int = 0
asset_id: str = ""
text_content: str = ""
start_time: float = 0.0
duration: float = 0.0
transition_effect: str = "cut"
status: str = "ready"
config: dict[str, Any] = field(default_factory=dict)
def _oss_bucket() -> oss2.Bucket | None:
"""获取 OSS Bucket"""
settings = _oss_settings()
if settings is None:
return None
access_key_id, access_key_secret, endpoint, bucket_name = settings
return oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)
def _build_plan_and_clips_from_task(
task_id: str,
downloaded_paths: list[Path],
mode: str,
) -> tuple[_VirtualPlan, list[_VirtualClip], dict[str, Path]]:
"""根据模式和下载的素材路径,构建虚拟 plan + clips + asset_path_map。
模式 → clip_type 映射:
ONE_TAKE: N 个 main clips
PIP: 1 main + N-1 overlay
VOICE_OVER: N 个 main(config.role=b_roll)
VOICE_PIP: 1 background + 1 corner_voice + N-2 b_roll
def _normalize_storage_key(storage_key_or_url: str) -> str:
"""标准化存储键"""
if storage_key_or_url.startswith(("http://", "https://")):
return urlparse(storage_key_or_url).path.lstrip("/")
return storage_key_or_url.lstrip("/")
Returns:
(virtual_plan, virtual_clips, asset_path_map)
"""
plan = _VirtualPlan(id=task_id, name=f"Generated-{task_id[:8]}")
# 为每个下载路径生成合成 asset_id
asset_path_map: dict[str, Path] = {}
path_to_asset_id: dict[Path, str] = {}
for i, p in enumerate(downloaded_paths):
asset_id = f"gen_{task_id[:8]}_{i:03d}{p.suffix or '.mp4'}"
asset_path_map[asset_id] = p
path_to_asset_id[p] = asset_id
def _download_asset(asset_storage_key: str, local_path: Path) -> bool:
"""下载素材文件"""
bucket = _oss_bucket()
if bucket is None:
return False
try:
bucket.get_object_to_file(_normalize_storage_key(asset_storage_key), str(local_path))
return local_path.exists() and local_path.stat().st_size > 0
except Exception:
return False
clips: list[_VirtualClip] = []
n = len(downloaded_paths)
if mode == "pip":
# 1 main + N-1 overlay
for i, p in enumerate(downloaded_paths):
clip_type = "main" if i == 0 else "overlay"
clips.append(
_VirtualClip(
id=f"vc_{i:03d}",
plan_id=task_id,
clip_type=clip_type,
order=i,
asset_id=path_to_asset_id[p],
)
)
elif mode == "voice_over":
# N 个 main(config.role=b_roll)
for i, p in enumerate(downloaded_paths):
clips.append(
_VirtualClip(
id=f"vc_{i:03d}",
plan_id=task_id,
clip_type="main",
order=i,
asset_id=path_to_asset_id[p],
config={"role": "b_roll"},
)
)
elif mode == "voice_pip":
# 1 background + 1 corner_voice + N-2 b_roll
for i, p in enumerate(downloaded_paths):
if i == 0:
clip_type = "background"
elif i == 1:
clip_type = "corner_voice"
else:
clip_type = "b_roll"
clips.append(
_VirtualClip(
id=f"vc_{i:03d}",
plan_id=task_id,
clip_type=clip_type,
order=i,
asset_id=path_to_asset_id[p],
)
)
else:
# ONE_TAKE (default): N 个 main clips
for i, p in enumerate(downloaded_paths):
clips.append(
_VirtualClip(
id=f"vc_{i:03d}",
plan_id=task_id,
clip_type="main",
order=i,
asset_id=path_to_asset_id[p],
)
)
def _probe_duration(local_path: Path) -> float:
"""获取视频时长"""
try:
result = subprocess.run(
[
FFPROBE_BIN,
"-v",
"error",
"-show_entries",
"format=duration",
"-of",
"default=noprint_wrappers=1:nokey=1",
str(local_path),
],
check=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
) # nosec B603
return round(float(result.stdout.strip()), 3)
except Exception:
return OUTPUT_DURATION_SECONDS
return plan, clips, asset_path_map
def _create_fallback_clip(output_path: Path, title: str) -> None:
"""创建 fallback 视频(无素材时)"""
safe_title = title.replace(":", "\\:").replace("'", "\\'")[:80]
_run_ffmpeg(
run_ffmpeg(
[
FFMPEG_BIN,
"-y",
@@ -173,19 +235,43 @@ def _create_fallback_clip(output_path: Path, title: str) -> None:
)
def _mux_audio_track(video_path: Path, audio_path: str, output_path: Path) -> None:
"""将音频轨混入已渲染的视频(后处理步骤)。
使用 FFmpeg 将视频和音频合并,视频时长为准,音频不足则循环,
音频过长则截断。
"""
command = [
FFMPEG_BIN,
"-y",
"-i",
str(video_path),
"-i",
audio_path,
"-c:v",
"copy",
"-c:a",
"aac",
"-b:a",
"192k",
"-shortest",
"-map",
"0:v:0",
"-map",
"1:a:0",
"-movflags",
"+faststart",
str(output_path),
]
run_ffmpeg(command)
def _download_voice_asset(voice_library_id: str, local_path: Path) -> bool:
"""下载配音文件"""
if not voice_library_id:
return False
bucket = _oss_bucket()
if bucket is None:
return False
storage_key = f"voice/{voice_library_id}.mp3"
try:
bucket.get_object_to_file(_normalize_storage_key(storage_key), str(local_path))
return local_path.exists() and local_path.stat().st_size > 0
except Exception:
return False
return download_asset(storage_key, local_path)
def _download_library_assets(
@@ -193,9 +279,8 @@ def _download_library_assets(
temp_path: Path,
video_extensions: tuple = (".mp4", ".mov", ".avi", ".mkv", ".webm"),
asset_ids: list[str] | None = None,
) -> list[str]:
"""
从素材库下载视频素材
) -> list[Path]:
"""从素材库下载视频素材。
Args:
asset_library_id: 素材库 ID
@@ -204,91 +289,63 @@ def _download_library_assets(
asset_ids: 指定素材 ID 列表,为空则下载全部 ready 视频素材
Returns:
下载成功的视频文件路径列表
下载成功的视频文件 Path 列表
"""
# 导入模型和会话
try:
from packages.adapters.sqlalchemy_impl.models import AssetModel
session = SessionLocal()
try:
# 查询素材库中的视频素材
query = session.query(AssetModel).filter(
AssetModel.asset_library_id == asset_library_id,
AssetModel.status == "ready",
AssetModel.file_type.in_(["video", "video/mp4", "video/quicktime"]),
)
# 如果指定了 asset_ids,则只下载这些素材
if asset_ids:
query = query.filter(AssetModel.id.in_(asset_ids))
assets = query.order_by(AssetModel.created_at).all()
if not assets:
logger.info(f"No video assets found in library {asset_library_id}")
logger.info("No video assets found in library %s", asset_library_id)
return []
downloaded_videos = []
downloaded: list[Path] = []
for i, asset in enumerate(assets):
# 获取文件 URL 或 storage_key
storage_key = asset.file_url if asset.file_url else None
if not storage_key:
continue
local_file = temp_path / f"asset_{i}_{asset.id}.mp4"
if _download_asset(storage_key, local_file):
downloaded_videos.append(str(local_file))
logger.info(f"Downloaded asset: {asset.name} -> {local_file}")
ext = Path(storage_key).suffix or ".mp4"
local_file = temp_path / f"asset_{i:03d}_{asset.id}{ext}"
if download_asset(storage_key, local_file):
downloaded.append(local_file)
logger.info("Downloaded asset: %s -> %s", asset.name, local_file)
else:
logger.warning(f"Failed to download asset: {asset.name}")
logger.warning("Failed to download asset: %s", asset.name)
return downloaded_videos
return downloaded
finally:
session.close()
except Exception as e:
logger.error(f"Error downloading library assets: {e}")
logger.error("Error downloading library assets: %s", e)
return []
def _process_with_editing_mode(
video_paths: list[str],
audio_path: Optional[str],
mode: str,
output_path: Path,
) -> None:
"""根据剪辑模式处理视频"""
from video_processing.editing_modes import (
EditingMode,
EditingModeConfig,
EditingModeProcessor,
PIPPosition,
)
config = EditingModeConfig(
mode=EditingMode(mode),
output_width=OUTPUT_WIDTH,
output_height=OUTPUT_HEIGHT,
output_fps=int(OUTPUT_FPS),
pip_position=PIPPosition.TOP_RIGHT,
pip_scale=0.25,
transition_duration=0.5,
)
processor = EditingModeProcessor(config=config)
processor.process(
video_paths=video_paths,
audio_path=audio_path,
output_path=str(output_path),
)
# ── Celery Task ──────────────────────────────────────────────────────────────
@celery_app.task(bind=True, name="worker.generate_video", max_retries=2)
def generate_video(self, task_id: str) -> dict:
"""
生成视频任务
"""生成视频任务 — 使用 UnifiedRenderService 统一渲染。
流程:
1. 加载 GenerationTask 信息
2. 从素材库下载视频素材
3. 根据模式构建虚拟 plan + clips
4. 使用 UnifiedRenderService 渲染
5. 如有配音,后处理混音
6. 上传 OSS + 查重
7. 更新 GenerationTask 状态
Args:
task_id: 任务 ID(从数据库加载完整任务信息)
@@ -337,35 +394,63 @@ def generate_video(self, task_id: str) -> dict:
temp_path = Path(temp_dir)
output_path = temp_path / output_name
# 从素材库下载视频素材(如果任务指定了 asset_ids 则只下载这些)
# 1. 从素材库下载视频素材
downloaded_videos = _download_library_assets(asset_library_id, temp_path, asset_ids=task_asset_ids or None)
audio_path = None
# 2. 下载配音(如有)
audio_path: str | None = None
if voice_library_id:
local_audio = temp_path / "voice.mp3"
if _download_voice_asset(voice_library_id, local_audio):
audio_path = str(local_audio)
# 3. 渲染
if downloaded_videos:
_process_with_editing_mode(
video_paths=downloaded_videos,
audio_path=audio_path,
# 构建虚拟 plan + clips + asset_path_map
virtual_plan, virtual_clips, asset_path_map = _build_plan_and_clips_from_task(
task_id=task_id,
downloaded_paths=downloaded_videos,
mode=editing_mode.value,
output_path=output_path,
)
# 使用 UnifiedRenderService 渲染
render_service = UnifiedRenderService(
plan=virtual_plan,
clips=virtual_clips,
asset_path_map=asset_path_map,
work_dir=temp_path,
output_width=OUTPUT_WIDTH,
output_height=OUTPUT_HEIGHT,
output_fps=int(OUTPUT_FPS),
)
render_result = render_service.render()
# 4. 如有配音,后处理混音
if audio_path:
final_path = temp_path / f"final-{task_id}.mp4"
try:
_mux_audio_track(render_result.output_path, audio_path, final_path)
# 混音成功,使用混音后的文件
output_path = final_path
except Exception as mux_err:
logger.warning("音频混合失败,使用无音频版本: %s", mux_err)
output_path = render_result.output_path
else:
output_path = render_result.output_path
else:
# 无素材,生成 fallback 视频
_create_fallback_clip(output_path, f"Generated Video {task_id[:8]}")
file_size = output_path.stat().st_size
duration = _probe_duration(output_path)
duration = probe_duration(output_path)
# 上传到 OSS
bucket = _oss_bucket()
# 5. 上传到 OSS
bucket = oss_bucket()
if bucket:
try:
bucket.put_object_from_file(storage_key, str(output_path))
except Exception as oss_err:
logger.warning(f"OSS upload failed: {oss_err}")
logger.warning("OSS upload failed: %s", oss_err)
# 构建视频 URL
if bucket:
@@ -373,19 +458,24 @@ def generate_video(self, task_id: str) -> dict:
else:
file_url = f"{GENERATED_FILES_URL_PREFIX}/{task_id}/{output_name}"
# 创建 GeneratedVideo 记录 + 查重
video_count = _create_video_record_and_dedup(
task_id=task_id,
project_id=project_id,
batch_id=batch_id,
file_url=file_url,
file_size=file_size,
duration=duration,
video_path=str(output_path),
mode=editing_mode.value,
)
# 6. 创建 GeneratedVideo 记录 + 查重
dedup_session = SessionLocal()
try:
video_count = create_video_record_and_dedup(
generation_task_id=task_id,
project_id=project_id,
batch_id=batch_id,
file_url=file_url,
file_size=file_size,
duration=duration,
video_path=str(output_path),
mode=editing_mode.value,
session=dedup_session,
)
finally:
dedup_session.close()
# 标记任务为 completed
# 7. 标记任务为 completed
_update_task_status(task_id, "mark_completed", result_count=video_count or 1)
logger.info("视频生成完成: task_id=%s duration=%.2fs file_size=%d", task_id, duration, file_size)
@@ -401,98 +491,10 @@ def generate_video(self, task_id: str) -> dict:
"mode": editing_mode.value,
}
except Exception as error:
logger.error(f"Video generation failed: {error}", exc_info=True)
# 标记任务为 failed
logger.error("Video generation failed: %s", error, exc_info=True)
_update_task_status(task_id, "mark_failed", error_message=str(error))
return {
"status": "failed",
"task_id": task_id,
"error": str(error),
}
def _create_video_record_and_dedup(
*,
task_id: str,
project_id: str,
batch_id: str,
file_url: str,
file_size: int,
duration: float,
video_path: str,
mode: str,
) -> int:
"""创建 GeneratedVideo 记录,计算指纹并执行查重(历史 + 批次)。
Returns:
创建的视频记录数量(1 表示成功,0 表示失败)
"""
from uuid import uuid4
from video_processing.dedup import VideoDeduplicator
from packages.adapters.sqlalchemy_impl.generated_video_repository import (
SQLAlchemyGeneratedVideoRepository,
)
from packages.domain import GeneratedVideo
session = SessionLocal()
try:
video_id = uuid4().hex
generated_video = GeneratedVideo(
id=video_id,
project_id=project_id,
generation_task_id=task_id,
name=f"generated-{task_id[:8]}.mp4",
file_url=file_url,
file_size=file_size,
duration=duration,
width=OUTPUT_WIDTH,
height=OUTPUT_HEIGHT,
fps=OUTPUT_FPS,
status="completed",
generation_params={"mode": mode},
)
video_repo = SQLAlchemyGeneratedVideoRepository(session)
video_repo.create(generated_video)
# 计算视频指纹
deduplicator = VideoDeduplicator()
try:
fingerprint = deduplicator.compute_fingerprint(video_path)
except Exception as fp_err:
logger.warning(f"Fingerprint computation failed for {video_id}: {fp_err}")
session.commit()
return 1
generated_video.video_fingerprint = fingerprint.to_dict()
# (a) 历史成片查重
duplicate_result = deduplicator.check_duplicate(fingerprint, project_id, session)
# (b) 批次内查重(仅当有 batch_id 时)
if not duplicate_result and batch_id:
duplicate_result = deduplicator.check_batch_duplicate(fingerprint, batch_id, video_id, session)
if duplicate_result:
generated_video.is_duplicate = True
generated_video.duplicate_of = duplicate_result["duplicate_of"]
logger.info(
f"Duplicate detected: {video_id} -> {duplicate_result['duplicate_of']} "
f"(reason={duplicate_result['reason']}, similarity={duplicate_result['similarity']:.3f})"
)
else:
generated_video.is_duplicate = False
generated_video.duplicate_of = None
video_repo.update(generated_video)
session.commit()
logger.info(f"GeneratedVideo record created: {video_id} (task={task_id}, dup={generated_video.is_duplicate})")
return 1
except Exception as e:
logger.error(f"Failed to create video record / dedup for task {task_id}: {e}")
session.rollback()
return 0
finally:
session.close()
+8
View File
@@ -1098,6 +1098,14 @@
"type": "VARCHAR(50)",
"unique": false
},
{
"index": false,
"name": "editing_mode",
"nullable": false,
"primary_key": false,
"type": "VARCHAR(20)",
"unique": false
},
{
"index": false,
"name": "config",
@@ -71,6 +71,7 @@ class SQLAlchemyEditTemplateRepository:
name=template.name,
description=template.description,
template_type=template.template_type,
editing_mode=template.editing_mode,
config=template.config,
preview_url=template.preview_url,
sort_weight=template.sort_weight,
@@ -89,6 +90,7 @@ class SQLAlchemyEditTemplateRepository:
model.name = template.name
model.description = template.description
model.template_type = template.template_type
model.editing_mode = template.editing_mode
model.config = template.config
model.preview_url = template.preview_url
model.sort_weight = template.sort_weight
@@ -128,6 +130,7 @@ class SQLAlchemyEditTemplateRepository:
name=model.name,
description=model.description or "",
template_type=model.template_type or "default",
editing_mode=model.editing_mode or "one_take",
config=model.config or {},
preview_url=model.preview_url or "",
sort_weight=model.sort_weight or 0,
@@ -126,6 +126,7 @@ class EditTemplateModel(Base):
name = Column(String(120), nullable=False)
description = Column(Text, nullable=False, default="")
template_type = Column(String(50), nullable=False, default="default", index=True)
editing_mode = Column(String(20), nullable=False, default="one_take")
config = Column(JSON, nullable=False, default=dict)
preview_url = Column(String(1000), nullable=False, default="")
sort_weight = Column(Integer, nullable=False, default=0, index=True)
+44 -7
View File
@@ -8,6 +8,7 @@
from __future__ import annotations
import copy
from enum import Enum
from typing import List, Optional
@@ -129,24 +130,29 @@ class EditPlanConfigSchema(BaseModel):
用于 API 层校验和默认值填充。所有子结构均可选,
未传入时使用各自默认值。
editing_mode 记录计划使用的剪辑模式。
"""
cover: CoverConfig = Field(default_factory=CoverConfig, description="封面配置")
title: TitleConfig = Field(default_factory=TitleConfig, description="标题配置")
subtitle: SubtitleConfig = Field(default_factory=SubtitleConfig, description="字幕配置")
bgm: BGMConfig = Field(default_factory=BGMConfig, description="BGM 配置")
editing_mode: str = Field(default="one_take", description="剪辑模式")
class EditTemplateConfigSchema(BaseModel):
"""EditTemplate.config 完整结构
模板级别的默认配置,创建计划时可作为初始值继承。
editing_mode 指定模板对应的剪辑模式,transition_enabled 控制是否启用转场。
"""
cover: CoverConfig = Field(default_factory=CoverConfig, description="封面默认配置")
title: TitleConfig = Field(default_factory=TitleConfig, description="标题默认配置")
subtitle: SubtitleConfig = Field(default_factory=SubtitleConfig, description="字幕默认配置")
bgm: BGMConfig = Field(default_factory=BGMConfig, description="BGM 默认配置")
editing_mode: str = Field(default="one_take", description="剪辑模式")
transition_enabled: bool = Field(default=True, description="是否启用转场")
# ── 默认值常量 ────────────────────────────────────────────────────────────────
@@ -183,9 +189,13 @@ DEFAULT_EDIT_PLAN_CONFIG: dict = {
"asset_id": "",
"volume": 0.3,
},
"editing_mode": "one_take",
}
DEFAULT_EDIT_TEMPLATE_CONFIG: dict = DEFAULT_EDIT_PLAN_CONFIG.copy()
DEFAULT_EDIT_TEMPLATE_CONFIG: dict = {
**DEFAULT_EDIT_PLAN_CONFIG,
"transition_enabled": True,
}
# ── 工具函数 ──────────────────────────────────────────────────────────────────
@@ -197,9 +207,7 @@ def normalize_plan_config(raw: dict | None) -> dict:
用于创建/更新计划时确保 config 结构完整。
"""
if raw is None:
return DEFAULT_EDIT_PLAN_CONFIG.copy()
import copy
return copy.deepcopy(DEFAULT_EDIT_PLAN_CONFIG)
base = copy.deepcopy(DEFAULT_EDIT_PLAN_CONFIG)
@@ -209,14 +217,43 @@ def normalize_plan_config(raw: dict | None) -> dict:
base[section_key] = {}
base[section_key].update(raw[section_key])
# editing_mode 顶层字段
if "editing_mode" in raw and isinstance(raw["editing_mode"], str):
base["editing_mode"] = raw["editing_mode"]
# 保留非标准字段(如 generation_task_id
for key, value in raw.items():
if key not in ("cover", "title", "subtitle", "bgm"):
if key not in ("cover", "title", "subtitle", "bgm", "editing_mode"):
base[key] = value
return base
def normalize_template_config(raw: dict | None) -> dict:
"""将模板原始 config dict 标准化。逻辑同 normalize_plan_config。"""
return normalize_plan_config(raw)
"""将模板原始 config dict 标准化。
在 plan config 基础上额外支持 transition_enabled 字段。
"""
if raw is None:
return copy.deepcopy(DEFAULT_EDIT_TEMPLATE_CONFIG)
base = copy.deepcopy(DEFAULT_EDIT_TEMPLATE_CONFIG)
for section_key in ("cover", "title", "subtitle", "bgm"):
if section_key in raw and isinstance(raw[section_key], dict):
if section_key not in base:
base[section_key] = {}
base[section_key].update(raw[section_key])
# 顶层字段
if "editing_mode" in raw and isinstance(raw["editing_mode"], str):
base["editing_mode"] = raw["editing_mode"]
if "transition_enabled" in raw and isinstance(raw["transition_enabled"], bool):
base["transition_enabled"] = raw["transition_enabled"]
# 保留非标准字段
for key, value in raw.items():
if key not in ("cover", "title", "subtitle", "bgm", "editing_mode", "transition_enabled"):
base[key] = value
return base
+16
View File
@@ -18,6 +18,8 @@ else:
from typing import Any
from uuid import uuid4
from .editing_mode import EditingMode
class EditTemplateStatus(StrEnum):
"""模板状态"""
@@ -26,18 +28,25 @@ class EditTemplateStatus(StrEnum):
INACTIVE = "inactive"
_VALID_EDITING_MODES = {m.value for m in EditingMode}
@dataclass(slots=True)
class EditTemplate:
"""Phase 8 剪辑模板实体
全局模板库中的模板,定义剪辑风格、配置参数和预览信息。
不绑定到具体项目,可被多个 EditPlan 引用。
editing_mode 指定模板对应的剪辑模式(one_take / pip / voice_over / voice_pip),
决定剪辑计划生成时的片段结构。
"""
id: str
name: str
description: str = ""
template_type: str = "default"
editing_mode: str = EditingMode.ONE_TAKE.value
config: dict[str, Any] = field(default_factory=dict)
preview_url: str = ""
sort_weight: int = 0
@@ -52,6 +61,7 @@ class EditTemplate:
*,
description: str = "",
template_type: str = "default",
editing_mode: str = EditingMode.ONE_TAKE.value,
config: dict[str, Any] | None = None,
preview_url: str = "",
sort_weight: int = 0,
@@ -61,11 +71,17 @@ class EditTemplate:
clean_name = name.strip()
if not clean_name:
raise ValueError("模板名称不能为空")
clean_mode = editing_mode.strip() or EditingMode.ONE_TAKE.value
if clean_mode not in _VALID_EDITING_MODES:
raise ValueError(
f"无效的 editing_mode: {clean_mode}" f"允许值: {', '.join(sorted(_VALID_EDITING_MODES))}"
)
return cls(
id=uuid4().hex,
name=clean_name,
description=description.strip(),
template_type=template_type.strip() or "default",
editing_mode=clean_mode,
config=config or {},
preview_url=preview_url.strip(),
sort_weight=sort_weight,
@@ -0,0 +1,236 @@
"""四模式渲染集成测试.
验证 4 种剪辑模式(ONE_TAKE / PIP / VOICE_OVER / VOICE_PIP)通过
_build_plan_and_clips_from_task + UnifiedRenderService 的完整渲染流程。
需要 ffmpeg 可用;CI 无 ffmpeg 时自动跳过。
"""
from __future__ import annotations
import shutil
import subprocess
import tempfile
from pathlib import Path
import pytest
from video_processing.unified_render_service import (
RenderResult,
UnifiedRenderService,
_resolve_layer_role,
)
from worker_app.tasks.generation import _build_plan_and_clips_from_task
pytestmark = pytest.mark.skipif(
not shutil.which("ffmpeg"),
reason="ffmpeg not available",
)
# ── 辅助函数 ──────────────────────────────────────────────────────────────────
def _generate_test_video(path: Path, duration: float = 3.0, color: str = "red") -> None:
"""生成一个纯色测试视频。"""
cmd = [
"ffmpeg",
"-y",
"-f",
"lavfi",
"-i",
f"color=c={color}:s=640x360:d={duration}:r=25",
"-c:v",
"libx264",
"-pix_fmt",
"yuv420p",
"-movflags",
"+faststart",
str(path),
]
subprocess.run(cmd, check=True, capture_output=True, timeout=30)
def _render_with_mode(
mode: str,
num_clips: int = 3,
duration: float = 2.0,
) -> tuple[RenderResult, Path]:
"""用指定模式生成测试视频并渲染,返回 (result, work_dir)。
调用方负责清理 work_dir。
"""
work_dir = Path(tempfile.mkdtemp(prefix="test_4mode_"))
# 生成测试视频素材
colors = ["red", "green", "blue", "yellow", "purple"]
downloaded_paths: list[Path] = []
for i in range(num_clips):
p = work_dir / f"test_{i:03d}.mp4"
_generate_test_video(p, duration=duration, color=colors[i % len(colors)])
downloaded_paths.append(p)
# 构建虚拟 plan + clips
task_id = f"test_task_{mode}"
plan, clips, asset_path_map = _build_plan_and_clips_from_task(
task_id=task_id,
downloaded_paths=downloaded_paths,
mode=mode,
)
# 渲染
service = UnifiedRenderService(
plan=plan,
clips=clips,
asset_path_map=asset_path_map,
work_dir=work_dir,
output_width=640,
output_height=360,
output_fps=25,
)
result = service.render()
return result, work_dir
# ── 测试 _build_plan_and_clips_from_task ──────────────────────────────────────
class TestBuildPlanAndClips:
"""测试 4 种模式的虚拟 plan 构建。"""
def _make_paths(self, n: int) -> list[Path]:
return [Path(f"/tmp/test_{i}.mp4") for i in range(n)]
def test_one_take_mode(self):
paths = self._make_paths(3)
plan, clips, asset_map = _build_plan_and_clips_from_task("t1", paths, "one_take")
assert plan.id == "t1"
assert len(clips) == 3
assert all(c.clip_type == "main" for c in clips)
assert len(asset_map) == 3
def test_pip_mode(self):
paths = self._make_paths(3)
plan, clips, asset_map = _build_plan_and_clips_from_task("t2", paths, "pip")
assert len(clips) == 3
assert clips[0].clip_type == "main"
assert clips[1].clip_type == "overlay"
assert clips[2].clip_type == "overlay"
def test_voice_over_mode(self):
paths = self._make_paths(3)
plan, clips, asset_map = _build_plan_and_clips_from_task("t3", paths, "voice_over")
assert len(clips) == 3
assert all(c.clip_type == "main" for c in clips)
assert all(c.config.get("role") == "b_roll" for c in clips)
def test_voice_pip_mode(self):
paths = self._make_paths(4)
plan, clips, asset_map = _build_plan_and_clips_from_task("t4", paths, "voice_pip")
assert len(clips) == 4
assert clips[0].clip_type == "background"
assert clips[1].clip_type == "corner_voice"
assert clips[2].clip_type == "b_roll"
assert clips[3].clip_type == "b_roll"
def test_unknown_mode_defaults_to_one_take(self):
paths = self._make_paths(2)
plan, clips, asset_map = _build_plan_and_clips_from_task("t5", paths, "unknown_mode")
assert len(clips) == 2
assert all(c.clip_type == "main" for c in clips)
def test_asset_path_map_keys_match_clip_asset_ids(self):
paths = self._make_paths(3)
_, clips, asset_map = _build_plan_and_clips_from_task("t6", paths, "one_take")
clip_asset_ids = {c.asset_id for c in clips}
map_keys = set(asset_map.keys())
assert clip_asset_ids == map_keys
# ── 测试图层分组(4 模式) ────────────────────────────────────────────────────
class TestFourModeLayerGrouping:
"""验证 4 种模式的 clip_type 分布经 _resolve_layer_role 后产生正确的图层。"""
def test_one_take_layers(self):
"""ONE_TAKE: 3 main → 1 main layer。"""
paths = [Path(f"/tmp/ot_{i}.mp4") for i in range(3)]
_, clips, _ = _build_plan_and_clips_from_task("ot", paths, "one_take")
roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips}
assert roles == {"main"}
def test_pip_layers(self):
"""PIP: 1 main + 2 overlay → main + overlay。"""
paths = [Path(f"/tmp/pip_{i}.mp4") for i in range(3)]
_, clips, _ = _build_plan_and_clips_from_task("pip", paths, "pip")
roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips}
assert roles == {"main", "overlay"}
def test_voice_over_layers(self):
"""VOICE_OVER: 3 main(b_roll) → broll。"""
paths = [Path(f"/tmp/vo_{i}.mp4") for i in range(3)]
_, clips, _ = _build_plan_and_clips_from_task("vo", paths, "voice_over")
roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips}
assert roles == {"broll"}
def test_voice_pip_layers(self):
"""VOICE_PIP: 1 bg + 1 corner_voice + 2 b_roll → 3 个图层。"""
paths = [Path(f"/tmp/vpip_{i}.mp4") for i in range(4)]
_, clips, _ = _build_plan_and_clips_from_task("vpip", paths, "voice_pip")
roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips}
assert roles == {"background", "corner_voice", "broll"}
# ── 端到端渲染测试(需要 ffmpeg) ─────────────────────────────────────────────
class TestEndToEndRendering:
"""4 种模式的完整渲染测试,验证输出文件存在且时长合理。"""
def test_one_take_render(self):
result, work_dir = _render_with_mode("one_take", num_clips=2, duration=2.0)
try:
assert result.output_path.exists()
assert result.file_size > 0
assert result.duration > 0
assert result.width == 640
assert result.height == 360
finally:
shutil.rmtree(work_dir, ignore_errors=True)
def test_pip_render(self):
result, work_dir = _render_with_mode("pip", num_clips=2, duration=2.0)
try:
assert result.output_path.exists()
assert result.file_size > 0
assert result.duration > 0
finally:
shutil.rmtree(work_dir, ignore_errors=True)
def test_voice_over_render(self):
result, work_dir = _render_with_mode("voice_over", num_clips=2, duration=2.0)
try:
assert result.output_path.exists()
assert result.file_size > 0
assert result.duration > 0
finally:
shutil.rmtree(work_dir, ignore_errors=True)
def test_voice_pip_render(self):
result, work_dir = _render_with_mode("voice_pip", num_clips=3, duration=2.0)
try:
assert result.output_path.exists()
assert result.file_size > 0
assert result.duration > 0
finally:
shutil.rmtree(work_dir, ignore_errors=True)
+238
View File
@@ -0,0 +1,238 @@
"""全链路集成测试.
验证 PlanGeneratorService → UnifiedRenderService → 查重 的端到端流程。
"""
from __future__ import annotations
import shutil
import subprocess
import tempfile
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from video_processing.unified_render_service import (
RenderResult,
UnifiedRenderService,
)
from worker_app.tasks.generation import (
OUTPUT_HEIGHT,
OUTPUT_WIDTH,
_build_plan_and_clips_from_task,
_create_fallback_clip,
_mux_audio_track,
)
pytestmark = pytest.mark.skipif(
not shutil.which("ffmpeg"),
reason="ffmpeg not available",
)
def _generate_test_video(path: Path, duration: float = 3.0) -> None:
"""生成一个测试视频。"""
cmd = [
"ffmpeg",
"-y",
"-f",
"lavfi",
"-i",
f"color=c=blue:s=640x360:d={duration}:r=25",
"-c:v",
"libx264",
"-pix_fmt",
"yuv420p",
str(path),
]
subprocess.run(cmd, check=True, capture_output=True, timeout=30)
def _generate_test_audio(path: Path, duration: float = 5.0) -> None:
"""生成一个测试音频文件。"""
cmd = [
"ffmpeg",
"-y",
"-f",
"lavfi",
"-i",
f"sine=frequency=440:duration={duration}",
"-c:a",
"aac",
"-b:a",
"128k",
str(path),
]
subprocess.run(cmd, check=True, capture_output=True, timeout=30)
# ── 测试 _create_fallback_clip ────────────────────────────────────────────────
class TestFallbackClip:
"""测试 fallback 视频生成。"""
def test_fallback_clip_creates_video(self):
with tempfile.TemporaryDirectory() as tmpdir:
output = Path(tmpdir) / "fallback.mp4"
_create_fallback_clip(output, "Test Fallback")
assert output.exists()
assert output.stat().st_size > 0
# ── 测试 _mux_audio_track ────────────────────────────────────────────────────
class TestMuxAudioTrack:
"""测试视频+音频混合。"""
def test_mux_audio_into_video(self):
with tempfile.TemporaryDirectory() as tmpdir:
video_path = Path(tmpdir) / "video.mp4"
audio_path = Path(tmpdir) / "audio.aac"
output_path = Path(tmpdir) / "output.mp4"
_generate_test_video(video_path, duration=3.0)
_generate_test_audio(audio_path, duration=5.0)
_mux_audio_track(video_path, str(audio_path), output_path)
assert output_path.exists()
assert output_path.stat().st_size > 0
# 验证输出文件包含音频轨
probe_cmd = [
"ffprobe",
"-v",
"quiet",
"-show_streams",
"-select_streams",
"a",
"-of",
"csv=p=0",
str(output_path),
]
result = subprocess.run(probe_cmd, capture_output=True, text=True, timeout=10)
# 如果有音频流,输出非空
assert result.stdout.strip() != "" or result.returncode == 0
# ── 测试 PlanGenerator → UnifiedRenderService 全链路 ─────────────────────────
class TestFullPipeline:
"""验证从虚拟 plan 构建到渲染输出的完整流程。"""
def test_one_take_pipeline(self):
"""ONE_TAKE 模式完整流程。"""
with tempfile.TemporaryDirectory() as tmpdir:
work_dir = Path(tmpdir)
# 生成测试素材
paths = []
for i in range(3):
p = work_dir / f"clip_{i}.mp4"
_generate_test_video(p, duration=2.0)
paths.append(p)
# 构建虚拟 plan
plan, clips, asset_map = _build_plan_and_clips_from_task("pipeline_test", paths, "one_take")
# 渲染
service = UnifiedRenderService(
plan=plan,
clips=clips,
asset_path_map=asset_map,
work_dir=work_dir,
output_width=640,
output_height=360,
)
result = service.render()
assert result.output_path.exists()
assert result.duration > 0
assert result.file_size > 0
assert result.width == 640
assert result.height == 360
def test_pipeline_with_audio_mux(self):
"""渲染 + 混音后处理。"""
with tempfile.TemporaryDirectory() as tmpdir:
work_dir = Path(tmpdir)
# 生成测试素材
video_path = work_dir / "clip_0.mp4"
_generate_test_video(video_path, duration=3.0)
# 构建虚拟 plan
plan, clips, asset_map = _build_plan_and_clips_from_task("audio_test", [video_path], "one_take")
# 渲染
service = UnifiedRenderService(
plan=plan,
clips=clips,
asset_path_map=asset_map,
work_dir=work_dir,
output_width=640,
output_height=360,
)
render_result = service.render()
# 混音
audio_path = work_dir / "voice.aac"
_generate_test_audio(audio_path, duration=5.0)
final_path = work_dir / "final.mp4"
_mux_audio_track(render_result.output_path, str(audio_path), final_path)
assert final_path.exists()
assert final_path.stat().st_size > 0
def test_single_clip_pipeline(self):
"""单 clip 渲染(无转场)。"""
with tempfile.TemporaryDirectory() as tmpdir:
work_dir = Path(tmpdir)
video_path = work_dir / "single.mp4"
_generate_test_video(video_path, duration=5.0)
plan, clips, asset_map = _build_plan_and_clips_from_task("single_test", [video_path], "one_take")
service = UnifiedRenderService(
plan=plan,
clips=clips,
asset_path_map=asset_map,
work_dir=work_dir,
output_width=640,
output_height=360,
)
result = service.render()
assert result.output_path.exists()
assert result.duration > 0
def test_dedup_helper_integration(self):
"""验证 dedup_helpers.create_video_record_and_dedup 的导入和签名。"""
# 只验证函数存在且签名正确(不实际调用,需要数据库)
import inspect
from video_processing.dedup_helpers import create_video_record_and_dedup
sig = inspect.signature(create_video_record_and_dedup)
params = set(sig.parameters.keys())
expected = {
"generation_task_id",
"project_id",
"batch_id",
"file_url",
"file_size",
"duration",
"video_path",
"mode",
"session",
"width",
"height",
"fps",
}
assert expected.issubset(params), f"Missing params: {expected - params}"
@@ -105,11 +105,19 @@ class TestNormalizePlanConfig:
class TestNormalizeTemplateConfig:
def test_same_as_plan_config(self):
def test_same_as_plan_config_plus_template_fields(self):
"""template config 包含 plan config 的所有字段,外加 transition_enabled"""
from packages.domain.config_schemas import normalize_plan_config, normalize_template_config
raw = {"title": {"text": "模板标题"}}
assert normalize_template_config(raw) == normalize_plan_config(raw)
plan_cfg = normalize_plan_config(raw)
tpl_cfg = normalize_template_config(raw)
# plan config 的字段在 template config 中应一致
for key in plan_cfg:
assert tpl_cfg[key] == plan_cfg[key]
# template config 额外包含 transition_enabled
assert "transition_enabled" in tpl_cfg
assert tpl_cfg["transition_enabled"] is True
def test_none_returns_defaults(self):
from packages.domain.config_schemas import DEFAULT_EDIT_TEMPLATE_CONFIG, normalize_template_config
+1
View File
@@ -407,6 +407,7 @@ class TestResponseSchema:
"name",
"description",
"template_type",
"editing_mode",
"config",
"preview_url",
"sort_weight",
+598
View File
@@ -0,0 +1,598 @@
"""
PlanGeneratorService 单元测试
覆盖(6 组测试):
- ONE_TAKE 模式:素材顺序分配给 main clips
- PIP 模式:第1个素材→main,其余→overlay
- VOICE_OVER 模式:素材→main clips (B-roll)
- VOICE_PIP 模式:第1个→background, 第2个→corner_voice, 其余→b_roll
- 无 clip_configs 时自动生成默认结构
- 空素材列表时 clips 创建但无素材分配
"""
from __future__ import annotations
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, List, Optional
from unittest.mock import MagicMock, patch
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
from packages.domain.edit_plan import EditPlan, EditPlanStatus
from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus
from packages.domain.edit_template import EditTemplate, EditTemplateStatus
from packages.domain.editing_mode import EditingMode
from packages.domain.template_clip_config import ClipType, TemplateClipConfig, TransitionEffect
# ---------------------------------------------------------------------------
# Stub Repositories
# ---------------------------------------------------------------------------
class StubEditPlanRepository:
"""内存中的 EditPlan 仓储 stub"""
def __init__(self) -> None:
self._plans: dict[str, EditPlan] = {}
self._counter = 0
def _next_id(self) -> str:
self._counter += 1
return f"plan-{self._counter:03d}"
def get(self, plan_id: str) -> Optional[EditPlan]:
return self._plans.get(plan_id)
def create(self, plan: EditPlan) -> EditPlan:
if not plan.id:
plan = EditPlan(
id=self._next_id(),
template_id=plan.template_id,
name=plan.name,
status=plan.status,
total_duration=plan.total_duration,
source_edit_plan_id=plan.source_edit_plan_id,
project_id=plan.project_id,
created_by_user_id=plan.created_by_user_id,
config=plan.config,
created_at=plan.created_at,
updated_at=plan.updated_at,
)
self._plans[plan.id] = plan
return plan
def update(self, plan: EditPlan) -> EditPlan:
self._plans[plan.id] = plan
return plan
def list_all(self, **kwargs) -> List[EditPlan]:
return list(self._plans.values())
def count(self, **kwargs) -> int:
return len(self._plans)
def delete(self, plan_id: str) -> bool:
return self._plans.pop(plan_id, None) is not None
class StubEditPlanClipRepository:
"""内存中的 EditPlanClip 仓储 stub"""
def __init__(self) -> None:
self._clips: dict[str, EditPlanClip] = {}
self._counter = 0
def _next_id(self) -> str:
self._counter += 1
return f"clip-{self._counter:03d}"
def get(self, clip_id: str) -> Optional[EditPlanClip]:
return self._clips.get(clip_id)
def create(self, clip: EditPlanClip) -> EditPlanClip:
if not clip.id:
clip = EditPlanClip(
id=self._next_id(),
plan_id=clip.plan_id,
template_clip_config_id=clip.template_clip_config_id,
clip_type=clip.clip_type,
order=clip.order,
asset_id=clip.asset_id,
text_content=clip.text_content,
start_time=clip.start_time,
duration=clip.duration,
transition_effect=clip.transition_effect,
status=clip.status,
config=clip.config,
created_at=clip.created_at,
updated_at=clip.updated_at,
)
self._clips[clip.id] = clip
return clip
def update(self, clip: EditPlanClip) -> EditPlanClip:
self._clips[clip.id] = clip
return clip
def list_by_plan(
self,
plan_id: str,
*,
status: Optional[EditPlanClipStatus] = None,
skip: int = 0,
limit: int = 100,
) -> List[EditPlanClip]:
items = [c for c in self._clips.values() if c.plan_id == plan_id]
items.sort(key=lambda c: c.order)
if status:
items = [c for c in items if c.status == status]
return items[skip : skip + limit]
def delete(self, clip_id: str) -> bool:
return self._clips.pop(clip_id, None) is not None
# ---------------------------------------------------------------------------
# Helper: 构建 PlanGeneratorServicepatch 仓储)
# ---------------------------------------------------------------------------
def _make_generator():
"""创建使用 stub 仓储的 PlanGeneratorService"""
from apps.api.app.services.plan_generator_service import PlanGeneratorService
plan_repo = StubEditPlanRepository()
clip_repo = StubEditPlanClipRepository()
with (
patch(
"apps.api.app.services.plan_generator_service.SQLAlchemyEditPlanRepository",
return_value=plan_repo,
),
patch(
"apps.api.app.services.plan_generator_service.SQLAlchemyEditPlanClipRepository",
return_value=clip_repo,
),
):
db = MagicMock()
svc = PlanGeneratorService(db)
# 替换为 stub
svc._plan_repo = plan_repo
svc._clip_repo = clip_repo
return svc, plan_repo, clip_repo
def _make_template(
editing_mode: str = "one_take",
config: Optional[dict] = None,
) -> EditTemplate:
"""创建测试用 EditTemplate"""
return EditTemplate(
id="tpl-001",
name="测试模板",
description="",
template_type="default",
editing_mode=editing_mode,
config=config or {},
preview_url="",
sort_weight=0,
status=EditTemplateStatus.ACTIVE,
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
)
def _make_clip_configs(
template_id: str = "tpl-001",
specs: Optional[List[dict]] = None,
) -> List[TemplateClipConfig]:
"""创建测试用 TemplateClipConfig 列表
specs 示例: [{"clip_type": ClipType.INTRO, "order": 0}, ...]
"""
if specs is None:
specs = [
{"clip_type": ClipType.INTRO, "order": 0, "min_duration": 2.0, "max_duration": 4.0},
{"clip_type": ClipType.MAIN, "order": 1, "min_duration": 3.0, "max_duration": 7.0},
{"clip_type": ClipType.OUTRO, "order": 2, "min_duration": 2.0, "max_duration": 4.0},
]
configs = []
for i, spec in enumerate(specs):
cfg = TemplateClipConfig(
id=f"cfg-{i:03d}",
template_id=template_id,
clip_type=spec.get("clip_type", ClipType.MAIN),
order=spec.get("order", i),
min_duration=spec.get("min_duration", 0.0),
max_duration=spec.get("max_duration", 0.0),
text_template=spec.get("text_template", ""),
material_requirements=spec.get("material_requirements"),
transition_effect=spec.get("transition_effect", TransitionEffect.CUT),
config=spec.get("config"),
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
)
configs.append(cfg)
return configs
# ---------------------------------------------------------------------------
# 测试:ONE_TAKE 模式
# ---------------------------------------------------------------------------
class TestGenerateOneTakePlan:
"""ONE_TAKE 模式:素材顺序分配给 main clips"""
def test_generate_one_take_plan(self):
"""3个clip_configs + 3个asset_ids → 按顺序分配"""
svc, plan_repo, clip_repo = _make_generator()
template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)
clip_configs = _make_clip_configs()
asset_ids = ["asset-1", "asset-2", "asset-3"]
result = svc.generate_from_template(
template=template,
clip_configs=clip_configs,
asset_ids=asset_ids,
project_id="proj-001",
created_by_user_id="user-001",
)
plan = result["plan"]
clips = result["clips"]
assert plan.template_id == "tpl-001"
assert plan.config["editing_mode"] == "one_take"
assert plan.status == EditPlanStatus.EDITING
assert len(clips) == 3
# 按 order 排序后检查素材分配
sorted_clips = sorted(clips, key=lambda c: c.order)
# intro clip (order=0) 不是 main 类型,不分配素材
# main clip (order=1) → asset-2ONE_TAKE 只分配给 main clips
# outro clip (order=2) 不是 main 类型
main_clips = [c for c in sorted_clips if c.clip_type == ClipType.MAIN.value]
assert len(main_clips) == 1
assert main_clips[0].asset_id == "asset-1"
def test_one_take_plan_name_from_template(self):
"""name 为空时自动取模板名"""
svc, _, _ = _make_generator()
template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)
result = svc.generate_from_template(
template=template,
clip_configs=[],
asset_ids=["asset-1"],
)
assert "测试模板" in result["plan"].name
def test_one_take_custom_name(self):
"""指定 name 时使用自定义名称"""
svc, _, _ = _make_generator()
template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)
result = svc.generate_from_template(
template=template,
clip_configs=[],
asset_ids=["asset-1"],
name="我的剪辑",
)
assert result["plan"].name == "我的剪辑"
# ---------------------------------------------------------------------------
# 测试:PIP 模式
# ---------------------------------------------------------------------------
class TestGeneratePipPlan:
"""PIP 模式:第1个素材→main,其余→overlay"""
def test_generate_pip_plan(self):
"""4个素材 → 第1个→main,其余→overlay"""
svc, _, _ = _make_generator()
template = _make_template(editing_mode=EditingMode.PIP.value)
# 无 clip_configs,自动生成默认结构
asset_ids = ["bg-asset", "overlay-1", "overlay-2", "overlay-3"]
result = svc.generate_from_template(
template=template,
clip_configs=[],
asset_ids=asset_ids,
)
plan = result["plan"]
clips = result["clips"]
assert plan.config["editing_mode"] == "pip"
# 自动生成: 1 main + 3 overlay
assert len(clips) == 4
main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
overlay_clips = [c for c in clips if c.clip_type == "overlay"]
assert len(main_clips) == 1
assert len(overlay_clips) == 3
# 第1个素材 → main
assert main_clips[0].asset_id == "bg-asset"
# 其余 → overlay
assert overlay_clips[0].asset_id == "overlay-1"
assert overlay_clips[1].asset_id == "overlay-2"
assert overlay_clips[2].asset_id == "overlay-3"
# ---------------------------------------------------------------------------
# 测试:VOICE_OVER 模式
# ---------------------------------------------------------------------------
class TestGenerateVoiceOverPlan:
"""VOICE_OVER 模式:素材→main clips (B-roll)"""
def test_generate_voice_over_plan(self):
"""3个素材 → 3个 main clips,每个标记为 b_roll"""
svc, _, _ = _make_generator()
template = _make_template(editing_mode=EditingMode.VOICE_OVER.value)
asset_ids = ["video-1", "video-2", "video-3"]
result = svc.generate_from_template(
template=template,
clip_configs=[],
asset_ids=asset_ids,
)
plan = result["plan"]
clips = result["clips"]
assert plan.config["editing_mode"] == "voice_over"
assert len(clips) == 3
# 所有 clips 都是 main 类型
for clip in clips:
assert clip.clip_type == ClipType.MAIN.value
# 素材按顺序分配
assert clips[0].asset_id == "video-1"
assert clips[1].asset_id == "video-2"
assert clips[2].asset_id == "video-3"
# 每个 clip 的 config 标记为 b_roll
for clip in clips:
assert clip.config.get("role") == "b_roll"
# ---------------------------------------------------------------------------
# 测试:VOICE_PIP 模式
# ---------------------------------------------------------------------------
class TestGenerateVoicePipPlan:
"""VOICE_PIP 模式:第1个→background, 第2个→corner_voice, 其余→b_roll"""
def test_generate_voice_pip_plan(self):
"""4个素材 → background + corner_voice + 2 b_roll"""
svc, _, _ = _make_generator()
template = _make_template(editing_mode=EditingMode.VOICE_PIP.value)
asset_ids = ["bg-video", "corner-video", "broll-1", "broll-2"]
result = svc.generate_from_template(
template=template,
clip_configs=[],
asset_ids=asset_ids,
)
plan = result["plan"]
clips = result["clips"]
assert plan.config["editing_mode"] == "voice_pip"
assert len(clips) == 4
bg_clips = [c for c in clips if c.clip_type == "background"]
corner_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"]
assert len(bg_clips) == 1
assert len(corner_clips) == 1
assert len(broll_clips) == 2
# 素材分配
assert bg_clips[0].asset_id == "bg-video"
assert corner_clips[0].asset_id == "corner-video"
assert broll_clips[0].asset_id == "broll-1"
assert broll_clips[1].asset_id == "broll-2"
# ---------------------------------------------------------------------------
# 测试:无 clip_configs 时自动生成默认结构
# ---------------------------------------------------------------------------
class TestGenerateWithoutClipConfigs:
"""无 clip_configs 时根据 editing_mode 生成默认 clip 结构"""
def test_one_take_default_clips(self):
"""ONE_TAKE + 3个素材 → 3个 main clips"""
svc, _, _ = _make_generator()
template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)
result = svc.generate_from_template(
template=template,
clip_configs=[],
asset_ids=["a1", "a2", "a3"],
)
clips = result["clips"]
assert len(clips) == 3
for clip in clips:
assert clip.clip_type == ClipType.MAIN.value
def test_pip_default_clips(self):
"""PIP + 3个素材 → 1 main + 2 overlay"""
svc, _, _ = _make_generator()
template = _make_template(editing_mode=EditingMode.PIP.value)
result = svc.generate_from_template(
template=template,
clip_configs=[],
asset_ids=["a1", "a2", "a3"],
)
clips = result["clips"]
assert len(clips) == 3
main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
overlay_clips = [c for c in clips if c.clip_type == "overlay"]
assert len(main_clips) == 1
assert len(overlay_clips) == 2
def test_voice_pip_default_clips(self):
"""VOICE_PIP + 4个素材 → 1 background + 1 corner_voice + 2 b_roll"""
svc, _, _ = _make_generator()
template = _make_template(editing_mode=EditingMode.VOICE_PIP.value)
result = svc.generate_from_template(
template=template,
clip_configs=[],
asset_ids=["a1", "a2", "a3", "a4"],
)
clips = result["clips"]
assert len(clips) == 4
# ---------------------------------------------------------------------------
# 测试:空素材列表
# ---------------------------------------------------------------------------
class TestGenerateEmptyAssets:
"""空素材列表时 clips 创建但无素材分配"""
def test_empty_assets(self):
"""空 asset_ids → clips 创建但 asset_id 为空"""
svc, _, _ = _make_generator()
template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)
clip_configs = _make_clip_configs()
result = svc.generate_from_template(
template=template,
clip_configs=clip_configs,
asset_ids=[],
)
plan = result["plan"]
clips = result["clips"]
assert plan.total_duration > 0 # clips 有默认时长
assert len(clips) == 3
for clip in clips:
assert clip.asset_id == ""
def test_empty_assets_pip(self):
"""PIP 模式空素材 → 1个 main clip(至少1个)"""
svc, _, _ = _make_generator()
template = _make_template(editing_mode=EditingMode.PIP.value)
result = svc.generate_from_template(
template=template,
clip_configs=[],
asset_ids=[],
)
clips = result["clips"]
# 至少1个 main clipn = max(asset_count, 1) = 1
assert len(clips) == 1
assert clips[0].clip_type == ClipType.MAIN.value
# ---------------------------------------------------------------------------
# 测试:plan config 继承模板配置
# ---------------------------------------------------------------------------
class TestPlanConfigInheritance:
"""plan config 继承模板的 cover/title/subtitle/bgm"""
def test_inherit_template_config(self):
"""模板有 cover/title/bgm 配置 → plan 继承"""
svc, _, _ = _make_generator()
template_config = {
"editing_mode": "one_take",
"cover": {"type": "ai_frame"},
"title": {"text": "测试标题", "font": "思源黑体"},
"bgm": {"url": "https://example.com/bgm.mp3"},
}
template = _make_template(
editing_mode=EditingMode.ONE_TAKE.value,
config=template_config,
)
result = svc.generate_from_template(
template=template,
clip_configs=[],
asset_ids=["a1"],
)
plan_config = result["plan"].config
assert plan_config["editing_mode"] == "one_take"
assert plan_config["cover"]["type"] == "ai_frame"
assert plan_config["title"]["text"] == "测试标题"
assert plan_config["bgm"]["url"] == "https://example.com/bgm.mp3"
# ---------------------------------------------------------------------------
# 测试:total_duration 计算
# ---------------------------------------------------------------------------
class TestTotalDuration:
"""total_duration 正确计算"""
def test_duration_from_clip_configs(self):
"""有 clip_configs 时,duration 取 min/max 中间值"""
svc, _, _ = _make_generator()
template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)
clip_configs = _make_clip_configs(
specs=[
{"clip_type": ClipType.INTRO, "order": 0, "min_duration": 2.0, "max_duration": 4.0},
{"clip_type": ClipType.MAIN, "order": 1, "min_duration": 4.0, "max_duration": 6.0},
{"clip_type": ClipType.OUTRO, "order": 2, "min_duration": 2.0, "max_duration": 4.0},
]
)
result = svc.generate_from_template(
template=template,
clip_configs=clip_configs,
asset_ids=["a1"],
)
# intro: (2+4)/2=3, main: (4+6)/2=5, outro: (2+4)/2=3 → total=11
assert result["plan"].total_duration == 11.0
def test_duration_from_default_clips(self):
"""无 clip_configs 时,每个 clip 默认 5 秒"""
svc, _, _ = _make_generator()
template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)
result = svc.generate_from_template(
template=template,
clip_configs=[],
asset_ids=["a1", "a2", "a3"],
)
# 3 个 clips × 5 秒 = 15 秒
assert result["plan"].total_duration == 15.0
+190
View File
@@ -0,0 +1,190 @@
"""
EditTemplate editing_mode 字段单元测试
覆盖:
- EditTemplate.create() 带 editing_mode
- 默认值 "one_take"
- 无效 editing_mode 抛 ValueError
- EditingMode 枚举值完整性
- config_schemas 中 editing_mode 和 transition_enabled 字段
"""
from __future__ import annotations
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
from packages.domain.config_schemas import (
DEFAULT_EDIT_PLAN_CONFIG,
DEFAULT_EDIT_TEMPLATE_CONFIG,
normalize_plan_config,
normalize_template_config,
)
from packages.domain.edit_template import EditTemplate, EditTemplateStatus
from packages.domain.editing_mode import EditingMode
# ---------------------------------------------------------------------------
# 测试:EditTemplate.create() 带 editing_mode
# ---------------------------------------------------------------------------
class TestEditTemplateEditingMode:
"""EditTemplate editing_mode 字段测试"""
def test_create_with_editing_mode(self):
"""create() 指定 editing_mode"""
tpl = EditTemplate.create(
name="测试模板",
editing_mode=EditingMode.PIP.value,
)
assert tpl.editing_mode == "pip"
def test_create_default_editing_mode(self):
"""create() 不指定 editing_mode → 默认 one_take"""
tpl = EditTemplate.create(name="默认模式模板")
assert tpl.editing_mode == "one_take"
def test_create_all_editing_modes(self):
"""所有 EditingMode 枚举值均可创建"""
for mode in EditingMode:
tpl = EditTemplate.create(
name=f"模板-{mode.value}",
editing_mode=mode.value,
)
assert tpl.editing_mode == mode.value
def test_create_invalid_editing_mode(self):
"""无效 editing_mode 抛 ValueError"""
with pytest.raises(ValueError, match="无效的 editing_mode"):
EditTemplate.create(
name="无效模板",
editing_mode="invalid_mode",
)
def test_editing_mode_case_sensitive(self):
"""editing_mode 大小写敏感"""
with pytest.raises(ValueError):
EditTemplate.create(
name="大写模式",
editing_mode="ONE_TAKE", # 应小写
)
# ---------------------------------------------------------------------------
# 测试:EditingMode 枚举完整性
# ---------------------------------------------------------------------------
class TestEditingModeEnum:
"""EditingMode 枚举值测试"""
def test_enum_values(self):
"""枚举包含4种模式"""
assert EditingMode.ONE_TAKE.value == "one_take"
assert EditingMode.PIP.value == "pip"
assert EditingMode.VOICE_OVER.value == "voice_over"
assert EditingMode.VOICE_PIP.value == "voice_pip"
def test_enum_count(self):
"""枚举共4个成员"""
assert len(EditingMode) == 4
def test_str_enum(self):
"""EditingMode 是 StrEnum"""
assert isinstance(EditingMode.ONE_TAKE, str)
assert EditingMode.ONE_TAKE == "one_take"
# ---------------------------------------------------------------------------
# 测试:config_schemas 中的 editing_mode
# ---------------------------------------------------------------------------
class TestConfigSchemasEditingMode:
"""config_schemas editing_mode 和 transition_enabled 字段测试"""
def test_default_plan_config_has_editing_mode(self):
"""DEFAULT_EDIT_PLAN_CONFIG 包含 editing_mode"""
assert "editing_mode" in DEFAULT_EDIT_PLAN_CONFIG
assert DEFAULT_EDIT_PLAN_CONFIG["editing_mode"] == "one_take"
def test_default_template_config_has_editing_mode(self):
"""DEFAULT_EDIT_TEMPLATE_CONFIG 包含 editing_mode"""
assert "editing_mode" in DEFAULT_EDIT_TEMPLATE_CONFIG
assert DEFAULT_EDIT_TEMPLATE_CONFIG["editing_mode"] == "one_take"
def test_default_template_config_has_transition_enabled(self):
"""DEFAULT_EDIT_TEMPLATE_CONFIG 包含 transition_enabled"""
assert "transition_enabled" in DEFAULT_EDIT_TEMPLATE_CONFIG
assert DEFAULT_EDIT_TEMPLATE_CONFIG["transition_enabled"] is True
def test_normalize_plan_config_editing_mode(self):
"""normalize_plan_config 处理 editing_mode"""
config = normalize_plan_config({"editing_mode": "pip"})
assert config["editing_mode"] == "pip"
def test_normalize_plan_config_default_editing_mode(self):
"""normalize_plan_config 空输入 → 默认 editing_mode"""
config = normalize_plan_config({})
assert config["editing_mode"] == "one_take"
def test_normalize_template_config_editing_mode(self):
"""normalize_template_config 处理 editing_mode"""
config = normalize_template_config({"editing_mode": "voice_pip"})
assert config["editing_mode"] == "voice_pip"
def test_normalize_template_config_transition_enabled(self):
"""normalize_template_config 处理 transition_enabled"""
config = normalize_template_config({"transition_enabled": False})
assert config["transition_enabled"] is False
def test_normalize_template_config_defaults(self):
"""normalize_template_config 空输入 → 默认值"""
config = normalize_template_config({})
assert config["editing_mode"] == "one_take"
assert config["transition_enabled"] is True
# ---------------------------------------------------------------------------
# 测试:EditTemplate 实体直接构造
# ---------------------------------------------------------------------------
class TestEditTemplateEntityDirect:
"""直接构造 EditTemplate 实体测试 editing_mode"""
def test_direct_construction(self):
"""直接构造带 editing_mode 的实体"""
now = datetime.now(timezone.utc)
tpl = EditTemplate(
id="tpl-test",
name="直接构造",
description="",
template_type="default",
editing_mode="voice_over",
config={},
preview_url="",
sort_weight=0,
status=EditTemplateStatus.ACTIVE,
created_at=now,
updated_at=now,
)
assert tpl.editing_mode == "voice_over"
def test_all_enum_values_accepted(self):
"""所有 EditingMode 枚举值均可通过 create() 验证"""
for mode in EditingMode:
tpl = EditTemplate.create(
name=f"模板-{mode.value}",
editing_mode=mode.value,
)
assert tpl.editing_mode == mode.value
+404
View File
@@ -0,0 +1,404 @@
"""UnifiedRenderService 单元测试.
测试图层分组算法、filter_complex 构建、以及渲染流程。
"""
from __future__ import annotations
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from unittest.mock import MagicMock, patch
import pytest
from video_processing.unified_render_service import (
RenderLayer,
RenderResult,
ResolvedClip,
UnifiedRenderService,
_resolve_layer_role,
)
# ── Fixtures ──────────────────────────────────────────────────────────────────
@dataclass
class FakeClip:
"""模拟 EditPlanClip。"""
id: str
plan_id: str = "plan_001"
clip_type: str = "main"
order: int = 0
asset_id: str = ""
text_content: str = ""
start_time: float = 0.0
duration: float = 0.0
transition_effect: str = "cut"
status: str = "ready"
config: dict[str, Any] = field(default_factory=dict)
@dataclass
class FakePlan:
"""模拟 EditPlan。"""
id: str = "plan_001"
name: str = "测试计划"
def _make_clip(
clip_id: str,
clip_type: str = "main",
order: int = 0,
asset_id: str = "",
duration: float = 0.0,
transition_effect: str = "cut",
config: dict[str, Any] | None = None,
) -> FakeClip:
return FakeClip(
id=clip_id,
clip_type=clip_type,
order=order,
asset_id=asset_id or f"asset_{clip_id}.mp4",
duration=duration,
transition_effect=transition_effect,
config=config or {},
)
def _make_service(
clips: list[FakeClip] | None = None,
asset_paths: dict[str, Path] | None = None,
work_dir: Path | None = None,
) -> UnifiedRenderService:
"""创建测试用的 UnifiedRenderService 实例。
如果未提供 asset_paths,自动从 clips 生成默认映射
asset_id → /tmp/asset_{clip_id}.mp4)。
"""
plan = FakePlan()
clips = clips or []
work_dir = work_dir or Path("/tmp/test_render")
if asset_paths is None:
asset_paths = {}
for c in clips:
if c.asset_id:
asset_paths[c.asset_id] = Path(f"/tmp/{c.asset_id}")
return UnifiedRenderService(
plan=plan,
clips=clips,
asset_path_map=asset_paths,
work_dir=work_dir,
)
def _patch_path_exists():
"""Patch Path.exists() 让测试路径返回 True。"""
return patch("pathlib.Path.exists", return_value=True)
# ── 测试 _resolve_layer_role ─────────────────────────────────────────────────
class TestResolveLayerRole:
"""测试 clip_type → layer role 映射。"""
def test_main_default(self):
assert _resolve_layer_role("main", {}) == "main"
def test_main_with_b_roll_role(self):
assert _resolve_layer_role("main", {"role": "b_roll"}) == "broll"
def test_overlay(self):
assert _resolve_layer_role("overlay", {}) == "overlay"
def test_background(self):
assert _resolve_layer_role("background", {}) == "background"
def test_corner_voice(self):
assert _resolve_layer_role("corner_voice", {}) == "corner_voice"
def test_b_roll(self):
assert _resolve_layer_role("b_roll", {}) == "broll"
def test_intro(self):
assert _resolve_layer_role("intro", {}) == "main"
def test_outro(self):
assert _resolve_layer_role("outro", {}) == "main"
# ── 测试图层分组 ──────────────────────────────────────────────────────────────
class TestGroupClipsIntoLayers:
"""测试 _group_clips_into_layers 方法。"""
def test_group_clips_one_take(self):
"""4 个 main clips → 1 个 main_layer。"""
clips = [
_make_clip("c1", "main", order=0),
_make_clip("c2", "main", order=1),
_make_clip("c3", "main", order=2),
_make_clip("c4", "main", order=3),
]
svc = _make_service(clips)
with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
resolved = svc._resolve_clips()
layers = svc._group_clips_into_layers(resolved)
assert len(layers) == 1
assert layers[0].role == "main"
assert len(layers[0].clips) == 4
assert layers[0].z_index == 0
def test_group_clips_pip(self):
"""1 main + 2 overlay → main_layer + overlay_layer。"""
clips = [
_make_clip("c1", "main", order=0),
_make_clip("c2", "overlay", order=1),
_make_clip("c3", "overlay", order=2),
]
svc = _make_service(clips)
with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
resolved = svc._resolve_clips()
layers = svc._group_clips_into_layers(resolved)
roles = {lyr.role for lyr in layers}
assert "main" in roles
assert "overlay" in roles
main_layer = next(lyr for lyr in layers if lyr.role == "main")
overlay_layer = next(lyr for lyr in layers if lyr.role == "overlay")
assert len(main_layer.clips) == 1
assert len(overlay_layer.clips) == 2
assert overlay_layer.z_index > main_layer.z_index
def test_group_clips_voice_over(self):
"""3 个 main(b_roll) clips → 1 个 broll_layer。"""
clips = [
_make_clip("c1", "main", order=0, config={"role": "b_roll"}),
_make_clip("c2", "main", order=1, config={"role": "b_roll"}),
_make_clip("c3", "main", order=2, config={"role": "b_roll"}),
]
svc = _make_service(clips)
with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
resolved = svc._resolve_clips()
layers = svc._group_clips_into_layers(resolved)
assert len(layers) == 1
assert layers[0].role == "broll"
assert len(layers[0].clips) == 3
def test_group_clips_voice_pip(self):
"""1 background + 1 corner_voice + 2 b_roll → 3 layers。"""
clips = [
_make_clip("c1", "background", order=0),
_make_clip("c2", "corner_voice", order=1),
_make_clip("c3", "b_roll", order=2),
_make_clip("c4", "b_roll", order=3),
]
svc = _make_service(clips)
with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
resolved = svc._resolve_clips()
layers = svc._group_clips_into_layers(resolved)
roles = {lyr.role for lyr in layers}
assert roles == {"background", "corner_voice", "broll"}
assert len(layers) == 3
# z_index 排序
assert layers[0].z_index <= layers[1].z_index <= layers[2].z_index
def test_group_clips_intro_outro(self):
"""intro + 2 main + outro → 1 main_layer4 clips,按 order 排序)。"""
clips = [
_make_clip("intro", "intro", order=0),
_make_clip("c1", "main", order=1),
_make_clip("c2", "main", order=2),
_make_clip("outro", "outro", order=3),
]
svc = _make_service(clips)
with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
resolved = svc._resolve_clips()
layers = svc._group_clips_into_layers(resolved)
assert len(layers) == 1
assert layers[0].role == "main"
assert len(layers[0].clips) == 4
# 按 order 排序
orders = [c.order for c in layers[0].clips]
assert orders == [0, 1, 2, 3]
# ── 测试 _resolve_clips ──────────────────────────────────────────────────────
class TestResolveClips:
"""测试 _resolve_clips 方法。"""
def test_skip_missing_asset(self):
"""跳过 asset_id 在 asset_path_map 中找不到的 clip。"""
clips = [
_make_clip("c1", "main", order=0, asset_id="asset_1.mp4"),
_make_clip("c2", "main", order=1, asset_id="missing.mp4"),
]
# 只有 asset_1.mp4 存在
asset_paths = {"asset_1.mp4": Path("/tmp/asset_1.mp4")}
svc = _make_service(clips, asset_paths)
with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
resolved = svc._resolve_clips()
assert len(resolved) == 1
assert resolved[0].clip_id == "c1"
def test_skip_empty_asset_id(self):
"""跳过 asset_id 为空的 clip。"""
clips = [
_make_clip("c1", "main", order=0, asset_id=""),
_make_clip("c2", "main", order=1, asset_id="asset_2.mp4"),
]
asset_paths = {"asset_2.mp4": Path("/tmp/asset_2.mp4")}
svc = _make_service(clips, asset_paths)
with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
resolved = svc._resolve_clips()
assert len(resolved) == 1
assert resolved[0].clip_id == "c2"
def test_sort_by_order(self):
"""解析后的 clips 按 order 排序。"""
clips = [
_make_clip("c3", "main", order=3, asset_id="a3.mp4"),
_make_clip("c1", "main", order=1, asset_id="a1.mp4"),
_make_clip("c2", "main", order=2, asset_id="a2.mp4"),
]
asset_paths = {
"a1.mp4": Path("/tmp/a1.mp4"),
"a2.mp4": Path("/tmp/a2.mp4"),
"a3.mp4": Path("/tmp/a3.mp4"),
}
svc = _make_service(clips, asset_paths)
with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
resolved = svc._resolve_clips()
orders = [c.order for c in resolved]
assert orders == [1, 2, 3]
# ── 测试 _build_filter_complex ───────────────────────────────────────────────
class TestBuildFilterComplex:
"""测试 _build_filter_complex 方法。"""
def test_single_layer_single_clip(self):
"""只有 1 个 main clip → 简单 scale + setpts。"""
clips = [_make_clip("c1", "main", order=0)]
asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
svc = _make_service(clips, asset_paths)
with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
resolved = svc._resolve_clips()
layers = svc._group_clips_into_layers(resolved)
fc, input_args = svc._build_filter_complex(layers)
assert "-i" in input_args
assert "/tmp/asset_c1.mp4" in input_args
assert "scale=" in fc
assert "[final_video]" in fc
def test_single_layer_multi_clips(self):
"""多个 main clips → xfade 串联。"""
clips = [
_make_clip("c1", "main", order=0, duration=3.0),
_make_clip("c2", "main", order=1, duration=3.0),
]
asset_paths = {
"asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
"asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
}
svc = _make_service(clips, asset_paths)
with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
resolved = svc._resolve_clips()
layers = svc._group_clips_into_layers(resolved)
fc, input_args = svc._build_filter_complex(layers)
assert input_args.count("-i") == 2
assert "xfade=" in fc
assert "[final_video]" in fc
def test_with_overlay(self):
"""main + overlay → overlay 滤镜。"""
clips = [
_make_clip("c1", "main", order=0),
_make_clip("c2", "overlay", order=1),
]
asset_paths = {
"asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
"asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
}
svc = _make_service(clips, asset_paths)
with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
resolved = svc._resolve_clips()
layers = svc._group_clips_into_layers(resolved)
fc, input_args = svc._build_filter_complex(layers)
assert "overlay=" in fc
assert "[final_video]" in fc
def test_empty_layers_raises(self):
"""空图层列表抛出 ValueError。"""
svc = _make_service()
with pytest.raises(ValueError, match="没有可渲染的图层"):
svc._build_filter_complex([])
# ── 测试 render 方法 ─────────────────────────────────────────────────────────
class TestRender:
"""测试 render 方法。"""
def test_render_empty_clips_raises(self):
"""没有 clips 时抛出 ValueError。"""
svc = _make_service(clips=[], asset_paths={})
with pytest.raises(ValueError, match="没有可渲染的片段"):
svc.render()
def test_render_with_missing_assets_raises(self):
"""所有 clips 素材缺失时抛出 ValueError。"""
clips = [_make_clip("c1", "main", order=0, asset_id="missing.mp4")]
svc = _make_service(clips, asset_paths={})
with pytest.raises(ValueError, match="没有可渲染的片段"):
svc.render()
def test_render_success(self):
"""正常渲染流程。"""
clips = [_make_clip("c1", "main", order=0)]
asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
svc = _make_service(clips, asset_paths)
with (
_patch_path_exists(),
patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
patch.object(svc, "_execute_ffmpeg") as mock_exec,
patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)),
):
result = svc.render()
assert isinstance(result, RenderResult)
assert result.duration == 5.0
assert result.file_size == 1024
assert result.width == 1280
assert result.height == 720
mock_exec.assert_called_once()