KawasakiAkasei
ad3c86b8ba
- Add get_model_urls() and download_model() to api.py and api_complete.py supporting all 14 discovered urlResult format keys (glb, obj, pbr maps, etc.) - Update generator.py to extract full urlResult dict instead of just modelUrl - Add CLI subcommands: formats (list available formats) and download (fetch by key) - Update reverse engineering docs with complete format key table and CLI examples
249 lines
8.6 KiB
Python
249 lines
8.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
腾讯混元3D 图生3D 完整自动化脚本
|
|
使用 cloakbrowser 在浏览器内完成所有操作（包括签名生成）
|
|
"""
|
|
|
|
import contextlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from datetime import datetime
|
|
from cloakbrowser import launch_persistent_context
|
|
|
|
from ..config import get_profile_dir
|
|
|
|
PROFILE_DIR = str(get_profile_dir())
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def _temp_persistent_context(headless=True):
|
|
"""
|
|
将标准 profile 复制到临时目录后启动持久化上下文，
|
|
避免与 login 进程发生 Chromium SingletonLock 冲突。
|
|
"""
|
|
standard = str(get_profile_dir())
|
|
if not os.path.exists(standard):
|
|
raise FileNotFoundError(f"未找到登录状态目录: {standard}")
|
|
|
|
temp_root = tempfile.mkdtemp(prefix="hunyuan3dweb-profile-")
|
|
temp_profile = os.path.join(temp_root, "profile")
|
|
shutil.copytree(standard, temp_profile, dirs_exist_ok=True)
|
|
|
|
context = None
|
|
try:
|
|
context = launch_persistent_context(temp_profile, headless=headless)
|
|
yield context
|
|
finally:
|
|
if context is not None:
|
|
try:
|
|
context.close()
|
|
except Exception:
|
|
pass
|
|
try:
|
|
shutil.rmtree(temp_root)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def generate_3d(image_path, wait_for_complete=False, timeout=300):
|
|
"""
|
|
上传图片并触发生成3D模型
|
|
|
|
Args:
|
|
image_path: 图片路径
|
|
wait_for_complete: 是否等待生成完成
|
|
timeout: 最长等待时间（秒）
|
|
|
|
Returns:
|
|
dict: 包含 creationsId 和状态信息
|
|
"""
|
|
if not os.path.exists(image_path):
|
|
raise FileNotFoundError(f"图片不存在: {image_path}")
|
|
|
|
with _temp_persistent_context(headless=True) as context:
|
|
page = context.new_page()
|
|
|
|
# 用于存储结果
|
|
result = {"creationsId": None, "status": None, "urlResult": {}}
|
|
|
|
# 监听响应
|
|
def handle_response(response):
|
|
url = response.url
|
|
if "creations/generations" in url and response.status == 200:
|
|
try:
|
|
body = response.json()
|
|
if "creationsId" in body:
|
|
result["creationsId"] = body["creationsId"]
|
|
print(f"✅ 生成已触发，creationsId: {body['creationsId']}")
|
|
except Exception:
|
|
pass
|
|
elif "creations/detail" in url and response.status == 200:
|
|
try:
|
|
body = response.json()
|
|
result["status"] = body.get("status")
|
|
if "result" in body and isinstance(body["result"], list) and len(body["result"]) > 0:
|
|
model_data = body["result"][0]
|
|
url_result = model_data.get("urlResult", {})
|
|
if url_result:
|
|
result["urlResult"] = {
|
|
k: v for k, v in url_result.items()
|
|
if v not in (None, "", {})
|
|
}
|
|
except Exception:
|
|
pass
|
|
|
|
page.on("response", handle_response)
|
|
|
|
print("[1/6] 打开首页...")
|
|
page.goto("https://3d.hunyuan.tencent.com/", timeout=60000)
|
|
page.wait_for_timeout(3000)
|
|
|
|
print("[2/6] 点击 AI创作...")
|
|
page.locator("button").filter(has_text="AI创作").first.click()
|
|
page.locator("text=图生3D").wait_for(state="visible", timeout=10000)
|
|
|
|
print("[3/6] 点击 图生3D...")
|
|
page.locator("text=图生3D").first.click()
|
|
page.locator("text=单张图片").wait_for(state="visible", timeout=10000)
|
|
|
|
print("[4/6] 点击 单张图片...")
|
|
page.locator("text=单张图片").first.click()
|
|
page.locator('input[type=file]').first.wait_for(state="visible", timeout=10000)
|
|
|
|
print("[5/6] 上传图片...")
|
|
page.locator('input[type=file]').first.set_input_files(image_path)
|
|
page.locator("text=立即生成").wait_for(state="visible", timeout=10000)
|
|
|
|
print("[6/6] 点击 立即生成...")
|
|
page.locator("text=立即生成").first.click()
|
|
|
|
# 等待生成触发
|
|
page.wait_for_timeout(5000)
|
|
|
|
if not result["creationsId"]:
|
|
print("⚠️ 未获取到 creationsId，可能生成失败")
|
|
return result
|
|
|
|
if wait_for_complete:
|
|
print(f"\n⏳ 等待生成完成（最长 {timeout} 秒）...")
|
|
start = time.time()
|
|
last_status = None
|
|
while time.time() - start < timeout:
|
|
# 轮询状态
|
|
status_result = page.evaluate('''
|
|
async (creationsId) => {
|
|
const resp = await fetch(`https://3d.hunyuan.tencent.com/api/3d/creations/detail?creationsId=${creationsId}`, {
|
|
headers: {'X-Source': 'web', 'Referer': 'https://3d.hunyuan.tencent.com/'}
|
|
});
|
|
return await resp.json();
|
|
}
|
|
''', result["creationsId"])
|
|
|
|
status = status_result.get("status", "unknown")
|
|
progress = status_result.get("progress", 0)
|
|
|
|
if status != last_status:
|
|
print(f" 状态: {status} (进度: {progress}%)")
|
|
last_status = status
|
|
|
|
if status == "success":
|
|
# 提取所有可用格式URL
|
|
if "result" in status_result and len(status_result["result"]) > 0:
|
|
model_data = status_result["result"][0]
|
|
url_result = model_data.get("urlResult", {})
|
|
if url_result:
|
|
result["urlResult"] = {
|
|
k: v for k, v in url_result.items()
|
|
if v not in (None, "", {})
|
|
}
|
|
print(f"\n✅ 生成完成！可用格式: {', '.join(result['urlResult'].keys())}")
|
|
break
|
|
elif status == "fail":
|
|
print(f"\n❌ 生成失败")
|
|
break
|
|
|
|
time.sleep(3)
|
|
else:
|
|
print(f"\n⏰ 超时，当前状态: {last_status}")
|
|
|
|
return result
|
|
|
|
|
|
def get_quota():
|
|
"""获取当前配额信息"""
|
|
with _temp_persistent_context(headless=True) as context:
|
|
page = context.new_page()
|
|
|
|
page.goto("https://3d.hunyuan.tencent.com/", timeout=60000)
|
|
page.wait_for_timeout(3000)
|
|
|
|
result = page.evaluate('''
|
|
async () => {
|
|
const resp = await fetch('https://3d.hunyuan.tencent.com/api/3d/quotainfo', {
|
|
method: 'POST',
|
|
headers: {'Content-Type': 'application/json', 'X-Source': 'web', 'Referer': 'https://3d.hunyuan.tencent.com/'},
|
|
body: '{"sceneType":"3dCreations"}'
|
|
});
|
|
return await resp.json();
|
|
}
|
|
''')
|
|
return result
|
|
|
|
|
|
def get_creations_list():
|
|
"""获取作品列表"""
|
|
with _temp_persistent_context(headless=True) as context:
|
|
page = context.new_page()
|
|
|
|
page.goto("https://3d.hunyuan.tencent.com/", timeout=60000)
|
|
page.wait_for_timeout(3000)
|
|
|
|
result = page.evaluate('''
|
|
async () => {
|
|
const resp = await fetch('https://3d.hunyuan.tencent.com/api/3d/creations/list', {
|
|
method: 'POST',
|
|
headers: {'Content-Type': 'application/json', 'X-Source': 'web', 'Referer': 'https://3d.hunyuan.tencent.com/'},
|
|
body: '{"page":1,"pageSize":20}'
|
|
});
|
|
return await resp.json();
|
|
}
|
|
''')
|
|
return result
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) < 2:
|
|
print("用法:")
|
|
print(f" python {sys.argv[0]} quota # 查询配额")
|
|
print(f" python {sys.argv[0]} list # 查询作品列表")
|
|
print(f" python {sys.argv[0]} generate <图片路径> [wait] # 生成3D模型")
|
|
sys.exit(1)
|
|
|
|
cmd = sys.argv[1]
|
|
|
|
if cmd == "quota":
|
|
quota = get_quota()
|
|
print(json.dumps(quota, indent=2, ensure_ascii=False))
|
|
|
|
elif cmd == "list":
|
|
creations = get_creations_list()
|
|
print(json.dumps(creations, indent=2, ensure_ascii=False))
|
|
|
|
elif cmd == "generate":
|
|
if len(sys.argv) < 3:
|
|
print("请提供图片路径")
|
|
sys.exit(1)
|
|
|
|
image_path = sys.argv[2]
|
|
wait = len(sys.argv) > 3 and sys.argv[3] == "wait"
|
|
|
|
result = generate_3d(image_path, wait_for_complete=wait)
|
|
print(json.dumps(result, indent=2, ensure_ascii=False))
|
|
|
|
else:
|
|
print(f"未知命令: {cmd}")
|