""" 视频生成任务 — 使用 UnifiedRenderService 统一渲染引擎. 支持四种剪辑模式：一镜到底、画中画、口播、口播+画中画。 模式差异体现在虚拟剪辑计划的 clip_type 分布上，渲染引擎不判断模式。 模式 → clip_type 映射： ONE_TAKE: N 个 main clips PIP: 1 main + N-1 overlay VOICE_OVER: N 个 main(config.role=b_roll) VOICE_PIP: 1 background + 1 corner_voice + N-2 b_roll """ from __future__ import annotations import logging import os import tempfile import time from dataclasses import dataclass, field from pathlib import Path from typing import Any from worker_app.celery_app import celery_app from worker_app.db import SessionLocal OUTPUT_WIDTH = 1280 OUTPUT_HEIGHT = 720 OUTPUT_FPS = 25.0 OUTPUT_DURATION_SECONDS = 5.0 GENERATED_FILES_DIR = Path(os.getenv("GENERATED_FILES_DIR", "/app/generated")) logger = logging.getLogger(__name__) # ── 状态更新辅助函数 ────────────────────────────────────────────────────────── def _update_task_status(task_id: str, status_action: str, **kwargs) -> bool: """更新 GenerationTask 状态（独立 session，异常不向外抛出）。 Args: task_id: 任务 ID status_action: 状态动作名，如 "mark_processing" / "mark_completed" / "mark_failed" **kwargs: 传递给对应方法的参数 Returns: True 表示更新成功，False 表示更新失败 """ try: from packages.adapters.sqlalchemy_impl.generation_task_repository import ( SQLAlchemyGenerationTaskRepository, ) session = SessionLocal() try: repo = SQLAlchemyGenerationTaskRepository(session) task = repo.get(task_id) if task is None: logger.warning("更新任务状态失败：任务不存在 task_id=%s", task_id) return False action = getattr(task, status_action, None) if action is None: logger.warning("未知的状态动作: %s", status_action) return False action(**kwargs) repo.update(task) logger.info( "GenerationTask 状态更新成功: task_id=%s action=%s", task_id, status_action, ) return True finally: session.close() except Exception as e: logger.error( "更新 GenerationTask 状态异常: task_id=%s action=%s error=%s", task_id, status_action, e, exc_info=True, ) return False def _build_error_info(error: Exception, stage: str = "render") -> dict: """构建结构化错误信息。 Args: error: 异常对象 stage: 发生错误的阶段（download/render/merge/upload等） Returns: 包含 error_type, message, stack_trace, stage, failed_at 的字典 """ import traceback from datetime import datetime, timezone tb_str = traceback.format_exc() # 截取堆栈前20行，避免字段过大 tb_lines = tb_str.strip().splitlines() if len(tb_lines) > 20: tb_summary = "\n".join(tb_lines[:20]) + f"\n... (truncated, total {len(tb_lines)} lines)" else: tb_summary = tb_str return { "error_type": type(error).__name__, "message": str(error), "stack_trace": tb_summary, "stage": stage, "failed_at": datetime.now(timezone.utc).isoformat(), } # ── 日志持久化辅助 ──────────────────────────────────────────────────────────── def _flush_logs(task_id: str, gen_task) -> None: """将 gen_task.logs 持久化到 DB（独立 session，失败不抛异常）。""" try: session = SessionLocal() try: from packages.adapters.sqlalchemy_impl.models import GenerationTaskModel model = session.query(GenerationTaskModel).filter(GenerationTaskModel.id == task_id).first() if model: model.logs = gen_task.logs session.commit() finally: session.close() except Exception: logger.warning("[task_id=%s] 日志持久化失败", task_id, exc_info=True) # ── 共享工具模块导入 ────────────────────────────────────────────────────────── from services.asr_service_factory import get_asr_service from video_processing.dedup_helpers import create_video_record_and_dedup from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg from video_processing.oss_helpers import ( download_asset, get_signed_download_url, upload_to_oss, ) from video_processing.render_engine_resolver import ENGINE_LEGACY, ENGINE_UNIFIED from video_processing.unified_render_service import UnifiedRenderService # ── 虚拟 Plan / Clip（内存中构建，不写数据库） ──────────────────────────────── @dataclass class _VirtualPlan: """内存中的虚拟剪辑计划，供 UnifiedRenderService 使用。""" id: str name: str = "" config: dict[str, Any] = field(default_factory=dict) @dataclass class _VirtualClip: """内存中的虚拟剪辑片段，供 UnifiedRenderService 使用。""" id: str plan_id: str = "" clip_type: str = "main" order: int = 0 asset_id: str = "" text_content: str = "" start_time: float = 0.0 duration: float = 0.0 transition_effect: str = "cut" status: str = "ready" config: dict[str, Any] = field(default_factory=dict) def _build_plan_and_clips_from_task( task_id: str, downloaded_paths: list[Path], mode: str, ) -> tuple[_VirtualPlan, list[_VirtualClip], dict[str, Path]]: """根据模式和下载的素材路径，构建虚拟 plan + clips + asset_path_map。 模式 → clip_type 映射： ONE_TAKE: N 个 main clips PIP: 1 main + N-1 overlay VOICE_OVER: N 个 main(config.role=b_roll) VOICE_PIP: 1 background + 1 corner_voice + N-2 b_roll Returns: (virtual_plan, virtual_clips, asset_path_map) """ plan = _VirtualPlan(id=task_id, name=f"Generated-{task_id[:8]}") # 为每个下载路径生成合成 asset_id，并预探测素材时长 asset_path_map: dict[str, Path] = {} path_to_asset_id: dict[Path, str] = {} path_duration: dict[Path, float] = {} for i, p in enumerate(downloaded_paths): asset_id = f"gen_{task_id[:8]}_{i:03d}{p.suffix or '.mp4'}" asset_path_map[asset_id] = p path_to_asset_id[p] = asset_id path_duration[p] = probe_duration(p) clips: list[_VirtualClip] = [] if mode == "pip": # 1 main + N-1 overlay for i, p in enumerate(downloaded_paths): clip_type = "main" if i == 0 else "overlay" clips.append( _VirtualClip( id=f"vc_{i:03d}", plan_id=task_id, clip_type=clip_type, order=i, asset_id=path_to_asset_id[p], duration=path_duration[p], ) ) elif mode == "voice_over": # N 个 main(config.role=b_roll) for i, p in enumerate(downloaded_paths): clips.append( _VirtualClip( id=f"vc_{i:03d}", plan_id=task_id, clip_type="main", order=i, asset_id=path_to_asset_id[p], duration=path_duration[p], config={"role": "b_roll"}, ) ) elif mode == "voice_pip": # 1 background + 1 corner_voice + N-2 b_roll for i, p in enumerate(downloaded_paths): if i == 0: clip_type = "background" elif i == 1: clip_type = "corner_voice" else: clip_type = "b_roll" clips.append( _VirtualClip( id=f"vc_{i:03d}", plan_id=task_id, clip_type=clip_type, order=i, asset_id=path_to_asset_id[p], duration=path_duration[p], ) ) else: # ONE_TAKE (default): N 个 main clips for i, p in enumerate(downloaded_paths): clips.append( _VirtualClip( id=f"vc_{i:03d}", plan_id=task_id, clip_type="main", order=i, asset_id=path_to_asset_id[p], duration=path_duration[p], ) ) return plan, clips, asset_path_map def _create_fallback_clip(output_path: Path, title: str) -> None: """创建 fallback 视频（无素材时）""" safe_title = title.replace(":", "\\:").replace("'", "\\'")[:80] run_ffmpeg( [ FFMPEG_BIN, "-y", "-f", "lavfi", "-i", f"color=c=#111827:s={OUTPUT_WIDTH}x{OUTPUT_HEIGHT}:d={OUTPUT_DURATION_SECONDS}:r={int(OUTPUT_FPS)}", "-vf", f"drawtext=text='{safe_title}':fontcolor=white:fontsize=48:x=(w-text_w)/2:y=(h-text_h)/2", "-c:v", "libx264", "-pix_fmt", "yuv420p", "-movflags", "+faststart", str(output_path), ] ) def _mux_audio_track(video_path: Path, audio_path: str, output_path: Path) -> None: """将音频轨混入已渲染的视频（后处理步骤）。 使用 FFmpeg 将视频和音频合并，视频时长为准，音频不足则循环， 音频过长则截断。 """ command = [ FFMPEG_BIN, "-y", "-i", str(video_path), "-i", audio_path, "-c:v", "copy", "-c:a", "aac", "-b:a", "192k", "-shortest", "-map", "0:v:0", "-map", "1:a:0", "-movflags", "+faststart", str(output_path), ] run_ffmpeg(command) def _download_voice_asset(voice_library_id: str, local_path: Path) -> bool: """下载配音文件""" if not voice_library_id: return False storage_key = f"voice/{voice_library_id}.mp3" return download_asset(storage_key, local_path) def _prepare_bgm_track( *, bgm_config: dict, temp_path: Path, task_id: str = "", ) -> str | None: """准备 BGM 音频文件（下载到本地）. 支持 3 种来源（按优先级）： 1. audio_url — 外部直链 URL（最高优先级） 2. asset_id — 素材库中的音频素材 3. preset_id — 预设 BGM 库 Returns: BGM 本地文件路径，准备失败返回 None """ from urllib.parse import urlparse audio_url = bgm_config.get("audio_url", "") or "" asset_id = bgm_config.get("asset_id", "") or "" preset_id = bgm_config.get("preset_id", "") or "" bgm_file = temp_path / f"bgm_{task_id or 'track'}.mp3" # 优先级1：外部直链 URL if audio_url: try: parsed = urlparse(audio_url) if parsed.scheme in ("http", "https"): from video_processing.url_security import ( ALLOWED_AUDIO_MIME_TYPES, safe_download_file, ) logger.info("[task_id=%s] [BGM] 从URL下载: %s", task_id, audio_url[:80]) safe_download_file( audio_url, str(bgm_file), purpose="bgm_download", allowed_mime_types=ALLOWED_AUDIO_MIME_TYPES, timeout=60.0, ) if bgm_file.exists() and bgm_file.stat().st_size > 0: return str(bgm_file) except Exception as e: logger.warning("[task_id=%s] [BGM] URL下载失败: %s", task_id, e) # 优先级2：素材库素材 if asset_id: try: from app.core.db import SessionLocal from packages.adapters.sqlalchemy_impl.models import AssetModel session = SessionLocal() try: model = session.query(AssetModel).filter(AssetModel.id == asset_id).first() if model and model.file_url: storage_key = model.file_url logger.info("[task_id=%s] [BGM] 从素材库下载: asset_id=%s", task_id, asset_id) ok = download_asset(storage_key, bgm_file) if ok and bgm_file.exists() and bgm_file.stat().st_size > 0: return str(bgm_file) finally: session.close() except Exception as e: logger.warning("[task_id=%s] [BGM] 素材库下载失败: %s", task_id, e) # 优先级3：预设 BGM 库 if preset_id: try: from packages.domain.preset_bgm import get_preset_bgm preset = get_preset_bgm(preset_id) if preset and preset.audio_url: from video_processing.url_security import ( ALLOWED_AUDIO_MIME_TYPES, safe_download_file, ) logger.info("[task_id=%s] [BGM] 从预设库下载: preset_id=%s", task_id, preset_id) safe_download_file( preset.audio_url, str(bgm_file), purpose="bgm_preset_download", allowed_mime_types=ALLOWED_AUDIO_MIME_TYPES, timeout=60.0, ) if bgm_file.exists() and bgm_file.stat().st_size > 0: return str(bgm_file) except Exception as e: logger.warning("[task_id=%s] [BGM] 预设库下载失败: %s", task_id, e) # 所有来源都失败 logger.warning("[task_id=%s] [BGM] 所有来源都无法获取BGM，跳过", task_id) return None def _verify_url_accessible( url: str, timeout: float = 10.0, retries: int = 2, max_redirects: int = 5, ) -> bool: """HEAD 请求校验 URL 可访问（含重试，防止 OSS 抖动误报）。 安全增强： - 请求前先做 SSRF 安全校验（内网IP/回环地址/链路本地地址等） - scheme 仅允许 http/https - 端口仅允许 80/443 - 手动跟随重定向，每一跳 URL 都做 SSRF 校验，避免重定向到内网地址绕过 Args: url: 待校验的 URL timeout: 单次请求超时时间（秒） retries: 最大重试次数（默认 2 次，首次失败后间隔 1s 重试） max_redirects: 最大重定向次数（默认 5 次） Returns: True 表示 URL 可访问（HTTP 2xx/3xx），False 表示所有尝试均失败或安全校验不通过。 """ import time import urllib.request from urllib.parse import urljoin from video_processing.url_security import UrlSecurityError, validate_url_safety # P0-1 SSRF 防护：请求前先校验 URL 安全性 try: validate_url_safety(url, purpose="url_verify") except UrlSecurityError as e: logger.warning("URL 安全校验失败，拒绝访问: url=%s error=%s", url[:80], e) return False last_error: Exception | None = None def _do_verify(current_url: str) -> bool: """单次校验：手动跟随重定向，每跳都做 SSRF 检查.""" redirect_count = 0 url_being_checked = current_url # 禁止自动重定向的 handler，手动控制每一跳 class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: N802 return None opener = urllib.request.build_opener(NoRedirect()) while redirect_count <= max_redirects: # 每一跳都做 SSRF 安全校验 try: safe_url = validate_url_safety(url_being_checked, purpose="url_verify") except UrlSecurityError as e: logger.warning( "URL校验跳转地址不安全: redirect=%d url=%s error=%s", redirect_count, url_being_checked, e, ) raise req = urllib.request.Request(safe_url, method="HEAD") req.add_header("User-Agent", "xiaoxia-saas-worker/1.0") with opener.open(req, timeout=timeout) as resp: # noqa: S310 if 200 <= resp.status < 300: return True if resp.status in (301, 302, 303, 307, 308): location = resp.headers.get("Location", "") if not location: raise Exception(f"HTTP {resp.status} 但无 Location 头") # 相对路径转绝对 url_being_checked = urljoin(safe_url, location) redirect_count += 1 continue if resp.status < 400: return True raise Exception(f"HTTP {resp.status}") raise Exception(f"重定向次数超过上限 ({max_redirects})") for attempt in range(1 + retries): try: if _do_verify(url): return True except Exception as e: last_error = e if attempt < retries: logger.warning( "URL 校验失败，1s 后重试: url=%s attempt=%d/%d error=%s", url, attempt + 1, retries, last_error, ) time.sleep(1) logger.warning("URL 可访问性校验最终失败: url=%s error=%s", url, last_error) return False def _download_library_assets( temp_path: Path, *, asset_library_id: str = "", project_id: str = "", asset_ids: list[str] | None = None, strict: bool = True, task_id: str = "", gen_task=None, ) -> list[Path]: """下载视频素材 — 同时支持素材库模式和项目级模式。 两种查询路径： - 素材库模式：asset_library_id 非空时，按 asset_library_id + asset_ids 查 - 项目级模式：project_id 非空时，按 project_id + asset_ids 查 - 两者都提供时优先素材库模式；两者都为空时抛异常 归属校验与下载在同一 DB session 中完成，避免多次连接开销（P3-2）。 Args: temp_path: 临时目录路径 asset_library_id: 素材库 ID（可选，与 project_id 二选一） project_id: 项目 ID（可选，与 asset_library_id 二选一） asset_ids: 指定素材 ID 列表，为空则下载全部 ready 视频素材 strict: 严格模式（默认 True）。 True — 任何素材下载失败立即抛 RuntimeError； False — 跳过失败素材，返回成功列表（调用方可通过日志感知失败）。 Returns: 下载成功的视频文件 Path 列表 Raises: ValueError: 当 asset_library_id 和 project_id 都为空时 RuntimeError: strict=True 时任何下载失败；或指定了 asset_ids 但全部下载失败 """ if not asset_library_id and not project_id: raise ValueError("asset_library_id 和 project_id 至少需要提供一个") try: from packages.adapters.sqlalchemy_impl.models import AssetModel session = SessionLocal() try: # 构建查询 query = session.query(AssetModel).filter( AssetModel.status == "ready", AssetModel.file_type.in_(["video", "video/mp4", "video/quicktime"]), ) if asset_ids: # 明确指定了 asset_ids：直接按 ID 查，不预先按 library/project 过滤 # 避免项目级素材或跨库素材因为 library_id 不匹配而查不到 # 归属安全由后面的归属校验保证 query = query.filter(AssetModel.id.in_(asset_ids)) logger.info( "下载指定素材: asset_ids=%d 个, asset_library_id=%s, project_id=%s", len(asset_ids), asset_library_id or "none", project_id or "none", ) else: # 未指定 asset_ids：按 library 或 project 下载全部 ready 视频 if asset_library_id: query = query.filter(AssetModel.asset_library_id == asset_library_id) logger.info( "下载素材库全部视频: asset_library_id=%s", asset_library_id, ) else: query = query.filter(AssetModel.project_id == project_id) logger.info( "下载项目全部视频: project_id=%s", project_id, ) assets = query.order_by(AssetModel.created_at).all() if not assets: mode_desc = f"素材库 {asset_library_id}" if asset_library_id else f"项目 {project_id}" msg = f"未找到视频素材: {mode_desc}, asset_ids={asset_ids or 'all'}" logger.error(msg) raise RuntimeError(msg) # P3-2: 归属校验合并到同一 session if asset_ids: found_ids = {a.id for a in assets} missing_ids = set(asset_ids) - found_ids if missing_ids: raise ValueError(f"素材不存在: asset_ids={sorted(missing_ids)}") for asset in assets: # 校验素材库归属（只要传了 asset_library_id 就校验） if asset_library_id and asset.asset_library_id != asset_library_id: raise ValueError( f"素材不属于指定素材库: asset_id={asset.id}, " f"expected_asset_library_id={asset_library_id}, " f"actual_asset_library_id={asset.asset_library_id}" ) # 校验项目归属（只要传了 project_id 就校验） if project_id and asset.project_id != project_id: raise ValueError( f"素材不属于指定项目: asset_id={asset.id}, " f"expected_project_id={project_id}, " f"actual_project_id={asset.project_id}" ) logger.info( "素材归属校验通过 (同 session): %d 个 asset_ids", len(asset_ids), ) downloaded: list[Path] = [] failed_assets: list[str] = [] for i, asset in enumerate(assets): storage_key = asset.file_url if asset.file_url else None if not storage_key: failed_assets.append(f"{asset.name}({asset.id})") logger.warning( "[task_id=%s] 素材缺少 file_url, 跳过: asset_id=%s name=%s", task_id, asset.id, asset.name, ) if gen_task: gen_task.append_log( "下载素材", "素材缺少file_url, 跳过", level="WARN", asset_id=asset.id, asset_name=asset.name, success=False, file_size=0, duration=0.0, ) if strict: raise RuntimeError(f"素材缺少 file_url: asset_id={asset.id}, name={asset.name}") continue ext = Path(storage_key).suffix or ".mp4" local_file = temp_path / f"asset_{i:03d}_{asset.id}{ext}" asset_start = time.monotonic() download_ok = download_asset(storage_key, local_file) asset_elapsed = time.monotonic() - asset_start if download_ok: file_size = local_file.stat().st_size if local_file.exists() else 0 downloaded.append(local_file) logger.info( "[task_id=%s] Downloaded asset: %s -> %s (size=%d, time=%.1fs)", task_id, asset.name, local_file, file_size, asset_elapsed, ) if gen_task: gen_task.append_log( "下载素材", f"下载成功: {asset.name}", asset_id=asset.id, asset_name=asset.name, success=True, file_size=file_size, duration=round(asset_elapsed, 2), ) else: failed_assets.append(f"{asset.name}({asset.id})") logger.warning( "[task_id=%s] Failed to download asset: %s (id=%s)", task_id, asset.name, asset.id, ) if gen_task: gen_task.append_log( "下载素材", f"下载失败: {asset.name}", level="WARN", asset_id=asset.id, asset_name=asset.name, success=False, file_size=0, duration=round(asset_elapsed, 2), ) if strict: raise RuntimeError(f"素材下载失败: asset_id={asset.id}, name={asset.name}") # 指定了 asset_ids 但全部下载失败 → 无论 strict 与否都报错 if asset_ids and not downloaded: msg = f"指定的 {len(asset_ids)} 个素材全部下载失败, failed={failed_assets}" logger.error(msg) raise RuntimeError(msg) # 非严格模式有部分失败，记录警告 if failed_assets and not strict: logger.warning( "素材下载部分失败 (非严格模式): failed=%s, succeeded=%d", failed_assets, len(downloaded), ) return downloaded finally: session.close() except (ValueError, RuntimeError): raise except Exception as e: logger.error("Error downloading library assets: %s", e, exc_info=True) raise RuntimeError(f"素材下载异常: {e}") from e # ── P1 校验函数 ────────────────────────────────────────────────────────────── def _validate_template_exists(template_id: str) -> None: """校验 template_id 是否存在且可用。 Raises: ValueError: template_id 不存在或已禁用时抛出 """ from packages.adapters.sqlalchemy_impl.models import TemplateModel session = SessionLocal() try: template = ( session.query(TemplateModel) .filter( TemplateModel.id == template_id, TemplateModel.is_active.is_(True), ) .first() ) if template is None: raise ValueError(f"模板不存在或已禁用: template_id={template_id}") logger.info("模板校验通过: template_id=%s name=%s", template_id, template.name) finally: session.close() # ── 渲染引擎选择 ───────────────────────────────────────────────────────────── def _resolve_render_engine(user_id: str) -> str: """根据 Feature Flag 决定使用哪个渲染引擎。 Returns: "legacy" 或 "unified" """ try: from video_processing.render_engine_resolver import get_render_engine_resolver resolver = get_render_engine_resolver() return resolver.get_engine(user_id=user_id) except Exception as exc: logger.warning("获取渲染引擎配置失败，fallback 到 unified: %s", exc) return ENGINE_UNIFIED # ── 旧引擎渲染（FFmpeg filter_complex） ──────────────────────────────────────── def _render_with_legacy_engine( task_id: str, virtual_clips: list[_VirtualClip], asset_path_map: dict[str, Path], work_dir: Path, output_path: Path, ) -> tuple[float, int]: """旧引擎渲染路径：手动构建 FFmpeg filter_complex 命令。 说明：generate_video 任务使用虚拟 clips（无 EditPlan 数据库记录）， 因此无法直接复用 VideoComposeService。这里手动构建等价的 filter_complex 命令，与旧引擎行为一致（scale → crop → setpts → trim → setpts， 无 fps 归一化，保持原帧率）。 支持模式：one_take / pip / voice_over / voice_pip - 所有模式统一走 concat 滤镜（与旧引擎多片段逻辑一致） Returns: (duration_seconds, file_size_bytes) """ import subprocess main_clips = [ c for c in virtual_clips if c.clip_type in ("main", "b_roll", "background") or (c.clip_type == "main" and c.config.get("role") == "b_roll") ] if not main_clips: main_clips = virtual_clips[:1] input_args: list[str] = [] video_filters: list[str] = [] audio_filters: list[str] = [] for i, clip in enumerate(main_clips): local_path = asset_path_map.get(clip.asset_id) if not local_path: continue input_args.extend(["-i", str(local_path)]) duration = clip.duration or 0.0 # 视频滤镜：scale → crop → setpts → trim → setpts（与旧引擎一致） vf = ( f"[{i}:v]" f"scale={OUTPUT_WIDTH}:{OUTPUT_HEIGHT}:force_original_aspect_ratio=increase," f"crop={OUTPUT_WIDTH}:{OUTPUT_HEIGHT}," f"setpts=PTS-STARTPTS," f"trim=0:{duration:.3f}," f"setpts=PTS-STARTPTS" f"[v{i}]" ) video_filters.append(vf) # 音频滤镜：atrim → asetpts af = f"[{i}:a]atrim=0:{duration:.3f},asetpts=PTS-STARTPTS[a{i}]" audio_filters.append(af) n = len(main_clips) if n == 1: video_label = "[v0]" audio_label = "[a0]" else: # concat 视频 v_inputs = "".join(f"[v{i}]" for i in range(n)) video_filters.append(f"{v_inputs}concat=n={n}:v=1:a=0[outv]") # concat 音频 a_inputs = "".join(f"[a{i}]" for i in range(n)) audio_filters.append(f"{a_inputs}concat=n={n}:v=0:a=1[outa]") video_label = "[outv]" audio_label = "[outa]" # 组装 filter_complex fc_parts = video_filters + audio_filters filter_complex = ";".join(fc_parts) command = [ FFMPEG_BIN, "-y", *input_args, "-filter_complex", filter_complex, "-map", video_label, "-map", audio_label, "-c:v", "libx264", "-crf", "23", "-preset", "medium", "-c:a", "aac", "-b:a", "192k", "-movflags", "+faststart", str(output_path), ] logger.info("[task_id=%s] [渲染] legacy 引擎 FFmpeg 开始: clips=%d", task_id, n) try: run_ffmpeg(command) except subprocess.CalledProcessError as e: logger.error( "[task_id=%s] [渲染] legacy 引擎 FFmpeg 失败: %s\nfilter_complex: %s", task_id, e, filter_complex[:500], ) raise file_size = output_path.stat().st_size if output_path.exists() else 0 duration = probe_duration(output_path) 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) # ── 准备 BGM 音频 ── bgm_path: str | None = None plan_config = virtual_plan.config or {} bgm_config = plan_config.get("bgm", {}) or {} if bgm_config.get("enabled", False): try: bgm_path = _prepare_bgm_track( bgm_config=bgm_config, temp_path=temp_path, task_id=task_id, ) except Exception as bgm_err: logger.warning("[task_id=%s] [BGM] 准备失败，跳过BGM: %s", task_id, bgm_err) bgm_path = None 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), asr_service=get_asr_service(), bgm_path=bgm_path, ) 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 ────────────────────────────────────────────────────────────── @celery_app.task(bind=True, name="worker.generate_video", max_retries=2) def generate_video(self, task_id: str) -> dict: """生成视频任务 — 使用 UnifiedRenderService 统一渲染。 流程： 1. 加载 GenerationTask 信息 2. 从素材库下载视频素材 3. 根据模式构建虚拟 plan + clips 4. 使用 UnifiedRenderService 渲染 5. 如有配音，后处理混音 6. 上传 OSS + 查重 7. 更新 GenerationTask 状态 Args: task_id: 任务 ID（从数据库加载完整任务信息） Returns: 生成结果字典 """ from packages.adapters.sqlalchemy_impl.generation_task_repository import ( SQLAlchemyGenerationTaskRepository, ) from packages.domain import EditingMode logger.info("[task_id=%s] [接收任务] 开始生成视频任务", task_id) # ── 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: _repo = SQLAlchemyGenerationTaskRepository(_session) gen_task = _repo.get(task_id) finally: _session.close() # 记录接收任务日志 if gen_task: gen_task.append_log( "接收任务", 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) # 标记任务为 running _update_task_status(task_id, "mark_processing") try: 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" try: if template_id: _validate_template_exists(template_id) with tempfile.TemporaryDirectory(prefix="xiaoxia-generation-") as temp_dir: temp_path = Path(temp_dir) # ── 2. 下载素材 ────────────────────────────────────────────────── downloaded_videos, audio_path = _download_all_assets( temp_path, asset_library_id=asset_library_id, project_id=project_id, task_asset_ids=task_asset_ids, voice_library_id=voice_library_id, task_id=task_id, ) if gen_task: gen_task.append_log( "下载素材", f"成功下载 {len(downloaded_videos)} 个视频素材", count=len(downloaded_videos), ) _flush_logs(task_id, gen_task) # ── 3. 渲染 + 混音 ─────────────────────────────────────────────── output_path, render_duration = _render_video( task_id=task_id, 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"渲染完成, 时长={render_duration:.1f}s") _flush_logs(task_id, gen_task) # ── 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}", file_size=file_size, file_url=file_url, ) _flush_logs(task_id, gen_task) # ── 5. 标记完成 ────────────────────────────────────────────────── _update_task_status(task_id, "mark_completed", result_count=video_count) # 5.1 更新标题使用次数 try: _title_session = SessionLocal() try: from packages.adapters.sqlalchemy_impl.generation_task_repository import ( SQLAlchemyGenerationTaskRepository, ) from packages.adapters.sqlalchemy_impl.title_library_repository import ( SQLAlchemyTitleLibraryRepository, ) _task_repo = SQLAlchemyGenerationTaskRepository(_title_session) _gen_task = _task_repo.get(task_id) if _gen_task and _gen_task.title_ids and _gen_task.created_by_user_id: _title_repo = SQLAlchemyTitleLibraryRepository(_title_session) for _tid in _gen_task.title_ids: try: _title_repo.increment_usage_count(_tid, _gen_task.created_by_user_id) except Exception: logger.warning( "[task_id=%s] 更新标题使用次数失败: title_id=%s", task_id, _tid, exc_info=True, ) finally: _title_session.close() except Exception: logger.warning("[task_id=%s] 更新标题使用次数异常（不影响主流程）", task_id, exc_info=True) # 5.2 更新素材使用次数 + 最近使用时间 try: from worker_app.core.asset_usage import mark_asset_used_for_generation _asset_session = SessionLocal() try: from packages.adapters.sqlalchemy_impl.asset_repository import ( SQLAlchemyAssetRepository, ) from packages.adapters.sqlalchemy_impl.generation_task_repository import ( SQLAlchemyGenerationTaskRepository, ) _task_repo = SQLAlchemyGenerationTaskRepository(_asset_session) _asset_repo = SQLAlchemyAssetRepository(_asset_session) _gen_task = _task_repo.get(task_id) if _gen_task and _gen_task.asset_ids: for _aid in _gen_task.asset_ids: try: _asset = _asset_repo.get(_aid) if _asset: mark_asset_used_for_generation(_asset) _asset_repo.update(_asset) except Exception: logger.warning( "[task_id=%s] 更新素材使用次数失败: asset_id=%s", task_id, _aid, exc_info=True, ) finally: _asset_session.close() except Exception: logger.warning("[task_id=%s] 更新素材使用次数异常（不影响主流程）", task_id, exc_info=True) if gen_task: gen_task.append_log( "任务完成", f"视频生成完成: 时长={duration:.2f}s, 大小={file_size}", duration=round(duration, 2), file_size=file_size, video_count=video_count, ) _flush_logs(task_id, gen_task) logger.info( "[task_id=%s] [任务完成] duration=%.2fs file_size=%d", task_id, duration, file_size, ) return { "status": "completed", "task_id": task_id, "output_path": str(output_path), "file_size": file_size, "duration": duration, "width": OUTPUT_WIDTH, "height": OUTPUT_HEIGHT, "mode": editing_mode.value, } except Exception as error: logger.error("[task_id=%s] [任务失败] %s", task_id, error, exc_info=True) # 构建结构化错误信息 error_info = _build_error_info(error, stage="render") # 记录失败日志 try: _session = SessionLocal() try: _repo = SQLAlchemyGenerationTaskRepository(_session) gen_task = _repo.get(task_id) if gen_task: gen_task.append_log( "任务失败", str(error), level="ERROR", error_type=type(error).__name__, stage="render", ) _flush_logs(task_id, gen_task) finally: _session.close() except Exception: logger.warning("[task_id=%s] 记录失败日志异常", task_id, exc_info=True) _update_task_status( task_id, "mark_failed", error_message=str(error), error_info=error_info, ) # ── 自动重试逻辑 ────────────────────────────────────────────────── try: from packages.adapters.sqlalchemy_impl.generation_task_repository import ( SQLAlchemyGenerationTaskRepository, ) _s = SessionLocal() try: _r = SQLAlchemyGenerationTaskRepository(_s) _task = _r.get(task_id) if _task and _task.auto_retry_enabled and _task.auto_retry_max > 0: current_retry = _task.retry_count or 0 if current_retry < _task.auto_retry_max: logger.info( "[task_id=%s] 触发自动重试: 当前重试次数=%d, 最大重试次数=%d", task_id, current_retry, _task.auto_retry_max, ) # 计算退避延迟（指数退避，基础5s，最大60s） backoff_seconds = min(5 * (2**current_retry), 60) # 原地重试 _task.mark_pending_from_failed() _r.update(_task) # 延迟重新入队 celery_app.send_task( "worker.generate_video", args=[task_id], countdown=backoff_seconds, ) logger.info( "[task_id=%s] 自动重试已入队: 延迟=%ds, 第%d次重试", task_id, backoff_seconds, current_retry + 1, ) finally: _s.close() except Exception as retry_err: logger.warning( "[task_id=%s] 自动重试逻辑执行失败: %s", task_id, retry_err, exc_info=True, ) return { "status": "failed", "task_id": task_id, "error": str(error), }