#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ LIPSYNC-ADAPTER 口型适配器 — 接视频改口型或 Wav2Lip，解决人物真正说台词的问题。 功能: 1. 输入视频 + 对白音频 2. 用 Wav2Lip 开源工具做口型同步 3. 支持批量处理 4. 封装为统一接口 依赖: pip install librosa opencv-python numpy # Wav2Lip 需要单独安装: https://github.com/Rudrabha/Wav2Lip 用法: python lipsync-adapter.py --video input.mp4 --audio dialogue.mp3 --output output.mp4 python lipsync-adapter.py --batch video_list.json """ import os import sys import json import argparse from pathlib import Path from datetime import datetime PROJECT_ROOT = Path(__file__).parent.parent sys.path.insert(0, str(PROJECT_ROOT / "engines")) class LipSyncAdapter: """口型适配器 — 支持多路由: Wav2Lip(本地GPU) → 火山改口型(云端) → 回退""" def __init__(self, wav2lip_path=None, allow_fallback=False): self.wav2lip_path = wav2lip_path or PROJECT_ROOT / "tools" / "Wav2Lip" self.allow_fallback = allow_fallback self.available = self._check_wav2lip() self._cloud_available = None def _check_wav2lip(self): """检查 Wav2Lip 是否可用""" if not self.wav2lip_path.exists(): print(f"⚠️ Wav2Lip 未安装: {self.wav2lip_path}") print(f" 安装方法: git clone https://github.com/Rudrabha/Wav2Lip.git {self.wav2lip_path}") return False # 检查 infer.py 是否存在 infer_script = self.wav2lip_path / "infer.py" if not infer_script.exists(): print(f"⚠️ Wav2Lip infer.py 未找到: {infer_script}") return False print(f"✅ Wav2Lip 已安装: {self.wav2lip_path}") return True def sync_lips(self, video_path, audio_path, output_path=None, allow_fallback=None): """ 口型同步 参数: video_path: 输入视频路径 audio_path: 对白音频路径 output_path: 输出视频路径 (可选，默认加 _synced 后缀) 返回: { "success": bool, "output_path": str, "method": str, # "wav2lip" | "fallback" "warning": str } """ print(f"\n🎤 口型同步") print(f" 视频: {Path(video_path).name}") print(f" 音频: {Path(audio_path).name}") video_path = Path(video_path) audio_path = Path(audio_path) if not video_path.exists(): return {"success": False, "error": f"视频不存在: {video_path}"} if not audio_path.exists(): return {"success": False, "error": f"音频不存在: {audio_path}"} # 确定输出路径 if output_path is None: output_path = video_path.parent / f"{video_path.stem}_synced{video_path.suffix}" else: output_path = Path(output_path) # 确保输出目录存在 output_path.parent.mkdir(parents=True, exist_ok=True) # 方法1: Wav2Lip (本地GPU) if self.available: print(f" 🔧 使用 Wav2Lip...") result = self._run_wav2lip(video_path, audio_path, output_path) return result # 方法2: 火山引擎云端改口型 cloud = self._check_cloud() if cloud["available"]: print(f" ☁️ 尝试火山改口型...") result = self._run_cloud_lipsync(video_path, audio_path, output_path) if result.get("success"): return result print(f" ⚠️ 云端改口型失败: {result.get('error')}") # 方法3: Edge-TTS 配音替代（角色不说话时可用） print(f" 💡 当前可用路由:") print(f" Route1 Wav2Lip: {'✅' if self.available else '❌ 需要CUDA GPU'}") print(f" Route2 火山改口型: {'STUB 需开通' if not cloud.get('available') else '可用'}") print(f" Route3 Edge-TTS: ✅ (voice-emotion-compiler已通过)") fallback_enabled = self.allow_fallback if allow_fallback is None else allow_fallback if not fallback_enabled: print(f" ❌ Wav2Lip 不可用，口型同步未执行") print(f" 💡 如需调试占位输出，可显式传入 --allow-fallback-copy") return { "success": False, "error": "Wav2Lip 不可用，未执行真实口型同步", "method": "not-run", "requires": "Install Wav2Lip and checkpoint wav2lip_gan.pth" } # 调试回退: 不处理，只复制视频。必须显式开启，避免生产线假阳性。 print(f" ⚠️ Wav2Lip 不可用，按 --allow-fallback-copy 复制原视频") print(f" 💡 注意: 该输出不是口型同步成品") import shutil shutil.copy2(video_path, output_path) return { "success": True, "output_path": str(output_path), "method": "fallback(copy)", "warning": "Wav2Lip 不可用，口型未同步。该文件只能用于调试占位。" } def _check_cloud(self): """检查云端改口型服务是否可用""" if self._cloud_available is not None: return self._cloud_available try: import importlib.util spec = importlib.util.spec_from_file_location( 'volce', str(PROJECT_ROOT / 'engines' / 'volce-lipsync-adapter.py') ) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) self._cloud_available = mod.check_available() except Exception as e: self._cloud_available = {"available": False, "reason": f"云端适配器不可用: {e}"} return self._cloud_available def _run_cloud_lipsync(self, video_path, audio_path, output_path): """通过火山引擎云端改口型""" try: import importlib.util spec = importlib.util.spec_from_file_location( 'volce', str(PROJECT_ROOT / 'engines' / 'volce-lipsync-adapter.py') ) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod.submit_lipsync(video_path, audio_path) except Exception as e: return {"success": False, "error": f"云端适配器调用失败: {e}", "method": "cloud-stub"} def _run_wav2lip(self, video_path, audio_path, output_path): """运行 Wav2Lip""" import subprocess infer_script = self.wav2lip_path / "infer.py" # Wav2Lip 命令 cmd = [ "python", str(infer_script), "--checkpoint_path", str(self.wav2lip_path / "checkpoints" / "wav2lip_gan.pth"), "--face", str(video_path), "--audio", str(audio_path), "--outfile", str(output_path) ] print(f" 📤 执行命令: {' '.join(cmd[:6])}...") try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=300 # 5分钟超时 ) if result.returncode == 0: print(f" ✅ 口型同步完成: {output_path.name}") return { "success": True, "output_path": str(output_path), "method": "wav2lip", "stdout": result.stdout[-500:] # 最后500字符 } else: print(f" ❌ Wav2Lip 失败: {result.stderr}") return { "success": False, "error": result.stderr, "stdout": result.stdout } except subprocess.TimeoutExpired: print(f" ❌ Wav2Lip 超时 (5分钟)") return {"success": False, "error": "Timeout"} except Exception as e: print(f" ❌ Wav2Lip 执行失败: {e}") return {"success": False, "error": str(e)} def batch_sync(self, video_audio_pairs, output_dir): """ 批量口型同步 参数: video_audio_pairs: list of (video_path, audio_path) output_dir: 输出目录 返回: list of result dicts """ print(f"\n📦 批量口型同步: {len(video_audio_pairs)} 个任务") print(f" 输出目录: {output_dir}") output_dir = Path(output_dir) output_dir.mkdir(parents=True, exist_ok=True) results = [] for i, (video_path, audio_path) in enumerate(video_audio_pairs): print(f"\n 进度: [{i+1}/{len(video_audio_pairs)}]") output_path = output_dir / f"{Path(video_path).stem}_synced.mp4" result = self.sync_lips(video_path, audio_path, output_path) results.append(result) # 统计 success_count = sum(1 for r in results if r.get("success")) print(f"\n✅ 批量完成: {success_count}/{len(results)} 成功") # 保存报告 report_path = output_dir / "lipsync_report.json" with open(report_path, "w", encoding="utf-8") as f: json.dump({ "total": len(results), "success": success_count, "results": results, "generated_at": datetime.now().isoformat() }, f, ensure_ascii=False, indent=2) print(f" 报告已保存: {report_path}") return results def check_audio_sync(self, video_path, tolerance_ms=100): """ 检查口型同步质量 简单方法: 检测音频能量峰值，与视频画面变化对比 返回: { "synced": bool, "offset_ms": float, "score": float # 0-1, 1=完美同步 } """ print(f"\n🔍 检查口型同步质量: {Path(video_path).name}") if not self.available: print(f" ⚠️ Wav2Lip 不可用，跳过质量检查") return {"synced": None, "score": None, "warning": "Wav2Lip 不可用"} # TODO: 实现口型同步质量检查 # 1. 提取音频能量包络 # 2. 检测视频中嘴部区域的运动 # 3. 计算相关性 # 4. 返回偏移量和分数 print(f" ⚠️ 质量检查未实现 (需要 librosa + OpenCV 嘴部检测)") return { "synced": None, "offset_ms": 0, "score": None, "warning": "Quality check not implemented yet" } def main(): parser = argparse.ArgumentParser(description="LIPSYNC-ADAPTER") parser.add_argument("--video", type=str, help="输入视频路径") parser.add_argument("--audio", type=str, help="对白音频路径") parser.add_argument("--output", type=str, help="输出视频路径") parser.add_argument("--batch", type=str, help="批量处理配置文件 (JSON)") parser.add_argument("--wav2lip-path", type=str, help="Wav2Lip 安装路径") parser.add_argument("--check-sync", action="store_true", help="检查口型同步质量") parser.add_argument("--allow-fallback-copy", action="store_true", help="仅调试用: Wav2Lip 不可用时复制原视频并标记为 fallback") args = parser.parse_args() if args.check_sync: if not args.video: print("❌ --check-sync 需要 --video") sys.exit(1) adapter = LipSyncAdapter(args.wav2lip_path) result = adapter.check_audio_sync(args.video) print(json.dumps(result, ensure_ascii=False, indent=2)) sys.exit(0) if args.batch: # 批量模式 batch_file = Path(args.batch) if not batch_file.exists(): print(f"❌ 批量配置文件不存在: {batch_file}") sys.exit(1) with open(batch_file, "r", encoding="utf-8") as f: config = json.load(f) video_audio_pairs = [] for item in config.get("tasks", []): video_audio_pairs.append((item["video"], item["audio"])) output_dir = config.get("output_dir", "./outputs/lipsync/") adapter = LipSyncAdapter(args.wav2lip_path, allow_fallback=args.allow_fallback_copy) results = adapter.batch_sync(video_audio_pairs, output_dir) sys.exit(0) if not args.video or not args.audio: parser.print_help() sys.exit(1) adapter = LipSyncAdapter(args.wav2lip_path, allow_fallback=args.allow_fallback_copy) result = adapter.sync_lips(args.video, args.audio, args.output) if result["success"]: print(f"\n✅ 成功: {result['output_path']}") sys.exit(0) else: print(f"\n❌ 失败: {result.get('error', 'Unknown error')}") sys.exit(1) if __name__ == "__main__": main()