847 lines
35 KiB
Python
847 lines
35 KiB
Python
from typing import Optional, Any, List
|
||
from pathlib import Path
|
||
import asyncio
|
||
import time
|
||
import traceback
|
||
import httpx
|
||
from loguru import logger
|
||
|
||
from app.core.config import settings
|
||
from app.services.tts_service import TTSService
|
||
from app.services.video_service import VideoService
|
||
from app.services.lipsync_service import LipSyncService
|
||
from app.services.voice_clone_service import voice_clone_service
|
||
from app.services.assets_service import (
|
||
get_style,
|
||
get_default_style,
|
||
resolve_bgm_path,
|
||
prepare_style_for_remotion,
|
||
)
|
||
from app.services.storage import storage_service
|
||
from app.services.whisper_service import whisper_service
|
||
from app.services.remotion_service import remotion_service
|
||
|
||
from .schemas import GenerateRequest
|
||
from .task_store import task_store
|
||
|
||
# 全局并发限制：最多同时运行 2 个视频生成任务
|
||
_generation_semaphore = asyncio.Semaphore(2)
|
||
|
||
|
||
def _locale_to_whisper_lang(locale: str) -> str:
|
||
"""'en-US' → 'en', 'zh-CN' → 'zh'"""
|
||
return locale.split("-")[0] if "-" in locale else locale
|
||
|
||
|
||
def _locale_to_tts_lang(locale: str) -> str:
|
||
"""'zh-CN' → 'Chinese', 'en-US' → 'English', 其他 → 'Auto'"""
|
||
mapping = {"zh": "Chinese", "en": "English"}
|
||
return mapping.get(locale.split("-")[0], "Auto")
|
||
|
||
|
||
_lipsync_service: Optional[LipSyncService] = None
|
||
_lipsync_ready: Optional[bool] = None
|
||
_lipsync_last_check: float = 0
|
||
|
||
|
||
def _get_lipsync_service() -> LipSyncService:
|
||
"""获取或创建 LipSync 服务实例（单例模式，避免重复初始化）"""
|
||
global _lipsync_service
|
||
if _lipsync_service is None:
|
||
_lipsync_service = LipSyncService()
|
||
return _lipsync_service
|
||
|
||
|
||
async def _check_lipsync_ready(force: bool = False) -> bool:
|
||
"""检查 LipSync 是否就绪（带缓存，5分钟内不重复检查）"""
|
||
global _lipsync_ready, _lipsync_last_check
|
||
|
||
now = time.time()
|
||
if not force and _lipsync_ready is not None and (now - _lipsync_last_check) < 300:
|
||
return bool(_lipsync_ready)
|
||
|
||
lipsync = _get_lipsync_service()
|
||
health = await lipsync.check_health()
|
||
_lipsync_ready = health.get("ready", False)
|
||
_lipsync_last_check = now
|
||
print(f"[LipSync] Health check: ready={_lipsync_ready}")
|
||
return bool(_lipsync_ready)
|
||
|
||
|
||
async def _download_material(path_or_url: str, temp_path: Path):
|
||
"""下载素材到临时文件 (流式下载，节省内存)"""
|
||
if path_or_url.startswith("http"):
|
||
timeout = httpx.Timeout(None)
|
||
async with httpx.AsyncClient(timeout=timeout) as client:
|
||
async with client.stream("GET", path_or_url) as resp:
|
||
resp.raise_for_status()
|
||
with open(temp_path, "wb") as f:
|
||
async for chunk in resp.aiter_bytes():
|
||
f.write(chunk)
|
||
else:
|
||
src = Path(path_or_url)
|
||
if not src.is_absolute():
|
||
src = settings.BASE_DIR.parent / path_or_url
|
||
|
||
if src.exists():
|
||
import shutil
|
||
shutil.copy(src, temp_path)
|
||
else:
|
||
raise FileNotFoundError(f"Material not found: {path_or_url}")
|
||
|
||
|
||
def _update_task(task_id: str, **updates: Any) -> None:
|
||
task_store.update(task_id, updates)
|
||
|
||
|
||
async def _run_blocking(func, *args):
|
||
"""在线程池执行阻塞函数，避免卡住事件循环。"""
|
||
loop = asyncio.get_running_loop()
|
||
return await loop.run_in_executor(None, func, *args)
|
||
|
||
|
||
# ── 多素材辅助函数 ──
|
||
|
||
|
||
def _split_equal(segments: List[dict], material_paths: List[str]) -> List[dict]:
|
||
"""按素材数量均分音频时长，对齐到最近的 Whisper 字边界。
|
||
|
||
Args:
|
||
segments: Whisper 产出的 segment 列表, 每个包含 words (字级时间戳)
|
||
material_paths: 素材路径列表
|
||
|
||
Returns:
|
||
[{"material_path": "...", "start": 0.0, "end": 5.2, "index": 0}, ...]
|
||
"""
|
||
# 展平所有 Whisper 字符
|
||
all_chars: List[dict] = []
|
||
for seg in segments:
|
||
for w in seg.get("words", []):
|
||
all_chars.append(w)
|
||
|
||
n = len(material_paths)
|
||
|
||
if not all_chars or n == 0:
|
||
return [{"material_path": material_paths[0] if material_paths else "",
|
||
"start": 0.0, "end": 99999.0, "index": 0}]
|
||
|
||
# 素材数不能超过字符数，否则边界会重复
|
||
if n > len(all_chars):
|
||
logger.warning(f"[MultiMat] 素材数({n}) > 字符数({len(all_chars)})，裁剪为 {len(all_chars)}")
|
||
n = len(all_chars)
|
||
|
||
total_start = all_chars[0]["start"]
|
||
total_end = all_chars[-1]["end"]
|
||
seg_dur = (total_end - total_start) / n
|
||
|
||
# 计算 N-1 个分割点，对齐到最近的字边界
|
||
boundaries = [0] # 第一段从第 0 个字开始
|
||
for i in range(1, n):
|
||
target_time = total_start + i * seg_dur
|
||
# 找到 start 时间最接近 target_time 的字
|
||
best_idx = boundaries[-1] + 1 # 至少比上一个边界后移 1
|
||
best_diff = float("inf")
|
||
for j in range(boundaries[-1] + 1, len(all_chars)):
|
||
diff = abs(all_chars[j]["start"] - target_time)
|
||
if diff < best_diff:
|
||
best_diff = diff
|
||
best_idx = j
|
||
elif diff > best_diff:
|
||
break # 时间递增，差值开始变大后可以停了
|
||
boundaries.append(min(best_idx, len(all_chars) - 1))
|
||
boundaries.append(len(all_chars)) # 最后一段到末尾
|
||
|
||
# 按边界生成分配结果
|
||
assignments: List[dict] = []
|
||
for i in range(n):
|
||
s_idx = boundaries[i]
|
||
e_idx = boundaries[i + 1]
|
||
if s_idx >= len(all_chars) or s_idx >= e_idx:
|
||
continue
|
||
assignments.append({
|
||
"material_path": material_paths[i],
|
||
"start": all_chars[s_idx]["start"],
|
||
"end": all_chars[e_idx - 1]["end"],
|
||
"text": "".join(c["word"] for c in all_chars[s_idx:e_idx]),
|
||
"index": len(assignments),
|
||
})
|
||
|
||
if not assignments:
|
||
return [{"material_path": material_paths[0], "start": 0.0, "end": 99999.0, "index": 0}]
|
||
|
||
logger.info(f"[MultiMat] 均分 {len(all_chars)} 字为 {len(assignments)} 段")
|
||
for a in assignments:
|
||
dur = a["end"] - a["start"]
|
||
logger.info(f" 段{a['index']}: [{a['start']:.2f}-{a['end']:.2f}s] ({dur:.1f}s) {a['text'][:20]}")
|
||
|
||
return assignments
|
||
|
||
|
||
async def process_video_generation(task_id: str, req: GenerateRequest, user_id: str):
|
||
_update_task(task_id, message="排队中...")
|
||
async with _generation_semaphore:
|
||
await _process_video_generation_inner(task_id, req, user_id)
|
||
|
||
|
||
async def _process_video_generation_inner(task_id: str, req: GenerateRequest, user_id: str):
|
||
temp_files = []
|
||
try:
|
||
start_time = time.time()
|
||
|
||
# ── 确定素材列表 ──
|
||
material_paths: List[str] = []
|
||
if req.custom_assignments and len(req.custom_assignments) > 1:
|
||
material_paths = [a.material_path for a in req.custom_assignments if a.material_path]
|
||
elif req.material_paths and len(req.material_paths) > 1:
|
||
material_paths = req.material_paths
|
||
else:
|
||
material_paths = [req.material_path]
|
||
|
||
is_multi = len(material_paths) > 1
|
||
target_resolution = (1080, 1920) if req.output_aspect_ratio == "9:16" else (1920, 1080)
|
||
|
||
logger.info(
|
||
f"[Render] 输出画面比例: {req.output_aspect_ratio}, "
|
||
f"目标分辨率: {target_resolution[0]}x{target_resolution[1]}"
|
||
)
|
||
|
||
_update_task(task_id, status="processing", progress=5, message="正在下载素材...")
|
||
|
||
temp_dir = settings.UPLOAD_DIR / "temp"
|
||
temp_dir.mkdir(parents=True, exist_ok=True)
|
||
video = VideoService()
|
||
input_material_path: Optional[Path] = None
|
||
|
||
# 单素材模式：下载主素材
|
||
if not is_multi:
|
||
input_material_path = temp_dir / f"{task_id}_input.mp4"
|
||
temp_files.append(input_material_path)
|
||
await _download_material(material_paths[0], input_material_path)
|
||
|
||
# 归一化旋转元数据（如 iPhone MOV 1920x1080 + rotation=-90）
|
||
normalized_input_path = temp_dir / f"{task_id}_input_norm.mp4"
|
||
normalized_result = await _run_blocking(
|
||
video.normalize_orientation,
|
||
str(input_material_path),
|
||
str(normalized_input_path),
|
||
)
|
||
if normalized_result != str(input_material_path):
|
||
temp_files.append(normalized_input_path)
|
||
input_material_path = normalized_input_path
|
||
|
||
_update_task(task_id, message="正在生成语音...", progress=10)
|
||
|
||
audio_path = temp_dir / f"{task_id}_audio.wav"
|
||
temp_files.append(audio_path)
|
||
|
||
if req.generated_audio_id:
|
||
# 新流程：使用预生成的配音
|
||
_update_task(task_id, message="正在下载配音...", progress=12)
|
||
audio_url = await storage_service.get_signed_url(
|
||
bucket="generated-audios",
|
||
path=req.generated_audio_id,
|
||
)
|
||
await _download_material(audio_url, audio_path)
|
||
|
||
# 从元数据获取 language
|
||
meta_path = req.generated_audio_id.replace("_audio.wav", "_audio.json")
|
||
try:
|
||
meta_url = await storage_service.get_signed_url(
|
||
bucket="generated-audios", path=meta_path,
|
||
)
|
||
import httpx as _httpx
|
||
async with _httpx.AsyncClient(timeout=5.0) as client:
|
||
resp = await client.get(meta_url)
|
||
if resp.status_code == 200:
|
||
meta = resp.json()
|
||
req.language = meta.get("language", req.language)
|
||
# 无条件用配音元数据覆盖文案，确保字幕与配音语言一致
|
||
meta_text = meta.get("text", "")
|
||
if meta_text:
|
||
req.text = meta_text
|
||
except Exception as e:
|
||
logger.warning(f"读取配音元数据失败: {e}")
|
||
|
||
elif req.tts_mode == "voiceclone":
|
||
if not req.ref_audio_id or not req.ref_text:
|
||
raise ValueError("声音克隆模式需要提供参考音频和参考文字")
|
||
|
||
_update_task(task_id, message="正在下载参考音频...")
|
||
|
||
ref_audio_local = temp_dir / f"{task_id}_ref.wav"
|
||
temp_files.append(ref_audio_local)
|
||
|
||
ref_audio_url = await storage_service.get_signed_url(
|
||
bucket="ref-audios",
|
||
path=req.ref_audio_id
|
||
)
|
||
await _download_material(ref_audio_url, ref_audio_local)
|
||
|
||
_update_task(task_id, message="正在克隆声音...")
|
||
await voice_clone_service.generate_audio(
|
||
text=req.text,
|
||
ref_audio_path=str(ref_audio_local),
|
||
ref_text=req.ref_text,
|
||
output_path=str(audio_path),
|
||
language=_locale_to_tts_lang(req.language)
|
||
)
|
||
else:
|
||
_update_task(task_id, message="正在生成语音 (EdgeTTS)...")
|
||
tts = TTSService()
|
||
await tts.generate_audio(req.text, req.voice, str(audio_path))
|
||
|
||
tts_time = time.time() - start_time
|
||
print(f"[Pipeline] TTS completed in {tts_time:.1f}s")
|
||
|
||
lipsync = _get_lipsync_service()
|
||
lipsync_video_path = temp_dir / f"{task_id}_lipsync.mp4"
|
||
temp_files.append(lipsync_video_path)
|
||
|
||
captions_path = None
|
||
|
||
async def _whisper_and_split():
|
||
"""Whisper 对齐 → _split_equal 均分素材（公共逻辑）"""
|
||
_update_task(task_id, message="正在生成字幕 (Whisper)...")
|
||
_captions_path = temp_dir / f"{task_id}_captions.json"
|
||
temp_files.append(_captions_path)
|
||
captions_data = None
|
||
try:
|
||
captions_data = await whisper_service.align(
|
||
audio_path=str(audio_path),
|
||
text=req.text,
|
||
output_path=str(_captions_path),
|
||
language=_locale_to_whisper_lang(req.language),
|
||
original_text=req.text,
|
||
)
|
||
print(f"[Pipeline] Whisper alignment completed (multi-material)")
|
||
except Exception as e:
|
||
logger.warning(f"Whisper alignment failed: {e}")
|
||
_captions_path = None
|
||
|
||
_update_task(task_id, progress=15, message="正在分配素材...")
|
||
|
||
if captions_data and captions_data.get("segments"):
|
||
result = _split_equal(captions_data["segments"], material_paths)
|
||
else:
|
||
logger.warning("[MultiMat] Whisper 无数据，按时长均分")
|
||
audio_dur = await _run_blocking(video._get_duration, str(audio_path))
|
||
if audio_dur <= 0:
|
||
audio_dur = 30.0
|
||
seg_dur = audio_dur / len(material_paths)
|
||
result = [
|
||
{"material_path": material_paths[i], "start": i * seg_dur,
|
||
"end": (i + 1) * seg_dur, "index": i}
|
||
for i in range(len(material_paths))
|
||
]
|
||
return result, _captions_path
|
||
|
||
if is_multi:
|
||
# ══════════════════════════════════════
|
||
# 多素材流水线
|
||
# ══════════════════════════════════════
|
||
_update_task(task_id, progress=12, message="正在分配素材...")
|
||
|
||
if req.custom_assignments and len(req.custom_assignments) == len(material_paths):
|
||
# 用户自定义分配，跳过 Whisper 均分
|
||
assignments = [
|
||
{
|
||
"material_path": a.material_path,
|
||
"start": a.start,
|
||
"end": a.end,
|
||
"source_start": a.source_start,
|
||
"source_end": a.source_end,
|
||
"index": i,
|
||
}
|
||
for i, a in enumerate(req.custom_assignments)
|
||
]
|
||
# 仍然需要 Whisper 生成字幕（如果启用）
|
||
captions_path = temp_dir / f"{task_id}_captions.json"
|
||
temp_files.append(captions_path)
|
||
if req.enable_subtitles:
|
||
_update_task(task_id, message="正在生成字幕 (Whisper)...")
|
||
try:
|
||
await whisper_service.align(
|
||
audio_path=str(audio_path),
|
||
text=req.text,
|
||
output_path=str(captions_path),
|
||
language=_locale_to_whisper_lang(req.language),
|
||
original_text=req.text,
|
||
)
|
||
print(f"[Pipeline] Whisper alignment completed (custom assignments)")
|
||
except Exception as e:
|
||
logger.warning(f"Whisper alignment failed: {e}")
|
||
captions_path = None
|
||
else:
|
||
captions_path = None
|
||
elif req.custom_assignments:
|
||
logger.warning(
|
||
f"[MultiMat] custom_assignments 数量({len(req.custom_assignments)})"
|
||
f" 与素材数量({len(material_paths)})不一致，回退自动分配"
|
||
)
|
||
|
||
assignments, captions_path = await _whisper_and_split()
|
||
|
||
else:
|
||
assignments, captions_path = await _whisper_and_split()
|
||
|
||
# 扩展段覆盖完整音频范围：首段从0开始，末段到音频结尾
|
||
audio_duration = await _run_blocking(video._get_duration, str(audio_path))
|
||
if assignments and audio_duration > 0:
|
||
assignments[0]["start"] = 0.0
|
||
assignments[-1]["end"] = audio_duration
|
||
|
||
num_segments = len(assignments)
|
||
print(f"[Pipeline] Multi-material: {num_segments} segments, {len(material_paths)} materials")
|
||
|
||
if num_segments == 0:
|
||
raise RuntimeError("Multi-material: no valid segments after splitting")
|
||
|
||
lipsync_start = time.time()
|
||
|
||
# ── 第一步：并行下载所有素材并检测分辨率 ──
|
||
material_locals: List[Path] = []
|
||
resolutions = []
|
||
|
||
async def _download_and_normalize(i: int, assignment: dict):
|
||
"""下载单个素材并归一化方向"""
|
||
material_local = temp_dir / f"{task_id}_material_{i}.mp4"
|
||
temp_files.append(material_local)
|
||
await _download_material(assignment["material_path"], material_local)
|
||
|
||
normalized_material = temp_dir / f"{task_id}_material_{i}_norm.mp4"
|
||
normalized_result = await _run_blocking(
|
||
video.normalize_orientation,
|
||
str(material_local),
|
||
str(normalized_material),
|
||
)
|
||
if normalized_result != str(material_local):
|
||
temp_files.append(normalized_material)
|
||
material_local = normalized_material
|
||
|
||
res = video.get_resolution(str(material_local))
|
||
return material_local, res
|
||
|
||
download_tasks = [
|
||
_download_and_normalize(i, assignment)
|
||
for i, assignment in enumerate(assignments)
|
||
]
|
||
download_results = await asyncio.gather(*download_tasks)
|
||
for local, res in download_results:
|
||
material_locals.append(local)
|
||
resolutions.append(res)
|
||
|
||
# 按用户选择的画面比例统一分辨率
|
||
base_res = target_resolution
|
||
need_scale = any(r != base_res for r in resolutions)
|
||
if need_scale:
|
||
logger.info(f"[MultiMat] 素材分辨率不一致，统一到 {base_res[0]}x{base_res[1]}")
|
||
|
||
# ── 第二步：并行裁剪每段素材到对应时长 ──
|
||
prepared_segments: List[Optional[Path]] = [None] * num_segments
|
||
|
||
async def _prepare_one_segment(i: int, assignment: dict):
|
||
"""将单个素材裁剪/循环到对应时长"""
|
||
seg_dur = assignment["end"] - assignment["start"]
|
||
prepared_path = temp_dir / f"{task_id}_prepared_{i}.mp4"
|
||
temp_files.append(prepared_path)
|
||
prepare_target_res = None if resolutions[i] == base_res else base_res
|
||
|
||
await _run_blocking(
|
||
video.prepare_segment,
|
||
str(material_locals[i]),
|
||
seg_dur,
|
||
str(prepared_path),
|
||
prepare_target_res,
|
||
assignment.get("source_start", 0.0),
|
||
assignment.get("source_end"),
|
||
25,
|
||
)
|
||
return i, prepared_path
|
||
|
||
_update_task(
|
||
task_id,
|
||
progress=15,
|
||
message=f"正在并行准备 {num_segments} 个素材片段..."
|
||
)
|
||
|
||
prepare_tasks = [
|
||
_prepare_one_segment(i, assignment)
|
||
for i, assignment in enumerate(assignments)
|
||
]
|
||
prepare_results = await asyncio.gather(*prepare_tasks)
|
||
for i, path in prepare_results:
|
||
prepared_segments[i] = path
|
||
|
||
# ── 第二步：拼接所有素材片段 ──
|
||
_update_task(task_id, progress=50, message="正在拼接素材片段...")
|
||
concat_path = temp_dir / f"{task_id}_concat.mp4"
|
||
temp_files.append(concat_path)
|
||
prepared_segment_paths = [str(p) for p in prepared_segments if p is not None]
|
||
if len(prepared_segment_paths) != num_segments:
|
||
raise RuntimeError("Multi-material: prepared segments mismatch")
|
||
await _run_blocking(
|
||
video.concat_videos,
|
||
prepared_segment_paths,
|
||
str(concat_path),
|
||
25,
|
||
)
|
||
|
||
# ── 第三步：一次 LatentSync 推理 ──
|
||
is_ready = await _check_lipsync_ready()
|
||
|
||
if is_ready:
|
||
_update_task(task_id, progress=55, message="正在合成唇形 (LatentSync)...")
|
||
print(f"[LipSync] Multi-material: single LatentSync on concatenated video")
|
||
try:
|
||
await lipsync.generate(
|
||
str(concat_path),
|
||
str(audio_path),
|
||
str(lipsync_video_path),
|
||
model_mode=req.lipsync_model,
|
||
)
|
||
except Exception as e:
|
||
logger.warning(f"[LipSync] Failed, fallback to concat without lipsync: {e}")
|
||
import shutil
|
||
shutil.copy(str(concat_path), str(lipsync_video_path))
|
||
else:
|
||
print(f"[LipSync] Not ready, using concatenated video without lipsync")
|
||
import shutil
|
||
shutil.copy(str(concat_path), str(lipsync_video_path))
|
||
|
||
lipsync_time = time.time() - lipsync_start
|
||
print(f"[Pipeline] Multi-material prepare + concat + LipSync completed in {lipsync_time:.1f}s")
|
||
_update_task(task_id, progress=80)
|
||
|
||
# 如果用户关闭了字幕，清除 captions_path（Whisper 仅用于句子切分）
|
||
if not req.enable_subtitles:
|
||
captions_path = None
|
||
|
||
else:
|
||
# ══════════════════════════════════════
|
||
# 单素材流水线（原有逻辑）
|
||
# ══════════════════════════════════════
|
||
|
||
if input_material_path is None:
|
||
raise RuntimeError("单素材流程缺少输入素材")
|
||
|
||
# 单素材：按用户选择画面比例统一到目标分辨率，并应用 source_start
|
||
single_source_start = 0.0
|
||
single_source_end = None
|
||
if req.custom_assignments and len(req.custom_assignments) == 1:
|
||
single_source_start = req.custom_assignments[0].source_start
|
||
single_source_end = req.custom_assignments[0].source_end
|
||
|
||
_update_task(task_id, progress=20, message="正在准备素材片段...")
|
||
audio_dur = await _run_blocking(video._get_duration, str(audio_path))
|
||
if audio_dur <= 0:
|
||
audio_dur = 30.0
|
||
single_res = await _run_blocking(video.get_resolution, str(input_material_path))
|
||
single_target_res = None if single_res == target_resolution else target_resolution
|
||
prepared_single_path = temp_dir / f"{task_id}_prepared_single.mp4"
|
||
temp_files.append(prepared_single_path)
|
||
await _run_blocking(
|
||
video.prepare_segment,
|
||
str(input_material_path),
|
||
audio_dur,
|
||
str(prepared_single_path),
|
||
single_target_res,
|
||
single_source_start,
|
||
single_source_end,
|
||
None,
|
||
)
|
||
input_material_path = prepared_single_path
|
||
|
||
_update_task(task_id, progress=25)
|
||
_update_task(task_id, message="正在合成唇形 (LatentSync)...", progress=30)
|
||
|
||
lipsync_start = time.time()
|
||
is_ready = await _check_lipsync_ready()
|
||
|
||
if is_ready:
|
||
print(f"[LipSync] Starting LatentSync inference...")
|
||
_update_task(task_id, progress=35, message="正在运行 LatentSync 推理...")
|
||
try:
|
||
await lipsync.generate(
|
||
str(input_material_path),
|
||
str(audio_path),
|
||
str(lipsync_video_path),
|
||
model_mode=req.lipsync_model,
|
||
)
|
||
except Exception as e:
|
||
logger.warning(f"[LipSync] Failed on single-material, fallback to prepared video: {e}")
|
||
_update_task(task_id, message="唇形同步失败，使用原始视频...")
|
||
import shutil
|
||
shutil.copy(str(input_material_path), str(lipsync_video_path))
|
||
else:
|
||
print(f"[LipSync] LatentSync not ready, copying original video")
|
||
_update_task(task_id, message="唇形同步不可用，使用原始视频...")
|
||
import shutil
|
||
shutil.copy(str(input_material_path), lipsync_video_path)
|
||
|
||
lipsync_time = time.time() - lipsync_start
|
||
print(f"[Pipeline] LipSync completed in {lipsync_time:.1f}s")
|
||
_update_task(task_id, progress=80)
|
||
|
||
# 单素材模式：Whisper 延迟到下方与 BGM 并行执行
|
||
if not req.enable_subtitles:
|
||
captions_path = None
|
||
|
||
_update_task(task_id, progress=85)
|
||
|
||
# ── Whisper 字幕 + BGM 混音 并行（两者都只依赖 audio_path）──
|
||
final_audio_path = audio_path
|
||
_whisper_task = None
|
||
_bgm_task = None
|
||
mix_output_path: Optional[Path] = None
|
||
|
||
# 单素材模式下 Whisper 尚未执行，这里与 BGM 并行启动
|
||
need_whisper = not is_multi and req.enable_subtitles and captions_path is None
|
||
if need_whisper:
|
||
captions_path = temp_dir / f"{task_id}_captions.json"
|
||
temp_files.append(captions_path)
|
||
_captions_path_str = str(captions_path)
|
||
|
||
async def _run_whisper():
|
||
_update_task(task_id, message="正在生成字幕 (Whisper)...", progress=82)
|
||
try:
|
||
await whisper_service.align(
|
||
audio_path=str(audio_path),
|
||
text=req.text,
|
||
output_path=_captions_path_str,
|
||
language=_locale_to_whisper_lang(req.language),
|
||
original_text=req.text,
|
||
)
|
||
print(f"[Pipeline] Whisper alignment completed")
|
||
return True
|
||
except Exception as e:
|
||
logger.warning(f"Whisper alignment failed, skipping subtitles: {e}")
|
||
return False
|
||
|
||
_whisper_task = _run_whisper()
|
||
|
||
if req.bgm_id:
|
||
bgm_path = resolve_bgm_path(req.bgm_id)
|
||
if bgm_path:
|
||
mix_output_path = temp_dir / f"{task_id}_audio_mix.wav"
|
||
temp_files.append(mix_output_path)
|
||
volume = req.bgm_volume if req.bgm_volume is not None else 0.2
|
||
volume = max(0.0, min(float(volume), 1.0))
|
||
_mix_output = str(mix_output_path)
|
||
_bgm_path = str(bgm_path)
|
||
_voice_path = str(audio_path)
|
||
_volume = volume
|
||
|
||
async def _run_bgm():
|
||
_update_task(task_id, message="正在合成背景音乐...", progress=86)
|
||
try:
|
||
await _run_blocking(
|
||
video.mix_audio,
|
||
_voice_path,
|
||
_bgm_path,
|
||
_mix_output,
|
||
_volume,
|
||
)
|
||
return True
|
||
except Exception as e:
|
||
logger.warning(f"BGM mix failed, fallback to voice only: {e}")
|
||
return False
|
||
|
||
_bgm_task = _run_bgm()
|
||
else:
|
||
logger.warning(f"BGM not found: {req.bgm_id}")
|
||
|
||
# 并行等待 Whisper + BGM
|
||
parallel_tasks = [t for t in (_whisper_task, _bgm_task) if t is not None]
|
||
if parallel_tasks:
|
||
results = await asyncio.gather(*parallel_tasks)
|
||
result_idx = 0
|
||
if _whisper_task is not None:
|
||
if not results[result_idx]:
|
||
captions_path = None
|
||
result_idx += 1
|
||
if _bgm_task is not None:
|
||
if results[result_idx] and mix_output_path is not None:
|
||
final_audio_path = mix_output_path
|
||
|
||
|
||
use_remotion = (captions_path and captions_path.exists()) or req.title or req.secondary_title
|
||
|
||
subtitle_style = None
|
||
title_style = None
|
||
secondary_title_style = None
|
||
if req.enable_subtitles:
|
||
subtitle_style = get_style("subtitle", req.subtitle_style_id) or get_default_style("subtitle")
|
||
if req.title:
|
||
title_style = get_style("title", req.title_style_id) or get_default_style("title")
|
||
if req.secondary_title:
|
||
secondary_title_style = get_style("title", req.secondary_title_style_id) or get_default_style("title")
|
||
|
||
if req.subtitle_font_size and req.enable_subtitles:
|
||
if subtitle_style is None:
|
||
subtitle_style = {}
|
||
subtitle_style["font_size"] = int(req.subtitle_font_size)
|
||
|
||
if req.title_font_size and req.title:
|
||
if title_style is None:
|
||
title_style = {}
|
||
title_style["font_size"] = int(req.title_font_size)
|
||
|
||
if req.title_top_margin is not None and req.title:
|
||
if title_style is None:
|
||
title_style = {}
|
||
title_style["top_margin"] = int(req.title_top_margin)
|
||
|
||
if req.subtitle_bottom_margin is not None and req.enable_subtitles:
|
||
if subtitle_style is None:
|
||
subtitle_style = {}
|
||
subtitle_style["bottom_margin"] = int(req.subtitle_bottom_margin)
|
||
|
||
if req.secondary_title_font_size and req.secondary_title:
|
||
if secondary_title_style is None:
|
||
secondary_title_style = {}
|
||
secondary_title_style["font_size"] = int(req.secondary_title_font_size)
|
||
|
||
if req.secondary_title_top_margin is not None and req.secondary_title:
|
||
if secondary_title_style is None:
|
||
secondary_title_style = {}
|
||
secondary_title_style["top_margin"] = int(req.secondary_title_top_margin)
|
||
|
||
if use_remotion:
|
||
subtitle_style = prepare_style_for_remotion(
|
||
subtitle_style,
|
||
temp_dir,
|
||
f"{task_id}_subtitle_font"
|
||
)
|
||
title_style = prepare_style_for_remotion(
|
||
title_style,
|
||
temp_dir,
|
||
f"{task_id}_title_font"
|
||
)
|
||
secondary_title_style = prepare_style_for_remotion(
|
||
secondary_title_style,
|
||
temp_dir,
|
||
f"{task_id}_secondary_title_font"
|
||
)
|
||
|
||
# 清理字体临时文件
|
||
for prefix in [f"{task_id}_subtitle_font", f"{task_id}_title_font", f"{task_id}_secondary_title_font"]:
|
||
for ext in [".ttf", ".otf", ".woff", ".woff2"]:
|
||
font_tmp = temp_dir / f"{prefix}{ext}"
|
||
if font_tmp.exists():
|
||
temp_files.append(font_tmp)
|
||
|
||
final_output_local_path = temp_dir / f"{task_id}_output.mp4"
|
||
temp_files.append(final_output_local_path)
|
||
needs_audio_compose = str(final_audio_path) != str(audio_path)
|
||
|
||
if use_remotion:
|
||
_update_task(task_id, message="正在合成视频 (Remotion)...", progress=87)
|
||
remotion_input_path = lipsync_video_path
|
||
|
||
if needs_audio_compose:
|
||
composed_video_path = temp_dir / f"{task_id}_composed.mp4"
|
||
temp_files.append(composed_video_path)
|
||
await video.compose(str(lipsync_video_path), str(final_audio_path), str(composed_video_path))
|
||
remotion_input_path = composed_video_path
|
||
else:
|
||
logger.info("[Pipeline] Audio unchanged, skip pre-Remotion compose")
|
||
|
||
remotion_health = await remotion_service.check_health()
|
||
if remotion_health.get("ready"):
|
||
try:
|
||
def on_remotion_progress(percent):
|
||
mapped = 87 + int(percent * 0.08)
|
||
_update_task(task_id, progress=mapped)
|
||
|
||
title_display_mode = (
|
||
req.title_display_mode
|
||
if req.title_display_mode in ("short", "persistent")
|
||
else "short"
|
||
)
|
||
title_duration = max(0.5, min(float(req.title_duration or 4.0), 30.0))
|
||
|
||
await remotion_service.render(
|
||
video_path=str(remotion_input_path),
|
||
output_path=str(final_output_local_path),
|
||
captions_path=str(captions_path) if captions_path else None,
|
||
title=req.title,
|
||
title_duration=title_duration,
|
||
title_display_mode=title_display_mode,
|
||
fps=25,
|
||
enable_subtitles=req.enable_subtitles,
|
||
subtitle_style=subtitle_style,
|
||
title_style=title_style,
|
||
secondary_title=req.secondary_title,
|
||
secondary_title_style=secondary_title_style,
|
||
on_progress=on_remotion_progress
|
||
)
|
||
print(f"[Pipeline] Remotion render completed")
|
||
except Exception as e:
|
||
logger.warning(f"Remotion render failed, using FFmpeg fallback: {e}")
|
||
import shutil
|
||
shutil.copy(str(remotion_input_path), str(final_output_local_path))
|
||
else:
|
||
logger.warning(f"Remotion not ready: {remotion_health.get('error')}, using FFmpeg")
|
||
import shutil
|
||
shutil.copy(str(remotion_input_path), str(final_output_local_path))
|
||
else:
|
||
_update_task(task_id, message="正在合成最终视频...", progress=90)
|
||
if needs_audio_compose:
|
||
await video.compose(str(lipsync_video_path), str(final_audio_path), str(final_output_local_path))
|
||
else:
|
||
import shutil
|
||
shutil.copy(str(lipsync_video_path), str(final_output_local_path))
|
||
|
||
total_time = time.time() - start_time
|
||
|
||
_update_task(task_id, message="正在上传结果...", progress=95)
|
||
|
||
storage_path = f"{user_id}/{task_id}_output.mp4"
|
||
await storage_service.upload_file_from_path(
|
||
bucket=storage_service.BUCKET_OUTPUTS,
|
||
storage_path=storage_path,
|
||
local_file_path=str(final_output_local_path),
|
||
content_type="video/mp4"
|
||
)
|
||
|
||
signed_url = await storage_service.get_signed_url(
|
||
bucket=storage_service.BUCKET_OUTPUTS,
|
||
path=storage_path
|
||
)
|
||
|
||
print(f"[Pipeline] Total generation time: {total_time:.1f}s")
|
||
|
||
_update_task(
|
||
task_id,
|
||
status="completed",
|
||
progress=100,
|
||
message=f"生成完成！耗时 {total_time:.0f} 秒",
|
||
output=storage_path,
|
||
download_url=signed_url,
|
||
)
|
||
|
||
except Exception as e:
|
||
_update_task(
|
||
task_id,
|
||
status="failed",
|
||
message=f"错误: {str(e)}",
|
||
error=traceback.format_exc(),
|
||
)
|
||
logger.error(f"Generate video failed: {e}")
|
||
finally:
|
||
for f in temp_files:
|
||
try:
|
||
if f.exists():
|
||
f.unlink()
|
||
except Exception as e:
|
||
print(f"Error cleaning up {f}: {e}")
|
||
|
||
|
||
async def get_lipsync_health():
|
||
lipsync = _get_lipsync_service()
|
||
return await lipsync.check_health()
|
||
|
||
|
||
async def get_voiceclone_health():
|
||
return await voice_clone_service.check_health()
|