CI Bot
f8a4c59cd0
- 三个入口(compose_video/edit_plan/generate_video)都增加flag配置详情日志 （enabled/percentage/whitelist/default_engine），灰度期间便于排查 - 统一fallback行为：异常时全部fallback到legacy（保守策略） 之前generate_video异常时fallback到unified，与其他入口不一致
1492 lines
54 KiB
Python
Executable File
1492 lines
54 KiB
Python
Executable File
"""
|
||
视频生成任务 — 使用 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()
|
||
engine = resolver.get_engine(user_id=user_id)
|
||
# 灰度期间打印详细 flag 配置，便于排查
|
||
config = resolver.get_config_snapshot()
|
||
logger.info(
|
||
"[渲染引擎] flag 解析: user_id=%s engine=%s enabled=%s percentage=%s whitelist=%d default=%s",
|
||
user_id,
|
||
engine,
|
||
config.get("enabled"),
|
||
config.get("percentage"),
|
||
len(config.get("whitelist", [])),
|
||
config.get("default_engine"),
|
||
)
|
||
return engine
|
||
except Exception as exc:
|
||
# 异常时 fallback 到 legacy（保守策略，与 edit_plan_generation 一致）
|
||
logger.warning("获取渲染引擎配置失败，fallback 到 legacy: %s", exc, exc_info=True)
|
||
return ENGINE_LEGACY
|
||
|
||
|
||
# ── 旧引擎渲染（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),
|
||
}
|