xiaoxia
c48ddeef7d
CI/CD Pipeline / Validate Code Quality And Tests (push) Successful in 1m27s
CI/CD Pipeline / Unit Tests (push) Successful in 2m12s
CI/CD Pipeline / Integration Tests (push) Successful in 3m23s
CI/CD Pipeline / Frontend Lint (push) Successful in 3m27s
CI/CD Pipeline / Build Production Runtime Images (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (push) Failing after 5s
CI/CD Pipeline / Staging E2E Tests (push) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (push) Has been skipped
1077 lines
40 KiB
Python
1077 lines
40 KiB
Python
"""
|
||
视频生成任务 — 使用 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 json
|
||
import logging
|
||
import os
|
||
import tempfile
|
||
import time
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any, Optional
|
||
|
||
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 _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 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] = []
|
||
n = len(downloaded_paths)
|
||
|
||
if mode == "pip":
|
||
# 1 main + N-1 overlay
|
||
for i, p in enumerate(downloaded_paths):
|
||
clip_type = "main" if i == 0 else "overlay"
|
||
clips.append(
|
||
_VirtualClip(
|
||
id=f"vc_{i:03d}",
|
||
plan_id=task_id,
|
||
clip_type=clip_type,
|
||
order=i,
|
||
asset_id=path_to_asset_id[p],
|
||
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 _verify_url_accessible(url: str, timeout: float = 10.0, retries: int = 2) -> bool:
|
||
"""HEAD 请求校验 URL 可访问（含重试，防止 OSS 抖动误报）。
|
||
|
||
Args:
|
||
url: 待校验的 URL
|
||
timeout: 单次请求超时时间（秒）
|
||
retries: 最大重试次数（默认 2 次，首次失败后间隔 1s 重试）
|
||
|
||
Returns:
|
||
True 表示 URL 可访问（HTTP 2xx/3xx），False 表示所有尝试均失败。
|
||
"""
|
||
import time
|
||
import urllib.request
|
||
|
||
last_error: Exception | None = None
|
||
for attempt in range(1 + retries):
|
||
try:
|
||
req = urllib.request.Request(url, method="HEAD")
|
||
req.add_header("User-Agent", "xiaoxia-saas-worker/1.0")
|
||
with urllib.request.urlopen(req, timeout=timeout) as resp: # nosec B310
|
||
if resp.status < 400:
|
||
return True
|
||
last_error = Exception(f"HTTP {resp.status}")
|
||
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,
|
||
video_extensions: tuple = (".mp4", ".mov", ".avi", ".mkv", ".webm"),
|
||
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 视频素材
|
||
video_extensions: 支持的视频扩展名（保留兼容，当前按 file_type 过滤）
|
||
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
|
||
|
||
|
||
# ── 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.domain import EditingMode
|
||
|
||
logger.info("[task_id=%s] [接收任务] 开始生成视频任务", task_id)
|
||
|
||
# 从数据库加载任务信息
|
||
session = SessionLocal()
|
||
try:
|
||
from packages.adapters.sqlalchemy_impl.generation_task_repository import (
|
||
SQLAlchemyGenerationTaskRepository,
|
||
)
|
||
|
||
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 ""
|
||
|
||
# 记录接收任务日志
|
||
gen_task.append_log(
|
||
"接收任务",
|
||
f"模式={mode}, 模板={template_id}, 素材数={len(task_asset_ids)}",
|
||
mode=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)
|
||
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(
|
||
temp_path,
|
||
asset_library_id=asset_library_id,
|
||
project_id=project_id,
|
||
asset_ids=task_asset_ids or None,
|
||
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(
|
||
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,
|
||
)
|
||
|
||
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),
|
||
)
|
||
_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_file_size = render_result.file_size
|
||
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,
|
||
)
|
||
|
||
if gen_task:
|
||
gen_task.append_log(
|
||
"OSS上传",
|
||
f"上传成功, 大小={file_size}, 耗时={upload_elapsed:.1f}s",
|
||
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()
|
||
|
||
# 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,
|
||
)
|
||
_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)
|
||
|
||
# 记录失败日志
|
||
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__,
|
||
)
|
||
_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))
|
||
return {
|
||
"status": "failed",
|
||
"task_id": task_id,
|
||
"error": str(error),
|
||
}
|