diff --git a/apps/api/app/api/routes/templates_editor/draft.py b/apps/api/app/api/routes/templates_editor/draft.py index 9e83c4bbc..adbe50e77 100755 --- a/apps/api/app/api/routes/templates_editor/draft.py +++ b/apps/api/app/api/routes/templates_editor/draft.py @@ -17,6 +17,8 @@ from fastapi import APIRouter, Depends, HTTPException, Query, status from .dependencies import get_draft_plan_id, get_editor_services from .schemas import ( + EditorClipBatchUpdateRequest, + EditorClipBatchUpdateResponse, EditorDraftResponse, EditorPublishResponse, EditorRollbackRequest, @@ -126,11 +128,7 @@ def list_template_versions( clip_count=len(v.clip_configs), change_note=v.change_note, published_by=v.published_by, - created_at=( - v.created_at.isoformat() - if hasattr(v.created_at, "isoformat") - else str(v.created_at) - ), + created_at=(v.created_at.isoformat() if hasattr(v.created_at, "isoformat") else str(v.created_at)), ) for v in versions ] @@ -162,3 +160,35 @@ def rollback_template( new_version=tpl.version, clip_count=len(clip_configs), ) + + +@router.put("/clips", response_model=EditorClipBatchUpdateResponse) +def batch_update_clips( + template_id: str, + req: EditorClipBatchUpdateRequest, + plan_id: str = Depends(get_draft_plan_id), + services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), + _: AuthenticatedUser = Depends(get_current_user), +): + """批量替换草稿clips（全量覆盖，用于前端选择素材后同步片段） + + 事务保证：清空→创建→标记ready 在同一数据库事务内完成， + 任何步骤失败时自动回滚，避免数据不一致。 + """ + _, plan_svc = services + plan_svc.get_plan_or_raise(plan_id) + + clips_data = [] + for clip_item in req.clips: + item = { + "asset_id": clip_item.asset_id, + "start_time": clip_item.start_time, + "duration": clip_item.duration, + } + if clip_item.order is not None: + item["order"] = clip_item.order + clips_data.append(item) + + plan_svc.replace_all_clips_transactional(plan_id, clips_data) + + return EditorClipBatchUpdateResponse(plan_id=plan_id, clip_count=len(req.clips)) diff --git a/apps/api/app/api/routes/templates_editor/generation.py b/apps/api/app/api/routes/templates_editor/generation.py index 7c8009df5..aa9912f8a 100755 --- a/apps/api/app/api/routes/templates_editor/generation.py +++ b/apps/api/app/api/routes/templates_editor/generation.py @@ -197,7 +197,7 @@ def generate_editor_draft( plan_svc.update_plan_config(plan_id, {"generation_task_id": gen_task.id}) plan_svc.transition_status(plan_id, EditPlanStatus.RENDERING) - celery_app.send_task("worker.render_edit_plan", args=[plan_id]) + celery_app.send_task("worker.generate_video", args=[gen_task.id]) updated_plan = plan_svc.get_plan_or_raise(plan_id) diff --git a/apps/api/app/api/routes/templates_editor/schemas.py b/apps/api/app/api/routes/templates_editor/schemas.py index bc441d333..d282a42ab 100755 --- a/apps/api/app/api/routes/templates_editor/schemas.py +++ b/apps/api/app/api/routes/templates_editor/schemas.py @@ -46,6 +46,7 @@ class EditPlanGenerationStatusResponse(BaseModel): class EditPlanGenerateRequest(BaseModel): """模板编辑器触发生成请求体""" + title_config: Optional[Dict[str, Any]] = Field( default_factory=dict, description="标题配置（可选），渲染时烧录到视频中。支持字段: text/font/font_size/font_color/position/bold/stroke/shadow", @@ -75,12 +76,8 @@ class AIRecommendRequest(BaseModel): """AI 推荐片段方案请求体""" asset_ids: List[str] = Field(default_factory=list, description="素材 ID 列表") - editing_mode: str = Field( - default="one_take", description="剪辑模式: one_take / pip / voice_over / voice_pip" - ) - target_duration: float = Field( - default=30.0, ge=1.0, le=600.0, description="目标时长（秒）" - ) + editing_mode: str = Field(default="one_take", description="剪辑模式: one_take / pip / voice_over / voice_pip") + target_duration: float = Field(default=30.0, ge=1.0, le=600.0, description="目标时长（秒）") class AIRecommendClipItem(BaseModel): @@ -107,8 +104,6 @@ class AIRecommendResponse(BaseModel): confidence: float = Field(..., ge=0.0, le=1.0, description="AI 推荐置信度 (0~1)") - - # ── BGM ──────────────────────────────────────────────────────────────────── @@ -224,9 +219,7 @@ class ClipBatchDeleteResponse(BaseModel): class ClipsFromAssetsRequest(BaseModel): """从素材批量创建片段请求""" - asset_ids: List[str] = Field( - ..., min_length=1, max_length=200, description="素材 ID 列表，按顺序追加到时间线末尾" - ) + asset_ids: List[str] = Field(..., min_length=1, max_length=200, description="素材 ID 列表，按顺序追加到时间线末尾") clip_type: str = Field(default="main", description="片段类型，默认 main") @@ -501,6 +494,28 @@ class EditorClipUpdateRequest(BaseModel): config: Optional[dict[str, Any]] = None +class EditorClipBatchItem(BaseModel): + """批量更新clips的单个片段""" + + asset_id: str = Field(default="", max_length=100, description="关联素材ID，可为空（占位片段）") + start_time: float = Field(default=0.0, ge=0.0) + duration: float = Field(default=0.0, ge=0.0) + order: Optional[int] = Field(default=None, ge=0, description="排序，None表示按数组顺序") + + +class EditorClipBatchUpdateRequest(BaseModel): + """批量替换clips请求（全量覆盖）""" + + clips: List[EditorClipBatchItem] = Field(default_factory=list) + + +class EditorClipBatchUpdateResponse(BaseModel): + """批量更新clips响应""" + + plan_id: str + clip_count: int + + class EditorPublishResponse(BaseModel): """发布草稿响应""" diff --git a/apps/api/app/services/edit_plan_service.py b/apps/api/app/services/edit_plan_service.py index f28a2d08a..41f0afff6 100755 --- a/apps/api/app/services/edit_plan_service.py +++ b/apps/api/app/services/edit_plan_service.py @@ -371,6 +371,85 @@ class EditPlanService: logger.info("删除所有片段: plan_id=%s count=%d", plan_id, count) return count + def replace_all_clips_transactional( + self, + plan_id: str, + clips_data: list[dict], + ) -> int: + """事务性地替换所有片段：清空→创建→标记ready，单事务保证原子性。 + + Args: + plan_id: 计划 ID + clips_data: 片段数据列表，每项包含 asset_id/start_time/duration/order + + Returns: + int: 创建的片段数量 + + Raises: + Exception: 任何步骤失败时自动回滚 + """ + from packages.adapters.sqlalchemy_impl.models import EditPlanClipModel + + db = self._clip_repo.session + try: + # 1. 清空现有 clips（不 commit） + deleted_count = db.query(EditPlanClipModel).filter(EditPlanClipModel.plan_id == plan_id).delete() + + # 2. 批量创建新 clips（不 commit） + for i, clip_item in enumerate(clips_data): + order = clip_item.get("order") or i + clip = EditPlanClip.create( + plan_id=plan_id, + clip_type="main", + order=order, + asset_id=clip_item.get("asset_id", ""), + start_time=clip_item.get("start_time", 0.0), + duration=clip_item.get("duration", 0.0), + ) + model = EditPlanClipModel( + id=clip.id, + plan_id=clip.plan_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, + transition_duration=clip.transition_duration, + playback_speed=clip.playback_speed, + status=clip.status.value, + config=clip.config, + ) + db.add(model) + + # 3. 标记有 asset_id 的 clips 为 ready（不 commit） + pending_with_asset = ( + db.query(EditPlanClipModel) + .filter( + EditPlanClipModel.plan_id == plan_id, + EditPlanClipModel.status == "pending", + EditPlanClipModel.asset_id != "", + ) + .all() + ) + for m in pending_with_asset: + m.status = "ready" + + # 4. 一次性提交 + db.commit() + logger.info( + "事务性替换片段: plan_id=%s deleted=%d created=%d", + plan_id, + deleted_count, + len(clips_data), + ) + return len(clips_data) + except Exception: + db.rollback() + logger.exception("事务性替换片段失败: plan_id=%s", plan_id) + raise + # ── 片段分割与合并 ────────────────────────────────────────────────────── def split_clip(self, clip_id: str, split_time: float) -> Dict[str, Any]: diff --git a/apps/worker/video_processing/__init__.py b/apps/worker/video_processing/__init__.py index d45325dae..d7627cdec 100644 --- a/apps/worker/video_processing/__init__.py +++ b/apps/worker/video_processing/__init__.py @@ -7,7 +7,7 @@ VideoProcessor 等）按需从子模块导入，避免 __init__ 阶段引入 packages / DB 等重依赖。 """ -# 共享工具模块（零外部依赖，供 editing_modes / generation / edit_plan_generation 等复用） +# 共享工具模块（零外部依赖，供 editing_modes / generation 等复用） from . import dedup_helpers, ffmpeg_utils, oss_helpers, url_security __all__ = [ diff --git a/apps/worker/video_processing/dedup_helpers.py b/apps/worker/video_processing/dedup_helpers.py index d339f7144..401c09f87 100755 --- a/apps/worker/video_processing/dedup_helpers.py +++ b/apps/worker/video_processing/dedup_helpers.py @@ -1,6 +1,6 @@ """查重辅助函数 — 从 generation.py 提取的 GeneratedVideo 记录 + 查重逻辑. -供 render_edit_plan 和 generate_video 共同复用， +供 generate_video 共同复用， 创建 GeneratedVideo 记录后计算指纹并执行项目级 + 批次内查重。 """ diff --git a/apps/worker/video_processing/oss_helpers.py b/apps/worker/video_processing/oss_helpers.py index 05a58685a..b042e5aa7 100755 --- a/apps/worker/video_processing/oss_helpers.py +++ b/apps/worker/video_processing/oss_helpers.py @@ -1,4 +1,4 @@ -"""OSS 工具函数 — 从 generation.py / edit_plan_generation.py 提取的共享 OSS 操作. +"""OSS 工具函数 — 从 generation.py 提取的共享 OSS 操作. 提供 OSS 配置读取、Bucket 创建、素材上传/下载、asset_id → 本地路径解析 等能力，供 render_edit_plan 和 generate_video 共同复用。 diff --git a/apps/worker/video_processing/render_adapter.py b/apps/worker/video_processing/render_adapter.py index a8c2301f5..9d4a48808 100755 --- a/apps/worker/video_processing/render_adapter.py +++ b/apps/worker/video_processing/render_adapter.py @@ -128,6 +128,7 @@ class RenderAdapter: job_id: str = "", work_dir: Path | None = None, progress_cb: ProgressCallback | None = None, + voiceover_audio_path: str | None = None, ) -> RenderAdapterResult: """渲染一个 EditPlan。 @@ -142,6 +143,7 @@ class RenderAdapter: job_id: 关联的 Job ID（用于结果存储路径） work_dir: 工作目录，不传则使用临时目录 progress_cb: 进度回调函数 + voiceover_audio_path: 配音音频本地路径（一键生成场景使用） Returns: RenderAdapterResult @@ -208,6 +210,7 @@ class RenderAdapter: progress_cb=progress_cb, rendered_clip_ids=rendered_clip_ids, failed_clip_ids=failed_clip_ids, + voiceover_audio_path=voiceover_audio_path, ) except subprocess.CalledProcessError as exc: diff --git a/apps/worker/worker_app/celery_app.py b/apps/worker/worker_app/celery_app.py index 2cf626277..751e4da05 100755 --- a/apps/worker/worker_app/celery_app.py +++ b/apps/worker/worker_app/celery_app.py @@ -14,8 +14,6 @@ celery_app.conf.imports = ( "worker_app.tasks.voice_extraction", "worker_app.tasks.voice_clone", "worker_app.tasks.tts_synthesis", - "worker_app.tasks.edit_plan_generation", - "worker_app.tasks.compose_video", "worker_app.tasks.batch_download", "worker_app.tasks._startup", "apps.worker.video_processing.dedup", diff --git a/apps/worker/worker_app/tasks/__init__.py b/apps/worker/worker_app/tasks/__init__.py index 7c9c315a4..87b7b613f 100755 --- a/apps/worker/worker_app/tasks/__init__.py +++ b/apps/worker/worker_app/tasks/__init__.py @@ -25,10 +25,6 @@ def __getattr__(name: str): from .voice_extraction import extract_voice_task return extract_voice_task - elif name == "compose_video": - from .compose_video import compose_video - - return compose_video elif name == "extract_background_task": from .voice_extraction import extract_background_task @@ -58,7 +54,6 @@ def __getattr__(name: str): __all__ = [ "classify_asset", - "compose_video", "generate_video", "healthcheck", "ingest_asset", diff --git a/apps/worker/worker_app/tasks/compose_video.py b/apps/worker/worker_app/tasks/compose_video.py deleted file mode 100644 index 5cb97b80e..000000000 --- a/apps/worker/worker_app/tasks/compose_video.py +++ /dev/null @@ -1,198 +0,0 @@ -"""视频合成 Celery 任务 — Phase 8 任务 2.10. - -使用 JobService 管理任务生命周期，通过 RenderAdapter 调用 UnifiedRenderService 执行合成。 -""" - -from __future__ import annotations - -import os -import subprocess # pragma: no cover -import tempfile -from pathlib import Path - -from celery.exceptions import SoftTimeLimitExceeded # pragma: no cover -from celery.utils.log import get_task_logger -from worker_app.celery_app import celery_app -from worker_app.db import SessionLocal - -logger = get_task_logger(__name__) - -# 任务超时时间（秒）：超过此时间 Celery 会抛出 SoftTimeLimitExceeded -RENDER_TASK_SOFT_TIME_LIMIT = 600 # pragma: no cover # 10 分钟 -# 硬超时：超过此时间进程会被强制 kill -RENDER_TASK_TIME_LIMIT = 660 # pragma: no cover # 10 分钟 + 1 分钟清理缓冲 - - -def _get_job_service(): - """延迟导入 JobService，避免循环依赖。""" - from apps.api.app.services.job_service import JobService - from packages.adapters.sqlalchemy_impl.job_repository import SQLAlchemyJobRepository - - db = SessionLocal() - repo = SQLAlchemyJobRepository(db) - return JobService(repo), db - - -@celery_app.task( - name="worker.compose_video", - bind=True, - max_retries=3, - default_retry_delay=60, - soft_time_limit=RENDER_TASK_SOFT_TIME_LIMIT, - time_limit=RENDER_TASK_TIME_LIMIT, -) -def compose_video(self, job_id: str, **kwargs): # pragma: no cover - """视频合成任务。 - - 使用 UnifiedRenderService（图层架构）进行渲染。 - - Args: - job_id: JobService 中的任务 ID - **kwargs: 来自 Job.payload 的额外参数（plan_id, output_path 等） - """ - job_service, db = _get_job_service() - - try: - job = job_service.get_job(job_id) - if job is None: - logger.error("Job not found: %s", job_id) - return {"status": "error", "message": f"Job {job_id} not found"} - - plan_id = job.payload.get("plan_id", "") - if not plan_id: - job_service.fail_job(job_id, "Missing plan_id in job payload") - return {"status": "error", "message": "Missing plan_id"} - - # 使用 unified 渲染引擎 - return _compose_with_unified_engine(self, job_service, job, plan_id, db) - - except SoftTimeLimitExceeded: - # Celery 软超时：任务执行超过 soft_time_limit - error_msg = f"渲染任务超时（超过 {RENDER_TASK_SOFT_TIME_LIMIT // 60} 分钟）" - logger.error("视频合成超时: job_id=%s", job_id) - try: - job_service.fail_job(job_id, error_msg) - except Exception: - logger.exception("更新 Job 超时失败状态时出错") - # 超时不重试 - return {"status": "error", "message": error_msg, "error_type": "timeout"} - - except subprocess.TimeoutExpired as exc: - # FFmpeg 子进程超时 - error_msg = f"FFmpeg 渲染超时（{exc.timeout}s）" - logger.error("视频合成 FFmpeg 超时: job_id=%s timeout=%s", job_id, exc.timeout) - try: - job_service.fail_job(job_id, error_msg) - except Exception: - logger.exception("更新 Job 超时失败状态时出错") - # 超时不重试 - return {"status": "error", "message": error_msg, "error_type": "ffmpeg_timeout"} - - except self.retry_exc as exc: - logger.warning("视频合成重试中: job_id=%s, exc=%s", job_id, exc) - raise - - except subprocess.CalledProcessError as exc: - # FFmpeg 执行失败，提取有意义的错误信息 - from video_processing.video_validation import get_exit_code_message - - exit_msg = get_exit_code_message(exc.returncode) - stderr_text = (exc.stderr or "").strip() - stderr_tail = stderr_text[-300:] if len(stderr_text) > 300 else stderr_text - error_msg = f"渲染失败: {exit_msg}" - if stderr_tail: - error_msg += f" | {stderr_tail[:200]}" - - logger.error("视频合成 FFmpeg 失败: job_id=%s %s", job_id, exit_msg) - try: - job_service.fail_job(job_id, error_msg[:500]) - except Exception: - logger.exception("更新 Job 失败状态时出错") - # FFmpeg 错误不重试（通常是素材或配置问题） - return {"status": "error", "message": error_msg, "error_type": "ffmpeg_error", "exit_code": exc.returncode} - - except Exception as exc: - logger.exception("视频合成异常: job_id=%s", job_id) - try: - job_service.fail_job(job_id, str(exc)[:500]) - except Exception: - logger.exception("更新 Job 失败状态时出错") - raise self.retry(exc=exc, countdown=60) from exc - finally: - db.close() - - -def _compose_with_unified_engine(task, job_service, job, plan_id: str, db) -> dict: # pragma: no cover - """新引擎渲染路径（UnifiedRenderService + RenderAdapter）。""" - job_id = job.id - - # 标记为 running - job_service.update_progress(job_id, progress=10.0, current_stage="初始化统一渲染引擎") - - from video_processing.render_adapter import RenderAdapter - - adapter = RenderAdapter(db) - - # 校验合成条件 - job_service.update_progress(job_id, progress=15.0, current_stage="校验合成条件") - valid, errors, warnings, ready_count, total_count = adapter.validate_plan(plan_id) - if not valid: - error_msg = "; ".join(errors) - job_service.fail_job(job_id, f"合成校验失败: {error_msg}") - return {"status": "error", "message": error_msg} - - # 进度回调 - def progress_cb(progress: float, stage: str) -> None: - try: - job_service.update_progress(job_id, progress=progress, current_stage=stage) - except Exception: - logger.exception("更新进度失败") - - # 执行渲染 - job_service.update_progress(job_id, progress=20.0, current_stage="开始渲染") - logger.info("统一渲染引擎开始: job_id=%s plan_id=%s", job_id, plan_id) - - result = adapter.render_plan( - plan_id=plan_id, - job_id=job_id, - progress_cb=progress_cb, - ) - - if not result.success: - error_msg = f"渲染失败: {result.error_message}" - job_service.fail_job(job_id, error_msg[:500]) - raise RuntimeError(result.error_message) - - # 更新 Job 状态为完成 - result_data = { - "plan_id": plan_id, - "output_path": str(result.output_path) if result.output_path else "", - "storage_key": f"rendered/{plan_id}/{job_id}.mp4", - "output_url": result.output_url, - "estimated_duration": result.duration, - "clip_count": result.clip_count, - "engine": "unified", - "width": result.width, - "height": result.height, - "file_size": result.file_size, - } - job_service.complete_job(job_id, result=result_data) - - logger.info( - "视频合成完成(unified): job_id=%s plan_id=%s duration=%.2fs", - job_id, - plan_id, - result.duration, - ) - return {"status": "completed", "job_id": job_id, "result": result_data} - - -def _cleanup_output(job_id: str) -> None: - """清理临时输出文件。""" - try: - _output_dir = os.environ.get("VIDEO_OUTPUT_DIR", os.path.join(tempfile.gettempdir(), "video_output")) - output_path = os.path.join(_output_dir, f"{job_id}.mp4") - if Path(output_path).exists(): - Path(output_path).unlink() - except Exception as e: - logger.warning(f"清理输出文件失败: {e}", exc_info=True) diff --git a/apps/worker/worker_app/tasks/edit_plan_generation.py b/apps/worker/worker_app/tasks/edit_plan_generation.py deleted file mode 100644 index 69ce7311d..000000000 --- a/apps/worker/worker_app/tasks/edit_plan_generation.py +++ /dev/null @@ -1,452 +0,0 @@ -"""剪辑计划渲染任务 — 使用 UnifiedRenderService 统一渲染引擎. - -Celery 任务 worker.render_edit_plan: - 1. 加载 EditPlan + EditPlanClips - 2. 通过 RenderAdapter 调用 UnifiedRenderService 渲染 - 3. 下载各片段素材 + 渲染 - 4. 上传渲染结果到 OSS - 5. 创建 GeneratedVideo 记录 + 查重 - 6. 更新 EditPlan / EditPlanClip 状态 - 7. 更新 GenerationTask 进度 -""" - -from __future__ import annotations - -import logging -from datetime import datetime, timezone -from pathlib import Path - -from worker_app.celery_app import celery_app -from worker_app.db import SessionLocal - -logger = logging.getLogger(__name__) - -OUTPUT_WIDTH = 1280 -OUTPUT_HEIGHT = 720 -OUTPUT_FPS = 25.0 - - -# ── 共享工具模块导入 ────────────────────────────────────────────────────────── - -from video_processing.dedup_helpers import create_video_record_and_dedup - -# ── Repository imports (延迟导入避免循环依赖) ───────────────────────────────── - - -def _get_repos(): - """获取数据库仓储实例""" - from packages.adapters.sqlalchemy_impl import SQLAlchemyEditPlanRepository - from packages.adapters.sqlalchemy_impl.edit_plan_clip_repository import ( - SQLAlchemyEditPlanClipRepository, - ) - from packages.adapters.sqlalchemy_impl.generation_task_repository import ( - SQLAlchemyGenerationTaskRepository, - ) - - db = SessionLocal() - try: - plan_repo = SQLAlchemyEditPlanRepository(db) - clip_repo = SQLAlchemyEditPlanClipRepository(db) - gen_task_repo = SQLAlchemyGenerationTaskRepository(db) - yield plan_repo, clip_repo, gen_task_repo, db - finally: - db.close() - - -# ── Celery Task ─────────────────────────────────────────────────────────────── - - -def _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, error_msg: str): - """统一的计划失败标记工具。""" - plan = plan_repo.get(plan_id) - if plan and plan.status.value == "rendering": - plan.mark_failed() - plan_repo.update(plan) - if generation_task_id: - gen_task = gen_task_repo.get(generation_task_id) - if gen_task and gen_task.status.value != "failed": - gen_task.status = "failed" - gen_task.error_message = error_msg - gen_task.completed_at = datetime.now(timezone.utc) - try: - gen_task.append_log( - stage="render_failed", - message=error_msg[:500], - level="ERROR", - ) - except Exception: - pass - gen_task_repo.update(gen_task) - - -def _finalize_render_success( - plan, - plan_repo, - clip_repo, - gen_task_repo, - db, - plan_id: str, - output_url: str, - storage_key: str, - duration: float, - file_size: int, - width: int, - height: int, - rendered_clip_ids: list[str], - failed_clip_ids: list[str], - generation_task_id: str, - output_path: Path, - engine: str, - thumbnail_url: str = "", - cover_candidates: list[dict] | None = None, -) -> dict: - """渲染成功后的统一收尾：查重 + 更新状态 + 返回结果。""" - # 创建 GeneratedVideo 记录 + 查重 - project_id = plan.project_id or "" - batch_id = plan.config.get("batch_id", "") - mode = plan.config.get("mode", "edit_plan") - # 从 plan.config.title.text 读取视频名称 - plan_config = plan.config or {} - title_cfg = plan_config.get("title", {}) or {} - if not isinstance(title_cfg, dict): - title_cfg = {} - video_name = (title_cfg.get("text") or "").strip() or f"generated-{generation_task_id[:8]}.mp4" - if generation_task_id: - try: - create_video_record_and_dedup( - generation_task_id=generation_task_id, - project_id=project_id, - user_id=plan.created_by_user_id or "", - batch_id=batch_id, - file_url=output_url or "", - file_size=file_size, - duration=duration, - video_path=str(output_path), - mode=mode, - session=db, - width=width, - height=height, - fps=OUTPUT_FPS, - name=video_name, - thumbnail_url=thumbnail_url, - ) - except Exception as dedup_err: - logger.warning("查重失败（不影响渲染结果）: %s", dedup_err) - - # 更新片段状态为 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) - - # 更新 EditPlan 状态为 completed + 回写实际渲染时长 + 结果数 - plan.config["rendered_url"] = output_url or "" - plan.config["rendered_storage_key"] = storage_key - if hasattr(plan, "total_duration") and duration > 0: - plan.total_duration = duration - plan.mark_completed() - plan_repo.update(plan) - - # 更新 GenerationTask 状态为 completed - if generation_task_id: - gen_task = gen_task_repo.get(generation_task_id) - if gen_task: - gen_task.status = "completed" - gen_task.progress = 100.0 - # 剪辑计划是多片段合成 1 个成片，result_count = 1 - gen_task.result_count = 1 - gen_task.append_log( - stage="render_complete", - message=f"渲染完成，输出时长 {duration:.1f}s", - level="INFO", - engine=engine, - clip_count=len(rendered_clip_ids), - ) - gen_task.completed_at = datetime.now(timezone.utc) - - # 回写封面 URL 到 GenerationTask，供封面生成接口读取 - if cover_candidates: - first_cover = cover_candidates[0].get("image_url") or cover_candidates[0].get("url") or "" - if first_cover: - gen_task.cover_url = first_cover - logger.info( - "预览渲染完成，回写 cover_url: plan_id=%s task_id=%s url=%s", - plan_id, - generation_task_id, - first_cover[:80], - ) - - gen_task_repo.update(gen_task) - - logger.info( - "剪辑计划渲染完成: plan_id=%s engine=%s rendered=%d failed=%d duration=%.1fs", - plan_id, - engine, - len(rendered_clip_ids), - len(failed_clip_ids), - duration, - ) - - return { - "status": "completed", - "plan_id": plan_id, - "rendered_count": len(rendered_clip_ids), - "failed_count": len(failed_clip_ids), - "output_url": output_url, - "duration": duration, - } - - -def _render_with_unified( - plan, - clips, - plan_id: str, - generation_task_id: str, - plan_repo, - clip_repo, - gen_task_repo, - db, -) -> dict: - """统一渲染引擎路径（通过 RenderAdapter 调用 UnifiedRenderService）。 - - RenderAdapter 内部处理：素材下载、BGM 准备、ASR 自动字幕、渲染执行、OSS 上传。 - 本函数只负责：业务状态更新、查重、收尾。 - """ - from video_processing.render_adapter import RenderAdapter - - adapter = RenderAdapter(db) - - # 进度回调：更新 GenerationTask 进度 - def _progress_cb(progress: float, stage: str): - if not generation_task_id: - return - try: - gen_task = gen_task_repo.get(generation_task_id) - if gen_task: - # 映射到 30%~90% 区间（素材下载前已到 30%） - mapped_progress = 30.0 + progress * 0.6 - gen_task.progress = min(mapped_progress, 95.0) - gen_task.append_log( - stage="render_progress", - message=stage, - level="INFO", - progress=mapped_progress, - ) - gen_task_repo.update(gen_task) - except Exception: - pass - - try: - result = adapter.render_plan( - plan_id=plan_id, - job_id=generation_task_id or plan_id, - progress_cb=_progress_cb, - ) - except Exception as render_err: - logger.error("渲染失败(unified): %s — %s", plan_id, render_err) - _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, f"渲染失败: {render_err}") - return {"status": "error", "message": f"渲染失败: {render_err}"} - - if not result.success: - full_error = result.error_message or "渲染失败" - if result.error_detail: - full_error = f"{full_error}\n--- stderr ---\n{result.error_detail}" - logger.error("渲染失败(unified): %s — %s", plan_id, result.error_message) - _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, full_error) - return {"status": "error", "message": result.error_message or "渲染失败"} - - output_path = result.output_path or Path("") - output_url = result.output_url or "" - thumbnail_url = result.thumbnail_url or "" - # adapter 上传到 rendered/{plan_id}/{job_id}.mp4，从 URL 提取实际 key - # 不能用 output.mp4 硬编码，否则 cover 等下游通过 key 构造的 URL 指向不存在的文件 - if output_url: - from packages.shared.storage import get_shared_storage_service - - storage_key = get_shared_storage_service().normalize_storage_key(output_url) - else: - storage_key = f"rendered/{plan_id}/{generation_task_id or plan_id}.mp4" - - # 用 adapter 返回的 clip 明细（以 adapter 的结果为准） - rendered_clip_ids = result.rendered_clip_ids or [] - failed_clip_ids = result.failed_clip_ids or [] - - # 将封面候选帧写入 plan.config（供封面 API 直接使用，跳过 MediaKit 抽帧） - if result.cover_candidates: - plan_config = plan.config or {} - plan_config["cover_candidates"] = result.cover_candidates - plan.config = plan_config - logger.info( - "封面候选帧已写入 plan.config: plan_id=%s count=%d", - plan_id, - len(result.cover_candidates), - ) - - return _finalize_render_success( - plan=plan, - plan_repo=plan_repo, - clip_repo=clip_repo, - gen_task_repo=gen_task_repo, - db=db, - plan_id=plan_id, - output_url=output_url or "", - storage_key=storage_key, - duration=result.duration, - file_size=result.file_size, - width=result.width, - height=result.height, - rendered_clip_ids=rendered_clip_ids, - failed_clip_ids=failed_clip_ids, - generation_task_id=generation_task_id, - output_path=output_path, - engine="unified", - thumbnail_url=thumbnail_url, - cover_candidates=result.cover_candidates, - ) - - -@celery_app.task( - name="worker.render_edit_plan", - bind=True, - max_retries=2, - soft_time_limit=600, # 10 分钟软超时 - time_limit=660, # 11 分钟硬超时 -) -def render_edit_plan(self, plan_id: str) -> dict: # pragma: no cover - """渲染剪辑计划 - - 流程： - 1. 加载 EditPlan + EditPlanClips - 2. 通过 RenderAdapter 调用 UnifiedRenderService 渲染 - 3. 下载素材 + 渲染 - 4. 上传渲染结果到 OSS - 5. 创建 GeneratedVideo 记录 + 查重 - 6. 更新 EditPlan → completed, EditPlanClips → rendered - 7. 更新 GenerationTask 进度 - """ - logger.info("开始渲染剪辑计划: plan_id=%s", plan_id) - - generation_task_id = "" - - for repos in _get_repos(): - plan_repo, clip_repo, gen_task_repo, db = repos - - try: - # 1. 加载 EditPlan - plan = plan_repo.get(plan_id) - if plan is None: - logger.error("剪辑计划不存在: %s", plan_id) - return {"status": "error", "message": f"计划不存在: {plan_id}"} - - # 获取 generation_task_id（提前读取，确保 except 块可用） - generation_task_id = plan.config.get("generation_task_id", "") - - # 2. 准备渲染（使用 unified 渲染引擎） - - # 3. 加载片段列表（按 order 排序） - clips = clip_repo.list_by_plan(plan_id, skip=0, limit=10000) - if not clips: - logger.warning("剪辑计划没有片段: %s", plan_id) - plan.mark_failed() - plan_repo.update(plan) - return {"status": "error", "message": "没有可渲染的片段"} - - # 更新 GenerationTask 状态为 running - if generation_task_id: - gen_task = gen_task_repo.get(generation_task_id) - if gen_task: - gen_task.status = "running" - gen_task.started_at = datetime.now(timezone.utc) - gen_task.append_log( - stage="render_start", - message=f"开始渲染，片段数 {len(clips)}", - level="INFO", - engine="unified", - clip_count=len(clips), - ) - gen_task_repo.update(gen_task) - - # 3. 渲染前取消检查 - if generation_task_id: - current_task = gen_task_repo.get(generation_task_id) - if current_task: - task_status = ( - current_task.status.value if hasattr(current_task.status, "value") else str(current_task.status) - ) - if task_status == "cancelled": - logger.info("任务已被取消，中止渲染: plan_id=%s task_id=%s", plan_id, generation_task_id) - - if plan.status.value == "rendering": - try: - plan.resume_editing() - plan_repo.update(plan) - except ValueError: - pass - return {"status": "cancelled", "plan_id": plan_id, "message": "任务已取消"} - - # 4. 渲染（unified 引擎：RenderAdapter 统一处理下载 + BGM + ASR + 渲染 + 上传） - result = _render_with_unified( - plan=plan, - clips=clips, - plan_id=plan_id, - generation_task_id=generation_task_id, - plan_repo=plan_repo, - clip_repo=clip_repo, - gen_task_repo=gen_task_repo, - db=db, - ) - - result["engine"] = "unified" - return result - - except Exception as exc: - # 超时异常不重试，直接标记失败 - from celery.exceptions import SoftTimeLimitExceeded - - is_timeout = isinstance(exc, SoftTimeLimitExceeded) - if is_timeout: - logger.error("渲染剪辑计划超时: plan_id=%s", plan_id) - else: - logger.exception("渲染剪辑计划异常: %s", plan_id) - - # 尝试标记计划和 GenerationTask 为失败 - error_msg = "渲染任务超时（超过10分钟）" if is_timeout else f"渲染异常: {type(exc).__name__}: {exc}" - try: - plan = plan_repo.get(plan_id) - if plan and plan.status.value == "rendering": - plan.mark_failed() - plan_repo.update(plan) - except Exception as e: - logger.warning("标记计划失败时异常: plan_id=%s error=%s", plan_id, e, exc_info=True) - # 更新 GenerationTask 状态为 failed，前端轮询能看到失败状态 - try: - if generation_task_id: - gen_task = gen_task_repo.get(generation_task_id) - if gen_task and gen_task.status.value != "failed": - gen_task.status = "failed" - gen_task.error_message = error_msg - gen_task.completed_at = datetime.now(timezone.utc) - try: - log_stage = "render_timeout" if is_timeout else "render_failed" - gen_task.append_log( - stage=log_stage, - message=error_msg[:500], - level="ERROR", - exception_type=type(exc).__name__, - ) - except Exception: - pass - gen_task_repo.update(gen_task) - logger.info( - "GenerationTask 已标记为 failed: task_id=%s plan_id=%s", - generation_task_id, - plan_id, - ) - except Exception as e: - logger.warning( - "更新 GenerationTask 失败状态时异常: task_id=%s error=%s", generation_task_id, e, exc_info=True - ) - raise self.retry(exc=exc, countdown=60) from exc - - return {"status": "error", "message": "数据库连接失败"} diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index e9b02bb75..da4f3773b 100644 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -223,6 +223,7 @@ def _load_template_segment_durations(template_id: str) -> list[float]: return [] +# DEPRECATED: 仅兼容无 source_edit_plan_id 的旧调用，后续移除 def _build_plan_and_clips_from_task( task_id: str, downloaded_paths: list[Path], @@ -889,6 +890,7 @@ def _load_task_info(task_id: str) -> dict | None: "custom_title": getattr(gen_task, "custom_title", "") or "", "title_config": dict(getattr(gen_task, "title_config", {}) or {}), "voice_ids": list(getattr(gen_task, "voice_ids", []) or []), + "source_edit_plan_id": getattr(gen_task, "source_edit_plan_id", "") or "", } finally: session.close() @@ -1211,6 +1213,135 @@ def _upload_and_record( soft_time_limit=600, # 10 分钟软超时 time_limit=660, # 11 分钟硬超时 ) +def _sync_task_config_to_plan(source_edit_plan_id: str, task_info: dict, db) -> str | None: + """将 GenerationTask 的配置同步到 EditPlan.config，返回配音本地路径（如果有）。 + + 包括：title_config、BGM、输出分辨率。配音单独处理（需下载到本地）。 + """ + from packages.adapters.sqlalchemy_impl.edit_plan_repository import ( + SQLAlchemyEditPlanRepository, + ) + + plan_repo = SQLAlchemyEditPlanRepository(db) + plan = plan_repo.get(source_edit_plan_id) + if plan is None: + logger.error("[task] EditPlan not found: %s", source_edit_plan_id) + return None + + plan_config = dict(plan.config or {}) + changed = False + + # 标题配置 + title_config = task_info.get("title_config") or {} + if title_config and isinstance(title_config, dict) and title_config.get("text", "").strip(): + cfg = dict(title_config) + # 字段名归一化 + if "font_size" in cfg and "size" not in cfg: + cfg["size"] = cfg["font_size"] + if "font_color" in cfg and "color" not in cfg: + cfg["color"] = cfg["font_color"] + plan_config["title"] = cfg + changed = True + logger.info("[task] title_config synced to plan: %s", cfg.get("text", "")[:30]) + + # BGM 配置 + bgm_config = task_info.get("bgm_config") or {} + if bgm_config: + from packages.domain.bgm_utils import merge_bgm_config + + existing_bgm = plan_config.get("bgm", {}) or {} + plan_config["bgm"] = merge_bgm_config(existing_bgm, bgm_config) + changed = True + + # 输出分辨率 + ow = task_info.get("output_width") or OUTPUT_WIDTH + oh = task_info.get("output_height") or OUTPUT_HEIGHT + if ow >= 100 and oh >= 100: + export_cfg = dict(plan_config.get("export", {}) or {}) + export_cfg["resolution"] = f"{ow}x{oh}" + plan_config["export"] = export_cfg + changed = True + + if changed: + plan.config = plan_config + plan_repo.update(plan) + logger.info("[task] plan.config synced: plan_id=%s", source_edit_plan_id) + + # 配音下载 + voiceover_path: str | None = None + voice_library_id = task_info.get("voice_library_id", "") + voice_ids = task_info.get("voice_ids", []) or [] + effective_voice_id = voice_library_id or (voice_ids[0] if voice_ids else "") + + if effective_voice_id: + import tempfile + + voice_tmp = Path(tempfile.gettempdir()) / f"voice_{source_edit_plan_id}_{id(task_info)}.mp3" + try: + if _download_voice_asset(effective_voice_id, voice_tmp): + voiceover_path = str(voice_tmp) + logger.info("[task] voice downloaded: %s -> %s", effective_voice_id, voiceover_path) + except Exception: + logger.warning("[task] voice download failed: %s", effective_voice_id, exc_info=True) + + return voiceover_path + + +def _render_from_edit_plan( + task_id: str, + source_edit_plan_id: str, + task_info: dict, +) -> tuple[Path, float, list[dict] | None]: + """从 EditPlan 数据库记录直接渲染（不再内存重建clips）。 + + Returns: + (output_path, render_duration, cover_candidates) + """ + from video_processing.render_adapter import RenderAdapter + from worker_app.db import SessionLocal + + db = SessionLocal() + try: + # 同步配置到 plan.config + 下载配音 + voiceover_path = _sync_task_config_to_plan(source_edit_plan_id, task_info, db) + + # 进度回调 + def _progress_cb(progress: float, stage: str): + mapped = 40.0 + progress * 0.4 + _update_task_progress(task_id, min(mapped, 80.0), stage) + + adapter = RenderAdapter(db) + render_start = time.monotonic() + logger.info("[task_id=%s] [渲染] RenderAdapter.render_plan 开始 (plan_id=%s)", task_id, source_edit_plan_id) + + result = adapter.render_plan( + plan_id=source_edit_plan_id, + job_id=task_id, + progress_cb=_progress_cb, + voiceover_audio_path=voiceover_path, + ) + + if not result.success: + raise RuntimeError(f"渲染失败: {result.error_message}") + + render_elapsed = time.monotonic() - render_start + logger.info( + "[task_id=%s] [渲染] RenderAdapter.render_plan 完成: 耗时=%.1fs, 时长=%.2fs", + task_id, + render_elapsed, + result.duration, + ) + + output_path = result.output_path + cover_candidates = getattr(result, "cover_candidates", None) + + return output_path, result.duration, cover_candidates + finally: + db.close() + # 清理临时配音文件 + # voiceover_path 在外部作用域，这里不直接引用 + + def generate_video(self, task_id: str) -> dict: """生成视频任务 — 使用 UnifiedRenderService 统一渲染。 @@ -1294,6 +1425,178 @@ def generate_video(self, task_id: str) -> dict: if template_id: _validate_template_exists(template_id) + # ── 新路径：有 source_edit_plan_id 时直接从数据库 EditPlan 渲染 ── + source_edit_plan_id = task_info.get("source_edit_plan_id", "") + if source_edit_plan_id: + logger.info( + "[task_id=%s] 使用 EditPlan 数据库路径渲染: plan_id=%s", + task_id, + source_edit_plan_id, + ) + _update_task_progress(task_id, 30, "加载草稿数据") + + if gen_task: + gen_task.append_log("渲染模式", "从草稿数据渲染（与预览一致）") + _flush_logs(task_id, gen_task) + + output_path, render_duration, cover_candidates = _render_from_edit_plan( + task_id=task_id, + source_edit_plan_id=source_edit_plan_id, + task_info=task_info, + ) + + if gen_task: + gen_task.append_log("渲染", f"渲染完成, 时长={render_duration:.1f}s") + _flush_logs(task_id, gen_task) + + _update_task_progress(task_id, 80, "渲染完成") + + # ── 4. 上传 OSS + 查重记录 ─────────────────────────────── + _update_task_progress(task_id, 85, "开始上传") + file_url, duration, file_size, video_count = _upload_and_record( + task_id=task_id, + output_path=output_path, + project_id=project_id, + batch_id=batch_id, + editing_mode=editing_mode, + user_id=user_id, + video_name=task_info.get("video_title", ""), + ) + + if gen_task: + gen_task.append_log( + "OSS上传", + f"上传成功, 大小={file_size}", + file_size=file_size, + file_url=file_url, + ) + _flush_logs(task_id, gen_task) + + _update_task_progress(task_id, 95, "上传完成") + + # ── 4.5 封面帧持久化 ──────────────────────────────────────────── + try: + if cover_candidates: + first = cover_candidates[0] + cover_frame_url = first.get("image_url") or first.get("url") or "" + if cover_frame_url: + _cover_session = SessionLocal() + try: + from packages.adapters.sqlalchemy_impl.models import ( + GenerationTaskModel, + ) + + _cover_model = ( + _cover_session.query(GenerationTaskModel) + .filter(GenerationTaskModel.id == task_id) + .first() + ) + if _cover_model: + _cover_model.cover_url = cover_frame_url + meta = dict(_cover_model.metadata or {}) + meta["cover_candidates"] = cover_candidates + _cover_model.metadata = meta + _cover_session.commit() + finally: + _cover_session.close() + except Exception: + logger.warning("[task_id=%s] 封面帧持久化失败", task_id, exc_info=True) + + # ── 5. 标记完成 ────────────────────────────────────────────────── + _update_task_status(task_id, "mark_completed", result_count=video_count) + + # 5.1 更新标题使用次数 + try: + _title_session = SessionLocal() + try: + from packages.adapters.sqlalchemy_impl.generation_task_repository import ( + SQLAlchemyGenerationTaskRepository, + ) + from packages.adapters.sqlalchemy_impl.title_library_repository import ( + SQLAlchemyTitleLibraryRepository, + ) + + _task_repo = SQLAlchemyGenerationTaskRepository(_title_session) + _gen_task = _task_repo.get(task_id) + if _gen_task and _gen_task.title_ids and _gen_task.created_by_user_id: + _title_repo = SQLAlchemyTitleLibraryRepository(_title_session) + for _tid in _gen_task.title_ids: + try: + _title_repo.increment_usage_count(_tid, _gen_task.created_by_user_id) + except Exception: + logger.warning( + "[task_id=%s] 更新标题使用次数失败: title_id=%s", + task_id, + _tid, + exc_info=True, + ) + finally: + _title_session.close() + except Exception: + logger.warning("[task_id=%s] 更新标题使用次数异常", task_id, exc_info=True) + + # 5.2 更新素材使用次数 + try: + from worker_app.core.asset_usage import mark_asset_used_for_generation + + _asset_session = SessionLocal() + try: + from packages.adapters.sqlalchemy_impl.asset_repository import ( + SQLAlchemyAssetRepository, + ) + from packages.adapters.sqlalchemy_impl.generation_task_repository import ( + SQLAlchemyGenerationTaskRepository, + ) + + _task_repo = SQLAlchemyGenerationTaskRepository(_asset_session) + _asset_repo = SQLAlchemyAssetRepository(_asset_session) + _gen_task = _task_repo.get(task_id) + if _gen_task and _gen_task.asset_ids: + for _aid in _gen_task.asset_ids: + try: + _asset = _asset_repo.get(_aid) + if _asset: + mark_asset_used_for_generation(_asset) + _asset_repo.update(_asset) + except Exception: + logger.warning( + "[task_id=%s] 更新素材使用次数失败: asset_id=%s", + task_id, + _aid, + exc_info=True, + ) + finally: + _asset_session.close() + except Exception: + logger.warning("[task_id=%s] 更新素材使用次数异常", task_id, exc_info=True) + + if gen_task: + gen_task.append_log( + "任务完成", + f"视频生成完成: 时长={duration:.2f}s, 大小={file_size}", + duration=round(duration, 2), + file_size=file_size, + video_count=video_count, + ) + _flush_logs(task_id, gen_task) + + logger.info( + "[task_id=%s] [任务完成] duration=%.2fs file_size=%d (edit_plan path)", + task_id, + duration, + file_size, + ) + + return { + "status": "completed", + "task_id": task_id, + "output_path": str(output_path), + "file_size": file_size, + "duration": duration, + "mode": editing_mode.value, + } + + # DEPRECATED: 以下为旧路径，仅兼容无 source_edit_plan_id 的旧调用，后续移除 with tempfile.TemporaryDirectory(prefix="xiaoxia-generation-") as temp_dir: temp_path = Path(temp_dir) diff --git a/tests/unit/test_batch_clips_endpoint.py b/tests/unit/test_batch_clips_endpoint.py new file mode 100755 index 000000000..696f8b492 --- /dev/null +++ b/tests/unit/test_batch_clips_endpoint.py @@ -0,0 +1,135 @@ +"""Tests for PUT /templates/{id}/editor/clips batch update endpoint. + +Updated for transactional replace_all_clips_transactional method. +""" + +from __future__ import annotations + +import os +from unittest.mock import MagicMock + +import pytest + +os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing") + + +@pytest.fixture +def mock_services(): + plan_svc = MagicMock() + tpl_svc = MagicMock() + plan_svc.get_plan_or_raise.return_value = MagicMock(id="plan-1", template_id="tpl-1") + plan_svc.replace_all_clips_transactional.return_value = 2 + return tpl_svc, plan_svc + + +class TestBatchUpdateClips: + def test_batch_update_calls_transactional_replace(self, mock_services): + """验证批量更新调用事务性替换方法，传入正确的参数。""" + from app.api.routes.templates_editor.draft import batch_update_clips + from app.api.routes.templates_editor.schemas import ( + EditorClipBatchItem, + EditorClipBatchUpdateRequest, + ) + + _, plan_svc = mock_services + req = EditorClipBatchUpdateRequest( + clips=[ + EditorClipBatchItem(asset_id="a1", start_time=0.0, duration=3.0, order=0), + EditorClipBatchItem(asset_id="a2", start_time=3.0, duration=5.0, order=1), + ] + ) + + result = batch_update_clips( + template_id="tpl-1", + req=req, + plan_id="plan-1", + services=mock_services, + _=MagicMock(), + ) + + assert result.plan_id == "plan-1" + assert result.clip_count == 2 + plan_svc.replace_all_clips_transactional.assert_called_once() + call_args = plan_svc.replace_all_clips_transactional.call_args + assert call_args[0][0] == "plan-1" + clips_data = call_args[0][1] + assert len(clips_data) == 2 + assert clips_data[0]["asset_id"] == "a1" + assert clips_data[0]["start_time"] == 0.0 + assert clips_data[0]["duration"] == 3.0 + assert clips_data[1]["asset_id"] == "a2" + + def test_batch_update_empty_clips(self, mock_services): + """空 clips 列表也能正常处理。""" + from app.api.routes.templates_editor.draft import batch_update_clips + from app.api.routes.templates_editor.schemas import EditorClipBatchUpdateRequest + + _, plan_svc = mock_services + req = EditorClipBatchUpdateRequest(clips=[]) + + result = batch_update_clips( + template_id="tpl-1", + req=req, + plan_id="plan-1", + services=mock_services, + _=MagicMock(), + ) + + assert result.clip_count == 0 + plan_svc.replace_all_clips_transactional.assert_called_once() + call_args = plan_svc.replace_all_clips_transactional.call_args + assert call_args[0][1] == [] + + def test_batch_update_passes_order_correctly(self, mock_services): + """验证 order 字段正确传递。""" + from app.api.routes.templates_editor.draft import batch_update_clips + from app.api.routes.templates_editor.schemas import ( + EditorClipBatchItem, + EditorClipBatchUpdateRequest, + ) + + _, plan_svc = mock_services + req = EditorClipBatchUpdateRequest( + clips=[ + EditorClipBatchItem(asset_id="a1", start_time=0.0, duration=3.0, order=5), + ] + ) + + batch_update_clips( + template_id="tpl-1", + req=req, + plan_id="plan-1", + services=mock_services, + _=MagicMock(), + ) + + clips_data = plan_svc.replace_all_clips_transactional.call_args[0][1] + assert clips_data[0]["order"] == 5 + assert clips_data[0]["asset_id"] == "a1" + assert clips_data[0]["start_time"] == 0.0 + assert clips_data[0]["duration"] == 3.0 + + +class TestEditorClipBatchItemValidation: + """验证 schema 校验规则。""" + + def test_asset_id_empty_string_allowed(self): + """asset_id 空字符串允许通过（占位片段场景）。""" + from app.api.routes.templates_editor.schemas import EditorClipBatchItem + + item = EditorClipBatchItem(asset_id="", start_time=0.0, duration=3.0, order=0) + assert item.asset_id == "" + + def test_asset_id_valid(self): + """有效 asset_id 应通过校验。""" + from app.api.routes.templates_editor.schemas import EditorClipBatchItem + + item = EditorClipBatchItem(asset_id="abc123", start_time=0.0, duration=3.0, order=0) + assert item.asset_id == "abc123" + + def test_order_none_by_default(self): + """order 默认为 None，表示按数组顺序。""" + from app.api.routes.templates_editor.schemas import EditorClipBatchItem + + item = EditorClipBatchItem(asset_id="a1", start_time=0.0, duration=3.0) + assert item.order is None diff --git a/tests/unit/test_cover_url_finalize.py b/tests/unit/test_cover_url_finalize.py deleted file mode 100644 index d47513e62..000000000 --- a/tests/unit/test_cover_url_finalize.py +++ /dev/null @@ -1,132 +0,0 @@ -"""Tests for cover_url backfill to GenerationTask. - -Verifies _finalize_render_success correctly writes cover_url -from cover_candidates to gen_task.cover_url. -""" - -from __future__ import annotations - -import sys -from pathlib import Path -from unittest.mock import MagicMock, patch - -import pytest - -# Add worker app to sys.path -_WORKER_ROOT = Path(__file__).resolve().parents[2] / "apps" / "worker" -if str(_WORKER_ROOT) not in sys.path: - sys.path.insert(0, str(_WORKER_ROOT)) - - -class FakeGenTask: - """Simple stand-in for GenerationTask that tracks attribute assignment.""" - - def __init__(self): - object.__setattr__(self, "_assigned", {}) - self.id = "task-1" - self.status = MagicMock() - self.status.value = "running" - - def __setattr__(self, name, value): - if not name.startswith("_"): - self._assigned[name] = value - object.__setattr__(self, name, value) - - def append_log(self, **kwargs): - pass - - -def _make_plan(): - plan = MagicMock() - plan.project_id = "proj-1" - plan.created_by_user_id = "user-1" - plan.config = {"batch_id": "batch-1", "mode": "edit_plan", "title": {"text": "test"}} - plan.mark_completed = MagicMock() - return plan - - -def _call_finalize(cover_candidates=None, gen_task=None, plan=None): - from worker_app.tasks.edit_plan_generation import _finalize_render_success - - plan = plan or _make_plan() - gen_task = gen_task or FakeGenTask() - - plan_repo = MagicMock() - clip_repo = MagicMock() - gen_task_repo = MagicMock() - gen_task_repo.get.return_value = gen_task - db = MagicMock() - - with patch("worker_app.tasks.edit_plan_generation.create_video_record_and_dedup"): - result = _finalize_render_success( - plan=plan, - plan_repo=plan_repo, - clip_repo=clip_repo, - gen_task_repo=gen_task_repo, - db=db, - plan_id="plan-1", - output_url="https://oss.example.com/output.mp4", - storage_key="rendered/plan-1/task-1.mp4", - duration=10.0, - file_size=1024, - width=1280, - height=720, - rendered_clip_ids=["clip-1"], - failed_clip_ids=[], - generation_task_id="task-1", - output_path=Path("/tmp/output.mp4"), - engine="unified", - thumbnail_url="", - cover_candidates=cover_candidates, - ) - - return result, gen_task, gen_task_repo - - -class TestFinalizeCoverUrl: - - def test_cover_url_set_from_image_url(self): - """cover_candidates with image_url should set gen_task.cover_url""" - candidates = [ - {"image_url": "https://oss.example.com/cover1.jpg", "frame_time": 1.5}, - {"image_url": "https://oss.example.com/cover2.jpg", "frame_time": 3.0}, - ] - _, gen_task, gen_task_repo = _call_finalize(cover_candidates=candidates) - assert gen_task.cover_url == "https://oss.example.com/cover1.jpg" - gen_task_repo.update.assert_called() - - def test_cover_url_fallback_to_url_key(self): - """Should fallback to 'url' key when 'image_url' is absent""" - candidates = [{"url": "https://oss.example.com/cover_url_key.jpg"}] - _, gen_task, _ = _call_finalize(cover_candidates=candidates) - assert gen_task.cover_url == "https://oss.example.com/cover_url_key.jpg" - - def test_cover_url_not_set_when_empty_list(self): - """Empty cover_candidates should not set cover_url""" - _, gen_task, _ = _call_finalize(cover_candidates=[]) - assert "cover_url" not in gen_task._assigned - - def test_cover_url_not_set_when_none(self): - """None cover_candidates should not set cover_url""" - _, gen_task, _ = _call_finalize(cover_candidates=None) - assert "cover_url" not in gen_task._assigned - - def test_cover_url_not_set_when_url_empty(self): - """Empty URL strings in candidates should not set cover_url""" - candidates = [{"image_url": "", "url": ""}] - _, gen_task, _ = _call_finalize(cover_candidates=candidates) - assert "cover_url" not in gen_task._assigned - - def test_no_generation_task_no_crash(self): - """Should not crash when gen_task is None""" - candidates = [{"image_url": "https://oss.example.com/cover.jpg"}] - gen_task_repo = MagicMock() - gen_task_repo.get.return_value = None - result, _, _ = _call_finalize(cover_candidates=candidates) - assert result["status"] == "completed" - - def test_image_url_priority_over_url(self): - """image_url should take priority over url key""" - candidates = [{"image_url": "https://a.jpg", "url": "https://b.jpg"}] - _, gen_task, _ = _call_finalize(cover_candidates=candidates) - assert gen_task.cover_url == "https://a.jpg" diff --git a/tests/unit/test_edit_plan_worker_failure.py b/tests/unit/test_edit_plan_worker_failure.py deleted file mode 100755 index 00c33b163..000000000 --- a/tests/unit/test_edit_plan_worker_failure.py +++ /dev/null @@ -1,333 +0,0 @@ -"""P0-2: Celery 任务 render_edit_plan 失败时更新 GenerationTask 状态。 - -验证： - - 异常发生时 GenerationTask 状态更新为 failed - - error_message 记录了异常类型和描述 - - completed_at 被设置 - - 即使 generation_task_id 为空也不崩溃 - - 即使更新 GenerationTask 本身失败也不影响 retry -""" - -from __future__ import annotations - -import os -import sys -from dataclasses import dataclass, field -from types import ModuleType -from typing import Any, 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 - -# ── Mock worker 模块以避免数据库连接 ────────────────────────────────────────── -# worker_app.db 在 import 时会尝试连接数据库，必须在导入 task 模块前 mock - -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "worker")) -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..")) - -# 预注册 mock 模块，阻止真实数据库初始化 -_mock_db_mod = ModuleType("worker_app.db") -_mock_db_mod.SessionLocal = MagicMock() -sys.modules.setdefault("worker_app.db", _mock_db_mod) - -_mock_celery_mod = ModuleType("worker_app.celery_app") -_mock_celery_app = MagicMock() -# 让 @celery_app.task(...) 装饰器透传原始函数，否则函数变成 MagicMock -_mock_celery_app.task = lambda **kwargs: lambda fn: fn -_mock_celery_mod.celery_app = _mock_celery_app -sys.modules.setdefault("worker_app.celery_app", _mock_celery_mod) - - -# ── Stub domain objects ─────────────────────────────────────────────────────── - - -@dataclass -class _StubStatus: - value: str - - def __eq__(self, other): - if isinstance(other, str): - return self.value == other - if isinstance(other, _StubStatus): - return self.value == other.value - return NotImplemented - - -@dataclass -class StubEditPlan: - id: str = "plan-001" - template_id: str = "tmpl-001" - status: Any = None - config: dict = field(default_factory=dict) - project_id: str = "" - created_by_user_id: str = "user-001" - - def mark_failed(self): - self.status = _StubStatus("failed") - - def mark_completed(self): - self.status = _StubStatus("completed") - - -@dataclass -class StubGenerationTask: - id: str = "gen-task-001" - status: Any = field(default_factory=lambda: _StubStatus("pending")) - error_message: str = "" - progress: float = 0.0 - result_count: int = 0 - started_at: Any = None - completed_at: Any = None - project_id: str = "" - created_by_user_id: str = "user-001" - - -@dataclass -class StubClip: - id: str = "clip-001" - plan_id: str = "plan-001" - asset_id: str = "assets/video.mp4" - order: int = 1 - status: Any = field(default_factory=lambda: _StubStatus("ready")) - transition_effect: str = "" - text_content: str = "" - clip_type: str = "MAIN" - duration: float = 0.0 - - def mark_failed(self): - self.status = _StubStatus("failed") - - def mark_rendered(self): - self.status = _StubStatus("rendered") - - -# ── Stub repositories ───────────────────────────────────────────────────────── - - -class StubPlanRepo: - def __init__(self, plan: StubEditPlan): - self._plan = plan - - def get(self, plan_id: str) -> Optional[StubEditPlan]: - if plan_id == self._plan.id: - return self._plan - return None - - def update(self, plan: StubEditPlan) -> StubEditPlan: - self._plan = plan - return plan - - -class StubClipRepo: - def __init__(self, clips: list[StubClip] | None = None): - self._clips = clips or [] - - def list_by_plan(self, plan_id: str, skip: int = 0, limit: int = 10000) -> list[StubClip]: - return [c for c in self._clips if c.plan_id == plan_id] - - def get(self, clip_id: str) -> Optional[StubClip]: - for c in self._clips: - if c.id == clip_id: - return c - return None - - def update(self, clip: StubClip) -> StubClip: - return clip - - -class StubGenTaskRepo: - def __init__(self, task: StubGenerationTask | None = None): - self._store: dict[str, StubGenerationTask] = {} - if task: - self._store[task.id] = task - - def get(self, task_id: str) -> Optional[StubGenerationTask]: - return self._store.get(task_id) - - def update(self, task: StubGenerationTask) -> StubGenerationTask: - self._store[task.id] = task - return task - - -# ── Import task module (after mocks are in place) ───────────────────────────── - -from worker_app.tasks.edit_plan_generation import render_edit_plan - -# ── Tests ───────────────────────────────────────────────────────────────────── - - -class TestRenderEditPlanFailureUpdatesGenTask: - """P0-2: render_edit_plan 异常时更新 GenerationTask 状态为 failed""" - - def _make_bound_task(self): - """构建绑定的 Celery task mock""" - task = MagicMock() - task.retry = MagicMock(side_effect=RuntimeError("retry called")) - return task - - def test_exception_marks_gen_task_failed(self): - """异常时 GenerationTask.status 被设为 failed""" - plan = StubEditPlan(status=_StubStatus("rendering")) - plan.config["generation_task_id"] = "gen-task-001" - gen_task = StubGenerationTask(id="gen-task-001", status=_StubStatus("running")) - - plan_repo = StubPlanRepo(plan) - gen_task_repo = StubGenTaskRepo(gen_task) - - # 让 clip_repo 抛异常以触发 except 路径 - clip_repo_bad = MagicMock() - clip_repo_bad.list_by_plan.side_effect = RuntimeError("OSS 连接失败") - - def fake_get_repos(): - yield plan_repo, clip_repo_bad, gen_task_repo, MagicMock() - - bound_task = self._make_bound_task() - - with patch( - "worker_app.tasks.edit_plan_generation._get_repos", - side_effect=fake_get_repos, - ): - with pytest.raises(RuntimeError, match="retry called"): - render_edit_plan(bound_task, "plan-001") - - # 核心断言：GenerationTask 状态为 failed（生产代码赋值为字符串） - assert gen_task.status == "failed" - - def test_exception_records_error_message(self): - """异常时 error_message 包含异常类型和描述""" - plan = StubEditPlan(status=_StubStatus("rendering")) - plan.config["generation_task_id"] = "gen-task-001" - gen_task = StubGenerationTask(id="gen-task-001", status=_StubStatus("running")) - - plan_repo = StubPlanRepo(plan) - clip_repo = MagicMock() - clip_repo.list_by_plan.side_effect = RuntimeError("DB 查询超时") - gen_task_repo = StubGenTaskRepo(gen_task) - - def fake_get_repos(): - yield plan_repo, clip_repo, gen_task_repo, MagicMock() - - bound_task = self._make_bound_task() - - with patch( - "worker_app.tasks.edit_plan_generation._get_repos", - side_effect=fake_get_repos, - ): - with pytest.raises(RuntimeError, match="retry called"): - render_edit_plan(bound_task, "plan-001") - - assert gen_task.status == "failed" - assert "DB 查询超时" in gen_task.error_message - assert "RuntimeError" in gen_task.error_message - - def test_exception_sets_completed_at(self): - """异常时 completed_at 被设置""" - plan = StubEditPlan(status=_StubStatus("rendering")) - plan.config["generation_task_id"] = "gen-task-001" - gen_task = StubGenerationTask(id="gen-task-001", status=_StubStatus("running")) - - plan_repo = StubPlanRepo(plan) - clip_repo = MagicMock() - clip_repo.list_by_plan.side_effect = RuntimeError("boom") - gen_task_repo = StubGenTaskRepo(gen_task) - - def fake_get_repos(): - yield plan_repo, clip_repo, gen_task_repo, MagicMock() - - bound_task = self._make_bound_task() - - with patch( - "worker_app.tasks.edit_plan_generation._get_repos", - side_effect=fake_get_repos, - ): - with pytest.raises(RuntimeError): - render_edit_plan(bound_task, "plan-001") - - assert gen_task.completed_at is not None - - def test_no_generation_task_id_does_not_crash(self): - """generation_task_id 为空时，异常处理不崩溃""" - plan = StubEditPlan(status=_StubStatus("rendering")) - plan.config = {} # 不设置 generation_task_id - - plan_repo = StubPlanRepo(plan) - clip_repo = MagicMock() - clip_repo.list_by_plan.side_effect = RuntimeError("boom") - gen_task_repo = StubGenTaskRepo() # 空 repo - - def fake_get_repos(): - yield plan_repo, clip_repo, gen_task_repo, MagicMock() - - bound_task = self._make_bound_task() - - with patch( - "worker_app.tasks.edit_plan_generation._get_repos", - side_effect=fake_get_repos, - ): - with pytest.raises(RuntimeError, match="retry called"): - render_edit_plan(bound_task, "plan-001") - - # 计划仍被标记为 failed - assert plan.status.value == "failed" - - def test_gen_task_update_failure_does_not_block_retry(self): - """更新 GenerationTask 失败时，不影响 retry 流程""" - plan = StubEditPlan(status=_StubStatus("rendering")) - plan.config["generation_task_id"] = "gen-task-001" - gen_task = StubGenerationTask(id="gen-task-001", status=_StubStatus("running")) - - plan_repo = StubPlanRepo(plan) - clip_repo = MagicMock() - clip_repo.list_by_plan.side_effect = RuntimeError("原始错误") - # gen_task_repo.update 也抛异常 - gen_task_repo = MagicMock() - gen_task_repo.get.return_value = gen_task - gen_task_repo.update.side_effect = RuntimeError("DB 写入失败") - - def fake_get_repos(): - yield plan_repo, clip_repo, gen_task_repo, MagicMock() - - bound_task = self._make_bound_task() - - with patch( - "worker_app.tasks.edit_plan_generation._get_repos", - side_effect=fake_get_repos, - ): - with pytest.raises(RuntimeError, match="retry called"): - render_edit_plan(bound_task, "plan-001") - - # retry 被调用说明流程正确 - bound_task.retry.assert_called_once() - - def test_already_failed_gen_task_not_overwritten(self): - """已经 failed 的 GenerationTask 不会被重复更新""" - plan = StubEditPlan(status=_StubStatus("rendering")) - plan.config["generation_task_id"] = "gen-task-001" - gen_task = StubGenerationTask( - id="gen-task-001", - status=_StubStatus("failed"), # 已经是 failed - error_message="之前的错误", - ) - - plan_repo = StubPlanRepo(plan) - clip_repo = MagicMock() - clip_repo.list_by_plan.side_effect = RuntimeError("新错误") - gen_task_repo = StubGenTaskRepo(gen_task) - - def fake_get_repos(): - yield plan_repo, clip_repo, gen_task_repo, MagicMock() - - bound_task = self._make_bound_task() - - with patch( - "worker_app.tasks.edit_plan_generation._get_repos", - side_effect=fake_get_repos, - ): - with pytest.raises(RuntimeError, match="retry called"): - render_edit_plan(bound_task, "plan-001") - - # error_message 应保持原值，不被覆盖 - assert gen_task.error_message == "之前的错误" diff --git a/tests/unit/test_replace_all_clips_transactional.py b/tests/unit/test_replace_all_clips_transactional.py new file mode 100644 index 000000000..bf04010bf --- /dev/null +++ b/tests/unit/test_replace_all_clips_transactional.py @@ -0,0 +1,147 @@ +"""Tests for EditPlanService.replace_all_clips_transactional.""" + +from __future__ import annotations + +import os +from unittest.mock import MagicMock, PropertyMock, patch + +import pytest + +os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing") + + +class TestReplaceAllClipsTransactional: + """事务性替换片段方法测试。""" + + @patch("packages.adapters.sqlalchemy_impl.models.EditPlanClipModel") + @patch("app.services.edit_plan_service.EditPlanClip") + def test_success_commits_once(self, mock_clip_cls, mock_model_cls): + """成功时单次 commit，不 rollback。""" + from app.services.edit_plan_service import EditPlanService + + db = MagicMock() + # Mock query chain for delete + query_mock = MagicMock() + query_mock.filter.return_value.delete.return_value = 3 + db.query.return_value = query_mock + + # Mock query chain for mark_ready (pending_with_asset) + # After the create loop, query returns empty list (no pending clips with asset) + ready_query = MagicMock() + ready_query.filter.return_value.filter.return_value.filter.return_value.all.return_value = [] + db.query.side_effect = [query_mock, ready_query] + + # Mock EditPlanClip.create to return a mock entity + mock_entity = MagicMock() + mock_entity.id = "clip-1" + mock_entity.plan_id = "plan-1" + mock_entity.clip_type = "main" + mock_entity.order = 0 + mock_entity.asset_id = "asset-1" + mock_entity.text_content = "" + mock_entity.start_time = 0.0 + mock_entity.duration = 3.0 + mock_entity.transition_effect = "cut" + mock_entity.transition_duration = 0.0 + mock_entity.playback_speed = 1.0 + mock_entity.status.value = "pending" + mock_entity.config = {} + mock_clip_cls.create.return_value = mock_entity + + # Mock the model constructor + mock_model_instance = MagicMock() + mock_model_cls.return_value = mock_model_instance + + # Mock clip_repo + clip_repo = MagicMock() + clip_repo.session = db + + svc = EditPlanService.__new__(EditPlanService) + svc._clip_repo = clip_repo + + result = svc.replace_all_clips_transactional( + "plan-1", + [{"asset_id": "asset-1", "start_time": 0.0, "duration": 3.0, "order": 0}], + ) + + assert result == 1 + db.commit.assert_called_once() + db.rollback.assert_not_called() + db.add.assert_called_once_with(mock_model_instance) + + @patch("packages.adapters.sqlalchemy_impl.models.EditPlanClipModel") + @patch("app.services.edit_plan_service.EditPlanClip") + def test_failure_rolls_back(self, mock_clip_cls, mock_model_cls): + """异常时自动 rollback。""" + from app.services.edit_plan_service import EditPlanService + + db = MagicMock() + query_mock = MagicMock() + query_mock.filter.return_value.delete.return_value = 0 + db.query.return_value = query_mock + + # Simulate failure during create + mock_clip_cls.create.side_effect = ValueError("模拟异常") + + clip_repo = MagicMock() + clip_repo.session = db + + svc = EditPlanService.__new__(EditPlanService) + svc._clip_repo = clip_repo + + with pytest.raises(ValueError, match="模拟异常"): + svc.replace_all_clips_transactional( + "plan-1", + [{"asset_id": "bad", "start_time": 0.0, "duration": 1.0, "order": 0}], + ) + + db.rollback.assert_called_once() + db.commit.assert_not_called() + + @patch("packages.adapters.sqlalchemy_impl.models.EditPlanClipModel") + @patch("app.services.edit_plan_service.EditPlanClip") + def test_order_defaults_to_index(self, mock_clip_cls, mock_model_cls): + """order=0 时使用索引值作为 order。""" + from app.services.edit_plan_service import EditPlanService + + db = MagicMock() + query_mock = MagicMock() + query_mock.filter.return_value.delete.return_value = 0 + db.query.return_value = query_mock + + ready_query = MagicMock() + ready_query.filter.return_value.filter.return_value.filter.return_value.all.return_value = [] + db.query.side_effect = [query_mock, ready_query] + + mock_entity = MagicMock() + mock_entity.id = "clip-1" + mock_entity.plan_id = "plan-1" + mock_entity.clip_type = "main" + mock_entity.order = 0 # order=0 → 使用 i=0 + mock_entity.asset_id = "a1" + mock_entity.text_content = "" + mock_entity.start_time = 0.0 + mock_entity.duration = 1.0 + mock_entity.transition_effect = "cut" + mock_entity.transition_duration = 0.0 + mock_entity.playback_speed = 1.0 + mock_entity.status.value = "pending" + mock_entity.config = {} + mock_clip_cls.create.return_value = mock_entity + + mock_model_cls.return_value = MagicMock() + + clip_repo = MagicMock() + clip_repo.session = db + + svc = EditPlanService.__new__(EditPlanService) + svc._clip_repo = clip_repo + + svc.replace_all_clips_transactional( + "plan-1", + [{"asset_id": "a1", "start_time": 0.0, "duration": 1.0, "order": 0}], + ) + + # order=0 → falsy → use index i=0 + create_call = mock_clip_cls.create.call_args + assert create_call.kwargs["order"] == 0