guanghulab/video-ai-system/engines/lipsync-adapter.py

#!/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()