From 6fc1abf2f5655a8a1a4ed1fd65c91f528c0b8899 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Mon, 13 Jul 2026 20:00:54 +0800 Subject: [PATCH 1/3] =?UTF-8?q?Phase=204=20Backend=20P2=20cleanup:=20?= =?UTF-8?q?=E5=BC=82=E5=B8=B8=E5=A4=84=E7=90=86/=E7=B1=BB=E5=9E=8B?= =?UTF-8?q?=E6=B3=A8=E8=A7=A3/=E6=A8=A1=E5=9D=97=E6=8B=86=E5=88=86/?= =?UTF-8?q?=E5=A4=A7=E5=87=BD=E6=95=B0=E6=8B=86=E5=88=86/=E9=85=8D?= =?UTF-8?q?=E7=BD=AE=E9=A1=B9=E7=A1=AE=E8=AE=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - P2-1: 10处 except 审查改进（细化异常类型、添加日志） - P2-2: 22个路由函数类型注解补全 - P2-3: edit_plans.py 拆分为 4 个模块（CRUD/generation/ai/timeline） - P2-4: generate_video + generate_plan 大函数拆分 - P2-5: unified_render_service.py 拆分（1517→984行） - render_audio.py: 音频混音模块（RenderContext + mix/merge 函数） - render_subtitles.py: ASS 字幕生成模块 - P2-6: 6个未使用配置项删除确认 Co-Authored-By: Claude Fable 5 --- apps/api/app/api/routes/auth.py | 22 +- apps/api/app/api/routes/edit_plans.py | 788 +----------------- apps/api/app/api/routes/edit_plans_ai.py | 199 +++++ .../app/api/routes/edit_plans_generation.py | 383 +++++++++ .../api/app/api/routes/edit_plans_timeline.py | 222 +++++ apps/api/app/api/routes/feature_flags.py | 10 +- apps/api/app/api/routes/projects.py | 2 +- apps/api/app/api/routes/subscription.py | 12 +- apps/api/app/api/routes/tts.py | 6 +- apps/api/app/middleware/auth.py | 10 +- apps/worker/video_processing/processor.py | 4 +- .../worker/video_processing/render_adapter.py | 4 +- apps/worker/video_processing/render_audio.py | 364 ++++++++ .../video_processing/render_subtitles.py | 255 ++++++ .../unified_render_service.py | 548 +----------- apps/worker/worker_app/tasks/generation.py | 505 ++++++----- .../application/tts_job/streaming_service.py | 6 +- 17 files changed, 1775 insertions(+), 1565 deletions(-) create mode 100644 apps/api/app/api/routes/edit_plans_ai.py create mode 100644 apps/api/app/api/routes/edit_plans_generation.py create mode 100644 apps/api/app/api/routes/edit_plans_timeline.py create mode 100644 apps/worker/video_processing/render_audio.py create mode 100644 apps/worker/video_processing/render_subtitles.py diff --git a/apps/api/app/api/routes/auth.py b/apps/api/app/api/routes/auth.py index bbac52f97..7132129b2 100755 --- a/apps/api/app/api/routes/auth.py +++ b/apps/api/app/api/routes/auth.py @@ -105,7 +105,7 @@ async def register( request: RegisterRequest, user_repository: UserRepository = Depends(get_user_repository), email_service=Depends(get_auth_email_service), -): +) -> RegisterResponse: use_case = RegisterUserUseCase( user_repository=user_repository, base_url=settings.APP_BASE_URL, @@ -136,7 +136,7 @@ async def login( request: LoginRequest, user_repository: UserRepository = Depends(get_user_repository), session_store=Depends(get_auth_session_store), -): +) -> LoginResponse: use_case = LoginUseCase( user_repository=user_repository, session_store=session_store, @@ -162,7 +162,7 @@ async def refresh( request: RefreshRequest, user_repository: UserRepository = Depends(get_user_repository), session_store=Depends(get_auth_session_store), -): +) -> LoginResponse: use_case = RefreshTokenUseCase( user_repository=user_repository, session_store=session_store, @@ -194,7 +194,7 @@ def _verify_email_token(token: str, user_repository: UserRepository) -> MessageR async def verify_email( token: str, user_repository: UserRepository = Depends(get_user_repository), -): +) -> MessageResponse: return _verify_email_token(token, user_repository) @@ -202,7 +202,7 @@ async def verify_email( async def verify_email_post( request: VerifyEmailRequestModel, user_repository: UserRepository = Depends(get_user_repository), -): +) -> MessageResponse: return _verify_email_token(request.token, user_repository) @@ -211,7 +211,7 @@ async def forgot_password( request: PasswordResetRequestModel, user_repository: UserRepository = Depends(get_user_repository), email_service=Depends(get_auth_email_service), -): +) -> MessageResponse: success, error = RequestPasswordResetUseCase( user_repository=user_repository, base_url=settings.APP_BASE_URL, @@ -227,7 +227,7 @@ async def forgot_password( async def reset_password( request: ResetPasswordModel, user_repository: UserRepository = Depends(get_user_repository), -): +) -> MessageResponse: success, error = ResetPasswordUseCase(user_repository=user_repository).execute( ResetPasswordRequest(token=request.token, new_password=request.new_password) ) @@ -241,7 +241,7 @@ async def reset_password( async def logout( credentials: HTTPAuthorizationCredentials | None = Depends(bearer_scheme), current_user: AuthenticatedUser = Depends(get_current_user), -): +) -> MessageResponse: """登出 - 将当前 token 加入黑名单""" if credentials: @@ -257,7 +257,7 @@ async def logout( @router.get("/me", response_model=CurrentUserResponse) async def get_current_user_info( authenticated_user: AuthenticatedUser = Depends(get_current_user), -): +) -> CurrentUserResponse: user = authenticated_user.user return CurrentUserResponse( user_id=user.id, @@ -323,7 +323,7 @@ def _get_internal_api_keys() -> list[str]: if content: return [k.strip() for k in content.split(",") if k.strip()] except Exception: - logger.debug("Failed to read internal API keys from file", exc_info=True) + logger.warning("无法读取内部 API 密钥文件，仅依赖环境变量配置", exc_info=True) return [] @@ -354,7 +354,7 @@ async def wechat_sync( request: WechatSyncRequest, user_repository: UserRepository = Depends(get_user_repository), _: bool = Depends(_verify_internal_api_key), -): +) -> WechatSyncResponse: """ 微信同步登录/注册（系统级内部接口） diff --git a/apps/api/app/api/routes/edit_plans.py b/apps/api/app/api/routes/edit_plans.py index 565c60981..9d7c9f2d3 100755 --- a/apps/api/app/api/routes/edit_plans.py +++ b/apps/api/app/api/routes/edit_plans.py @@ -6,12 +6,11 @@ RESTful CRUD for EditPlan: - POST /api/v1/edit-plans 创建 - PUT /api/v1/edit-plans/{id} 更新（含状态机流转） - DELETE /api/v1/edit-plans/{id} 删除 -- POST /api/v1/edit-plans/{id}/generate 触发剪辑渲染生成（任务 2.05） -- 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 基于模板+素材自动生成剪辑计划 + +拆分模块（各自独立 router，由本文件 include_router 聚合）: +- edit_plans_generation.py 生成相关（generate / generation-status / generations） +- edit_plans_ai.py AI 推荐 & 封面（ai-recommend / generate-cover） +- edit_plans_timeline.py 时间线 & 模板生成（timeline / generate-from-template） 业务逻辑委托给 EditPlanService 服务层。 """ @@ -23,29 +22,13 @@ from datetime import datetime from typing import Any, List, Optional from app.auth import AuthenticatedUser, get_current_user -from app.core.celery_app import celery_app -from app.core.task_enqueue import GLOBAL_PENDING_LIMIT, USER_PENDING_LIMIT -from app.dependencies import get_asset_library_repository, get_asset_repository, get_db_session, get_project_repository +from app.dependencies import get_db_session, get_project_repository from app.schemas.generation_task import GenerationTaskResponse -from app.services import EditPlanService, PlanGeneratorService +from app.services import EditPlanService from fastapi import APIRouter, Depends, HTTPException, Query, status from pydantic import BaseModel, Field from sqlalchemy.orm import Session -from packages.adapters.sqlalchemy_impl.generation_task_repository import ( - SQLAlchemyGenerationTaskRepository, -) -from packages.adapters.sqlalchemy_impl.template_clip_config_repository import ( - SQLAlchemyTemplateClipConfigRepository, -) -from packages.adapters.sqlalchemy_impl.template_repository import ( - SQLAlchemyTemplateRepository, -) -from packages.application.generation_tasks import ( - CreateGenerationTaskCommand, - CreateGenerationTaskUseCase, -) - from ._helpers import check_project_access from packages.domain.config_schemas import normalize_plan_config from packages.domain.edit_plan import EditPlan, EditPlanStatus @@ -258,7 +241,7 @@ def _to_response(p: EditPlan) -> EditPlanResponse: ) -# ── Routes ──────────────────────────────────────────────────────────────────── +# ── CRUD Routes ─────────────────────────────────────────────────────────────── @router.get("", response_model=EditPlanListResponse) @@ -464,753 +447,12 @@ def delete_plan( ) -# ── 生成相关端点（任务 2.05） ───────────────────────────────────────────────── +# ── Include sub-routers (拆分模块) ──────────────────────────────────────────── +from .edit_plans_ai import router as ai_router +from .edit_plans_generation import router as generation_router +from .edit_plans_timeline import router as timeline_router -@router.post("/{plan_id}/generate", response_model=EditPlanGenerateResponse) -def generate_plan( - plan_id: str, - db: Session = Depends(get_db_session), - current_user: AuthenticatedUser = Depends(get_current_user), - project_repository: Any = Depends(get_project_repository), - asset_library_repo: Any = Depends(get_asset_library_repository), - asset_repo: Any = Depends(get_asset_repository), -) -> EditPlanGenerateResponse: - """触发剪辑计划渲染生成 - - 前置条件：计划状态必须为 editing，且至少有一个片段。 - 流程： - 1. 验证计划状态为 editing - 2. 将 pending 片段标记为 ready - 3. 创建 GenerationTask - 4. 调度 Celery 任务 worker.render_edit_plan - 5. 将计划状态流转为 rendering - """ - svc = EditPlanService(db) - # 项目鉴权 - plan_check = svc.get_plan(plan_id) - if plan_check is None: - raise HTTPException(status_code=404, detail=f"剪辑计划不存在: {plan_id}") - if plan_check.project_id: - check_project_access(plan_check.project_id, current_user.user.id, project_repository) - - # ── 自动兜底 1: draft → editing ────────────────────────────────────── - if plan_check.status == EditPlanStatus.DRAFT: - logger.info("自动兜底: plan=%s draft→editing", plan_id) - svc.transition_status(plan_id, EditPlanStatus.EDITING) - - # ── 自动兜底 2: 无片段 + 有 template_id → 从模板复制片段配置 ────────── - existing_clips = svc.count_clips(plan_id) - if existing_clips == 0 and plan_check.template_id: - logger.info( - "自动兜底: plan=%s 无片段，从模板 %s 复制片段配置", - plan_id, - plan_check.template_id, - ) - # 优先从新模型 template_clip_configs 读取，若无则回退到旧模型 template_segments - clip_config_repo = SQLAlchemyTemplateClipConfigRepository(db) - configs = clip_config_repo.list_by_template(plan_check.template_id) - if configs: - for cfg in configs: - svc.create_clip( - plan_id=plan_id, - clip_type=cfg.clip_type.value if hasattr(cfg.clip_type, "value") else cfg.clip_type, - order=cfg.order, - template_clip_config_id=cfg.id, - duration=cfg.default_duration, - transition_effect=( - cfg.transition_effect.value - if hasattr(cfg.transition_effect, "value") - else cfg.transition_effect - ), - ) - logger.info("自动兜底: plan=%s 从新模型 template_clip_configs 复制了 %d 个片段", plan_id, len(configs)) - else: - # 回退到旧模型 template_segments - tpl_repo = SQLAlchemyTemplateRepository(db) - segments = tpl_repo.list_segments(plan_check.template_id) - for seg in segments: - avg_duration = (seg.duration_min + seg.duration_max) / 2 - svc.create_clip( - plan_id=plan_id, - clip_type="main", # 旧模型无结构角色，统一为主体片段 - order=seg.segment_order, - duration=avg_duration, - config={ - "material_type": seg.material_type or "", - "template_segment_id": seg.id, - }, - ) - logger.info("自动兜底: plan=%s 从旧模型 template_segments 复制了 %d 个片段", plan_id, len(segments)) - - # ── 自动兜底 3: 为没有素材的片段分配素材 ────────────────────────── - # 如果 plan.config.asset_ids 有素材，但 clips 没有 asset_id，自动按顺序分配 - all_clips = svc.list_clips(plan_id) - clips_without_asset = [c for c in all_clips if not c.asset_id] - config_asset_ids = (plan_check.config or {}).get("asset_ids", []) - material_mode = (plan_check.config or {}).get("material_mode", "manual") - - if clips_without_asset and config_asset_ids: - logger.info( - "自动兜底3: plan=%s 为 %d 个无素材片段分配 %d 个指定素材", - plan_id, - len(clips_without_asset), - len(config_asset_ids), - ) - for i, clip in enumerate(clips_without_asset): - asset_idx = i % len(config_asset_ids) - svc.assign_asset(clip.id, config_asset_ids[asset_idx]) - logger.info("自动兜底3: plan=%s 素材分配完成", plan_id) - clips_without_asset = [] # 已分配完 - - # ── 自动兜底 4: 自动素材模式 → 从项目默认视频素材库选取 ──────────── - if clips_without_asset and material_mode == "auto" and plan_check.project_id: - import random - - logger.info( - "自动兜底4: plan=%s 自动素材模式，从项目素材库选取素材 (%d 个片段需要)", - plan_id, - len(clips_without_asset), - ) - # 找到项目的视频素材库 - libs = asset_library_repo.find_by_project(plan_check.project_id) - video_lib = None - for lib in libs: - lib_kind = lib.kind.value if hasattr(lib.kind, "value") else lib.kind - if lib_kind == "video": - video_lib = lib - break - - if video_lib: - assets = asset_repo.find_by_library(video_lib.id) - # 筛选 ready 状态的视频素材 - ready_videos = [ - a - for a in assets - if (a.status.value if hasattr(a.status, "value") else a.status) == "ready" - and a.mime_type - and a.mime_type.startswith("video") - ] - if ready_videos: - # 随机选取，按片段数轮询分配 - random.shuffle(ready_videos) - for i, clip in enumerate(clips_without_asset): - asset = ready_videos[i % len(ready_videos)] - svc.assign_asset(clip.id, asset.id) - logger.info( - "自动兜底4: plan=%s 从素材库 %s 分配了 %d 个素材给 %d 个片段", - plan_id, - video_lib.name, - len(ready_videos), - len(clips_without_asset), - ) - else: - logger.warning("自动兜底4: plan=%s 素材库无可用视频素材", plan_id) - else: - logger.warning("自动兜底4: plan=%s 项目无视频素材库", plan_id) - - # 检查是否可生成 - try: - can_gen, reason = svc.can_generate(plan_id) - except ValueError as exc: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=str(exc), - ) - if not can_gen: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=reason, - ) - - # 核心生成流程：捕获异常返回明确错误信息，避免裸 500 - try: - # 将 pending 片段标记为 ready - clip_count = svc.mark_clips_ready(plan_id) - - # 创建 GenerationTask - gen_task_repo = SQLAlchemyGenerationTaskRepository(db) - - # 队列限流预检查（repository 不支持计数时跳过） - user_id = current_user.user.id - try: - has_count = hasattr(gen_task_repo, "count_pending_by_user") and hasattr( - gen_task_repo, "count_pending_total" - ) - if has_count: - user_pending = gen_task_repo.count_pending_by_user(user_id) - global_pending = gen_task_repo.count_pending_total() - if user_pending >= USER_PENDING_LIMIT: - raise HTTPException( - status_code=429, - detail=f"您的待处理任务过多（当前 {user_pending}/{USER_PENDING_LIMIT}），请等待完成后再提交", - ) - if global_pending >= GLOBAL_PENDING_LIMIT: - raise HTTPException( - status_code=503, - detail="系统繁忙，请稍后再试", - ) - except HTTPException: - raise - except Exception as e: - logger.warning("[队列限流] 剪辑计划限流检查失败，跳过: %s", e) - - gen_task_use_case = CreateGenerationTaskUseCase(gen_task_repo) - plan = svc.get_plan_or_raise(plan_id) - gen_task = gen_task_use_case.execute( - CreateGenerationTaskCommand( - project_id="", - template_id=plan.template_id, - created_by_user_id=current_user.user.id, - source_edit_plan_id=plan_id, - ) - ) - - # 将 generation_task_id 存入 plan config - svc.update_plan_config(plan_id, {"generation_task_id": gen_task.id}) - - # 流转状态为 rendering - svc.transition_status(plan_id, EditPlanStatus.RENDERING) - - # 调度 Celery 任务 - celery_app.send_task("worker.render_edit_plan", args=[plan_id]) - - # 获取最新状态 - updated_plan = svc.get_plan_or_raise(plan_id) - - logger.info( - "触发剪辑计划生成: plan_id=%s gen_task_id=%s clips=%d by user=%s", - plan_id, - gen_task.id, - clip_count, - current_user.user.id, - ) - - return EditPlanGenerateResponse( - plan_id=plan_id, - plan_status=updated_plan.status.value if hasattr(updated_plan.status, "value") else updated_plan.status, - generation_task_id=gen_task.id, - clip_count=clip_count, - ) - except HTTPException: - # 已处理的 HTTP 异常直接透传 - raise - except Exception: - logger.exception("触发剪辑计划生成失败: plan_id=%s", plan_id) - # 尝试将计划标记为失败（RENDERING → FAILED 是合法的状态流转） - try: - svc.transition_status(plan_id, EditPlanStatus.FAILED) - except Exception: - logger.warning("标记计划失败状态时异常: plan_id=%s", plan_id) - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail="生成失败，请稍后重试", - ) - - -@router.get( - "/{plan_id}/generation-status", - response_model=EditPlanGenerationStatusResponse, -) -def get_generation_status( - plan_id: str, - db: Session = Depends(get_db_session), - current_user: AuthenticatedUser = Depends(get_current_user), - project_repository: Any = Depends(get_project_repository), -) -> EditPlanGenerationStatusResponse: - """查询剪辑计划生成进度 - - 返回计划状态、关联的 GenerationTask ID、以及每个片段的状态。 - """ - svc = EditPlanService(db) - try: - gen_status = svc.get_generation_status(plan_id) - except ValueError as exc: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=str(exc), - ) - - plan = gen_status["plan"] - # 项目鉴权 - if plan.project_id: - check_project_access(plan.project_id, current_user.user.id, project_repository) - clips = gen_status["clips"] - - clip_items = [ - ClipStatusItem( - clip_id=c.id, - clip_type=c.clip_type, - order=c.order, - status=c.status.value if hasattr(c.status, "value") else c.status, - asset_id=c.asset_id or "", - text_content=c.text_content or "", - duration=c.duration, - ) - for c in clips - ] - - return EditPlanGenerationStatusResponse( - plan_id=plan_id, - plan_status=plan.status.value if hasattr(plan.status, "value") else plan.status, - generation_task_id=gen_status["generation_task_id"], - clips=clip_items, - ) - - -@router.get( - "/{plan_id}/generations", - response_model=EditPlanGenerationsResponse, -) -def list_plan_generations( - plan_id: str, - db: Session = Depends(get_db_session), - current_user: AuthenticatedUser = Depends(get_current_user), - project_repository: Any = Depends(get_project_repository), -) -> EditPlanGenerationsResponse: - """查询剪辑计划关联的所有生成记录 - - 返回该剪辑计划触发的所有 GenerationTask，按创建时间倒序。 - """ - svc = EditPlanService(db) - # 验证计划存在 + 项目鉴权 - plan = svc.get_plan_or_raise(plan_id) - if plan.project_id: - check_project_access(plan.project_id, current_user.user.id, project_repository) - - gen_task_repo = SQLAlchemyGenerationTaskRepository(db) - tasks = gen_task_repo.list_by_source_edit_plan(plan_id) - items = [ - GenerationTaskResponse( - id=t.id, - project_id=t.project_id, - asset_library_id=t.asset_library_id, - strategy_id=t.strategy_id, - voice_library_id=t.voice_library_id, - template_id=t.template_id, - asset_ids=t.asset_ids, - title_ids=t.title_ids, - voice_ids=t.voice_ids, - source_edit_plan_id=t.source_edit_plan_id or "", - status=t.status.value if hasattr(t.status, "value") else t.status, - progress=t.progress, - result_count=t.result_count, - error_message=t.error_message, - ) - for t in tasks - ] - return EditPlanGenerationsResponse(items=items, total=len(items)) - - -# ── AI 推荐 & 封面生成端点（任务 3.09） ──────────────────────────────────────── - - -@router.post( - "/{plan_id}/ai-recommend", - response_model=AIRecommendResponse, -) -def ai_recommend_clips( - plan_id: str, - body: AIRecommendRequest, - db: Session = Depends(get_db_session), - current_user: AuthenticatedUser = Depends(get_current_user), - project_repository: Any = Depends(get_project_repository), -) -> AIRecommendResponse: - """AI 推荐片段方案 - - 调用 AI 服务分析素材，自动生成片段编排方案并写入剪辑计划。 - - 流程： - 1. 验证计划存在且状态为 draft/editing - 2. 调用 AI 推荐服务（当前为 stub，后续接入真实 AI） - 3. 清除计划现有片段，按推荐方案重新创建 - 4. 更新计划 config（cover/title/subtitle/bgm）和 total_duration - 5. 返回推荐方案详情 - - 前端对接： - - 请求体只需传 asset_ids（必填），editing_mode 和 target_duration 可选 - - 返回的 clips 可直接渲染到时间线 - - 返回的 config 包含推荐的封面/标题/字幕/BGM 配置 - """ - svc = EditPlanService(db) - - # 验证计划存在 - try: - plan = svc.get_plan_or_raise(plan_id) - except ValueError as exc: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=str(exc), - ) - - # 项目鉴权 - if plan.project_id: - check_project_access(plan.project_id, current_user.user.id, project_repository) - - # 验证状态：只允许 draft 或 editing - plan_status = plan.status.value if hasattr(plan.status, "value") else plan.status - if plan_status not in ("draft", "editing"): - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail="当前计划状态不支持AI推荐，请先创建或编辑计划后再试", - ) - - # 调用 AI 推荐服务（同步调用 stub，后续改为 Celery 异步） - from apps.worker.worker_app.tasks.ai_tasks import run_ai_recommend - - result = run_ai_recommend( - plan_id=plan_id, - template_id=plan.template_id, - asset_ids=body.asset_ids, - editing_mode=body.editing_mode, - target_duration=body.target_duration, - ) - - # ── 事务保护：清除 → 重建 → 更新 必须在同一逻辑事务中 ── - # TODO: 当前各 repo 方法内部 commit()，无法真正回滚。 - # 后续重构 repo 为 flush() 模式后，此处改为统一 commit。 - try: - # 清除现有片段 - svc.delete_all_clips(plan_id) - - # 按推荐方案创建新片段 - for clip_data in result["clips"]: - svc.create_clip( - plan_id=plan_id, - clip_type=clip_data["clip_type"], - order=clip_data["order"], - text_content=clip_data.get("text_content", ""), - duration=clip_data["duration"], - transition_effect=clip_data.get("transition_effect", "cut"), - asset_id=clip_data.get("asset_id", ""), - start_time=clip_data.get("start_time", 0.0), - config=clip_data.get("config", {}), - ) - - # 更新计划 config 和 total_duration - normalized_config = normalize_plan_config(result.get("config", {})) - svc.update_plan( - plan_id, - config=normalized_config, - total_duration=result["total_duration"], - ) - except Exception: - logger.exception("AI 推荐写入失败，plan_id=%s 数据可能不一致", plan_id) - # 尝试回滚未提交的变更 - try: - db.rollback() - except Exception: - pass - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail="AI推荐结果保存失败，请稍后重试", - ) - - logger.info( - "AI 推荐片段方案: plan_id=%s clips=%d duration=%.1f by user=%s", - plan_id, - len(result["clips"]), - result["total_duration"], - current_user.user.id, - ) - - return AIRecommendResponse( - plan_id=plan_id, - clips=[ - AIRecommendClipItem( - clip_type=c["clip_type"], - order=c["order"], - text_content=c.get("text_content", ""), - duration=c["duration"], - transition_effect=c.get("transition_effect", "cut"), - asset_id=c.get("asset_id", ""), - start_time=c.get("start_time", 0.0), - config=c.get("config", {}), - ) - for c in result["clips"] - ], - config=normalized_config, - total_duration=result["total_duration"], - confidence=result["confidence"], - ) - - -@router.post( - "/{plan_id}/generate-cover", - response_model=GenerateCoverResponse, -) -def generate_cover( - plan_id: str, - body: GenerateCoverRequest, - db: Session = Depends(get_db_session), - current_user: AuthenticatedUser = Depends(get_current_user), - project_repository: Any = Depends(get_project_repository), -) -> GenerateCoverResponse: - """AI 生成封面 - - 调用 AI 服务从视频中选帧或生成封面图，并更新计划 config.cover。 - - 流程： - 1. 验证计划存在 - 2. 调用 AI 封面生成服务（当前为 stub，后续接入真实 AI） - 3. 更新 plan.config["cover"] 为生成的封面数据 - 4. 返回封面数据 - - 前端对接： - - cover_type=ai_frame: AI 智能选帧（默认） - - cover_type=manual: 手动选帧，需传 frame_time - - cover_type=upload: 用户上传，接口返回空 image_url，前端自行上传后更新 - - cover_type=ai_regenerate: AI 重新生成 - """ - svc = EditPlanService(db) - - # 验证计划存在 - try: - plan = svc.get_plan_or_raise(plan_id) - except ValueError as exc: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=str(exc), - ) - - # 项目鉴权 - if plan.project_id: - check_project_access(plan.project_id, current_user.user.id, project_repository) - - # 调用 AI 封面生成服务 - from apps.worker.worker_app.tasks.ai_tasks import run_generate_cover - - cover_data = run_generate_cover( - plan_id=plan_id, - asset_ids=body.asset_ids, - cover_type=body.cover_type, - frame_time=body.frame_time, - ) - - # 更新 plan.config["cover"] - current_config = dict(plan.config) - current_config["cover"] = cover_data - normalized = normalize_plan_config(current_config) - svc.update_plan_config(plan_id, {"cover": normalized["cover"]}) - - logger.info( - "AI 封面生成: plan_id=%s type=%s by user=%s", - plan_id, - body.cover_type, - current_user.user.id, - ) - - return GenerateCoverResponse( - plan_id=plan_id, - cover=cover_data, - ) - - -# ── Timeline / Scene 端点（P2-6） ───────────────────────────────────────────── - - -class TimelineSceneResponse(BaseModel): - """时间线场景""" - - scene: str = Field(..., description="场景描述") - time: str = Field(..., description='时间范围，如 "0:00 - 0:05"') - duration: float = Field(..., ge=0, description="时长（秒）") - color: str = Field(..., description="展示颜色") - clip_id: str = Field(default="", description="关联的片段 ID") - clip_type: str = Field(default="", description="片段类型") - - -class TimelineResponse(BaseModel): - """时间线响应""" - - plan_id: str - total_duration: float - scenes: List[TimelineSceneResponse] - - -# clip_type → 颜色映射 -_CLIP_TYPE_COLORS = { - "intro": "#6366f1", - "title": "#6366f1", - "product": "#818cf8", - "showcase": "#10b981", - "scene": "#10b981", - "subtitle": "#f59e0b", - "text": "#f59e0b", - "cta": "#ef4444", - "outro": "#ef4444", - "voiceover": "#8b5cf6", - "transition": "#64748b", -} - -_DEFAULT_COLOR = "#6366f1" - - -def _format_time(seconds: float) -> str: - """将秒数格式化为 M:SS""" - m = int(seconds) // 60 - s = int(seconds) % 60 - return f"{m}:{s:02d}" - - -def _clip_type_to_scene_label(clip_type: str, text_content: str) -> str: - """根据 clip_type 和 text_content 生成场景描述""" - type_labels = { - "intro": "开场", - "title": "标题", - "product": "产品展示", - "showcase": "场景展示", - "scene": "场景", - "subtitle": "字幕", - "text": "文字", - "cta": "结尾 CTA", - "outro": "结尾", - "voiceover": "配音", - "transition": "转场", - } - label = type_labels.get(clip_type, clip_type or "片段") - if text_content: - # 截取前 20 个字符作为副标题 - short = text_content[:20].strip() - if short: - return f"{label} - {short}" - return label - - -@router.get( - "/{plan_id}/timeline", - response_model=TimelineResponse, -) -def get_plan_timeline( - plan_id: str, - db: Session = Depends(get_db_session), - current_user: AuthenticatedUser = Depends(get_current_user), - project_repository: Any = Depends(get_project_repository), -) -> TimelineResponse: - """获取剪辑计划的时间线场景数据 - - 返回按计划片段排序的时间线场景列表，供前端 GeneratePage 渲染使用。 - """ - svc = EditPlanService(db) - plan = svc.get_plan_or_raise(plan_id) - # 项目鉴权 - if plan.project_id: - check_project_access(plan.project_id, current_user.user.id, project_repository) - - clips = svc.list_clips(plan_id=plan_id, skip=0, limit=200) - # 按 order 排序 - clips.sort(key=lambda c: c.order) - - scenes: List[TimelineSceneResponse] = [] - current_time = 0.0 - - for clip in clips: - start = current_time - end = start + clip.duration - color = _CLIP_TYPE_COLORS.get(clip.clip_type, _DEFAULT_COLOR) - scene_label = _clip_type_to_scene_label(clip.clip_type, clip.text_content) - - scenes.append( - TimelineSceneResponse( - scene=scene_label, - time=f"{_format_time(start)} - {_format_time(end)}", - duration=clip.duration, - color=color, - clip_id=clip.id, - clip_type=clip.clip_type, - ) - ) - current_time = end - - total_duration = sum(s.duration for s in scenes) or plan.total_duration - - return TimelineResponse( - plan_id=plan_id, - 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 - ], - ) +router.include_router(generation_router) +router.include_router(ai_router) +router.include_router(timeline_router) diff --git a/apps/api/app/api/routes/edit_plans_ai.py b/apps/api/app/api/routes/edit_plans_ai.py new file mode 100644 index 000000000..83c5b6bc4 --- /dev/null +++ b/apps/api/app/api/routes/edit_plans_ai.py @@ -0,0 +1,199 @@ +"""剪辑计划 AI 推荐 & 封面生成 API 端点。 + +从 edit_plans.py 拆分，包含： +- POST /{plan_id}/ai-recommend AI 推荐片段方案 +- POST /{plan_id}/generate-cover AI 生成封面 +""" + +from __future__ import annotations + +import logging +from typing import Any + +from app.api.routes._helpers import check_project_access +from app.api.routes.edit_plans import ( + AIRecommendClipItem, + AIRecommendRequest, + AIRecommendResponse, + GenerateCoverRequest, + GenerateCoverResponse, +) +from app.auth import AuthenticatedUser, get_current_user +from app.dependencies import get_db_session, get_project_repository +from app.services import EditPlanService +from fastapi import APIRouter, Depends, HTTPException, status +from sqlalchemy.orm import Session + +from packages.domain.config_schemas import normalize_plan_config + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +@router.post( + "/{plan_id}/ai-recommend", + response_model=AIRecommendResponse, +) +def ai_recommend_clips( + plan_id: str, + body: AIRecommendRequest, + db: Session = Depends(get_db_session), + current_user: AuthenticatedUser = Depends(get_current_user), + project_repository: Any = Depends(get_project_repository), +) -> AIRecommendResponse: + """AI 推荐片段方案 + + 调用 AI 服务分析素材，自动生成片段编排方案并写入剪辑计划。 + + 流程： + 1. 验证计划存在且状态为 draft/editing + 2. 调用 AI 推荐服务（当前为 stub，后续接入真实 AI） + 3. 清除计划现有片段，按推荐方案重新创建 + 4. 更新计划 config（cover/title/subtitle/bgm）和 total_duration + 5. 返回推荐方案详情 + """ + svc = EditPlanService(db) + + try: + plan = svc.get_plan_or_raise(plan_id) + except ValueError as exc: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) + + if plan.project_id: + check_project_access(plan.project_id, current_user.user.id, project_repository) + + plan_status = plan.status.value if hasattr(plan.status, "value") else plan.status + if plan_status not in ("draft", "editing"): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="当前计划状态不支持AI推荐，请先创建或编辑计划后再试", + ) + + from apps.worker.worker_app.tasks.ai_tasks import run_ai_recommend + + result = run_ai_recommend( + plan_id=plan_id, + template_id=plan.template_id, + asset_ids=body.asset_ids, + editing_mode=body.editing_mode, + target_duration=body.target_duration, + ) + + # 事务保护：清除 → 重建 → 更新 必须在同一逻辑事务中 + try: + svc.delete_all_clips(plan_id) + + for clip_data in result["clips"]: + svc.create_clip( + plan_id=plan_id, + clip_type=clip_data["clip_type"], + order=clip_data["order"], + text_content=clip_data.get("text_content", ""), + duration=clip_data["duration"], + transition_effect=clip_data.get("transition_effect", "cut"), + asset_id=clip_data.get("asset_id", ""), + start_time=clip_data.get("start_time", 0.0), + config=clip_data.get("config", {}), + ) + + normalized_config = normalize_plan_config(result.get("config", {})) + svc.update_plan( + plan_id, + config=normalized_config, + total_duration=result["total_duration"], + ) + except Exception: + logger.exception("AI 推荐写入失败，plan_id=%s 数据可能不一致", plan_id) + try: + db.rollback() + except Exception as rollback_err: + logger.error( + "AI 推荐回滚失败，数据库会话可能处于不一致状态: plan_id=%s error=%s", + plan_id, + rollback_err, + ) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="AI推荐结果保存失败，请稍后重试", + ) + + logger.info( + "AI 推荐片段方案: plan_id=%s clips=%d duration=%.1f by user=%s", + plan_id, + len(result["clips"]), + result["total_duration"], + current_user.user.id, + ) + + return AIRecommendResponse( + plan_id=plan_id, + clips=[ + AIRecommendClipItem( + clip_type=c["clip_type"], + order=c["order"], + text_content=c.get("text_content", ""), + duration=c["duration"], + transition_effect=c.get("transition_effect", "cut"), + asset_id=c.get("asset_id", ""), + start_time=c.get("start_time", 0.0), + config=c.get("config", {}), + ) + for c in result["clips"] + ], + config=normalized_config, + total_duration=result["total_duration"], + confidence=result["confidence"], + ) + + +@router.post( + "/{plan_id}/generate-cover", + response_model=GenerateCoverResponse, +) +def generate_cover( + plan_id: str, + body: GenerateCoverRequest, + db: Session = Depends(get_db_session), + current_user: AuthenticatedUser = Depends(get_current_user), + project_repository: Any = Depends(get_project_repository), +) -> GenerateCoverResponse: + """AI 生成封面 + + 调用 AI 服务从视频中选帧或生成封面图，并更新计划 config.cover。 + """ + svc = EditPlanService(db) + + try: + plan = svc.get_plan_or_raise(plan_id) + except ValueError as exc: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) + + if plan.project_id: + check_project_access(plan.project_id, current_user.user.id, project_repository) + + from apps.worker.worker_app.tasks.ai_tasks import run_generate_cover + + cover_data = run_generate_cover( + plan_id=plan_id, + asset_ids=body.asset_ids, + cover_type=body.cover_type, + frame_time=body.frame_time, + ) + + current_config = dict(plan.config) + current_config["cover"] = cover_data + normalized = normalize_plan_config(current_config) + svc.update_plan_config(plan_id, {"cover": normalized["cover"]}) + + logger.info( + "AI 封面生成: plan_id=%s type=%s by user=%s", + plan_id, + body.cover_type, + current_user.user.id, + ) + + return GenerateCoverResponse( + plan_id=plan_id, + cover=cover_data, + ) diff --git a/apps/api/app/api/routes/edit_plans_generation.py b/apps/api/app/api/routes/edit_plans_generation.py new file mode 100644 index 000000000..3aae7abfa --- /dev/null +++ b/apps/api/app/api/routes/edit_plans_generation.py @@ -0,0 +1,383 @@ +"""剪辑计划生成相关 API 端点。 + +从 edit_plans.py 拆分，包含： +- POST /{plan_id}/generate 触发剪辑渲染生成 +- GET /{plan_id}/generation-status 查询生成进度 +- GET /{plan_id}/generations 查询关联的生成记录 +""" + +from __future__ import annotations + +import logging +from typing import Any + +from app.api.routes._helpers import check_project_access +from app.api.routes.edit_plans import ( + ClipStatusItem, + EditPlanGenerateResponse, + EditPlanGenerationStatusResponse, + EditPlanGenerationsResponse, +) +from app.auth import AuthenticatedUser, get_current_user +from app.core.celery_app import celery_app +from app.core.task_enqueue import GLOBAL_PENDING_LIMIT, USER_PENDING_LIMIT +from app.dependencies import get_asset_library_repository, get_asset_repository, get_db_session, get_project_repository +from app.services import EditPlanService +from fastapi import APIRouter, Depends, HTTPException, status +from sqlalchemy.orm import Session + +from packages.adapters.sqlalchemy_impl.generation_task_repository import ( + SQLAlchemyGenerationTaskRepository, +) +from packages.adapters.sqlalchemy_impl.template_clip_config_repository import ( + SQLAlchemyTemplateClipConfigRepository, +) +from packages.adapters.sqlalchemy_impl.template_repository import ( + SQLAlchemyTemplateRepository, +) +from packages.application.generation_tasks import ( + CreateGenerationTaskCommand, + CreateGenerationTaskUseCase, +) +from packages.domain.edit_plan import EditPlanStatus + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +def _auto_fallback_draft_to_editing(svc: EditPlanService, plan_id: str, plan_check) -> None: + """自动兜底 1: draft → editing""" + if plan_check.status == EditPlanStatus.DRAFT: + logger.info("自动兜底: plan=%s draft→editing", plan_id) + svc.transition_status(plan_id, EditPlanStatus.EDITING) + + +def _auto_fallback_copy_template_clips(svc: EditPlanService, plan_id: str, plan_check, db: Session) -> None: + """自动兜底 2: 无片段 + 有 template_id → 从模板复制片段配置""" + existing_clips = svc.count_clips(plan_id) + if existing_clips == 0 and plan_check.template_id: + logger.info( + "自动兜底: plan=%s 无片段，从模板 %s 复制片段配置", + plan_id, + plan_check.template_id, + ) + clip_config_repo = SQLAlchemyTemplateClipConfigRepository(db) + configs = clip_config_repo.list_by_template(plan_check.template_id) + if configs: + for cfg in configs: + svc.create_clip( + plan_id=plan_id, + clip_type=cfg.clip_type.value if hasattr(cfg.clip_type, "value") else cfg.clip_type, + order=cfg.order, + template_clip_config_id=cfg.id, + duration=cfg.default_duration, + transition_effect=( + cfg.transition_effect.value + if hasattr(cfg.transition_effect, "value") + else cfg.transition_effect + ), + ) + logger.info("自动兜底: plan=%s 从新模型 template_clip_configs 复制了 %d 个片段", plan_id, len(configs)) + else: + tpl_repo = SQLAlchemyTemplateRepository(db) + segments = tpl_repo.list_segments(plan_check.template_id) + for seg in segments: + avg_duration = (seg.duration_min + seg.duration_max) / 2 + svc.create_clip( + plan_id=plan_id, + clip_type="main", + order=seg.segment_order, + duration=avg_duration, + config={ + "material_type": seg.material_type or "", + "template_segment_id": seg.id, + }, + ) + logger.info("自动兜底: plan=%s 从旧模型 template_segments 复制了 %d 个片段", plan_id, len(segments)) + + +def _auto_fallback_assign_assets( + svc: EditPlanService, + plan_id: str, + plan_check, +) -> list: + """自动兜底 3: 为没有素材的片段分配素材。返回剩余无素材片段列表。""" + all_clips = svc.list_clips(plan_id) + clips_without_asset = [c for c in all_clips if not c.asset_id] + config_asset_ids = (plan_check.config or {}).get("asset_ids", []) + + if clips_without_asset and config_asset_ids: + logger.info( + "自动兜底3: plan=%s 为 %d 个无素材片段分配 %d 个指定素材", + plan_id, + len(clips_without_asset), + len(config_asset_ids), + ) + for i, clip in enumerate(clips_without_asset): + asset_idx = i % len(config_asset_ids) + svc.assign_asset(clip.id, config_asset_ids[asset_idx]) + logger.info("自动兜底3: plan=%s 素材分配完成", plan_id) + clips_without_asset = [] + + return clips_without_asset + + +def _auto_fallback_auto_material_mode( + svc: EditPlanService, + plan_id: str, + plan_check, + clips_without_asset: list, + asset_library_repo: Any, + asset_repo: Any, +) -> None: + """自动兜底 4: 自动素材模式 → 从项目默认视频素材库选取""" + if not clips_without_asset: + return + material_mode = (plan_check.config or {}).get("material_mode", "manual") + if material_mode != "auto" or not plan_check.project_id: + return + + import random + + logger.info( + "自动兜底4: plan=%s 自动素材模式，从项目素材库选取素材 (%d 个片段需要)", + plan_id, + len(clips_without_asset), + ) + libs = asset_library_repo.find_by_project(plan_check.project_id) + video_lib = None + for lib in libs: + lib_kind = lib.kind.value if hasattr(lib.kind, "value") else lib.kind + if lib_kind == "video": + video_lib = lib + break + + if video_lib: + assets = asset_repo.find_by_library(video_lib.id) + ready_videos = [ + a + for a in assets + if (a.status.value if hasattr(a.status, "value") else a.status) == "ready" + and a.mime_type + and a.mime_type.startswith("video") + ] + if ready_videos: + random.shuffle(ready_videos) + for i, clip in enumerate(clips_without_asset): + asset = ready_videos[i % len(ready_videos)] + svc.assign_asset(clip.id, asset.id) + logger.info( + "自动兜底4: plan=%s 从素材库 %s 分配了 %d 个素材给 %d 个片段", + plan_id, + video_lib.name, + len(ready_videos), + len(clips_without_asset), + ) + else: + logger.warning("自动兜底4: plan=%s 素材库无可用视频素材", plan_id) + else: + logger.warning("自动兜底4: plan=%s 项目无视频素材库", plan_id) + + +def _check_queue_limits(gen_task_repo, user_id: str) -> None: + """队列限流预检查""" + try: + has_count = hasattr(gen_task_repo, "count_pending_by_user") and hasattr( + gen_task_repo, "count_pending_total" + ) + if has_count: + user_pending = gen_task_repo.count_pending_by_user(user_id) + global_pending = gen_task_repo.count_pending_total() + if user_pending >= USER_PENDING_LIMIT: + raise HTTPException( + status_code=429, + detail=f"您的待处理任务过多（当前 {user_pending}/{USER_PENDING_LIMIT}），请等待完成后再提交", + ) + if global_pending >= GLOBAL_PENDING_LIMIT: + raise HTTPException( + status_code=503, + detail="系统繁忙，请稍后再试", + ) + except HTTPException: + raise + except Exception as e: + logger.warning("[队列限流] 剪辑计划限流检查失败，跳过: %s", e) + + +@router.post("/{plan_id}/generate", response_model=EditPlanGenerateResponse) +def generate_plan( + plan_id: str, + db: Session = Depends(get_db_session), + current_user: AuthenticatedUser = Depends(get_current_user), + project_repository: Any = Depends(get_project_repository), + asset_library_repo: Any = Depends(get_asset_library_repository), + asset_repo: Any = Depends(get_asset_repository), +) -> EditPlanGenerateResponse: + """触发剪辑计划渲染生成 + + 前置条件：计划状态必须为 editing，且至少有一个片段。 + 流程： + 1. 验证计划状态为 editing + 2. 将 pending 片段标记为 ready + 3. 创建 GenerationTask + 4. 调度 Celery 任务 worker.render_edit_plan + 5. 将计划状态流转为 rendering + """ + svc = EditPlanService(db) + plan_check = svc.get_plan(plan_id) + if plan_check is None: + raise HTTPException(status_code=404, detail=f"剪辑计划不存在: {plan_id}") + if plan_check.project_id: + check_project_access(plan_check.project_id, current_user.user.id, project_repository) + + # 自动兜底流程 + _auto_fallback_draft_to_editing(svc, plan_id, plan_check) + _auto_fallback_copy_template_clips(svc, plan_id, plan_check, db) + clips_without_asset = _auto_fallback_assign_assets(svc, plan_id, plan_check) + _auto_fallback_auto_material_mode(svc, plan_id, plan_check, clips_without_asset, asset_library_repo, asset_repo) + + # 检查是否可生成 + try: + can_gen, reason = svc.can_generate(plan_id) + except ValueError as exc: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) + if not can_gen: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=reason) + + # 核心生成流程 + try: + clip_count = svc.mark_clips_ready(plan_id) + + gen_task_repo = SQLAlchemyGenerationTaskRepository(db) + user_id = current_user.user.id + _check_queue_limits(gen_task_repo, user_id) + + gen_task_use_case = CreateGenerationTaskUseCase(gen_task_repo) + plan = svc.get_plan_or_raise(plan_id) + gen_task = gen_task_use_case.execute( + CreateGenerationTaskCommand( + project_id="", + template_id=plan.template_id, + created_by_user_id=current_user.user.id, + source_edit_plan_id=plan_id, + ) + ) + + svc.update_plan_config(plan_id, {"generation_task_id": gen_task.id}) + svc.transition_status(plan_id, EditPlanStatus.RENDERING) + celery_app.send_task("worker.render_edit_plan", args=[plan_id]) + + updated_plan = svc.get_plan_or_raise(plan_id) + + logger.info( + "触发剪辑计划生成: plan_id=%s gen_task_id=%s clips=%d by user=%s", + plan_id, + gen_task.id, + clip_count, + current_user.user.id, + ) + + return EditPlanGenerateResponse( + plan_id=plan_id, + plan_status=updated_plan.status.value if hasattr(updated_plan.status, "value") else updated_plan.status, + generation_task_id=gen_task.id, + clip_count=clip_count, + ) + except HTTPException: + raise + except Exception: + logger.exception("触发剪辑计划生成失败: plan_id=%s", plan_id) + try: + svc.transition_status(plan_id, EditPlanStatus.FAILED) + except Exception: + logger.warning("标记计划失败状态时异常: plan_id=%s", plan_id) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="生成失败，请稍后重试", + ) + + +@router.get( + "/{plan_id}/generation-status", + response_model=EditPlanGenerationStatusResponse, +) +def get_generation_status( + plan_id: str, + db: Session = Depends(get_db_session), + current_user: AuthenticatedUser = Depends(get_current_user), + project_repository: Any = Depends(get_project_repository), +) -> EditPlanGenerationStatusResponse: + """查询剪辑计划生成进度""" + svc = EditPlanService(db) + try: + gen_status = svc.get_generation_status(plan_id) + except ValueError as exc: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) + + plan = gen_status["plan"] + if plan.project_id: + check_project_access(plan.project_id, current_user.user.id, project_repository) + clips = gen_status["clips"] + + clip_items = [ + ClipStatusItem( + clip_id=c.id, + clip_type=c.clip_type, + order=c.order, + status=c.status.value if hasattr(c.status, "value") else c.status, + asset_id=c.asset_id or "", + text_content=c.text_content or "", + duration=c.duration, + ) + for c in clips + ] + + return EditPlanGenerationStatusResponse( + plan_id=plan_id, + plan_status=plan.status.value if hasattr(plan.status, "value") else plan.status, + generation_task_id=gen_status["generation_task_id"], + clips=clip_items, + ) + + +@router.get( + "/{plan_id}/generations", + response_model=EditPlanGenerationsResponse, +) +def list_plan_generations( + plan_id: str, + db: Session = Depends(get_db_session), + current_user: AuthenticatedUser = Depends(get_current_user), + project_repository: Any = Depends(get_project_repository), +) -> EditPlanGenerationsResponse: + """查询剪辑计划关联的所有生成记录""" + svc = EditPlanService(db) + plan = svc.get_plan_or_raise(plan_id) + if plan.project_id: + check_project_access(plan.project_id, current_user.user.id, project_repository) + + from app.schemas.generation_task import GenerationTaskResponse + + gen_task_repo = SQLAlchemyGenerationTaskRepository(db) + tasks = gen_task_repo.list_by_source_edit_plan(plan_id) + items = [ + GenerationTaskResponse( + id=t.id, + project_id=t.project_id, + asset_library_id=t.asset_library_id, + strategy_id=t.strategy_id, + voice_library_id=t.voice_library_id, + template_id=t.template_id, + asset_ids=t.asset_ids, + title_ids=t.title_ids, + voice_ids=t.voice_ids, + source_edit_plan_id=t.source_edit_plan_id or "", + status=t.status.value if hasattr(t.status, "value") else t.status, + progress=t.progress, + result_count=t.result_count, + error_message=t.error_message, + ) + for t in tasks + ] + return EditPlanGenerationsResponse(items=items, total=len(items)) diff --git a/apps/api/app/api/routes/edit_plans_timeline.py b/apps/api/app/api/routes/edit_plans_timeline.py new file mode 100644 index 000000000..3d343ad06 --- /dev/null +++ b/apps/api/app/api/routes/edit_plans_timeline.py @@ -0,0 +1,222 @@ +"""剪辑计划时间线 & 模板生成 API 端点。 + +从 edit_plans.py 拆分，包含： +- GET /{plan_id}/timeline 时间线场景数据 +- POST /generate-from-template 基于模板+素材自动生成剪辑计划 +""" + +from __future__ import annotations + +import logging +from datetime import datetime +from typing import Any, List, Optional + +from app.api.routes._helpers import check_project_access +from app.api.routes.edit_plans import ( + EditPlanResponse, + GenerateFromTemplateRequest, + GenerateFromTemplateResponse, + _PlanClipItem, + _to_response, +) +from app.auth import AuthenticatedUser, get_current_user +from app.dependencies import get_db_session, get_project_repository +from app.services import EditPlanService, PlanGeneratorService +from fastapi import APIRouter, Depends, HTTPException, Query, status +from pydantic import BaseModel, Field +from sqlalchemy.orm import Session + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +# ── Timeline Schemas ────────────────────────────────────────────────────────── + + +class TimelineSceneResponse(BaseModel): + """时间线场景""" + + scene: str = Field(..., description="场景描述") + time: str = Field(..., description='时间范围，如 "0:00 - 0:05"') + duration: float = Field(..., ge=0, description="时长（秒）") + color: str = Field(..., description="展示颜色") + clip_id: str = Field(default="", description="关联的片段 ID") + clip_type: str = Field(default="", description="片段类型") + + +class TimelineResponse(BaseModel): + """时间线响应""" + + plan_id: str + total_duration: float + scenes: List[TimelineSceneResponse] + + +# clip_type → 颜色映射 +_CLIP_TYPE_COLORS = { + "intro": "#6366f1", + "title": "#6366f1", + "product": "#818cf8", + "showcase": "#10b981", + "scene": "#10b981", + "subtitle": "#f59e0b", + "text": "#f59e0b", + "cta": "#ef4444", + "outro": "#ef4444", + "voiceover": "#8b5cf6", + "transition": "#64748b", +} + +_DEFAULT_COLOR = "#6366f1" + + +def _format_time(seconds: float) -> str: + """将秒数格式化为 M:SS""" + m = int(seconds) // 60 + s = int(seconds) % 60 + return f"{m}:{s:02d}" + + +def _clip_type_to_scene_label(clip_type: str, text_content: str) -> str: + """根据 clip_type 和 text_content 生成场景描述""" + type_labels = { + "intro": "开场", + "title": "标题", + "product": "产品展示", + "showcase": "场景展示", + "scene": "场景", + "subtitle": "字幕", + "text": "文字", + "cta": "结尾 CTA", + "outro": "结尾", + "voiceover": "配音", + "transition": "转场", + } + label = type_labels.get(clip_type, clip_type or "片段") + if text_content: + short = text_content[:20].strip() + if short: + return f"{label} - {short}" + return label + + +# ── Routes ──────────────────────────────────────────────────────────────────── + + +@router.get( + "/{plan_id}/timeline", + response_model=TimelineResponse, +) +def get_plan_timeline( + plan_id: str, + db: Session = Depends(get_db_session), + current_user: AuthenticatedUser = Depends(get_current_user), + project_repository: Any = Depends(get_project_repository), +) -> TimelineResponse: + """获取剪辑计划的时间线场景数据""" + svc = EditPlanService(db) + plan = svc.get_plan_or_raise(plan_id) + if plan.project_id: + check_project_access(plan.project_id, current_user.user.id, project_repository) + + clips = svc.list_clips(plan_id=plan_id, skip=0, limit=200) + clips.sort(key=lambda c: c.order) + + scenes: List[TimelineSceneResponse] = [] + current_time = 0.0 + + for clip in clips: + start = current_time + end = start + clip.duration + color = _CLIP_TYPE_COLORS.get(clip.clip_type, _DEFAULT_COLOR) + scene_label = _clip_type_to_scene_label(clip.clip_type, clip.text_content) + + scenes.append( + TimelineSceneResponse( + scene=scene_label, + time=f"{_format_time(start)} - {_format_time(end)}", + duration=clip.duration, + color=color, + clip_id=clip.id, + clip_type=clip.clip_type, + ) + ) + current_time = end + + total_duration = sum(s.duration for s in scenes) or plan.total_duration + + return TimelineResponse( + plan_id=plan_id, + 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: + """基于模板 + 素材自动生成剪辑计划""" + 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) + + 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 + ], + ) diff --git a/apps/api/app/api/routes/feature_flags.py b/apps/api/app/api/routes/feature_flags.py index fada08be9..20a2dc53e 100755 --- a/apps/api/app/api/routes/feature_flags.py +++ b/apps/api/app/api/routes/feature_flags.py @@ -89,7 +89,7 @@ def _validate_flag_name(name: str) -> None: async def list_feature_flags( _: bool = Depends(_verify_internal_api_key), store: RedisFeatureFlagStore = Depends(_get_feature_flag_store), -): +) -> list[FeatureFlagResponse]: """列出所有 Feature Flag。""" try: flags = store.list_all() @@ -113,7 +113,7 @@ async def get_feature_flag( name: str, _: bool = Depends(_verify_internal_api_key), store: RedisFeatureFlagStore = Depends(_get_feature_flag_store), -): +) -> FeatureFlagResponse: """获取单个 Feature Flag 配置。""" try: config = store.get(name) @@ -129,7 +129,7 @@ async def check_feature_flag( identifier: Optional[str] = Query(None, description="标识符，如 user_id"), _: bool = Depends(_verify_internal_api_key), store: RedisFeatureFlagStore = Depends(_get_feature_flag_store), -): +) -> FeatureFlagCheckResponse: """检查某个标识符是否命中 Feature Flag。""" try: active = store.is_active(name, identifier=identifier) @@ -145,7 +145,7 @@ async def update_feature_flag( request: FeatureFlagUpdateRequest, _: bool = Depends(_verify_internal_api_key), store: RedisFeatureFlagStore = Depends(_get_feature_flag_store), -): +) -> FeatureFlagResponse: """更新 Feature Flag 配置。 只允许修改 ALLOWED_FLAGS 列表中的 flag。 @@ -178,7 +178,7 @@ async def delete_feature_flag( name: str, _: bool = Depends(_verify_internal_api_key), store: RedisFeatureFlagStore = Depends(_get_feature_flag_store), -): +) -> None: """删除 Feature Flag。 只允许删除 ALLOWED_FLAGS 列表中的 flag。 diff --git a/apps/api/app/api/routes/projects.py b/apps/api/app/api/routes/projects.py index 59be29640..17be8f438 100755 --- a/apps/api/app/api/routes/projects.py +++ b/apps/api/app/api/routes/projects.py @@ -77,7 +77,7 @@ def delete_project( project_id: str, authenticated_user: AuthenticatedUser = Depends(get_current_user), project_repository: Any = Depends(get_project_repository), -): +) -> dict: use_case = DeleteProjectUseCase(project_repository) try: deleted = use_case.execute(project_id, authenticated_user.user.id) diff --git a/apps/api/app/api/routes/subscription.py b/apps/api/app/api/routes/subscription.py index 1ff6ff013..47f5d44ff 100644 --- a/apps/api/app/api/routes/subscription.py +++ b/apps/api/app/api/routes/subscription.py @@ -92,7 +92,7 @@ def _build_subscription_info(user: AuthenticatedUser) -> SubscriptionInfo: @router.get("/current", response_model=SubscriptionInfo) async def get_current_subscription( current_user: AuthenticatedUser = Depends(get_current_user), -): +) -> SubscriptionInfo: """获取当前订阅信息""" return _build_subscription_info(current_user) @@ -100,7 +100,7 @@ async def get_current_subscription( @router.get("/billing-records", response_model=List[BillingRecord]) async def get_billing_records( current_user: AuthenticatedUser = Depends(get_current_user), -): +) -> List[BillingRecord]: """获取账单记录列表""" from packages.adapters.sqlalchemy_impl.billing_repository import SQLAlchemyBillingRepository from packages.adapters.sqlalchemy_impl.session import SessionLocal @@ -134,7 +134,7 @@ async def change_plan( request: ChangePlanRequest, current_user: AuthenticatedUser = Depends(get_current_user), user_repository: UserRepository = Depends(get_user_repository), -): +) -> ChangePlanResponse: """变更订阅套餐（升级/降级）""" # TODO: 接入支付验证（支付宝/微信支付） valid_plans = {"free", "standard", "pro", "enterprise"} @@ -186,7 +186,7 @@ async def change_plan( async def cancel_subscription( current_user: AuthenticatedUser = Depends(get_current_user), user_repository: UserRepository = Depends(get_user_repository), -): +) -> SimpleResponse: """取消订阅""" user = current_user.user if user.subscription_plan == "free": @@ -212,7 +212,7 @@ async def payment_callback( amount: float, payment_method: str = "alipay", payment_id: str = "", -): +) -> dict: """支付回调 - 在事务中更新账单和订阅状态 注意：生产环境需要验证支付签名 @@ -263,7 +263,7 @@ async def payment_callback( async def toggle_auto_renew( request: ToggleAutoRenewRequest, current_user: AuthenticatedUser = Depends(get_current_user), -): +) -> SimpleResponse: """切换自动续费""" # TODO: 实际需要在数据库中存储 auto_renew 字段 status_text = "已开启自动续费" if request.enabled else "已关闭自动续费" diff --git a/apps/api/app/api/routes/tts.py b/apps/api/app/api/routes/tts.py index 93b27aa4d..77f9991e2 100755 --- a/apps/api/app/api/routes/tts.py +++ b/apps/api/app/api/routes/tts.py @@ -372,10 +372,10 @@ async def tts_websocket_stream( streaming_service = TTSStreamingService(cosyvoice_service) await streaming_service.synthesize_and_stream(websocket, params) except WebSocketDisconnect: - logger.info("WebSocket 客户端断开连接") + logger.info("WebSocket 客户端主动断开连接") except Exception as e: logger.error(f"WebSocket 流式合成异常: {e}", exc_info=True) try: await websocket.send_json({"type": "error", "message": f"服务异常: {e}"}) - except Exception: - pass + except Exception as send_err: + logger.warning("WebSocket 错误消息发送失败（连接可能已断开）: %s", send_err) diff --git a/apps/api/app/middleware/auth.py b/apps/api/app/middleware/auth.py index 1bbf7b74d..6bcdfec48 100644 --- a/apps/api/app/middleware/auth.py +++ b/apps/api/app/middleware/auth.py @@ -7,12 +7,16 @@ common auth dependencies for backward compatibility. from __future__ import annotations +import logging + from app.auth import AuthenticatedUser from app.auth import get_current_user as get_authenticated_user from app.dependencies import get_user_repository from fastapi import Depends, HTTPException from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer +logger = logging.getLogger(__name__) + from packages.domain.entities import User from packages.ports.user_repository import UserRepository @@ -33,6 +37,10 @@ async def get_current_user_optional( return None try: authenticated_user = await get_authenticated_user(credentials, user_repository) - except HTTPException: + except HTTPException as exc: + if exc.status_code >= 500: + # 服务端错误不应被静默吞掉，记录日志 + logger.error("可选认证遇到服务端错误，status=%s", exc.status_code, exc_info=True) + # 4xx 认证失败（如 token 无效、用户不存在）属于正常流程，返回 None return None return authenticated_user.user diff --git a/apps/worker/video_processing/processor.py b/apps/worker/video_processing/processor.py index 6e9329d3a..8b2257280 100644 --- a/apps/worker/video_processing/processor.py +++ b/apps/worker/video_processing/processor.py @@ -130,8 +130,8 @@ class VideoProcessor: if concat_file is not None: try: concat_file.close() - except Exception: - pass # 忽略关闭时的错误 + except OSError as close_err: + logger.warning("临时文件关闭失败: %s", close_err) def generate_thumbnail( self, diff --git a/apps/worker/video_processing/render_adapter.py b/apps/worker/video_processing/render_adapter.py index 4fd5f76c0..b839a646d 100755 --- a/apps/worker/video_processing/render_adapter.py +++ b/apps/worker/video_processing/render_adapter.py @@ -210,8 +210,8 @@ class RenderAdapter: try: shutil.rmtree(temp_dir, ignore_errors=True) - except Exception: - pass + except Exception as cleanup_err: + logger.warning("临时目录清理失败: path=%s error=%s", temp_dir, cleanup_err) def validate_plan(self, plan_id: str) -> tuple[bool, list[str], list[str], int, int]: """校验计划是否可渲染（兼容 VideoComposeService.validate_compose 接口）。 diff --git a/apps/worker/video_processing/render_audio.py b/apps/worker/video_processing/render_audio.py new file mode 100644 index 000000000..6f114f5f6 --- /dev/null +++ b/apps/worker/video_processing/render_audio.py @@ -0,0 +1,364 @@ +"""音频混音模块 — 从 unified_render_service.py 拆分. + +职责： +- 主图层音频 concat 拼接 +- 独立音频轨 amix 混音 +- 音视频合并（mux） + +所有函数接收 RenderContext 获取共享依赖（work_dir、plan_id 等）， +避免直接依赖 UnifiedRenderService 类。 +""" + +from __future__ import annotations + +import logging +import subprocess +from dataclasses import dataclass, field +from pathlib import Path + +from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_has_audio, run_ffmpeg + +# 延迟导入避免循环依赖：unified_render_service 定义 ResolvedClip / RenderLayer， +# 本模块提供音频函数供 unified_render_service 调用。 +# 使用 from __future__ import annotations + TYPE_CHECKING 解决类型引用。 +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from video_processing.unified_render_service import ResolvedClip, RenderLayer + +logger = logging.getLogger(__name__) + + +@dataclass +class RenderContext: + """渲染上下文 — 提供音频混音所需的共享依赖.""" + + work_dir: Path + plan_id: str + # 音频探测缓存（避免同一 clip 被多次 ffprobe） + _audio_cache: dict[str, bool] = field(default_factory=dict) + + +# ── 工具函数 ────────────────────────────────────────────────────────────────── + + +def clip_effective_duration(clip: ResolvedClip) -> float: + """计算 clip 的有效时长. + + 与 UnifiedRenderService._clip_effective_duration 逻辑一致。 + """ + if clip.duration > 0: + return min(clip.duration, clip.actual_duration) if clip.actual_duration > 0 else clip.duration + return clip.actual_duration if clip.actual_duration > 0 else 0.0 + + +def clip_has_audio(ctx: RenderContext, clip: ResolvedClip) -> bool: + """探测 clip 是否有音频流（带缓存）. + + 避免同一个 clip 被多次 ffprobe 探测。 + """ + key = str(clip.local_path) + if key not in ctx._audio_cache: + ctx._audio_cache[key] = probe_has_audio(clip.local_path) + return ctx._audio_cache[key] + + +# ── 音频混音 ────────────────────────────────────────────────────────────────── + + +def mix_audio( + ctx: RenderContext, + layers: list[RenderLayer], + video_duration: float, +) -> Path | None: + """音频后处理混音. + + 处理逻辑： + 1. 主音频源按优先级查找：main > broll（background 不参与主音频，通常是图片无音轨） + 2. 主图层音频按顺序 concat 拼接 + 3. 独立音频轨（audio role）用 amix 混入 + 4. 输出时长截断到 video_duration + 5. 无音频流的 clip 会被自动跳过，避免 FFmpeg 引用 [i:a] 失败 + + Args: + ctx: 渲染上下文 + layers: 图层列表 + video_duration: 视频总时长（用于截断音频） + + Returns: + 混音后的音频文件路径，无音频时返回 None + """ + # 按优先级精确查找主音频图层：main > broll + # background 不参与主音频（通常是静态图片，无音轨） + layer_map = {layer.role: layer for layer in layers} + main_layer = None + for role in ("main", "broll"): + if role in layer_map and layer_map[role].clips: + main_layer = layer_map[role] + break + + main_clips: list[ResolvedClip] = main_layer.clips if main_layer else [] + + # 没有主视频图层时兜底：检查 overlay/corner_voice 层是否有带音频的素材 + if not main_clips: + for role in ("overlay", "corner_voice"): + if role in layer_map and layer_map[role].clips: + main_clips = layer_map[role].clips + break + + # 收集独立音频轨 + audio_clips: list[ResolvedClip] = [] + if "audio" in layer_map: + audio_clips = layer_map["audio"].clips + + # ── 防御：过滤掉无音频流的 clip ── + main_clips = [c for c in main_clips if clip_has_audio(ctx, c)] + audio_clips = [c for c in audio_clips if clip_has_audio(ctx, c)] + + if not main_clips and not audio_clips: + return None + + # 构建音频处理命令 + output_path = ctx.work_dir / f"audio_{ctx.plan_id}.aac" + + # 简单场景：只有主图层 + 无独立音频 → 直接从视频提取音频并拼接 + if main_clips and not audio_clips: + concat_main_audio(ctx, main_clips, output_path, video_duration) + return output_path + + # 有独立音频轨 → amix 混音 + mix_with_independent_audio(ctx, main_clips, audio_clips, output_path, video_duration) + return output_path + + +def concat_main_audio( + ctx: RenderContext, + clips: list[ResolvedClip], + output_path: Path, + video_duration: float, +) -> None: + """主图层音频 concat 拼接（对齐链路A行为）. + + 每个 clip 提取音频 → trim → 按顺序 concat。 + """ + if len(clips) == 1: + # 单 clip，直接提取音频，截断到 min(clip有效时长, 视频总时长) + clip = clips[0] + effective_duration = clip_effective_duration(clip) + # 最终时长：取 clip 有效时长和视频总时长的较小值 + # （视频总时长由主图层决定，但单 clip 场景下两者应该一致，仍做保护） + final_duration = effective_duration + if video_duration > 0 and (final_duration <= 0 or final_duration > video_duration): + final_duration = video_duration + + command = [ + FFMPEG_BIN, + "-y", + "-i", + str(clip.local_path), + "-vn", + "-acodec", + "aac", + "-b:a", + "128k", + ] + if final_duration > 0: + command.extend(["-t", f"{final_duration:.3f}"]) + command.append(str(output_path)) + run_ffmpeg(command) + return + + # 多 clip，用 filter_complex concat + input_args: list[str] = [] + filter_parts: list[str] = [] + + for i, clip in enumerate(clips): + input_args.extend(["-i", str(clip.local_path)]) + effective_duration = clip_effective_duration(clip) + if effective_duration > 0: + filter_parts.append(f"[{i}:a]atrim=0:{effective_duration:.3f},asetpts=PTS-STARTPTS[a{i}]") + else: + filter_parts.append(f"[{i}:a]asetpts=PTS-STARTPTS[a{i}]") + + audio_labels = "".join(f"[a{i}]" for i in range(len(clips))) + filter_parts.append(f"{audio_labels}concat=n={len(clips)}:v=0:a=1[outa]") + + # 截断到视频总时长 + if video_duration > 0: + filter_parts.append(f"[outa]atrim=0:{video_duration:.3f}[final_audio]") + final_label = "final_audio" + else: + final_label = "outa" + + filter_complex = ";".join(filter_parts) + + command = [ + FFMPEG_BIN, + "-y", + *input_args, + "-filter_complex", + filter_complex, + "-map", + f"[{final_label}]", + "-acodec", + "aac", + "-b:a", + "128k", + str(output_path), + ] + run_ffmpeg(command) + + +def mix_with_independent_audio( + ctx: RenderContext, + main_clips: list[ResolvedClip], + audio_clips: list[ResolvedClip], + output_path: Path, + video_duration: float, +) -> None: + """主音频 + 独立音频轨 amix 混音. + + Args: + ctx: 渲染上下文 + main_clips: 主视频 clips（提取音频后 concat） + audio_clips: 独立音频轨 clips + output_path: 输出路径 + video_duration: 视频总时长 + """ + input_args: list[str] = [] + filter_parts: list[str] = [] + mix_labels: list[str] = [] + + input_idx = 0 + + # 1. 主图层音频 concat + if main_clips: + for clip in main_clips: + input_args.extend(["-i", str(clip.local_path)]) + effective_duration = clip_effective_duration(clip) + if effective_duration > 0: + filter_parts.append( + f"[{input_idx}:a]atrim=0:{effective_duration:.3f},asetpts=PTS-STARTPTS[ma{input_idx}]" + ) + else: + filter_parts.append(f"[{input_idx}:a]asetpts=PTS-STARTPTS[ma{input_idx}]") + input_idx += 1 + + if len(main_clips) == 1: + mix_labels.append("ma0") + else: + main_labels = "".join(f"[ma{i}]" for i in range(len(main_clips))) + filter_parts.append(f"{main_labels}concat=n={len(main_clips)}:v=0:a=1[main_audio]") + mix_labels.append("main_audio") + + # 2. 独立音频轨 + for j, clip in enumerate(audio_clips): + input_args.extend(["-i", str(clip.local_path)]) + effective_duration = clip_effective_duration(clip) + volume = clip.config.get("volume", 1.0) if clip.config else 1.0 + label = f"ia{j}" + filters = [] + if effective_duration > 0: + filters.append(f"atrim=0:{effective_duration:.3f}") + filters.append("asetpts=PTS-STARTPTS") + if volume != 1.0: + filters.append(f"volume={volume}") + filter_parts.append(f"[{input_idx}:a]{','.join(filters)}[{label}]") + mix_labels.append(label) + input_idx += 1 + + # 3. amix 混音 + mix_inputs = "".join(f"[{label}]" for label in mix_labels) + n_inputs = len(mix_labels) + # normalized=0 保持音量，duration=shortest 取最短 + filter_parts.append(f"{mix_inputs}amix=inputs={n_inputs}:duration=longest:normalize=0[mixed_audio]") + + # 4. 截断到视频时长 + if video_duration > 0: + filter_parts.append(f"[mixed_audio]atrim=0:{video_duration:.3f}[final_audio]") + final_label = "final_audio" + else: + final_label = "mixed_audio" + + filter_complex = ";".join(filter_parts) + + command = [ + FFMPEG_BIN, + "-y", + *input_args, + "-filter_complex", + filter_complex, + "-map", + f"[{final_label}]", + "-acodec", + "aac", + "-b:a", + "128k", + str(output_path), + ] + + logger.info( + "音频混音: plan_id=%s main_clips=%d audio_clips=%d", + ctx.plan_id, + len(main_clips), + len(audio_clips), + ) + try: + run_ffmpeg(command) + except subprocess.CalledProcessError as e: + logger.error( + "音频混音失败: plan_id=%s exit_code=%d\nfilter_complex:\n%s", + ctx.plan_id, + e.returncode, + filter_complex[:3000], + ) + raise + + +def merge_audio_video( + ctx: RenderContext, + video_path: Path, + audio_path: Path, + output_path: Path, +) -> None: + """将音频合并到视频中（视频流拷贝，音频直接复用）. + + Args: + ctx: 渲染上下文 + video_path: 无声视频路径 + audio_path: 音频文件路径 + output_path: 输出文件路径 + """ + command = [ + FFMPEG_BIN, + "-y", + "-i", + str(video_path), + "-i", + str(audio_path), + "-c:v", + "copy", + "-c:a", + "aac", + "-b:a", + "128k", + "-map", + "0:v:0", + "-map", + "1:a:0", + "-shortest", + "-movflags", + "+faststart", + str(output_path), + ] + + logger.info("合并音视频: plan_id=%s", ctx.plan_id) + try: + run_ffmpeg(command) + except subprocess.CalledProcessError as e: + logger.error( + "合并音视频失败: plan_id=%s exit_code=%d", + ctx.plan_id, + e.returncode, + ) + raise diff --git a/apps/worker/video_processing/render_subtitles.py b/apps/worker/video_processing/render_subtitles.py new file mode 100644 index 000000000..0ace09a38 --- /dev/null +++ b/apps/worker/video_processing/render_subtitles.py @@ -0,0 +1,255 @@ +"""ASS 字幕生成模块 — 从 unified_render_service.py 拆分. + +职责： +- 将 title / subtitle 配置转换为 ASS 字幕文件 +- 提供样式计算（颜色、对齐、描边/阴影） +- 供 UnifiedRenderService._maybe_generate_ass 调用 +""" + +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Any + +logger = logging.getLogger(__name__) + + +# ── 常量 ────────────────────────────────────────────────────────────────────── + +# Title/Subtitle 默认边距（像素） +TITLE_MARGIN_TOP = 60 +TITLE_MARGIN_BOTTOM = 60 +TITLE_MARGIN_SIDE = 40 + + +# ── ASS 字幕工具 ───────────────────────────────────────────────────────────── + + +def _hex_to_ass_color(hex_color: str) -> str: + """将 HEX 颜色（#RRGGBB）转换为 ASS &HBBGGRR 格式。""" + hex_color = hex_color.lstrip("#") + if len(hex_color) != 6: + return "&H000000" + r, g, b = hex_color[0:2], hex_color[2:4], hex_color[4:6] + return f"&H{b.upper()}{g.upper()}{r.upper()}" + + +def _position_to_ass_alignment(position: str) -> int: + """将文字位置映射为 ASS \\an 对齐编号。 + + ASS 对齐编号（数字小键盘布局）： + 7 8 9 + 4 5 6 + 1 2 3 + """ + mapping = { + "top": 8, # 顶部居中 + "center": 5, # 居中 + "bottom": 2, # 底部居中 + } + return mapping.get(position, 8) + + +def _build_ass_style( + style_name: str, + *, + font_name: str = "思源黑体", + font_size: int = 48, + primary_color: str = "&H00FFFFFF", + outline_color: str = "&H00000000", + outline_width: float = 1.0, + shadow_blur: float = 0.0, + shadow_offset: tuple[int, int] = (0, 0), + bold: bool = False, + italic: bool = False, + alignment: int = 8, + margin_v: int = 60, + margin_l: int = 40, + margin_r: int = 40, +) -> str: + """构建 ASS Style 行。 + + Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, + Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, + BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding + """ + bold_val = -1 if bold else 0 + italic_val = -1 if italic else 0 + + # BackColour 用于阴影（BorderStyle=1 时 outline + shadow） + back_color = primary_color # 阴影颜色默认同文字色（带透明度由阴影模糊控制） + + # Shadow 值：ASS 中 Shadow 字段是阴影偏移距离（像素）， + # 我们用 shadow_offset[1] 作为纵向偏移，模糊由 BorderStyle=3 实现 + # 简化：BorderStyle=1（outline + drop shadow），Shadow 字段表示阴影深度 + shadow_depth = shadow_offset[1] if shadow_blur > 0 else 0 + + return ( + f"Style: {style_name},{font_name},{font_size},{primary_color}," + f"&H000000FF,{outline_color},{back_color}," + f"{bold_val},{italic_val},0,0,100,100,0,0," + f"1,{outline_width},{shadow_depth},{alignment}," + f"{margin_l},{margin_r},{margin_v},1" + ) + + +def _escape_ass_text(text: str) -> str: + r"""转义 ASS 文本中的特殊字符。 + + ASS 中换行用 \N（硬换行）或 \n（软换行）， + 大括号 {} 用于覆盖样式，需要转义。 + """ + # 将实际换行转为 ASS 硬换行 + text = text.replace("\r\n", "\\N").replace("\n", "\\N").replace("\r", "\\N") + # 转义大括号（ASS 用它做样式覆盖标签） + text = text.replace("{", "(").replace("}", ")") + return text + + +def _format_ass_time(seconds: float) -> str: + """将秒数格式化为 ASS 时间格式 H:MM:SS.cc。""" + hours = int(seconds // 3600) + minutes = int((seconds % 3600) // 60) + secs = seconds % 60 + return f"{hours}:{minutes:02d}:{secs:05.2f}" + + +def generate_ass_subtitles( + output_path: Path, + *, + video_width: int, + video_height: int, + video_duration: float, + title_text: str = "", + title_config: dict[str, Any] | None = None, + subtitle_text: str = "", + subtitle_config: dict[str, Any] | None = None, +) -> Path: + """生成 ASS 字幕文件。 + + 支持 Title（标题）和 Subtitle（字幕）两种字幕类型， + 各自可独立配置样式、位置和内容。 + + Args: + output_path: 输出 ASS 文件路径 + video_width: 视频宽度（用于 ASS PlayResX） + video_height: 视频高度（用于 ASS PlayResY） + video_duration: 视频总时长（秒），字幕显示整个时长 + title_text: 标题文本 + title_config: 标题样式配置（TitleConfig dict） + subtitle_text: 字幕文本 + subtitle_config: 字幕样式配置（SubtitleConfig dict） + + Returns: + 生成的 ASS 文件路径 + """ + title_config = title_config or {} + subtitle_config = subtitle_config or {} + + title_enabled = title_config.get("enabled", True) and bool(title_text.strip()) + subtitle_enabled = subtitle_config.get("enabled", True) and bool(subtitle_text.strip()) + + if not title_enabled and not subtitle_enabled: + # 没有字幕，生成空文件（仍返回路径，调用方自行判断是否使用） + output_path.write_text("", encoding="utf-8") + return output_path + + styles: list[str] = [] + events: list[str] = [] + + # ── Title 样式与事件 ────────────────────────────────────────────────── + if title_enabled: + title_color = _hex_to_ass_color(title_config.get("color", "#ffffff")) + title_stroke = title_config.get("stroke", {}) or {} + title_shadow = title_config.get("shadow", {}) or {} + stroke_color = _hex_to_ass_color(title_stroke.get("color", "#000000")) + stroke_width = float(title_stroke.get("width", 1)) if title_stroke.get("enabled", False) else 0.0 + shadow_blur = float(title_shadow.get("blur", 4)) if title_shadow.get("enabled", False) else 0.0 + shadow_offset = ( + title_shadow.get("offset_x", 2) if title_shadow.get("enabled", False) else 0, + title_shadow.get("offset_y", 2) if title_shadow.get("enabled", False) else 0, + ) + + title_alignment = _position_to_ass_alignment(title_config.get("position", "top")) + + styles.append( + _build_ass_style( + "TitleStyle", + font_name=title_config.get("font", "思源黑体"), + font_size=int(title_config.get("size", 48)), + primary_color=title_color, + outline_color=stroke_color, + outline_width=stroke_width, + shadow_blur=shadow_blur, + shadow_offset=shadow_offset, + bold=bool(title_config.get("bold", True)), + italic=bool(title_config.get("italic", False)), + alignment=title_alignment, + margin_v=TITLE_MARGIN_TOP, + margin_l=TITLE_MARGIN_SIDE, + margin_r=TITLE_MARGIN_SIDE, + ) + ) + + # 转义 ASS 特殊字符 + safe_title_text = _escape_ass_text(title_text) + + events.append( + "Dialogue: 0,0:00:00.00," f"{_format_ass_time(video_duration)}," "TitleStyle,,0,0,0,," f"{safe_title_text}" + ) + + # ── Subtitle 样式与事件 ─────────────────────────────────────────────── + if subtitle_enabled: + sub_color = _hex_to_ass_color(subtitle_config.get("color", "#ffffff")) + sub_alignment = _position_to_ass_alignment(subtitle_config.get("position", "bottom")) + + styles.append( + _build_ass_style( + "SubtitleStyle", + font_name=subtitle_config.get("font", "思源黑体"), + font_size=int(subtitle_config.get("size", 24)), + primary_color=sub_color, + outline_color="&H00000000", + outline_width=1.0, + shadow_blur=0.0, + shadow_offset=(0, 0), + bold=False, + italic=False, + alignment=sub_alignment, + margin_v=TITLE_MARGIN_BOTTOM, + margin_l=TITLE_MARGIN_SIDE, + margin_r=TITLE_MARGIN_SIDE, + ) + ) + + safe_subtitle_text = _escape_ass_text(subtitle_text) + + events.append( + "Dialogue: 0,0:00:00.00," + f"{_format_ass_time(video_duration)}," + "SubtitleStyle,,0,0,0,," + f"{safe_subtitle_text}" + ) + + # ── 组装 ASS 文件 ───────────────────────────────────────────────────── + ass_content = f"""[Script Info] +ScriptType: v4.00+ +PlayResX: {video_width} +PlayResY: {video_height} +ScaledBorderAndShadow: yes +WrapStyle: 2 +Encoding: UTF-8 + +[V4+ Styles] +Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding # noqa: E501 +{chr(10).join(styles)} + +[Events] +Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text +{chr(10).join(events)} +""" + + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(ass_content, encoding="utf-8") + return output_path diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index 63e717d81..8f43e0833 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -39,18 +39,12 @@ from video_processing.ffmpeg_utils import ( probe_video_info, run_ffmpeg, ) +from video_processing.render_audio import RenderContext, merge_audio_video, mix_audio +from video_processing.render_subtitles import generate_ass_subtitles logger = logging.getLogger(__name__) -# ── 常量 ────────────────────────────────────────────────────────────────────── - -# Title/Subtitle 默认边距（像素） -TITLE_MARGIN_TOP = 60 -TITLE_MARGIN_BOTTOM = 60 -TITLE_MARGIN_SIDE = 40 - - # ── 数据结构 ────────────────────────────────────────────────────────────────── @@ -97,238 +91,6 @@ class RenderResult: # ── clip_type → layer role 映射 ────────────────────────────────────────────── -# ── ASS 字幕工具 ───────────────────────────────────────────────────────────── - - -def _hex_to_ass_color(hex_color: str) -> str: - """将 HEX 颜色（#RRGGBB）转换为 ASS &HBBGGRR 格式。""" - hex_color = hex_color.lstrip("#") - if len(hex_color) != 6: - return "&H000000" - r, g, b = hex_color[0:2], hex_color[2:4], hex_color[4:6] - return f"&H{b.upper()}{g.upper()}{r.upper()}" - - -def _position_to_ass_alignment(position: str) -> int: - """将文字位置映射为 ASS \an 对齐编号。 - - ASS 对齐编号（数字小键盘布局）： - 7 8 9 - 4 5 6 - 1 2 3 - """ - mapping = { - "top": 8, # 顶部居中 - "center": 5, # 居中 - "bottom": 2, # 底部居中 - } - return mapping.get(position, 8) - - -def _build_ass_style( - style_name: str, - *, - font_name: str = "思源黑体", - font_size: int = 48, - primary_color: str = "&H00FFFFFF", - outline_color: str = "&H00000000", - outline_width: float = 1.0, - shadow_blur: float = 0.0, - shadow_offset: tuple[int, int] = (0, 0), - bold: bool = False, - italic: bool = False, - alignment: int = 8, - margin_v: int = 60, - margin_l: int = 40, - margin_r: int = 40, -) -> str: - """构建 ASS Style 行。 - - Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, - Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, - BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding - """ - bold_val = -1 if bold else 0 - italic_val = -1 if italic else 0 - - # BackColour 用于阴影（BorderStyle=1 时 outline + shadow） - back_color = primary_color # 阴影颜色默认同文字色（带透明度由阴影模糊控制） - - # Shadow 值：ASS 中 Shadow 字段是阴影偏移距离（像素）， - # 我们用 shadow_offset[1] 作为纵向偏移，模糊由 BorderStyle=3 实现 - # 简化：BorderStyle=1（outline + drop shadow），Shadow 字段表示阴影深度 - shadow_depth = shadow_offset[1] if shadow_blur > 0 else 0 - - return ( - f"Style: {style_name},{font_name},{font_size},{primary_color}," - f"&H000000FF,{outline_color},{back_color}," - f"{bold_val},{italic_val},0,0,100,100,0,0," - f"1,{outline_width},{shadow_depth},{alignment}," - f"{margin_l},{margin_r},{margin_v},1" - ) - - -def generate_ass_subtitles( - output_path: Path, - *, - video_width: int, - video_height: int, - video_duration: float, - title_text: str = "", - title_config: dict[str, Any] | None = None, - subtitle_text: str = "", - subtitle_config: dict[str, Any] | None = None, -) -> Path: - """生成 ASS 字幕文件。 - - 支持 Title（标题）和 Subtitle（字幕）两种字幕类型， - 各自可独立配置样式、位置和内容。 - - Args: - output_path: 输出 ASS 文件路径 - video_width: 视频宽度（用于 ASS PlayResX） - video_height: 视频高度（用于 ASS PlayResY） - video_duration: 视频总时长（秒），字幕显示整个时长 - title_text: 标题文本 - title_config: 标题样式配置（TitleConfig dict） - subtitle_text: 字幕文本 - subtitle_config: 字幕样式配置（SubtitleConfig dict） - - Returns: - 生成的 ASS 文件路径 - """ - title_config = title_config or {} - subtitle_config = subtitle_config or {} - - title_enabled = title_config.get("enabled", True) and bool(title_text.strip()) - subtitle_enabled = subtitle_config.get("enabled", True) and bool(subtitle_text.strip()) - - if not title_enabled and not subtitle_enabled: - # 没有字幕，生成空文件（仍返回路径，调用方自行判断是否使用） - output_path.write_text("", encoding="utf-8") - return output_path - - styles: list[str] = [] - events: list[str] = [] - - # ── Title 样式与事件 ────────────────────────────────────────────────── - if title_enabled: - title_color = _hex_to_ass_color(title_config.get("color", "#ffffff")) - title_stroke = title_config.get("stroke", {}) or {} - title_shadow = title_config.get("shadow", {}) or {} - stroke_color = _hex_to_ass_color(title_stroke.get("color", "#000000")) - stroke_width = float(title_stroke.get("width", 1)) if title_stroke.get("enabled", False) else 0.0 - shadow_blur = float(title_shadow.get("blur", 4)) if title_shadow.get("enabled", False) else 0.0 - shadow_offset = ( - title_shadow.get("offset_x", 2) if title_shadow.get("enabled", False) else 0, - title_shadow.get("offset_y", 2) if title_shadow.get("enabled", False) else 0, - ) - - title_alignment = _position_to_ass_alignment(title_config.get("position", "top")) - - styles.append( - _build_ass_style( - "TitleStyle", - font_name=title_config.get("font", "思源黑体"), - font_size=int(title_config.get("size", 48)), - primary_color=title_color, - outline_color=stroke_color, - outline_width=stroke_width, - shadow_blur=shadow_blur, - shadow_offset=shadow_offset, - bold=bool(title_config.get("bold", True)), - italic=bool(title_config.get("italic", False)), - alignment=title_alignment, - margin_v=TITLE_MARGIN_TOP, - margin_l=TITLE_MARGIN_SIDE, - margin_r=TITLE_MARGIN_SIDE, - ) - ) - - # 转义 ASS 特殊字符 - safe_title_text = _escape_ass_text(title_text) - - events.append( - "Dialogue: 0,0:00:00.00," f"{_format_ass_time(video_duration)}," "TitleStyle,,0,0,0,," f"{safe_title_text}" - ) - - # ── Subtitle 样式与事件 ─────────────────────────────────────────────── - if subtitle_enabled: - sub_color = _hex_to_ass_color(subtitle_config.get("color", "#ffffff")) - sub_alignment = _position_to_ass_alignment(subtitle_config.get("position", "bottom")) - - styles.append( - _build_ass_style( - "SubtitleStyle", - font_name=subtitle_config.get("font", "思源黑体"), - font_size=int(subtitle_config.get("size", 24)), - primary_color=sub_color, - outline_color="&H00000000", - outline_width=1.0, - shadow_blur=0.0, - shadow_offset=(0, 0), - bold=False, - italic=False, - alignment=sub_alignment, - margin_v=TITLE_MARGIN_BOTTOM, - margin_l=TITLE_MARGIN_SIDE, - margin_r=TITLE_MARGIN_SIDE, - ) - ) - - safe_subtitle_text = _escape_ass_text(subtitle_text) - - events.append( - "Dialogue: 0,0:00:00.00," - f"{_format_ass_time(video_duration)}," - "SubtitleStyle,,0,0,0,," - f"{safe_subtitle_text}" - ) - - # ── 组装 ASS 文件 ───────────────────────────────────────────────────── - ass_content = f"""[Script Info] -ScriptType: v4.00+ -PlayResX: {video_width} -PlayResY: {video_height} -ScaledBorderAndShadow: yes -WrapStyle: 2 -Encoding: UTF-8 - -[V4+ Styles] -Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding # noqa: E501 -{chr(10).join(styles)} - -[Events] -Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text -{chr(10).join(events)} -""" - - output_path.parent.mkdir(parents=True, exist_ok=True) - output_path.write_text(ass_content, encoding="utf-8") - return output_path - - -def _escape_ass_text(text: str) -> str: - r"""转义 ASS 文本中的特殊字符。 - - ASS 中换行用 \N（硬换行）或 \n（软换行）， - 大括号 {} 用于覆盖样式，需要转义。 - """ - # 将实际换行转为 ASS 硬换行 - text = text.replace("\r\n", "\\N").replace("\n", "\\N").replace("\r", "\\N") - # 转义大括号（ASS 用它做样式覆盖标签） - text = text.replace("{", "(").replace("}", ")") - return text - - -def _format_ass_time(seconds: float) -> str: - """将秒数格式化为 ASS 时间格式 H:MM:SS.cc。""" - hours = int(seconds // 3600) - minutes = int((seconds % 3600) // 60) - secs = seconds % 60 - return f"{hours}:{minutes:02d}:{secs:05.2f}" - - def _resolve_layer_role(clip_type: str, config: dict[str, Any]) -> str: """根据 clip_type 和 config.role 确定图层角色。 @@ -501,7 +263,8 @@ class UnifiedRenderService: # 直通场景已在一次调用中完成视频+音频 has_audio = pass_through_has_audio else: - audio_path = self._mix_audio(layers, video_duration) + ctx = RenderContext(work_dir=self.work_dir, plan_id=self.plan.id) + audio_path = mix_audio(ctx, layers, video_duration) t_audio_end = time.time() audio_mix_ms = int((t_audio_end - t_audio_start) * 1000) has_audio = audio_path is not None @@ -512,7 +275,7 @@ class UnifiedRenderService: audio_mix_ms, ) # 7. 合并音视频 - self._merge_audio_video(video_only_path, audio_path, output_path) + merge_audio_video(ctx, video_only_path, audio_path, output_path) else: # 无音频，直接用无声视频 import shutil @@ -798,8 +561,12 @@ class UnifiedRenderService: if output_path.exists(): try: output_path.unlink() - except OSError: - pass + except OSError as unlink_err: + logger.warning( + "[unified-render] 损坏输出文件清理失败: path=%s error=%s", + output_path, + unlink_err, + ) return False def _render_pass_through( @@ -1208,286 +975,6 @@ class UnifiedRenderService: info["height"], ) - # ── 音频后处理 ──────────────────────────────────────────────────────── - - def _mix_audio(self, layers: list[RenderLayer], video_duration: float) -> Path | None: - """音频后处理混音. - - 处理逻辑： - 1. 主音频源按优先级查找：main > broll（background 不参与主音频，通常是图片无音轨） - 2. 主图层音频按顺序 concat 拼接 - 3. 独立音频轨（audio role）用 amix 混入 - 4. 输出时长截断到 video_duration - 5. 无音频流的 clip 会被自动跳过，避免 FFmpeg 引用 [i:a] 失败 - - Args: - layers: 图层列表 - video_duration: 视频总时长（用于截断音频） - - Returns: - 混音后的音频文件路径，无音频时返回 None - """ - # 按优先级精确查找主音频图层：main > broll - # background 不参与主音频（通常是静态图片，无音轨） - layer_map = {layer.role: layer for layer in layers} - main_layer = None - for role in ("main", "broll"): - if role in layer_map and layer_map[role].clips: - main_layer = layer_map[role] - break - - main_clips: list[ResolvedClip] = main_layer.clips if main_layer else [] - - # 没有主视频图层时兜底：检查 overlay/corner_voice 层是否有带音频的素材 - if not main_clips: - for role in ("overlay", "corner_voice"): - if role in layer_map and layer_map[role].clips: - main_clips = layer_map[role].clips - break - - # 收集独立音频轨 - audio_clips: list[ResolvedClip] = [] - if "audio" in layer_map: - audio_clips = layer_map["audio"].clips - - # ── 防御：过滤掉无音频流的 clip ── - # 源视频可能没有音频流（如静音视频、纯图片转的视频），直接引用 [i:a] 会导致 FFmpeg 失败 - main_clips = [c for c in main_clips if self._clip_has_audio(c)] - audio_clips = [c for c in audio_clips if self._clip_has_audio(c)] - - if not main_clips and not audio_clips: - return None - - # 构建音频处理命令 - output_path = self.work_dir / f"audio_{self.plan.id}.aac" - - # 简单场景：只有主图层 + 无独立音频 → 直接从视频提取音频并拼接 - if main_clips and not audio_clips: - self._concat_main_audio(main_clips, output_path, video_duration) - return output_path - - # 有独立音频轨 → amix 混音 - self._mix_with_independent_audio(main_clips, audio_clips, output_path, video_duration) - return output_path - - def _concat_main_audio(self, clips: list[ResolvedClip], output_path: Path, video_duration: float) -> None: - """主图层音频 concat 拼接（对齐链路A行为）. - - 每个 clip 提取音频 → trim → 按顺序 concat。 - """ - if len(clips) == 1: - # 单 clip，直接提取音频，截断到 min(clip有效时长, 视频总时长) - clip = clips[0] - effective_duration = self._clip_effective_duration(clip) - # 最终时长：取 clip 有效时长和视频总时长的较小值 - # （视频总时长由主图层决定，但单 clip 场景下两者应该一致，仍做保护） - final_duration = effective_duration - if video_duration > 0 and (final_duration <= 0 or final_duration > video_duration): - final_duration = video_duration - - command = [ - FFMPEG_BIN, - "-y", - "-i", - str(clip.local_path), - "-vn", - "-acodec", - "aac", - "-b:a", - "128k", - ] - if final_duration > 0: - command.extend(["-t", f"{final_duration:.3f}"]) - command.append(str(output_path)) - run_ffmpeg(command) - return - - # 多 clip，用 filter_complex concat - input_args: list[str] = [] - filter_parts: list[str] = [] - - for i, clip in enumerate(clips): - input_args.extend(["-i", str(clip.local_path)]) - effective_duration = self._clip_effective_duration(clip) - if effective_duration > 0: - filter_parts.append(f"[{i}:a]atrim=0:{effective_duration:.3f},asetpts=PTS-STARTPTS[a{i}]") - else: - filter_parts.append(f"[{i}:a]asetpts=PTS-STARTPTS[a{i}]") - - audio_labels = "".join(f"[a{i}]" for i in range(len(clips))) - filter_parts.append(f"{audio_labels}concat=n={len(clips)}:v=0:a=1[outa]") - - # 截断到视频总时长 - if video_duration > 0: - filter_parts.append(f"[outa]atrim=0:{video_duration:.3f}[final_audio]") - final_label = "final_audio" - else: - final_label = "outa" - - filter_complex = ";".join(filter_parts) - - command = [ - FFMPEG_BIN, - "-y", - *input_args, - "-filter_complex", - filter_complex, - "-map", - f"[{final_label}]", - "-acodec", - "aac", - "-b:a", - "128k", - str(output_path), - ] - run_ffmpeg(command) - - def _mix_with_independent_audio( - self, - main_clips: list[ResolvedClip], - audio_clips: list[ResolvedClip], - output_path: Path, - video_duration: float, - ) -> None: - """主音频 + 独立音频轨 amix 混音. - - Args: - main_clips: 主视频 clips（提取音频后 concat） - audio_clips: 独立音频轨 clips - output_path: 输出路径 - video_duration: 视频总时长 - """ - input_args: list[str] = [] - filter_parts: list[str] = [] - mix_labels: list[str] = [] - - input_idx = 0 - - # 1. 主图层音频 concat - if main_clips: - for clip in main_clips: - input_args.extend(["-i", str(clip.local_path)]) - effective_duration = self._clip_effective_duration(clip) - if effective_duration > 0: - filter_parts.append( - f"[{input_idx}:a]atrim=0:{effective_duration:.3f},asetpts=PTS-STARTPTS[ma{input_idx}]" - ) - else: - filter_parts.append(f"[{input_idx}:a]asetpts=PTS-STARTPTS[ma{input_idx}]") - input_idx += 1 - - if len(main_clips) == 1: - mix_labels.append("ma0") - else: - main_labels = "".join(f"[ma{i}]" for i in range(len(main_clips))) - filter_parts.append(f"{main_labels}concat=n={len(main_clips)}:v=0:a=1[main_audio]") - mix_labels.append("main_audio") - - # 2. 独立音频轨 - for j, clip in enumerate(audio_clips): - input_args.extend(["-i", str(clip.local_path)]) - effective_duration = self._clip_effective_duration(clip) - volume = clip.config.get("volume", 1.0) if clip.config else 1.0 - label = f"ia{j}" - filters = [] - if effective_duration > 0: - filters.append(f"atrim=0:{effective_duration:.3f}") - filters.append("asetpts=PTS-STARTPTS") - if volume != 1.0: - filters.append(f"volume={volume}") - filter_parts.append(f"[{input_idx}:a]{','.join(filters)}[{label}]") - mix_labels.append(label) - input_idx += 1 - - # 3. amix 混音 - mix_inputs = "".join(f"[{label}]" for label in mix_labels) - n_inputs = len(mix_labels) - # normalized=0 保持音量，duration=shortest 取最短 - filter_parts.append(f"{mix_inputs}amix=inputs={n_inputs}:duration=longest:normalize=0[mixed_audio]") - - # 4. 截断到视频时长 - if video_duration > 0: - filter_parts.append(f"[mixed_audio]atrim=0:{video_duration:.3f}[final_audio]") - final_label = "final_audio" - else: - final_label = "mixed_audio" - - filter_complex = ";".join(filter_parts) - - command = [ - FFMPEG_BIN, - "-y", - *input_args, - "-filter_complex", - filter_complex, - "-map", - f"[{final_label}]", - "-acodec", - "aac", - "-b:a", - "128k", - str(output_path), - ] - - logger.info( - "音频混音: plan_id=%s main_clips=%d audio_clips=%d", - self.plan.id, - len(main_clips), - len(audio_clips), - ) - try: - run_ffmpeg(command) - except subprocess.CalledProcessError as e: - logger.error( - "音频混音失败: plan_id=%s exit_code=%d\nfilter_complex:\n%s", - self.plan.id, - e.returncode, - filter_complex[:3000], - ) - raise - - def _merge_audio_video(self, video_path: Path, audio_path: Path, output_path: Path) -> None: - """将音频合并到视频中（视频流拷贝，音频直接复用）. - - Args: - video_path: 无声视频路径 - audio_path: 音频文件路径 - output_path: 输出文件路径 - """ - command = [ - FFMPEG_BIN, - "-y", - "-i", - str(video_path), - "-i", - str(audio_path), - "-c:v", - "copy", - "-c:a", - "aac", - "-b:a", - "128k", - "-map", - "0:v:0", - "-map", - "1:a:0", - "-shortest", - "-movflags", - "+faststart", - str(output_path), - ] - - logger.info("合并音视频: plan_id=%s", self.plan.id) - try: - run_ffmpeg(command) - except subprocess.CalledProcessError as e: - logger.error( - "合并音视频失败: plan_id=%s exit_code=%d", - self.plan.id, - e.returncode, - ) - raise - @staticmethod def _clip_effective_duration(clip: ResolvedClip) -> float: """计算 clip 的有效时长.""" @@ -1495,16 +982,3 @@ class UnifiedRenderService: return min(clip.duration, clip.actual_duration) if clip.actual_duration > 0 else clip.duration return clip.actual_duration if clip.actual_duration > 0 else 0.0 - def _clip_has_audio(self, clip: ResolvedClip) -> bool: - """探测 clip 是否有音频流（带缓存）. - - 避免同一个 clip 被多次 ffprobe 探测。 - """ - if not hasattr(self, "_audio_cache"): - self._audio_cache: dict[str, bool] = {} - key = str(clip.local_path) - if key not in self._audio_cache: - from .ffmpeg_utils import probe_has_audio - - self._audio_cache[key] = probe_has_audio(clip.local_path) - return self._audio_cache[key] diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index 2e145eeed..443ddc464 100644 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -713,6 +713,236 @@ def _render_with_legacy_engine( return duration, file_size +# ── generate_video 阶段子函数 ───────────────────────────────────────────────── + + +def _load_task_info(task_id: str) -> dict | None: + """从数据库加载 GenerationTask 元数据。 + + Returns: + 包含任务元数据的字典，任务不存在时返回 None。 + """ + from packages.adapters.sqlalchemy_impl.generation_task_repository import ( + SQLAlchemyGenerationTaskRepository, + ) + + session = SessionLocal() + try: + task_repo = SQLAlchemyGenerationTaskRepository(session) + gen_task = task_repo.get(task_id) + if gen_task is None: + return None + + return { + "project_id": gen_task.project_id, + "asset_library_id": gen_task.asset_library_id, + "voice_library_id": gen_task.voice_library_id or "", + "template_id": getattr(gen_task, "template_id", "") or "", + "mode": gen_task.strategy_id or "one_take", + "task_asset_ids": list(gen_task.asset_ids or []), + "batch_id": getattr(gen_task, "batch_id", "") or "", + "user_id": getattr(gen_task, "created_by_user_id", "") or "", + } + finally: + session.close() + + +def _download_all_assets( + temp_path: Path, + asset_library_id: str, + project_id: str, + task_asset_ids: list[str], + voice_library_id: str, + task_id: str, +) -> tuple[list[Path], str | None]: + """下载视频素材和配音素材。 + + Returns: + (downloaded_videos, audio_path) + + Note: gen_task 不传入下载函数（session 已关闭）， + 主函数在下载前后已有汇总日志。 + """ + logger.info("[task_id=%s] [下载素材] 开始下载视频素材", task_id) + download_start = time.monotonic() + downloaded_videos = _download_library_assets( + temp_path, + asset_library_id=asset_library_id, + project_id=project_id, + asset_ids=task_asset_ids or None, + task_id=task_id, + ) + download_elapsed = time.monotonic() - download_start + logger.info( + "[task_id=%s] [下载素材] 完成: 成功=%d个, 耗时=%.1fs", + task_id, + len(downloaded_videos), + download_elapsed, + ) + + 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) + logger.info("[task_id=%s] [下载配音] 配音下载成功", task_id) + + return downloaded_videos, audio_path + + +def _render_video( + task_id: str, + downloaded_videos: list[Path], + voice_path: str | None, + editing_mode, + project_id: str, + template_id: str, + user_id: str, + temp_path: Path, + output_name: str, +) -> tuple[Path, float]: + """渲染视频（含配音混音）。 + + Returns: + (output_path, render_duration) + """ + if not downloaded_videos: + raise RuntimeError(f"素材下载结果为空: task_id={task_id}") + + # 构建虚拟 plan + clips + 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, + ) + + total_duration = sum(c.duration for c in virtual_clips) + logger.info( + "[task_id=%s] [剪辑计划] 片段数=%d, 总时长=%.1fs", + task_id, + len(virtual_clips), + total_duration, + ) + + # 选择渲染引擎 + engine = _resolve_render_engine(user_id) if user_id else ENGINE_UNIFIED + logger.info("[task_id=%s] [渲染] 引擎选择: %s (user_id=%s)", task_id, engine, user_id) + + render_start = time.monotonic() + render_output_path = temp_path / f"rendered-{task_id}.mp4" + + if engine == ENGINE_LEGACY: + render_duration, _ = _render_with_legacy_engine( + task_id=task_id, + virtual_clips=virtual_clips, + asset_path_map=asset_path_map, + work_dir=temp_path, + output_path=render_output_path, + ) + else: + logger.info("[task_id=%s] [渲染] unified 引擎 FFmpeg 渲染开始", task_id) + 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() + render_output_path = render_result.output_path + render_duration = render_result.duration + + render_elapsed = time.monotonic() - render_start + logger.info( + "[task_id=%s] [渲染] %s 引擎完成: 耗时=%.1fs, 时长=%.2fs", + task_id, + engine, + render_elapsed, + render_duration, + ) + + # 配音混音 + if voice_path: + final_path = temp_path / f"final-{task_id}.mp4" + try: + _mux_audio_track(render_output_path, voice_path, final_path) + output_path = final_path + except Exception as mux_err: + logger.warning("[task_id=%s] [混音] 音频混合失败，使用无音频版本: %s", task_id, mux_err) + output_path = render_output_path + else: + output_path = render_output_path + + return output_path, render_duration + + +def _upload_and_record( + task_id: str, + output_path: Path, + project_id: str, + batch_id: str, + editing_mode, +) -> tuple[str, float, int, int]: + """上传 OSS、创建视频记录并查重。 + + Returns: + (file_url, duration, file_size, video_count) + """ + storage_key = f"generated/projects/{project_id}/tasks/{task_id}/{output_path.name}" + file_size = output_path.stat().st_size + + # 上传 OSS + logger.info("[task_id=%s] [OSS上传] 开始上传: size=%d", task_id, file_size) + upload_start = time.monotonic() + file_url = upload_to_oss(output_path, storage_key) + upload_elapsed = time.monotonic() - upload_start + if not file_url: + raise RuntimeError(f"OSS 上传失败: task_id={task_id}, storage_key={storage_key}") + + # 校验 URL 可达性（P0-2: 私有 bucket 用预签名 + object_exists 降级） + verify_url = get_signed_download_url(file_url, expires_seconds=300) or file_url + if not _verify_url_accessible(verify_url): + from video_processing.oss_helpers import normalize_storage_key, oss_bucket + + bucket = oss_bucket() + key = normalize_storage_key(file_url) + if not (bucket and bucket.object_exists(key)): + raise RuntimeError( + f"OSS 上传后 URL 不可访问且 object_exists 失败: file_url={file_url}, " + f"storage_key={storage_key}" + ) + logger.info("URL 校验失败但 object_exists 确认文件存在，视为上传成功: storage_key=%s", key) + + logger.info( + "[task_id=%s] [OSS上传] 成功: 耗时=%.1fs, file_url=%s", + task_id, + upload_elapsed, + file_url, + ) + + # 创建 GeneratedVideo 记录 + 查重 + duration = probe_duration(output_path) + 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() + + return file_url, duration, file_size, video_count or 1 + + # ── Celery Task ────────────────────────────────────────────────────────────── @@ -735,293 +965,126 @@ def generate_video(self, task_id: str) -> dict: Returns: 生成结果字典 """ + from packages.adapters.sqlalchemy_impl.generation_task_repository import ( + SQLAlchemyGenerationTaskRepository, + ) from packages.domain import EditingMode logger.info("[task_id=%s] [接收任务] 开始生成视频任务", task_id) - # 从数据库加载任务信息 - session = SessionLocal() + # ── 1. 加载任务信息 ────────────────────────────────────────────────────── + task_info = _load_task_info(task_id) + if task_info is None: + logger.error("[task_id=%s] [接收任务] 任务不存在", task_id) + return {"status": "failed", "error": f"generation task {task_id} not found"} + + project_id = task_info["project_id"] + asset_library_id = task_info["asset_library_id"] + voice_library_id = task_info["voice_library_id"] + template_id = task_info["template_id"] + task_asset_ids = task_info["task_asset_ids"] + batch_id = task_info["batch_id"] + user_id = task_info["user_id"] + + # 加载 gen_task（用于全程进度日志；_flush_logs 使用独立 session 持久化） + _session = SessionLocal() try: - from packages.adapters.sqlalchemy_impl.generation_task_repository import ( - SQLAlchemyGenerationTaskRepository, - ) + _repo = SQLAlchemyGenerationTaskRepository(_session) + gen_task = _repo.get(task_id) + finally: + _session.close() - task_repo = SQLAlchemyGenerationTaskRepository(session) - gen_task = task_repo.get(task_id) - if gen_task is None: - logger.error("[task_id=%s] [接收任务] 任务不存在", task_id) - return {"status": "failed", "error": f"generation task {task_id} not found"} - project_id = gen_task.project_id - asset_library_id = gen_task.asset_library_id - voice_library_id = gen_task.voice_library_id or "" - template_id = getattr(gen_task, "template_id", "") or "" - mode = gen_task.strategy_id or "one_take" - task_asset_ids = list(gen_task.asset_ids or []) - batch_id = getattr(gen_task, "batch_id", "") or "" - - # 记录接收任务日志 + # 记录接收任务日志 + if gen_task: gen_task.append_log( "接收任务", - f"模式={mode}, 模板={template_id}, 素材数={len(task_asset_ids)}", - mode=mode, + f"模式={task_info['mode']}, 模板={template_id}, 素材数={len(task_asset_ids)}", + mode=task_info["mode"], template_id=template_id, asset_count=len(task_asset_ids), ) _flush_logs(task_id, gen_task) - finally: - session.close() # 标记任务为 running _update_task_status(task_id, "mark_processing") try: - editing_mode = EditingMode(mode) + editing_mode = EditingMode(mode) if (mode := task_info["mode"]) else EditingMode.ONE_TAKE except ValueError: editing_mode = EditingMode.ONE_TAKE output_name = f"generated-{task_id}.mp4" - storage_key = f"generated/projects/{project_id}/tasks/{task_id}/{output_name}" try: - # P1: template_id 存在性校验 if template_id: _validate_template_exists(template_id) - # P1: asset_ids 归属校验 — 已合并到 _download_library_assets 同一 session（P3-2） - with tempfile.TemporaryDirectory(prefix="xiaoxia-generation-") as temp_dir: temp_path = Path(temp_dir) - output_path = temp_path / output_name - # 1. 从素材库/项目下载视频素材 - logger.info("[task_id=%s] [下载素材] 开始下载视频素材", task_id) - download_start = time.monotonic() - downloaded_videos = _download_library_assets( + # ── 2. 下载素材 ────────────────────────────────────────────────── + downloaded_videos, audio_path = _download_all_assets( temp_path, asset_library_id=asset_library_id, project_id=project_id, - asset_ids=task_asset_ids or None, + task_asset_ids=task_asset_ids, + voice_library_id=voice_library_id, task_id=task_id, - gen_task=gen_task, ) - download_elapsed = time.monotonic() - download_start - logger.info( - "[task_id=%s] [下载素材] 完成: 成功=%d个, 耗时=%.1fs", - task_id, - len(downloaded_videos), - download_elapsed, - ) - - # 重新加载 gen_task 以追加日志（session 已关闭） - _session = SessionLocal() - try: - _repo = SQLAlchemyGenerationTaskRepository(_session) - gen_task = _repo.get(task_id) - finally: - _session.close() if gen_task: gen_task.append_log( "下载素材", f"成功下载 {len(downloaded_videos)} 个视频素材", count=len(downloaded_videos), - duration=round(download_elapsed, 2), ) _flush_logs(task_id, gen_task) - # 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) - logger.info("[task_id=%s] [下载配音] 配音下载成功", task_id) - - # 3. 渲染 - if not downloaded_videos: - # 素材下载为空（不应到达此处，_download_library_assets 已做校验） - raise RuntimeError( - f"素材下载结果为空: task_id={task_id}, " - f"asset_library_id={asset_library_id}, project_id={project_id}, " - f"asset_ids={task_asset_ids}" - ) - - # 构建虚拟 plan + clips + asset_path_map - virtual_plan, virtual_clips, asset_path_map = _build_plan_and_clips_from_task( + # ── 3. 渲染 + 混音 ─────────────────────────────────────────────── + output_path, render_duration = _render_video( task_id=task_id, - downloaded_paths=downloaded_videos, - mode=editing_mode.value, - ) - - total_duration = sum(c.duration for c in virtual_clips) - logger.info( - "[task_id=%s] [剪辑计划] 片段数=%d, 总时长=%.1fs", - task_id, - len(virtual_clips), - total_duration, + downloaded_videos=downloaded_videos, + voice_path=audio_path, + editing_mode=editing_mode, + project_id=project_id, + template_id=template_id, + user_id=user_id, + temp_path=temp_path, + output_name=output_name, ) if gen_task: - gen_task.append_log( - "剪辑计划", - f"片段数={len(virtual_clips)}, 总时长={total_duration:.1f}s", - segment_count=len(virtual_clips), - total_duration=round(total_duration, 2), - ) + gen_task.append_log("渲染", f"渲染完成, 时长={render_duration:.1f}s") _flush_logs(task_id, gen_task) - # 3. 根据 Feature Flag 选择渲染引擎 - user_id = getattr(gen_task, "created_by_user_id", "") if gen_task else "" - engine = _resolve_render_engine(user_id) if user_id else ENGINE_UNIFIED - logger.info("[task_id=%s] [渲染] 引擎选择: %s (user_id=%s)", task_id, engine, user_id) - - render_start = time.monotonic() - render_output_path = temp_path / f"rendered-{task_id}.mp4" - - if engine == ENGINE_LEGACY: - # 旧引擎：filter_complex + concat（保持原帧率，无 fps 归一化） - render_duration, render_file_size = _render_with_legacy_engine( - task_id=task_id, - virtual_clips=virtual_clips, - asset_path_map=asset_path_map, - work_dir=temp_path, - output_path=render_output_path, - ) - render_elapsed = time.monotonic() - render_start - logger.info( - "[task_id=%s] [渲染] legacy 引擎完成: 耗时=%.1fs, 时长=%.2fs", - task_id, - render_elapsed, - render_duration, - ) - else: - # 新引擎：UnifiedRenderService 图层架构 - logger.info("[task_id=%s] [渲染] unified 引擎 FFmpeg 渲染开始", task_id) - 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() - render_output_path = render_result.output_path - render_duration = render_result.duration - render_elapsed = time.monotonic() - render_start - logger.info( - "[task_id=%s] [渲染] unified 引擎完成: 耗时=%.1fs", - task_id, - render_elapsed, - ) - - if gen_task: - gen_task.append_log( - "渲染", - f"引擎={engine}, 耗时={render_elapsed:.1f}s", - duration=round(render_elapsed, 2), - engine=engine, - ) - _flush_logs(task_id, gen_task) - - # 4. 如有配音，后处理混音 - if audio_path: - final_path = temp_path / f"final-{task_id}.mp4" - try: - _mux_audio_track(render_output_path, audio_path, final_path) - # 混音成功，使用混音后的文件 - output_path = final_path - except Exception as mux_err: - logger.warning("[task_id=%s] [混音] 音频混合失败，使用无音频版本: %s", task_id, mux_err) - output_path = render_output_path - else: - output_path = render_output_path - - file_size = output_path.stat().st_size - duration = probe_duration(output_path) - - # 5. 上传到 OSS — 失败必须抛异常，不能静默忽略 - logger.info("[task_id=%s] [OSS上传] 开始上传: size=%d", task_id, file_size) - upload_start = time.monotonic() - file_url = upload_to_oss(output_path, storage_key) - upload_elapsed = time.monotonic() - upload_start - if not file_url: - # OSS 未配置或上传失败 - if gen_task: - gen_task.append_log("OSS上传", "上传失败", level="ERROR") - _flush_logs(task_id, gen_task) - raise RuntimeError( - f"OSS 上传失败: task_id={task_id}, storage_key={storage_key}, " f"output_path={output_path}" - ) - - # P0-2 修复：私有 bucket 下裸 URL 永远 403，改用预签名 URL 校验 - # 先用预签名 URL 校验，失败则降级为检查文件是否存在（object_exists） - verify_url = get_signed_download_url(file_url, expires_seconds=300) or file_url - if not _verify_url_accessible(verify_url): - # 预签名 URL 也访问失败时，退一步用 object_exists 确认上传成功 - from video_processing.oss_helpers import normalize_storage_key, oss_bucket - - bucket = oss_bucket() - key = normalize_storage_key(file_url) - if bucket and bucket.object_exists(key): - logger.info("URL 校验失败但 object_exists 确认文件存在，视为上传成功: storage_key=%s", key) - if gen_task: - gen_task.append_log("OSS上传", "URL校验降级: object_exists确认存在", level="WARN") - else: - if gen_task: - gen_task.append_log("OSS上传", "上传后URL不可访问", level="ERROR", file_url=file_url) - _flush_logs(task_id, gen_task) - raise RuntimeError( - f"OSS 上传后 URL 不可访问且 object_exists 失败: file_url={file_url}, " - f"storage_key={storage_key}" - ) - - logger.info( - "[task_id=%s] [OSS上传] 成功: 耗时=%.1fs, file_url=%s", - task_id, - upload_elapsed, - file_url, + # ── 4. 上传 OSS + 查重记录 ─────────────────────────────────────── + 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, ) if gen_task: gen_task.append_log( "OSS上传", - f"上传成功, 大小={file_size}, 耗时={upload_elapsed:.1f}s", + f"上传成功, 大小={file_size}", file_size=file_size, - duration=round(upload_elapsed, 2), file_url=file_url, ) _flush_logs(task_id, gen_task) - # 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() + # ── 5. 标记完成 ────────────────────────────────────────────────── + _update_task_status(task_id, "mark_completed", result_count=video_count) - # 7. 标记任务为 completed - _update_task_status(task_id, "mark_completed", result_count=video_count or 1) - - # 记录完成日志 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 or 1, + video_count=video_count, ) _flush_logs(task_id, gen_task) diff --git a/packages/application/tts_job/streaming_service.py b/packages/application/tts_job/streaming_service.py index aef880c2a..ffb4674f3 100644 --- a/packages/application/tts_job/streaming_service.py +++ b/packages/application/tts_job/streaming_service.py @@ -243,8 +243,8 @@ class TTSStreamingService: return total async def _send_json(self, websocket: Any, data: dict) -> None: - """安全发送 JSON 帧。""" + """发送 JSON 帧，失败时记录日志。""" try: await websocket.send_json(data) - except Exception: - pass + except Exception as e: + logger.warning("WebSocket JSON 发送失败: type=%s error=%s", data.get("type", "?"), e) -- 2.54.0 From 6f0a8253f6bdb4ebf3a1a8714ba3d6eadc801f62 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Mon, 13 Jul 2026 21:06:26 +0800 Subject: [PATCH 2/3] =?UTF-8?q?fix(P1-1/P1-2):=20processor.py=20=E8=A1=A5?= =?UTF-8?q?=20logger=20=E5=AE=9A=E4=B9=89=20+=20=E6=B5=8B=E8=AF=95?= =?UTF-8?q?=E6=96=87=E4=BB=B6=E5=AF=BC=E5=85=A5=E8=B7=AF=E5=BE=84=E5=90=8C?= =?UTF-8?q?=E6=AD=A5=E5=88=B0=E6=8B=86=E5=88=86=E5=90=8E=E6=A8=A1=E5=9D=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P1-1: processor.py 使用了 logger 但未定义，补 import logging + getLogger P1-2: test_unified_render_service.py 导入路径同步: - 音频函数 (mix_audio/merge_audio_video/clip_has_audio/RenderContext) → render_audio - 字幕函数 (_hex_to_ass_color/_position_to_ass_alignment/generate_ass_subtitles) → render_subtitles - patch.object(svc, '_mix_audio') → patch('video_processing.unified_render_service.mix_audio') - patch.object(svc, '_merge_audio_video') → patch('video_processing.unified_render_service.merge_audio_video') - probe_has_audio/run_ffmpeg patch 路径对齐到 render_audio 模块 全部 84 个测试通过。 --- apps/worker/video_processing/processor.py | 3 + tests/unit/test_unified_render_service.py | 118 +++++++++++++--------- 2 files changed, 75 insertions(+), 46 deletions(-) diff --git a/apps/worker/video_processing/processor.py b/apps/worker/video_processing/processor.py index 8b2257280..5b2e3e7ca 100644 --- a/apps/worker/video_processing/processor.py +++ b/apps/worker/video_processing/processor.py @@ -2,6 +2,7 @@ 视频处理核心类 """ +import logging import os import tempfile from dataclasses import dataclass @@ -9,6 +10,8 @@ from typing import List import ffmpeg +logger = logging.getLogger(__name__) + @dataclass class VideoResult: diff --git a/tests/unit/test_unified_render_service.py b/tests/unit/test_unified_render_service.py index b3223fa13..a9dc6d5a2 100755 --- a/tests/unit/test_unified_render_service.py +++ b/tests/unit/test_unified_render_service.py @@ -11,14 +11,22 @@ from typing import Any from unittest.mock import MagicMock, patch import pytest +from video_processing.render_audio import ( + RenderContext, + clip_has_audio, + merge_audio_video, + mix_audio, +) +from video_processing.render_subtitles import ( + _hex_to_ass_color, + _position_to_ass_alignment, + generate_ass_subtitles, +) from video_processing.unified_render_service import ( RenderResult, ResolvedClip, UnifiedRenderService, - _hex_to_ass_color, - _position_to_ass_alignment, _resolve_layer_role, - generate_ass_subtitles, ) # ── Fixtures ────────────────────────────────────────────────────────────────── @@ -101,6 +109,11 @@ def _patch_path_exists(): return patch("pathlib.Path.exists", return_value=True) +def _make_ctx() -> RenderContext: + """创建测试用 RenderContext。""" + return RenderContext(work_dir=Path("/tmp/test_render"), plan_id="plan_001") + + # ── 测试 _resolve_layer_role ───────────────────────────────────────────────── @@ -572,7 +585,7 @@ class TestPassThrough: patch("video_processing.unified_render_service.probe_duration", return_value=5.0), patch.object(svc, "_render_pass_through") as mock_pass, patch.object(svc, "_execute_ffmpeg") as mock_exec, - patch.object(svc, "_mix_audio", return_value=None), + patch("video_processing.unified_render_service.mix_audio", return_value=None), patch("shutil.copy2"), patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)), ): @@ -599,7 +612,7 @@ class TestPassThrough: patch("video_processing.unified_render_service.probe_duration", return_value=5.0), patch.object(svc, "_render_pass_through") as mock_pass, patch.object(svc, "_execute_ffmpeg") as mock_exec, - patch.object(svc, "_mix_audio", return_value=None), + patch("video_processing.unified_render_service.mix_audio", return_value=None), patch("shutil.copy2"), patch.object(svc, "_probe_output", return_value=(5.5, 2048, 1280, 720)), ): @@ -819,7 +832,7 @@ class TestRender: _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0), patch.object(svc, "_render_pass_through") as mock_pass, - patch.object(svc, "_mix_audio", return_value=None), + patch("video_processing.unified_render_service.mix_audio", return_value=None), patch("shutil.copy2"), patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)), ): @@ -848,7 +861,7 @@ class TestRender: _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, "_mix_audio", return_value=None), + patch("video_processing.unified_render_service.mix_audio", return_value=None), patch("shutil.copy2"), patch.object(svc, "_probe_output", return_value=(5.5, 2048, 1280, 720)), ): @@ -929,10 +942,11 @@ class TestAudioMixing: with ( _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0), - patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + patch("video_processing.render_audio.run_ffmpeg") as mock_run, ): layers = svc._group_clips_into_layers(svc._resolve_clips()) - result = svc._mix_audio(layers, 5.0) + ctx = _make_ctx() + result = mix_audio(ctx, layers, 5.0) assert result is not None assert result.name == "audio_plan_001.aac" @@ -957,10 +971,11 @@ class TestAudioMixing: with ( _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0), - patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + patch("video_processing.render_audio.run_ffmpeg") as mock_run, ): layers = svc._group_clips_into_layers(svc._resolve_clips()) - result = svc._mix_audio(layers, 4.5) + ctx = _make_ctx() + result = mix_audio(ctx, layers, 4.5) assert result is not None mock_run.assert_called_once() @@ -991,10 +1006,11 @@ class TestAudioMixing: with ( _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0), - patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + patch("video_processing.render_audio.run_ffmpeg") as mock_run, ): layers = svc._group_clips_into_layers(svc._resolve_clips()) - result = svc._mix_audio(layers, 5.0) + ctx = _make_ctx() + result = mix_audio(ctx, layers, 5.0) assert result is not None mock_run.assert_called_once() @@ -1018,7 +1034,8 @@ class TestAudioMixing: # 没有素材的clip会被跳过，layers为空 resolved = svc._resolve_clips() layers = svc._group_clips_into_layers(resolved) - result = svc._mix_audio(layers, 3.0) + ctx = _make_ctx() + result = mix_audio(ctx, layers, 3.0) assert result is None @@ -1037,10 +1054,11 @@ class TestAudioMixing: with ( _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0), - patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + patch("video_processing.render_audio.run_ffmpeg") as mock_run, ): layers = svc._group_clips_into_layers(svc._resolve_clips()) - result = svc._mix_audio(layers, 5.0) + ctx = _make_ctx() + result = mix_audio(ctx, layers, 5.0) assert result is not None mock_run.assert_called_once() @@ -1066,10 +1084,11 @@ class TestAudioMixing: with ( _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0), - patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + patch("video_processing.render_audio.run_ffmpeg") as mock_run, ): layers = svc._group_clips_into_layers(svc._resolve_clips()) - result = svc._mix_audio(layers, 5.0) + ctx = _make_ctx() + result = mix_audio(ctx, layers, 5.0) assert result is not None mock_run.assert_called_once() @@ -1088,10 +1107,11 @@ class TestAudioMixing: with ( _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0), - patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + patch("video_processing.render_audio.run_ffmpeg") as mock_run, ): layers = svc._group_clips_into_layers(svc._resolve_clips()) - result = svc._mix_audio(layers, 5.0) + ctx = _make_ctx() + result = mix_audio(ctx, layers, 5.0) assert result is not None mock_run.assert_called_once() @@ -1108,11 +1128,12 @@ class TestAudioMixing: with ( _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=10.0), - patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + patch("video_processing.render_audio.run_ffmpeg") as mock_run, ): layers = svc._group_clips_into_layers(svc._resolve_clips()) + ctx = _make_ctx() # video_duration 只有 3.0，小于 clip 的 10.0 - result = svc._mix_audio(layers, 3.0) + result = mix_audio(ctx, layers, 3.0) assert result is not None mock_run.assert_called_once() @@ -1125,13 +1146,13 @@ class TestAudioMixing: def test_merge_audio_video(self): """合并音视频命令正确。""" - svc = _make_service([], {}) video_path = Path("/tmp/video.mp4") audio_path = Path("/tmp/audio.aac") output_path = Path("/tmp/output.mp4") + ctx = _make_ctx() - with patch("video_processing.unified_render_service.run_ffmpeg") as mock_run: - svc._merge_audio_video(video_path, audio_path, output_path) + with patch("video_processing.render_audio.run_ffmpeg") as mock_run: + merge_audio_video(ctx, video_path, audio_path, output_path) mock_run.assert_called_once() cmd = mock_run.call_args[0][0] @@ -1156,8 +1177,8 @@ class TestAudioMixing: _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0), patch.object(svc, "_execute_ffmpeg"), - patch.object(svc, "_mix_audio", return_value=Path("/tmp/audio.aac")) as mock_mix, - patch.object(svc, "_merge_audio_video") as mock_merge, + patch("video_processing.unified_render_service.mix_audio", return_value=Path("/tmp/audio.aac")) as mock_mix, + patch("video_processing.unified_render_service.merge_audio_video") as mock_merge, patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)), ): result = svc.render() @@ -1182,7 +1203,7 @@ class TestAudioMixing: _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0), patch.object(svc, "_execute_ffmpeg"), - patch.object(svc, "_mix_audio", return_value=None), + patch("video_processing.unified_render_service.mix_audio", return_value=None), patch("shutil.copy2") as mock_copy, patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)), ): @@ -1201,8 +1222,8 @@ class TestAudioMixing: _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0), patch.object(svc, "_render_pass_through", return_value=True) as mock_pt, - patch.object(svc, "_mix_audio") as mock_mix, - patch.object(svc, "_merge_audio_video") as mock_merge, + patch("video_processing.unified_render_service.mix_audio") as mock_mix, + patch("video_processing.unified_render_service.merge_audio_video") as mock_merge, patch("shutil.copy2") as mock_copy, patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)), ): @@ -1266,11 +1287,12 @@ class TestAudioMixing: with ( _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0), - patch("video_processing.ffmpeg_utils.probe_has_audio", return_value=False), - patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + patch("video_processing.render_audio.probe_has_audio", return_value=False), + patch("video_processing.render_audio.run_ffmpeg") as mock_run, ): layers = svc._group_clips_into_layers(svc._resolve_clips()) - result = svc._mix_audio(layers, 5.0) + ctx = _make_ctx() + result = mix_audio(ctx, layers, 5.0) assert result is None # 没有音频流时不应调用 FFmpeg @@ -1295,11 +1317,12 @@ class TestAudioMixing: with ( _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0), - patch("video_processing.ffmpeg_utils.probe_has_audio", side_effect=fake_has_audio), - patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + patch("video_processing.render_audio.probe_has_audio", side_effect=fake_has_audio), + patch("video_processing.render_audio.run_ffmpeg") as mock_run, ): layers = svc._group_clips_into_layers(svc._resolve_clips()) - result = svc._mix_audio(layers, 5.0) + ctx = _make_ctx() + result = mix_audio(ctx, layers, 5.0) assert result is not None mock_run.assert_called_once() @@ -1333,11 +1356,12 @@ class TestAudioMixing: with ( _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0), - patch("video_processing.ffmpeg_utils.probe_has_audio", side_effect=fake_has_audio), - patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + patch("video_processing.render_audio.probe_has_audio", side_effect=fake_has_audio), + patch("video_processing.render_audio.run_ffmpeg") as mock_run, ): layers = svc._group_clips_into_layers(svc._resolve_clips()) - result = svc._mix_audio(layers, 5.0) + ctx = _make_ctx() + result = mix_audio(ctx, layers, 5.0) assert result is not None mock_run.assert_called_once() @@ -1367,11 +1391,12 @@ class TestAudioMixing: with ( _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0), - patch("video_processing.ffmpeg_utils.probe_has_audio", return_value=False), - patch("video_processing.unified_render_service.run_ffmpeg") as mock_run, + patch("video_processing.render_audio.probe_has_audio", return_value=False), + patch("video_processing.render_audio.run_ffmpeg") as mock_run, ): layers = svc._group_clips_into_layers(svc._resolve_clips()) - result = svc._mix_audio(layers, 5.0) + ctx = _make_ctx() + result = mix_audio(ctx, layers, 5.0) assert result is None mock_run.assert_not_called() @@ -1389,11 +1414,12 @@ class TestAudioMixing: resolved = svc._resolve_clips() clip = resolved[0] - with patch("video_processing.ffmpeg_utils.probe_has_audio", return_value=True) as mock_probe: + with patch("video_processing.render_audio.probe_has_audio", return_value=True) as mock_probe: # 调用 3 次 - r1 = svc._clip_has_audio(clip) - r2 = svc._clip_has_audio(clip) - r3 = svc._clip_has_audio(clip) + ctx = _make_ctx() + r1 = clip_has_audio(ctx, clip) + r2 = clip_has_audio(ctx, clip) + r3 = clip_has_audio(ctx, clip) assert r1 is True and r2 is True and r3 is True # 实际只探测了 1 次 -- 2.54.0 From a695342b3661a7d4bdaf37ffdd90a2537cd40939 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Mon, 13 Jul 2026 21:55:08 +0800 Subject: [PATCH 3/3] =?UTF-8?q?fix:=20edit=5Fplans=5Ftimeline.py=20?= =?UTF-8?q?=E6=B8=85=E7=90=86=E6=9C=AA=E4=BD=BF=E7=94=A8=20import=20(datet?= =?UTF-8?q?ime/EditPlanResponse/Optional)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/api/app/api/routes/edit_plans_timeline.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/apps/api/app/api/routes/edit_plans_timeline.py b/apps/api/app/api/routes/edit_plans_timeline.py index 3d343ad06..e75a1e899 100644 --- a/apps/api/app/api/routes/edit_plans_timeline.py +++ b/apps/api/app/api/routes/edit_plans_timeline.py @@ -8,12 +8,10 @@ from __future__ import annotations import logging -from datetime import datetime -from typing import Any, List, Optional +from typing import Any, List from app.api.routes._helpers import check_project_access from app.api.routes.edit_plans import ( - EditPlanResponse, GenerateFromTemplateRequest, GenerateFromTemplateResponse, _PlanClipItem, -- 2.54.0