udxandCursor
6550b8df4b
Bring in MaleExplorer assets/shaders, model tooling and Hunyuan updates, plus boot/YooAsset Art package support. Co-authored-by: Cursor <cursoragent@cursor.com>
331 lines
11 KiB
Python
331 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""腾讯混元3D 图生3D 工具。
|
||
|
||
支持：
|
||
- AI / Agent CLI 调用（参数或 JSON）
|
||
- 流程回调（HTTP webhook / 状态文件 / NDJSON 事件流）
|
||
- 未登录时弹出 Edge 微信扫码，取得凭证后继续生成
|
||
|
||
依赖：playwright + 本机 Microsoft Edge。见 Tools/requirements-hunyuan.txt
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import shutil
|
||
import sys
|
||
import uuid
|
||
from pathlib import Path
|
||
from typing import Any, Dict, Optional, Sequence
|
||
|
||
# 允许直接 `python Tools/hunyuan_3d_generate.py` 运行
|
||
_TOOLS_DIR = Path(__file__).resolve().parent
|
||
if str(_TOOLS_DIR) not in sys.path:
|
||
sys.path.insert(0, str(_TOOLS_DIR))
|
||
|
||
from hunyuan_3d.auth import ( # noqa: E402
|
||
ensure_login,
|
||
is_logged_in,
|
||
jobs_dir,
|
||
logout,
|
||
start_wechat_login_blocking,
|
||
status,
|
||
)
|
||
from hunyuan_3d.client import FACE_LABEL, FACE_LABELS, generate_from_image # noqa: E402
|
||
from hunyuan_3d.errors import HunyuanQuotaExceeded, HunyuanReloginNeeded # noqa: E402
|
||
from hunyuan_3d.events import CallbackHub # noqa: E402
|
||
|
||
|
||
class GenerateError(RuntimeError):
|
||
"""生成失败。"""
|
||
|
||
|
||
def _load_json_arg(raw: Optional[str]) -> Dict[str, Any]:
|
||
if not raw:
|
||
return {}
|
||
if raw == "-":
|
||
return json.load(sys.stdin)
|
||
path = Path(raw)
|
||
if path.is_file():
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
return json.loads(raw)
|
||
|
||
|
||
def _normalize_faces(value: Optional[str]) -> str:
|
||
text = (value or FACE_LABEL).strip().lower()
|
||
aliases = {
|
||
"50k": "50k",
|
||
"50": "50k",
|
||
"50000": "50k",
|
||
"500k": "500k",
|
||
"500000": "500k",
|
||
"1m": "1m",
|
||
"1000000": "1m",
|
||
"1": "1m",
|
||
}
|
||
label = aliases.get(text, text)
|
||
if label not in FACE_LABELS:
|
||
raise GenerateError(f"面数须为 {list(FACE_LABELS)} 之一，收到: {value}")
|
||
return label
|
||
|
||
|
||
def _make_update(callbacks: CallbackHub):
|
||
def update(**payload: Any) -> None:
|
||
event = payload.pop("event", "stage")
|
||
callbacks.emit(event, **payload)
|
||
|
||
return update
|
||
|
||
|
||
def run_generate(
|
||
*,
|
||
image: str,
|
||
output: str,
|
||
faces: str = FACE_LABEL,
|
||
interactive_login: bool = True,
|
||
login_timeout: int = 300,
|
||
callback_url: Optional[str] = None,
|
||
callback_file: Optional[str] = None,
|
||
quiet: bool = False,
|
||
max_login_retries: int = 3,
|
||
) -> Dict[str, Any]:
|
||
callbacks = CallbackHub(
|
||
callback_url=callback_url,
|
||
callback_file=callback_file,
|
||
quiet=quiet,
|
||
)
|
||
update = _make_update(callbacks)
|
||
|
||
image_path = Path(image).expanduser().resolve()
|
||
if not image_path.is_file():
|
||
raise GenerateError(f"参考图不存在: {image_path}")
|
||
|
||
output_path = Path(output).expanduser().resolve()
|
||
if output_path.suffix.lower() != ".glb":
|
||
if output_path.is_dir() or not output_path.suffix:
|
||
output_path = output_path / f"{image_path.stem}.glb"
|
||
else:
|
||
raise GenerateError("输出路径须为 .glb 文件或目录")
|
||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
face_label = _normalize_faces(faces)
|
||
job_id = uuid.uuid4().hex
|
||
work_dir = jobs_dir() / job_id
|
||
work_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
update(
|
||
event="started",
|
||
stage="开始混元图生3D",
|
||
progress=1,
|
||
job_id=job_id,
|
||
image=str(image_path),
|
||
output=str(output_path),
|
||
faces=face_label,
|
||
)
|
||
|
||
try:
|
||
ensure_login(
|
||
interactive=interactive_login,
|
||
timeout_sec=login_timeout,
|
||
update=update,
|
||
)
|
||
|
||
attempts = 0
|
||
while True:
|
||
attempts += 1
|
||
try:
|
||
mesh = generate_from_image(
|
||
image_path,
|
||
work_dir,
|
||
update,
|
||
face_label=face_label,
|
||
headed=False,
|
||
)
|
||
break
|
||
except HunyuanReloginNeeded as exc:
|
||
reason = getattr(exc, "reason", "auth") or "auth"
|
||
if isinstance(exc, HunyuanQuotaExceeded):
|
||
reason = "quota"
|
||
if not interactive_login or attempts > max_login_retries:
|
||
raise GenerateError(str(exc)) from exc
|
||
update(
|
||
event="login_required",
|
||
stage=(
|
||
"当前账号额度已满，打开登录页换号"
|
||
if reason == "quota"
|
||
else "凭证失效，重新微信登录"
|
||
),
|
||
progress=3,
|
||
error=str(exc),
|
||
reason=reason,
|
||
attempt=attempts,
|
||
)
|
||
print(
|
||
f"[hunyuan] {exc}\n"
|
||
f"[hunyuan] 即将打开浏览器登录页（原因={reason}，第 {attempts}/{max_login_retries} 次）",
|
||
flush=True,
|
||
)
|
||
if is_logged_in():
|
||
logout()
|
||
start_wechat_login_blocking(
|
||
timeout_sec=login_timeout,
|
||
update=update,
|
||
reason=reason,
|
||
clear_existing=True,
|
||
)
|
||
|
||
shutil.copy2(mesh, output_path)
|
||
result = {
|
||
"ok": True,
|
||
"job_id": job_id,
|
||
"status": "succeeded",
|
||
"progress": 100,
|
||
"faces": face_label,
|
||
"image": str(image_path),
|
||
"output": str(output_path),
|
||
"work_dir": str(work_dir),
|
||
"bytes": output_path.stat().st_size,
|
||
"hint": "生成完成后请在 Doc/ResDesc_Model.md 登记该资源",
|
||
}
|
||
callbacks.emit("succeeded", **result)
|
||
return result
|
||
except GenerateError:
|
||
raise
|
||
except Exception as exc:
|
||
err = {"ok": False, "error": str(exc), "job_id": job_id, "work_dir": str(work_dir)}
|
||
callbacks.emit("failed", **err)
|
||
raise GenerateError(str(exc)) from exc
|
||
|
||
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
p = argparse.ArgumentParser(
|
||
prog="hunyuan_3d_generate",
|
||
description="腾讯混元3D 图生3D（微信登录 + Playwright + GLB 下载）",
|
||
)
|
||
p.add_argument("--image", "-i", help="参考图本地路径")
|
||
p.add_argument("--output", "-o", help="输出 GLB 路径（文件或目录）")
|
||
p.add_argument(
|
||
"--faces",
|
||
default=FACE_LABEL,
|
||
help=f"原始面数：{'/'.join(FACE_LABELS)}，默认 {FACE_LABEL}",
|
||
)
|
||
p.add_argument(
|
||
"--login",
|
||
action="store_true",
|
||
help="仅弹出微信扫码登录并保存凭证",
|
||
)
|
||
p.add_argument("--logout", action="store_true", help="清除本地混元登录凭证")
|
||
p.add_argument("--status", action="store_true", help="打印当前登录状态 JSON")
|
||
p.add_argument(
|
||
"--no-interactive-login",
|
||
action="store_true",
|
||
help="未登录时不弹窗，直接失败（适合 CI）",
|
||
)
|
||
p.add_argument(
|
||
"--login-timeout",
|
||
type=int,
|
||
default=300,
|
||
help="微信扫码等待秒数，默认 300",
|
||
)
|
||
p.add_argument(
|
||
"--force-login",
|
||
action="store_true",
|
||
help="忽略本地凭证，强制弹出登录页（换号）",
|
||
)
|
||
p.add_argument("--callback-url", default=None, help="流程回调 URL：每个事件 POST JSON")
|
||
p.add_argument(
|
||
"--callback-file",
|
||
default=None,
|
||
help="流程回调状态文件（持续覆盖写入 latest/history）",
|
||
)
|
||
p.add_argument(
|
||
"--json",
|
||
dest="json_input",
|
||
default=None,
|
||
help='JSON 参数（对象字符串 / 文件路径 / "-" 读 stdin）',
|
||
)
|
||
p.add_argument(
|
||
"--result-only",
|
||
action="store_true",
|
||
help="只输出最终结果 JSON（抑制中间 NDJSON 事件到 stdout，仍写回调）",
|
||
)
|
||
p.add_argument(
|
||
"--quiet",
|
||
action="store_true",
|
||
help="不向 stdout 打印事件流（最终结果仍打印）",
|
||
)
|
||
return p
|
||
|
||
|
||
def main(argv: Optional[Sequence[str]] = None) -> int:
|
||
parser = build_parser()
|
||
args = parser.parse_args(argv)
|
||
payload = _load_json_arg(args.json_input)
|
||
quiet_events = bool(args.quiet or args.result_only or payload.get("quiet"))
|
||
|
||
if args.logout or payload.get("logout"):
|
||
print(json.dumps(logout(), ensure_ascii=False), flush=True)
|
||
return 0
|
||
|
||
if args.status or payload.get("status"):
|
||
print(json.dumps(status(), ensure_ascii=False), flush=True)
|
||
return 0
|
||
|
||
if args.login or payload.get("login"):
|
||
callbacks = CallbackHub(
|
||
callback_url=payload.get("callback_url") or args.callback_url,
|
||
callback_file=payload.get("callback_file") or args.callback_file,
|
||
quiet=quiet_events,
|
||
)
|
||
update = _make_update(callbacks)
|
||
force = bool(args.force_login or payload.get("force_login"))
|
||
reason = str(payload.get("reason") or ("quota" if force else "auth"))
|
||
try:
|
||
result = start_wechat_login_blocking(
|
||
timeout_sec=int(payload.get("login_timeout") or args.login_timeout),
|
||
update=update,
|
||
reason=reason,
|
||
clear_existing=force or reason == "quota",
|
||
)
|
||
except Exception as exc: # noqa: BLE001
|
||
err = {"ok": False, "error": str(exc)}
|
||
print(json.dumps(err, ensure_ascii=False), flush=True)
|
||
return 1
|
||
print(json.dumps({"ok": True, **result}, ensure_ascii=False), flush=True)
|
||
return 0
|
||
|
||
image = payload.get("image") or args.image
|
||
output = payload.get("output") or args.output
|
||
if not image or not output:
|
||
parser.error("生成模式必须提供 --image/-i 与 --output/-o（或 JSON 同名字段）")
|
||
|
||
try:
|
||
result = run_generate(
|
||
image=image,
|
||
output=output,
|
||
faces=payload.get("faces") or args.faces,
|
||
interactive_login=not (
|
||
args.no_interactive_login or payload.get("no_interactive_login")
|
||
),
|
||
login_timeout=int(payload.get("login_timeout") or args.login_timeout),
|
||
callback_url=payload.get("callback_url") or args.callback_url,
|
||
callback_file=payload.get("callback_file") or args.callback_file,
|
||
quiet=quiet_events,
|
||
)
|
||
except GenerateError as exc:
|
||
err = {"ok": False, "error": str(exc)}
|
||
print(json.dumps(err, ensure_ascii=False), flush=True)
|
||
return 1
|
||
except Exception as exc: # noqa: BLE001
|
||
err = {"ok": False, "error": str(exc)}
|
||
print(json.dumps(err, ensure_ascii=False), flush=True)
|
||
return 2
|
||
|
||
print(json.dumps(result, ensure_ascii=False), flush=True)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|