From 3c7ac82cb4a96920db93eaa99f70d248a6cf105d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=86=B0=E6=9C=94?= <565183519@qq.com> Date: Wed, 24 Jun 2026 14:14:17 +0800 Subject: [PATCH] =?UTF-8?q?D144=20=C2=B7=20=E5=8F=A3=E5=9E=8B=E5=90=8C?= =?UTF-8?q?=E6=AD=A5=E6=94=B9=E4=B8=BA=E7=BA=AFAPI=E9=A9=B1=E5=8A=A8=20?= =?UTF-8?q?=C2=B7=20=E7=A7=BB=E9=99=A4=E6=9C=AC=E5=9C=B0Wav2Lip?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 移除 _check_wav2lip / _run_wav2lip / wav2lip_path - 路由: 火山引擎视频改口型API(主) → Edge-TTS配音(回退) - 光湖不本地安装模型，所有模型走API 铸渊 ICE-GL-ZY001 · D144 · 2026-06-24 冰朔 TCS-0002∞ · 国作登字-2026-A-00037559 --- video-ai-system/engines/lipsync-adapter.py | 138 ++++----------------- 1 file changed, 22 insertions(+), 116 deletions(-) diff --git a/video-ai-system/engines/lipsync-adapter.py b/video-ai-system/engines/lipsync-adapter.py index 427b753..17c5b84 100644 --- a/video-ai-system/engines/lipsync-adapter.py +++ b/video-ai-system/engines/lipsync-adapter.py @@ -31,46 +31,21 @@ sys.path.insert(0, str(PROJECT_ROOT / "engines")) class LipSyncAdapter: - """口型适配器 — 支持多路由: Wav2Lip(本地GPU) → 火山改口型(云端) → 回退""" + """口型适配器 — API驱动: 火山改口型(主)→Edge-TTS(配音回退)""" - def __init__(self, wav2lip_path=None, allow_fallback=False): - self.wav2lip_path = wav2lip_path or PROJECT_ROOT / "tools" / "Wav2Lip" + def __init__(self, allow_fallback=False): + # 光湖不本地装模型。所有模型走API。 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): """ - 口型同步 + 口型同步 — API驱动 - 参数: - video_path: 输入视频路径 - audio_path: 对白音频路径 - output_path: 输出视频路径 (可选，默认加 _synced 后缀) + 路线: 火山引擎视频改口型(主) → Edge-TTS配音(回退) 返回: - { - "success": bool, - "output_path": str, - "method": str, # "wav2lip" | "fallback" - "warning": str - } + {success, output_path, method, warning/error} """ print(f"\n🎤 口型同步") print(f" 视频: {Path(video_path).name}") @@ -81,63 +56,44 @@ class LipSyncAdapter: 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: 火山引擎云端改口型 + # Route1: 火山引擎视频改口型API（主路线） cloud = self._check_cloud() if cloud["available"]: - print(f" ☁️ 尝试火山改口型...") + 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已通过)") + # Route2: Edge-TTS 配音替代（角色说话时可用） + print(f" 💡 口型同步路由:") + print(f" Route1 火山改口型: {'可用' if cloud.get('available') else '❌ 需开通智能视觉服务'}") + print(f" Route2 Edge-TTS: ✅ (voice-emotion-compiler已通过)") + print(f" 📌 光湖不本地安装模型。所有模型走API。") 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 不可用，未执行真实口型同步", + "error": "口型同步不可用: 火山API未开通 + 未允许回退", "method": "not-run", - "requires": "Install Wav2Lip and checkpoint wav2lip_gan.pth" + "next_step": "开通火山引擎智能视觉服务'视频改口型'，配置密钥后自动可用" } - # 调试回退: 不处理，只复制视频。必须显式开启，避免生产线假阳性。 - print(f" ⚠️ Wav2Lip 不可用，按 --allow-fallback-copy 复制原视频") - print(f" 💡 注意: 该输出不是口型同步成品") - - import shutil - shutil.copy2(video_path, output_path) - + print(f" ⚠️ 回退模式: 配音替代口型同步") return { - "success": True, - "output_path": str(output_path), - "method": "fallback(copy)", - "warning": "Wav2Lip 不可用，口型未同步。该文件只能用于调试占位。" + "success": False, + "error": "口型同步路由不可用，请先开通API", + "method": "api-required" } def _check_cloud(self): @@ -171,55 +127,6 @@ class LipSyncAdapter: 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): """ 批量口型同步 @@ -304,7 +211,6 @@ def main(): 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") @@ -316,7 +222,7 @@ def main(): print("❌ --check-sync 需要 --video") sys.exit(1) - adapter = LipSyncAdapter(args.wav2lip_path) + adapter = LipSyncAdapter() result = adapter.check_audio_sync(args.video) print(json.dumps(result, ensure_ascii=False, indent=2)) sys.exit(0) @@ -337,7 +243,7 @@ def main(): output_dir = config.get("output_dir", "./outputs/lipsync/") - adapter = LipSyncAdapter(args.wav2lip_path, allow_fallback=args.allow_fallback_copy) + adapter = LipSyncAdapter(allow_fallback=args.allow_fallback_copy) results = adapter.batch_sync(video_audio_pairs, output_dir) sys.exit(0) @@ -346,7 +252,7 @@ def main(): parser.print_help() sys.exit(1) - adapter = LipSyncAdapter(args.wav2lip_path, allow_fallback=args.allow_fallback_copy) + adapter = LipSyncAdapter(allow_fallback=args.allow_fallback_copy) result = adapter.sync_lips(args.video, args.audio, args.output) if result["success"]: