113 lines
3.2 KiB
Python
113 lines
3.2 KiB
Python
|
#!/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_FILES = [
|
|||
|
|
os.environ.get("VIDEO_AI_SECRETS_FILE", ""),
|
|||
|
|
"/root/guanghulab-local-secrets/cang-ying.env",
|
|||
|
|
os.path.expanduser("~/guanghulab-local-secrets/cang-ying.env"),
|
|||
|
|
"/Users/bingshuolingdianyuanhe/Documents/guanghulab-local-secrets/video-ai-system.env",
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _load_secrets():
|
|||
|
|
"""加载本地密钥文件"""
|
|||
|
|
secrets = {}
|
|||
|
|
for secrets_file in SECRETS_FILES:
|
|||
|
|
if not secrets_file or not os.path.isfile(secrets_file):
|
|||
|
|
continue
|
|||
|
|
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("'")
|
|||
|
|
break
|
|||
|
|
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))
|