#!/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.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 = 180, callback_url: Optional[str] = None, callback_file: Optional[str] = None, quiet: bool = False, ) -> 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 ) break except PermissionError as exc: if not interactive_login or attempts >= 2: raise GenerateError(str(exc)) from exc update( event="login_required", stage="凭证失效，重新微信登录", progress=3, error=str(exc), ) # 清掉失效会话再扫码 if is_logged_in(): logout() start_wechat_login_blocking( timeout_sec=login_timeout, update=update ) 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=180, help="微信扫码等待秒数，默认 180", ) 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) try: result = start_wechat_login_blocking( timeout_sec=int(payload.get("login_timeout") or args.login_timeout), update=update, ) 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())