xiaoxia
e5fffe40c1
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (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 Staging Web Image (push) Successful in 32s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 3m58s
CI/CD Pipeline / Validate Code Quality And Tests (push) Successful in 4m1s
CI/CD Pipeline / Frontend Lint (push) Successful in 4m28s
CI/CD Pipeline / Unit Tests (push) Failing after 5m2s
CI/CD Pipeline / Integration Tests (push) Successful in 1m55s
CI/CD Pipeline / Build Staging API Image (push) Successful in 6m51s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 8m33s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 44s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 34s
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
Co-authored-by: xiaoxia <dev@xiaoxiajianji.com> Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>
1652 lines
62 KiB
Python
1652 lines
62 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 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 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,
|
||
)
|
||
|
||
# ── 虚拟 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"
|
||
playback_speed: float = 1.0
|
||
status: str = "ready"
|
||
config: dict[str, Any] = field(default_factory=dict)
|
||
|
||
|
||
def _load_template_clip_configs(template_id: str) -> list:
|
||
"""从数据库读取模板的片段配置列表。
|
||
|
||
失败返回空列表，不阻断主流程。
|
||
"""
|
||
if not template_id:
|
||
return []
|
||
try:
|
||
from worker_app.db import SessionLocal
|
||
|
||
from packages.adapters.sqlalchemy_impl.template_clip_config_repository import (
|
||
SQLAlchemyTemplateClipConfigRepository,
|
||
)
|
||
|
||
session = SessionLocal()
|
||
try:
|
||
repo = SQLAlchemyTemplateClipConfigRepository(session)
|
||
configs = repo.list_by_template(template_id, limit=200)
|
||
logger.info("读取模板片段配置: template_id=%s count=%d", template_id, len(configs))
|
||
return configs
|
||
finally:
|
||
session.close()
|
||
except Exception as e:
|
||
logger.warning("读取模板片段配置失败，跳过效果层映射: template_id=%s error=%s", template_id, e)
|
||
return []
|
||
|
||
|
||
def _extract_intro_outro_from_clip_configs(clip_configs: list) -> dict[str, Any]:
|
||
"""从模板的 intro/outro 类型 clip_config 中提取 plan 级 intro_outro 配置。
|
||
|
||
UnifiedRenderService 已支持 plan.config.intro_outro 路径，
|
||
这里把 intro/outro 片段配置转为统一格式注入。
|
||
"""
|
||
intro_configs = [
|
||
c for c in clip_configs if (c.clip_type.value if hasattr(c.clip_type, "value") else c.clip_type) == "intro"
|
||
]
|
||
outro_configs = [
|
||
c for c in clip_configs if (c.clip_type.value if hasattr(c.clip_type, "value") else c.clip_type) == "outro"
|
||
]
|
||
|
||
result: dict[str, Any] = {}
|
||
|
||
if intro_configs:
|
||
intro = intro_configs[0]
|
||
intro_cfg = intro.config or {}
|
||
result["has_intro"] = True
|
||
result["intro_type"] = intro_cfg.get("intro_type", "text")
|
||
result["intro_duration"] = intro.default_duration or 3.0
|
||
if intro.text_template:
|
||
result["intro_text"] = intro.text_template
|
||
# 透传额外配置
|
||
for key in ("intro_text_color", "intro_bg_color", "intro_font_size", "intro_video_url", "intro_video_path"):
|
||
if key in intro_cfg:
|
||
result[key] = intro_cfg[key]
|
||
|
||
if outro_configs:
|
||
outro = outro_configs[0]
|
||
outro_cfg = outro.config or {}
|
||
result["has_outro"] = True
|
||
result["outro_type"] = outro_cfg.get("outro_type", "text")
|
||
result["outro_duration"] = outro.default_duration or 3.0
|
||
if outro.text_template:
|
||
result["outro_text"] = outro.text_template
|
||
for key in ("outro_text_color", "outro_bg_color", "outro_font_size", "outro_follow_text"):
|
||
if key in outro_cfg:
|
||
result[key] = outro_cfg[key]
|
||
|
||
return result
|
||
|
||
|
||
def _apply_template_clip_effects(
|
||
clips: list[_VirtualClip],
|
||
clip_configs: list,
|
||
mode: str,
|
||
) -> None:
|
||
"""将模板的 clip 级效果层映射到素材 clips 上（就地修改）。
|
||
|
||
映射规则：
|
||
- 只对素材主体 clips 做映射（ONE_TAKE: main, PIP: main+overlay, VOICE_OVER: main, VOICE_PIP: background+b_roll）
|
||
- 从模板中筛选 main 类型的 clip_config 作为效果模板
|
||
- 素材 clips 按顺序循环匹配模板 clip_config（素材多的话重复使用最后一个模板配置）
|
||
- 映射字段：transition_effect, config.color_grade, config.speed
|
||
"""
|
||
if not clip_configs or not clips:
|
||
return
|
||
|
||
# 筛选 main 类型的模板配置（作为效果模板池）
|
||
main_configs = [
|
||
c
|
||
for c in clip_configs
|
||
if (c.clip_type.value if hasattr(c.clip_type, "value") else c.clip_type) in ("main", "showcase", "b_roll")
|
||
]
|
||
if not main_configs:
|
||
return
|
||
|
||
# 确定需要映射的素材 clips（排除 corner_voice 等特殊层）
|
||
target_clips = [c for c in clips if c.clip_type not in ("corner_voice",)]
|
||
|
||
for i, clip in enumerate(target_clips):
|
||
# 循环匹配：素材多了用最后一个模板配置
|
||
cfg_idx = min(i, len(main_configs) - 1)
|
||
template_cfg = main_configs[cfg_idx]
|
||
|
||
# 1. 转场效果
|
||
transition = (
|
||
template_cfg.transition_effect.value
|
||
if hasattr(template_cfg.transition_effect, "value")
|
||
else template_cfg.transition_effect
|
||
)
|
||
if transition and transition != "cut":
|
||
clip.transition_effect = transition
|
||
|
||
# 2. clip 级效果配置（滤镜、调速等）
|
||
template_clip_config = template_cfg.config or {}
|
||
if template_clip_config:
|
||
# 合并到 clip.config（保留已有配置如 role 等）
|
||
existing_config = clip.config or {}
|
||
# 需要从模板复制的效果层 key
|
||
effect_keys = ("color_grade", "speed", "playback_speed", "reverse", "chroma_key", "filter")
|
||
for key in effect_keys:
|
||
if key in template_clip_config:
|
||
existing_config[key] = template_clip_config[key]
|
||
clip.config = existing_config
|
||
|
||
# 3. 调速：同步到 clip.playback_speed 顶级字段（渲染引擎读此字段）
|
||
template_speed = template_clip_config.get("playback_speed") or template_clip_config.get("speed")
|
||
if template_speed:
|
||
try:
|
||
speed_val = float(template_speed)
|
||
if speed_val > 0:
|
||
clip.playback_speed = speed_val
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
|
||
def _build_plan_and_clips_from_task(
|
||
task_id: str,
|
||
downloaded_paths: list[Path],
|
||
mode: str,
|
||
template_id: 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],
|
||
)
|
||
)
|
||
|
||
# ── P1: 模板效果层映射 ──
|
||
if template_id:
|
||
clip_configs = _load_template_clip_configs(template_id)
|
||
if clip_configs:
|
||
# 1. clip级效果层（转场、滤镜、调速等）
|
||
_apply_template_clip_effects(clips, clip_configs, mode)
|
||
|
||
# 2. 片头片尾（从 intro/outro 类型 clip 提取 plan 级配置）
|
||
intro_outro_config = _extract_intro_outro_from_clip_configs(clip_configs)
|
||
if intro_outro_config:
|
||
plan_config = plan.config or {}
|
||
plan_config["intro_outro"] = intro_outro_config
|
||
plan.config = plan_config
|
||
logger.info(
|
||
"模板片头片尾配置已注入: has_intro=%s has_outro=%s",
|
||
intro_outro_config.get("has_intro", False),
|
||
intro_outro_config.get("has_outro", False),
|
||
)
|
||
|
||
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:
|
||
"""下载配音文件。
|
||
|
||
支持两种来源（按优先级）：
|
||
1. 配音素材库 asset — 将 voice_library_id 当 asset_id 查 asset 表，
|
||
找到则用 asset.storage_key 下载（用户上传到配音库的音频）
|
||
2. 旧版 voice/{id}.mp3 路径 — 向后兼容
|
||
"""
|
||
if not voice_library_id:
|
||
return False
|
||
|
||
# 方式1：先尝试当 asset_id 查素材库（用户上传到配音库的音频）
|
||
try:
|
||
from worker_app.db import SessionLocal
|
||
|
||
from packages.adapters.sqlalchemy_impl.asset_repository import (
|
||
SQLAlchemyAssetRepository,
|
||
)
|
||
|
||
session = SessionLocal()
|
||
try:
|
||
repo = SQLAlchemyAssetRepository(session)
|
||
asset = repo.find_by_id(voice_library_id)
|
||
if asset and asset.storage_key:
|
||
# 是素材库的配音 asset，用 storage_key 下载
|
||
logger.info(
|
||
"配音素材来自素材库: asset_id=%s storage_key=%s",
|
||
voice_library_id,
|
||
asset.storage_key,
|
||
)
|
||
ok = download_asset(asset.storage_key, local_path)
|
||
if ok and local_path.exists() and local_path.stat().st_size > 0:
|
||
return True
|
||
finally:
|
||
session.close()
|
||
except Exception as e:
|
||
logger.warning("查询配音asset失败，fallback旧路径: %s", e)
|
||
|
||
# 方式2：旧版路径（向后兼容）
|
||
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.storage_key or model.file_url):
|
||
# 兼容存量数据：storage_key 为空时 fallback 到 file_url
|
||
storage_key = model.storage_key or 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 是否存在且可用。
|
||
|
||
优先读新模板系统（EditTemplate），找不到 fallback 到旧模板系统（TemplateModel）。
|
||
|
||
Raises:
|
||
ValueError: template_id 不存在或已禁用时抛出
|
||
"""
|
||
from packages.adapters.sqlalchemy_impl import (
|
||
SQLAlchemyEditTemplateRepository,
|
||
)
|
||
|
||
session = SessionLocal()
|
||
try:
|
||
# 优先读新模板系统
|
||
new_repo = SQLAlchemyEditTemplateRepository(session)
|
||
new_template = new_repo.get(template_id)
|
||
if new_template is not None:
|
||
status_val = new_template.status.value if hasattr(new_template.status, "value") else new_template.status
|
||
if status_val == "active":
|
||
logger.info("模板校验通过(新系统): template_id=%s name=%s", template_id, new_template.name)
|
||
return
|
||
else:
|
||
raise ValueError(f"模板已停用: template_id={template_id}")
|
||
|
||
# fallback: 旧模板系统
|
||
from packages.adapters.sqlalchemy_impl.models import TemplateModel
|
||
|
||
template = (
|
||
session.query(TemplateModel)
|
||
.filter(
|
||
TemplateModel.id == template_id,
|
||
TemplateModel.is_active.is_(True),
|
||
)
|
||
.first()
|
||
)
|
||
if template:
|
||
logger.info("模板校验通过(旧系统): template_id=%s name=%s", template_id, template.name)
|
||
return
|
||
|
||
raise ValueError(f"模板不存在或已禁用: template_id={template_id}")
|
||
finally:
|
||
session.close()
|
||
|
||
|
||
def _load_template_plan_config(template_id: str) -> dict:
|
||
"""从模板加载 plan 级配置（BGM、字幕、标题等效果层）。
|
||
|
||
优先读新模板系统（EditTemplate.config + TemplateClipConfig），
|
||
找不到 fallback 到旧模板系统（TemplateModel 独立字段）。
|
||
|
||
模板不存在时返回空 dict，不阻塞主流程。
|
||
"""
|
||
if not template_id:
|
||
return {}
|
||
try:
|
||
from packages.adapters.sqlalchemy_impl import (
|
||
SQLAlchemyEditTemplateRepository,
|
||
SQLAlchemyTemplateClipConfigRepository,
|
||
)
|
||
|
||
session = SessionLocal()
|
||
try:
|
||
# 优先读新模板系统
|
||
tpl_repo = SQLAlchemyEditTemplateRepository(session)
|
||
clip_repo = SQLAlchemyTemplateClipConfigRepository(session)
|
||
template = tpl_repo.get(template_id)
|
||
|
||
if template is not None:
|
||
# 新系统：config 直接就是 plan.config 格式
|
||
plan_config = dict(template.config or {})
|
||
|
||
# 从片段配置中提取 intro/outro 配置
|
||
clip_configs = clip_repo.list_by_template(template_id, limit=200)
|
||
if clip_configs:
|
||
intro_outro = _extract_intro_outro_from_clip_configs(clip_configs)
|
||
if intro_outro:
|
||
plan_config["intro_outro"] = intro_outro
|
||
|
||
# 把 editing_mode 也带过去
|
||
if template.editing_mode:
|
||
plan_config["editing_mode"] = template.editing_mode
|
||
|
||
logger.info(
|
||
"模板配置加载成功(新系统): template_id=%s keys=%s",
|
||
template_id,
|
||
list(plan_config.keys()),
|
||
)
|
||
return plan_config
|
||
|
||
# fallback: 旧模板系统
|
||
from packages.adapters.sqlalchemy_impl.models import TemplateModel
|
||
|
||
template = (
|
||
session.query(TemplateModel)
|
||
.filter(
|
||
TemplateModel.id == template_id,
|
||
TemplateModel.is_active.is_(True),
|
||
)
|
||
.first()
|
||
)
|
||
if template is None:
|
||
logger.warning("模板不存在，跳过配置加载: template_id=%s", template_id)
|
||
return {}
|
||
|
||
# 从独立字段组装成 plan.config 格式
|
||
plan_config: dict[str, Any] = {}
|
||
title_cfg = template.title_config or {}
|
||
subtitle_cfg = template.subtitle_config or {}
|
||
bgm_cfg = template.bgm_config or {}
|
||
|
||
if title_cfg:
|
||
plan_config["title"] = title_cfg
|
||
if subtitle_cfg:
|
||
plan_config["subtitle"] = subtitle_cfg
|
||
if bgm_cfg:
|
||
plan_config["bgm"] = bgm_cfg
|
||
|
||
logger.info(
|
||
"模板配置加载成功(旧系统): template_id=%s keys=%s",
|
||
template_id,
|
||
list(plan_config.keys()),
|
||
)
|
||
return plan_config
|
||
finally:
|
||
session.close()
|
||
except Exception as e:
|
||
logger.warning("加载模板配置失败，跳过: template_id=%s err=%s", template_id, e)
|
||
return {}
|
||
|
||
|
||
# ── 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 "",
|
||
"video_title": getattr(gen_task, "video_title", "") 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]:
|
||
"""渲染视频（含配音混音）。
|
||
|
||
使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/缩略图逻辑。
|
||
|
||
Returns:
|
||
(output_path, render_duration)
|
||
"""
|
||
if not downloaded_videos:
|
||
raise RuntimeError(f"素材下载结果为空: task_id={task_id}")
|
||
|
||
# 构建虚拟 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,
|
||
template_id=template_id,
|
||
)
|
||
|
||
# 注入模板配置（BGM、字幕等效果层）
|
||
if template_id:
|
||
template_config = _load_template_plan_config(template_id)
|
||
if template_config:
|
||
base_config = virtual_plan.config or {}
|
||
virtual_plan.config = {**template_config, **base_config}
|
||
logger.info(
|
||
"[task_id=%s] [渲染] 模板配置已注入: keys=%s",
|
||
task_id,
|
||
list(template_config.keys()),
|
||
)
|
||
|
||
# 确保输出分辨率配置存在（一键生成默认横屏 1280x720）
|
||
# RenderAdapter 从 plan.config.export.resolution 读取，
|
||
# 如果模板没有配置则用默认值，这里显式设置保持和旧逻辑一致
|
||
plan_cfg = virtual_plan.config or {}
|
||
export_cfg = plan_cfg.get("export", {}) or {}
|
||
if not export_cfg.get("resolution"):
|
||
export_cfg["resolution"] = f"{OUTPUT_WIDTH}x{OUTPUT_HEIGHT}"
|
||
plan_cfg["export"] = export_cfg
|
||
virtual_plan.config = plan_cfg
|
||
|
||
total_duration = sum(c.duration for c in virtual_clips)
|
||
logger.info(
|
||
"[task_id=%s] [剪辑计划] 片段数=%d, 总时长=%.1fs",
|
||
task_id,
|
||
len(virtual_clips),
|
||
total_duration,
|
||
)
|
||
|
||
render_start = time.monotonic()
|
||
logger.info("[task_id=%s] [渲染] RenderAdapter 统一渲染开始", task_id)
|
||
|
||
# 使用 RenderAdapter 统一渲染入口（复用 BGM/ASR/分辨率/缩略图逻辑）
|
||
from video_processing.render_adapter import RenderAdapter
|
||
from worker_app.db import SessionLocal
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
adapter = RenderAdapter(db)
|
||
render_result = adapter.render_from_memory(
|
||
plan=virtual_plan,
|
||
clips=virtual_clips,
|
||
asset_path_map=asset_path_map,
|
||
plan_id=f"gen_{task_id}",
|
||
job_id=task_id,
|
||
work_dir=temp_path,
|
||
)
|
||
finally:
|
||
db.close()
|
||
|
||
if not render_result.success:
|
||
raise RuntimeError(f"渲染失败: {render_result.error_message}")
|
||
|
||
render_output_path = render_result.output_path
|
||
render_duration = render_result.duration
|
||
|
||
render_elapsed = time.monotonic() - render_start
|
||
logger.info(
|
||
"[task_id=%s] [渲染] RenderAdapter 完成: 耗时=%.1fs, 时长=%.2fs",
|
||
task_id,
|
||
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,
|
||
user_id: str = "",
|
||
video_name: str = "",
|
||
) -> 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,
|
||
user_id=user_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,
|
||
name=video_name,
|
||
)
|
||
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)
|
||
|
||
# 调度时清理一次孤儿任务（其他 worker 崩溃留下的 running 任务）
|
||
try:
|
||
from worker_app.tasks._startup import cleanup_orphan_tasks
|
||
|
||
orphan_count = cleanup_orphan_tasks()
|
||
if orphan_count > 0:
|
||
logger.info("[task_id=%s] 调度前清理了 %d 个孤儿任务", task_id, orphan_count)
|
||
except Exception:
|
||
pass
|
||
|
||
# ── 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,
|
||
user_id=user_id,
|
||
video_name=task_info.get("video_title", ""),
|
||
)
|
||
|
||
if gen_task:
|
||
gen_task.append_log(
|
||
"OSS上传",
|
||
f"上传成功, 大小={file_size}",
|
||
file_size=file_size,
|
||
file_url=file_url,
|
||
)
|
||
_flush_logs(task_id, gen_task)
|
||
|
||
# ── 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( # type: ignore[misc]
|
||
"任务失败",
|
||
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),
|
||
}
|