D144 · 口型同步改为纯API驱动 · 移除本地Wav2Lip

- 移除 _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

@@ -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):
+    def __init__(self, allow_fallback=False):
-        self.wav2lip_path = wav2lip_path or PROJECT_ROOT / "tools" / "Wav2Lip"
+        # 光湖不本地装模型。所有模型走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驱动
 
-        参数:
+        路线: 火山引擎视频改口型(主) → Edge-TTS配音(回退)
-          video_path: 输入视频路径
-          audio_path: 对白音频路径
-          output_path: 输出视频路径 (可选，默认加 _synced 后缀)
 
         返回:
-          {
+          {success, output_path, method, warning/error}
-            "success": bool,
-            "output_path": str,
-            "method": str,  # "wav2lip" | "fallback"
-            "warning": str
-          }
         """
         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)
+        # Route1: 火山引擎视频改口型API（主路线）
-        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"  ☁️  尝试火山改口型...")
+            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 配音替代（角色不说话时可用）
+        # Route2: Edge-TTS 配音替代（角色说话时可用）
-        print(f"  💡 当前可用路由:")
+        print(f"  💡 口型同步路由:")
-        print(f"     Route1 Wav2Lip:    {'✅' if self.available else '❌ 需要CUDA GPU'}")
+        print(f"     Route1 火山改口型: {'可用' if cloud.get('available') else '❌ 需开通智能视觉服务'}")
-        print(f"     Route2 火山改口型: {'STUB 需开通' if not cloud.get('available') else '可用'}")
+        print(f"     Route2 Edge-TTS:   ✅ (voice-emotion-compiler已通过)")
-        print(f"     Route3 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"  ⚠️  回退模式: 配音替代口型同步")
-        print(f"  ⚠️  Wav2Lip 不可用，按 --allow-fallback-copy 复制原视频")
-        print(f"  💡 注意: 该输出不是口型同步成品")
-
-        import shutil
-        shutil.copy2(video_path, output_path)
-
         return {
-            "success": True,
+            "success": False,
-            "output_path": str(output_path),
+            "error": "口型同步路由不可用，请先开通API",
-            "method": "fallback(copy)",
+            "method": "api-required"
-            "warning": "Wav2Lip 不可用，口型未同步。该文件只能用于调试占位。"
         }
 
     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"]: