From bc2d56ccb25a8b4453d4a5fae738d074abd55ef3 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 13:57:18 +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=E4=B8=89=E8=B7=AF=E7=94=B1=E4=B8=8A=E7=BA=BF:=20Wav2L?= =?UTF-8?q?ip=E2=86=92=E7=81=AB=E5=B1=B1=E4=BA=91=E7=AB=AF=E2=86=92Edge-TT?= =?UTF-8?q?S?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - lipsync-adapter.py: 新增云端路由，Wav2Lip不可用时自动尝试火山改口型API - volce-lipsync-adapter.py: 火山引擎口型同步适配器（STUB），需开通智能视觉服务 - 当前状态: Wav2Lip需CUDA GPU(本地不可用)，火山需AK/SK配置，TTS配音可用 铸渊 ICE-GL-ZY001 · D144 · 2026-06-24 --- video-ai-system/engines/lipsync-adapter.py | 51 ++++++++- .../engines/volce-lipsync-adapter.py | 104 ++++++++++++++++++ 2 files changed, 153 insertions(+), 2 deletions(-) create mode 100644 video-ai-system/engines/volce-lipsync-adapter.py diff --git a/video-ai-system/engines/lipsync-adapter.py b/video-ai-system/engines/lipsync-adapter.py index e8c3e0b..427b753 100644 --- a/video-ai-system/engines/lipsync-adapter.py +++ b/video-ai-system/engines/lipsync-adapter.py @@ -31,12 +31,13 @@ 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 是否可用""" @@ -93,12 +94,27 @@ class LipSyncAdapter: # 确保输出目录存在 output_path.parent.mkdir(parents=True, exist_ok=True) - # 方法1: Wav2Lip + # 方法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 不可用，口型同步未执行") @@ -124,6 +140,37 @@ class LipSyncAdapter: "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 diff --git a/video-ai-system/engines/volce-lipsync-adapter.py b/video-ai-system/engines/volce-lipsync-adapter.py new file mode 100644 index 0000000..c470179 --- /dev/null +++ b/video-ai-system/engines/volce-lipsync-adapter.py @@ -0,0 +1,104 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +volce-lipsync-adapter.py · 火山引擎口型同步适配器 +D144 · 铸渊 ICE-GL-ZY001 + +对接火山引擎智能视觉服务"视频改口型"API。 +需要预先开通服务并获取 AK/SK。 + +文档: + https://www.volcengine.com/docs/85128/1462622 + +依赖: + pip install volcengine-python-sdk + +当前状态: STUB — 需要开通服务并配置 AK/SK 后才能激活 +""" + +import os +import json +import base64 +import hashlib +import hmac +import urllib.request +import urllib.parse +from datetime import datetime, timezone + + +SECRETS_FILE = "/Users/bingshuolingdianyuanhe/Documents/guanghulab-local-secrets/video-ai-system.env" + + +def _load_secrets(): + """加载本地密钥文件""" + secrets = {} + if os.path.isfile(SECRETS_FILE): + with open(SECRETS_FILE) as f: + for line in f: + line = line.strip() + if line and not line.startswith("#") and "=" in line: + k, v = line.split("=", 1) + secrets[k.strip()] = v.strip().strip('"').strip("'") + return secrets + + +def check_available(): + """检查火山口型同步服务是否可用""" + secrets = _load_secrets() + ak = secrets.get("VOLCE_ACCESS_KEY", "") + sk = secrets.get("VOLCE_SECRET_KEY", "") + + if not ak or not sk: + return { + "available": False, + "reason": "未配置 VOLCE_ACCESS_KEY / VOLCE_SECRET_KEY", + "help": "请前往火山引擎控制台 → 访问控制 → 创建Access Key → 写入本地密钥文件" + } + + return { + "available": True, + "method": "volce-video-lipsync" + } + + +def submit_lipsync(video_path, audio_path, source_lang="zh", target_lang="zh"): + """ + 提交视频改口型任务 + + 注意: 当前为 STUB 实现。火山引擎视频改口型 API 需要: + 1. 开通智能视觉服务 + 2. 获取独立 AK/SK（不是 JIMENG_API_KEY） + 3. 上传视频到火山 TOS 对象存储 + 4. 提交异步任务并轮询结果 + + :param video_path: 输入视频 + :param audio_path: 对白音频 + :param source_lang: 源语言 + :param target_lang: 目标语言 + :return: {success, output_path, method} + """ + status = check_available() + if not status["available"]: + return { + "success": False, + "error": status["reason"], + "method": "volce-stub" + } + + # STUB: 需要完整开通后才能调用 + return { + "success": False, + "error": "火山视频改口型 API 尚未完整接入（STUB）", + "method": "volce-stub", + "next_steps": [ + "开通火山引擎智能视觉服务", + "配置 VOLCE_ACCESS_KEY / VOLCE_SECRET_KEY", + "实现视频上传到 TOS 的逻辑", + "实现异步任务提交和轮询" + ] + } + + +if __name__ == "__main__": + result = check_available() + print(json.dumps(result, ensure_ascii=False, indent=2))