301 lines
9.5 KiB
Python
301 lines
9.5 KiB
Python
|
#!/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:
|
|||
|
|
"""口型适配器"""
|
|||
|
|
|
|||
|
|
def __init__(self, wav2lip_path=None):
|
|||
|
|
self.wav2lip_path = wav2lip_path or PROJECT_ROOT / "tools" / "Wav2Lip"
|
|||
|
|
self.available = self._check_wav2lip()
|
|||
|
|
|
|||
|
|
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):
|
|||
|
|
"""
|
|||
|
|
口型同步
|
|||
|
|
|
|||
|
|
参数:
|
|||
|
|
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
|
|||
|
|
if self.available:
|
|||
|
|
print(f" 🔧 使用 Wav2Lip...")
|
|||
|
|
result = self._run_wav2lip(video_path, audio_path, output_path)
|
|||
|
|
return result
|
|||
|
|
|
|||
|
|
# 方法2: 回退 (不处理，只复制视频)
|
|||
|
|
print(f" ⚠️ Wav2Lip 不可用，回退到不处理模式")
|
|||
|
|
print(f" 💡 提示: 安装 Wav2Lip 以获得口型同步能力")
|
|||
|
|
|
|||
|
|
import shutil
|
|||
|
|
shutil.copy2(video_path, output_path)
|
|||
|
|
|
|||
|
|
return {
|
|||
|
|
"success": True,
|
|||
|
|
"output_path": str(output_path),
|
|||
|
|
"method": "fallback(copy)",
|
|||
|
|
"warning": "Wav2Lip 不可用，口型未同步。请安装 Wav2Lip。"
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
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="检查口型同步质量")
|
|||
|
|
|
|||
|
|
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)
|
|||
|
|
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)
|
|||
|
|
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()
|