Chanjing Text To Digital Person

# 合规：根目录 合规规则.md §1–§2
name: chanjing-text-to-digital-person
version: 0.1.0
vendor: chanjing
runtime:
  interpreter: python3
  dependencies: []
env:
  required: []
  optional:
    - CHANJING_OPENAPI_CREDENTIALS_DIR
    - CHANJING_OPENAPI_BASE_URL
permissions:
  network_mode: allowlist
  allowed_hosts:
    - open-api.chanjing.cc
    - www.chanjing.cc
  filesystem:
    read_roots:
      - "${WORKSPACE_ROOT}"
      - "${SKILL_DIR}"
      - "${CHANJING_OPENAPI_CREDENTIALS_DIR}"
    write_roots:
      - "${WORKSPACE_ROOT}"
      - "${CHANJING_OPENAPI_CREDENTIALS_DIR}"
  allowed_commands:
    - python3

schemaVersion: 1
skill:
  id: chanjing-text-to-digital-person
  author: chan-skills
  category: 媒体处理
  tags:
    - 文生数字人
    - AIGC
    - ChanjingAPI
    - 蝉镜
  summary: >-
    文生图、图生说话视频、可选 LoRA 训练与轮询；可按需下载生成物。
  skillDoc: SKILL.md

environment:
  variables:
    - name: CHANJING_OPENAPI_CREDENTIALS_DIR
      required: false
      description: 存放 credentials.json 的目录（兼容 CHANJING_CONFIG_DIR），默认 ~/.chanjing
    - name: CHANJING_OPENAPI_BASE_URL
      required: false
      description: Open API 基址（兼容 CHANJING_API_BASE），默认 https://open-api.chanjing.cc

credentials:
  model: credentials_json
  defaultPath: "~/.chanjing/credentials.json"
  directoryEnv: CHANJING_OPENAPI_CREDENTIALS_DIR
  fileName: credentials.json
  sensitiveFields:
    - app_id
    - secret_key
    - access_token
    - expire_in
  persistAccessTokenOnDisk: true
  primaryEnvIntentionallyOmitted: true
  doNotCommitToVcs:
    - credentials.json

clientPermissions:
  network:
    httpsOutbound: true
    documentedHosts:
      - open-api.chanjing.cc
  filesystem:
    read:
      - "${CHANJING_OPENAPI_CREDENTIALS_DIR or CHANJING_CONFIG_DIR or ~/.chanjing}/credentials.json"
    write:
      - "${CHANJING_OPENAPI_CREDENTIALS_DIR or CHANJING_CONFIG_DIR or ~/.chanjing}/credentials.json"
  browser:
    mayOpenForAuth: true
    documentedHosts:
      - www.chanjing.cc
  subprocess:
    allowedPatterns:
      - python3
  userContent:
    mayDownloadFromApiResponseUrls: true

metadata:
  openclaw:
    homepage: https://doc.chanjing.cc

agentPolicy:
  alwaysSkill: false
  modifiesOtherSkillsOrGlobalAgent: false

#!/usr/bin/env python3
# 鉴权：与 chanjing-credentials-guard 使用同一配置文件（CONFIG_DIR/credentials.json）
# 无 AK/SK 时执行 open_login_page.py 打开注册/登录页
import json
import os
import subprocess
import sys
import time
import urllib.request
from pathlib import Path

_DEFAULT_OPENAPI_BASE = "https://open-api.chanjing.cc"


def credentials_config_dir() -> Path:
    raw = os.environ.get("CHANJING_OPENAPI_CREDENTIALS_DIR") or os.environ.get("CHANJING_CONFIG_DIR")
    return Path(raw).expanduser() if raw else Path.home() / ".chanjing"


def openapi_base_url() -> str:
    return (
        os.environ.get("CHANJING_OPENAPI_BASE_URL")
        or os.environ.get("CHANJING_API_BASE")
        or _DEFAULT_OPENAPI_BASE
    ).rstrip("/")


CONFIG_DIR = credentials_config_dir()
CONFIG_FILE = CONFIG_DIR / "credentials.json"
API_BASE = openapi_base_url()
BUFFER_SECONDS = 300
LOGIN_URL = "https://www.chanjing.cc/openapi/login"

NO_CREDENTIALS_MSG = """已在浏览器打开蝉镜登录/注册页。
获取秘钥后请执行：
  python skills/chanjing-credentials-guard/scripts/chanjing_config.py --ak <你的app_id> --sk <你的secret_key>
设置完毕后请重新执行您之前的操作。"""


def _run_open_login_page():
    """执行 credentials-guard 的 open_login_page.py，在默认浏览器打开注册/登录页。"""
    try:
        skills_dir = Path(__file__).resolve().parent.parent.parent
        script = skills_dir / "chanjing-credentials-guard" / "scripts" / "open_login_page.py"
        if script.exists():
            subprocess.run([sys.executable, str(script)], check=False, timeout=5)
        else:
            import webbrowser

            webbrowser.open(LOGIN_URL)
    except Exception:
        try:
            import webbrowser

            webbrowser.open(LOGIN_URL)
        except Exception:
            pass


def read_config():
    if CONFIG_FILE.exists():
        with open(CONFIG_FILE, "r", encoding="utf-8") as f:
            return json.load(f)
    return {}


def write_config(data):
    CONFIG_DIR.mkdir(parents=True, exist_ok=True)
    with open(CONFIG_FILE, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)


def get_token():
    """返回 (token, None) 或 (None, error_msg)。"""
    data = read_config()
    app_id = (data.get("app_id") or "").strip()
    secret_key = (data.get("secret_key") or "").strip()
    if not app_id or not secret_key:
        _run_open_login_page()
        return None, NO_CREDENTIALS_MSG

    now = int(time.time())
    token = data.get("access_token")
    expire_in = data.get("expire_in")
    try:
        expire_in = int(expire_in) if expire_in is not None else 0
    except (ValueError, TypeError):
        expire_in = 0

    if token and expire_in > now + BUFFER_SECONDS:
        return token, None

    url = API_BASE + "/open/v1/access_token"
    req = urllib.request.Request(
        url,
        data=json.dumps({"app_id": app_id, "secret_key": secret_key}).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            body = json.loads(resp.read().decode("utf-8"))
    except Exception as e:
        return None, str(e)

    if body.get("code") != 0:
        return None, body.get("msg", "获取 Token 失败")

    payload = body.get("data", {})
    new_token = payload.get("access_token")
    if not new_token:
        return None, "API 返回无 access_token"

    data["access_token"] = new_token
    data["expire_in"] = payload.get("expire_in")
    write_config(data)
    return new_token, None


def main():
    token, err = get_token()
    if err:
        raise SystemExit(err)
    print(token)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
import json
import urllib.parse
import urllib.request

API_BASE = (__import__("os").environ.get("CHANJING_OPENAPI_BASE_URL") or __import__("os").environ.get("CHANJING_API_BASE") or "https://open-api.chanjing.cc").rstrip("/")

PHOTO_RUNNING = {"Ready", "Generating", "Queued"}
PHOTO_SUCCESS = {"Success"}
PHOTO_FAILED = {"Error", "Fail"}

MOTION_RUNNING = {"Ready", "Generating", "Queued"}
MOTION_SUCCESS = {"Success"}
MOTION_FAILED = {"Error", "Fail"}

LORA_RUNNING = {"Queued", "Published", "Generating"}
LORA_SUCCESS = {"Success"}
LORA_FAILED = {"Fail"}


def api_get(token, path, query=None):
    query = query or {}
    suffix = ""
    if query:
        suffix = "?" + urllib.parse.urlencode(query)
    req = urllib.request.Request(
        f"{API_BASE}{path}{suffix}",
        headers={"access_token": token},
        method="GET",
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        body = json.loads(resp.read().decode("utf-8"))
    if body.get("code") != 0:
        raise RuntimeError(body.get("msg", body))
    return body.get("data")


def api_post(token, path, payload):
    req = urllib.request.Request(
        f"{API_BASE}{path}",
        data=json.dumps(payload).encode("utf-8"),
        headers={"access_token": token, "Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        body = json.loads(resp.read().decode("utf-8"))
    if body.get("code") != 0:
        raise RuntimeError(body.get("msg", body))
    return body.get("data")


def get_photo_task(token, unique_id):
    return api_get(token, "/open/v1/aigc/photo/task", {"unique_id": unique_id})


def list_photo_tasks(token, page=1, page_size=10):
    return api_get(token, "/open/v1/aigc/photo/task/page", {"page": page, "page_size": page_size})


def get_motion_task(token, unique_id):
    return api_get(token, "/open/v1/aigc/motion/task", {"unique_id": unique_id})


def get_lora_task(token, lora_id):
    return api_get(token, "/open/v1/aigc/lora/task", {"lora_id": lora_id})


def first_output_url(data):
    urls = (data or {}).get("output_url") or []
    if isinstance(urls, list) and urls:
        return urls[0]
    return None

#!/usr/bin/env python3
"""
创建 LoRA 训练任务。
用法:
  create_lora_task --name "我的LoRA" --photo-url https://a/1.jpg --photo-url https://a/2.jpg ...
输出: lora_id
"""
import argparse
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from _auth import get_token
from _task_api import api_post


def main():
    parser = argparse.ArgumentParser(description="创建文生数字人 LoRA 任务")
    parser.add_argument("--name", required=True, help="LoRA 名称")
    parser.add_argument("--photo-url", action="append", required=True, help="训练照片 URL，至少 5 张，可重复传参")
    parser.add_argument("--lora-id", help="失败任务重试时传入已有 lora_id")
    args = parser.parse_args()

    if len(args.photo_url) < 5 or len(args.photo_url) > 50:
        print("照片素材数量必须在 5 到 50 张之间", file=sys.stderr)
        sys.exit(1)

    body = {
        "name": args.name,
        "photos": args.photo_url,
    }
    if args.lora_id:
        body["lora_id"] = args.lora_id

    token, err = get_token()
    if err:
        print(err, file=sys.stderr)
        sys.exit(1)

    try:
        data = api_post(token, "/open/v1/aigc/lora/task/create", body)
    except Exception as exc:
        print(str(exc), file=sys.stderr)
        sys.exit(1)

    lora_id = (data or {}).get("lora_id")
    if not lora_id:
        print("响应无 lora_id", file=sys.stderr)
        sys.exit(1)
    print(lora_id)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
创建图生视频任务。
用法: create_motion_task.py --photo-unique-id <photo_task_id> --photo-path <image_url> [--emotion ...] [--gesture]
输出: 图生视频任务 unique_id
"""
import argparse
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from _auth import get_token
from _task_api import api_post


def main():
    parser = argparse.ArgumentParser(description="创建图生视频任务")
    parser.add_argument("--photo-unique-id", required=True, help="来源文生图任务 ID")
    parser.add_argument("--photo-path", required=True, help="图片地址")
    parser.add_argument("--emotion", help="情感提示词")
    parser.add_argument("--gesture", action="store_true", help="启用动作")
    args = parser.parse_args()

    body = {
        "photo_unique_id": args.photo_unique_id,
        "photo_path": args.photo_path,
        "gesture": args.gesture,
    }
    if args.emotion:
        body["emotion"] = args.emotion

    token, err = get_token()
    if err:
        print(err, file=sys.stderr)
        sys.exit(1)

    try:
        unique_id = api_post(token, "/open/v1/aigc/motion", body)
    except Exception as exc:
        print(str(exc), file=sys.stderr)
        sys.exit(1)

    if not unique_id:
        print("响应无图生视频任务 ID", file=sys.stderr)
        sys.exit(1)
    print(unique_id)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
创建文生图任务。
用法:
  create_photo_task --age "Young adult" --gender Female --number-of-images 1 [其他提示词]
输出: 文生图任务 unique_id
"""
import argparse
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from _auth import get_token
from _task_api import api_post


def main():
    parser = argparse.ArgumentParser(description="创建蝉镜文生图任务")
    parser.add_argument("--age", required=True, help="Young adult / Adult / Teenager / Elderly")
    parser.add_argument("--gender", required=True, choices=["Male", "Female"], help="性别")
    parser.add_argument("--number-of-images", required=True, type=int, choices=[1, 2, 3, 4], help="生成图片张数")
    parser.add_argument("--background", help="背景提示词")
    parser.add_argument("--detail", help="细节提示词")
    parser.add_argument("--talking-pose", help="讲话姿势提示词")
    parser.add_argument("--industry", help="行业提示词")
    parser.add_argument("--origin", help="人种提示词，例如 Chinese / European")
    parser.add_argument("--aspect-ratio", type=int, choices=[0, 1], default=0, help="0=9:16, 1=16:9")
    parser.add_argument("--ref-img-url", help="参考图链接")
    parser.add_argument("--ref-content", choices=["style", "appearance"], help="参考内容类型")
    args = parser.parse_args()

    body = {
        "age": args.age,
        "gender": args.gender,
        "number_of_images": args.number_of_images,
        "aspect_ratio": args.aspect_ratio,
    }
    if args.background:
        body["background"] = args.background
    if args.detail:
        body["detail"] = args.detail
    if args.talking_pose:
        body["talking_pose"] = args.talking_pose
    if args.industry:
        body["industry"] = args.industry
    if args.origin:
        body["origin"] = args.origin
    if args.ref_img_url:
        body["ref_img_url"] = args.ref_img_url
    if args.ref_content:
        body["ref_content"] = args.ref_content

    token, err = get_token()
    if err:
        print(err, file=sys.stderr)
        sys.exit(1)

    try:
        unique_id = api_post(token, "/open/v1/aigc/photo", body)
    except Exception as exc:
        print(str(exc), file=sys.stderr)
        sys.exit(1)

    if not unique_id:
        print("响应无文生图任务 ID", file=sys.stderr)
        sys.exit(1)
    print(unique_id)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
下载文生数字人生成结果到本地。
用法:
  download_result --url https://example.com/output.png
  download_result --url https://example.com/output.mp4 --output outputs/text-to-digital-person/demo.mp4
输出: 本地文件路径
"""
from __future__ import annotations

import argparse
import os
import sys
import urllib.parse
import urllib.request
from pathlib import Path


def infer_filename(url: str) -> str:
    parsed = urllib.parse.urlparse(url)
    name = Path(parsed.path).name or "text-to-digital-person.bin"
    if "." not in name:
        name += ".bin"
    return name


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="下载蝉镜文生数字人生成结果到本地目录")
    parser.add_argument("--url", required=True, help="图片或视频输出地址")
    parser.add_argument(
        "--output",
        help="输出文件路径；默认保存到 outputs/text-to-digital-person/<文件名>",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    default_dir = Path("outputs") / "text-to-digital-person"
    output_path = Path(args.output) if args.output else default_dir / infer_filename(args.url)

    output_path.parent.mkdir(parents=True, exist_ok=True)

    req = urllib.request.Request(
        args.url,
        headers={"User-Agent": "chanjing-text-to-digital-person-downloader"},
        method="GET",
    )

    try:
        with urllib.request.urlopen(req, timeout=120) as resp, open(output_path, "wb") as handle:
            handle.write(resp.read())
    except Exception as exc:
        print(f"下载失败: {exc}", file=sys.stderr)
        raise SystemExit(1)

    if not output_path.exists() or output_path.stat().st_size == 0:
        print("下载失败: 输出文件为空", file=sys.stderr)
        raise SystemExit(1)

    print(os.fspath(output_path))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
获取 LoRA 任务详情。
用法: get_lora_task.py --lora-id <id> [--field status]
默认输出: 完整 JSON
"""
import argparse
import json
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from _auth import get_token
from _task_api import get_lora_task


def main():
    parser = argparse.ArgumentParser(description="获取 LoRA 任务详情")
    parser.add_argument("--lora-id", required=True, help="LoRA 任务 ID")
    parser.add_argument("--field", help="只输出某个字段")
    args = parser.parse_args()

    token, err = get_token()
    if err:
        print(err, file=sys.stderr)
        sys.exit(1)

    try:
        data = get_lora_task(token, args.lora_id)
    except Exception as exc:
        print(str(exc), file=sys.stderr)
        sys.exit(1)

    if args.field:
        value = data.get(args.field)
        if value is None:
            print(f"字段不存在: {args.field}", file=sys.stderr)
            sys.exit(1)
        if isinstance(value, (dict, list)):
            print(json.dumps(value, ensure_ascii=False))
        else:
            print(value)
        return

    print(json.dumps(data, ensure_ascii=False))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
获取图生视频任务详情。
用法: get_motion_task.py --unique-id <id> [--field progress_desc] [--first-output]
默认输出: 完整 JSON
"""
import argparse
import json
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from _auth import get_token
from _task_api import first_output_url, get_motion_task


def main():
    parser = argparse.ArgumentParser(description="获取图生视频任务详情")
    parser.add_argument("--unique-id", required=True, help="图生视频任务 unique_id")
    parser.add_argument("--field", help="只输出某个字段")
    parser.add_argument("--first-output", action="store_true", help="只输出第一个输出地址")
    args = parser.parse_args()

    token, err = get_token()
    if err:
        print(err, file=sys.stderr)
        sys.exit(1)

    try:
        data = get_motion_task(token, args.unique_id)
    except Exception as exc:
        print(str(exc), file=sys.stderr)
        sys.exit(1)

    if args.first_output:
        value = first_output_url(data)
        if not value:
            print("当前无 output_url", file=sys.stderr)
            sys.exit(1)
        print(value)
        return

    if args.field:
        value = data.get(args.field)
        if value is None:
            print(f"字段不存在: {args.field}", file=sys.stderr)
            sys.exit(1)
        if isinstance(value, (dict, list)):
            print(json.dumps(value, ensure_ascii=False))
        else:
            print(value)
        return

    print(json.dumps(data, ensure_ascii=False))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
获取文生图任务详情。
用法: get_photo_task.py --unique-id <id> [--field progress_desc] [--first-output]
默认输出: 完整 JSON
"""
import argparse
import json
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from _auth import get_token
from _task_api import first_output_url, get_photo_task


def main():
    parser = argparse.ArgumentParser(description="获取文生图任务详情")
    parser.add_argument("--unique-id", required=True, help="文生图任务 unique_id")
    parser.add_argument("--field", help="只输出某个字段")
    parser.add_argument("--first-output", action="store_true", help="只输出第一张图片地址")
    args = parser.parse_args()

    token, err = get_token()
    if err:
        print(err, file=sys.stderr)
        sys.exit(1)

    try:
        data = get_photo_task(token, args.unique_id)
    except Exception as exc:
        print(str(exc), file=sys.stderr)
        sys.exit(1)

    if args.first_output:
        value = first_output_url(data)
        if not value:
            print("当前无 output_url", file=sys.stderr)
            sys.exit(1)
        print(value)
        return

    if args.field:
        value = data.get(args.field)
        if value is None:
            print(f"字段不存在: {args.field}", file=sys.stderr)
            sys.exit(1)
        if isinstance(value, (dict, list)):
            print(json.dumps(value, ensure_ascii=False))
        else:
            print(value)
        return

    print(json.dumps(data, ensure_ascii=False))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
列出文生数字人相关任务列表。
基于 GET /open/v1/aigc/photo/task/page，返回的 type=1 为 photo，type=2 为 motion。
"""
import argparse
import json
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from _auth import get_token
from _task_api import first_output_url, list_photo_tasks


def main():
    parser = argparse.ArgumentParser(description="列出文生数字人任务列表")
    parser.add_argument("--page", type=int, default=1, help="页码，默认 1")
    parser.add_argument("--page-size", type=int, default=10, help="每页数量，默认 10")
    parser.add_argument("--json", action="store_true", help="输出完整 JSON")
    args = parser.parse_args()

    token, err = get_token()
    if err:
        print(err, file=sys.stderr)
        sys.exit(1)

    try:
        items = list_photo_tasks(token, page=args.page, page_size=args.page_size)
    except Exception as exc:
        print(str(exc), file=sys.stderr)
        sys.exit(1)

    if args.json:
        print(json.dumps(items, ensure_ascii=False))
        return

    for item in items or []:
        row = [
            item.get("unique_id", ""),
            str(item.get("type", "")),
            item.get("progress_desc", ""),
            str(item.get("waiting_num", "")),
            first_output_url(item) or "",
        ]
        print("\t".join(row))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
轮询 LoRA 任务直到完成。
默认输出第一条 photo_task_id；可用 --json 输出完整详情。
"""
import argparse
import json
import sys
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from _auth import get_token
from _task_api import LORA_FAILED, LORA_RUNNING, LORA_SUCCESS, get_lora_task


def main():
    parser = argparse.ArgumentParser(description="轮询 LoRA 任务直到完成")
    parser.add_argument("--lora-id", required=True, help="LoRA 任务 ID")
    parser.add_argument("--interval", type=int, default=10, help="轮询间隔秒数，默认 10")
    parser.add_argument("--timeout", type=int, default=1800, help="轮询超时秒数，默认 1800")
    parser.add_argument("--json", action="store_true", help="成功时输出完整 JSON")
    args = parser.parse_args()

    token, err = get_token()
    if err:
        print(err, file=sys.stderr)
        sys.exit(1)

    deadline = time.monotonic() + args.timeout
    while time.monotonic() < deadline:
        try:
            data = get_lora_task(token, args.lora_id)
        except Exception as exc:
            print(str(exc), file=sys.stderr)
            sys.exit(1)

        status = data.get("status")
        if status in LORA_SUCCESS:
            if args.json:
                print(json.dumps(data, ensure_ascii=False))
                return
            photo_task_ids = data.get("photo_task_ids") or []
            if not photo_task_ids:
                print("任务成功但无 photo_task_ids", file=sys.stderr)
                sys.exit(1)
            print(photo_task_ids[0])
            return

        if status in LORA_FAILED:
            print(f"任务失败: {data.get('err_msg') or status}", file=sys.stderr)
            sys.exit(1)

        if status not in LORA_RUNNING:
            print(f"未知任务状态: {status}", file=sys.stderr)
            sys.exit(1)

        time.sleep(args.interval)

    print("轮询超时", file=sys.stderr)
    sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
轮询图生视频任务直到完成。
默认输出第一个视频地址；可用 --json 输出完整详情。
"""
import argparse
import json
import sys
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from _auth import get_token
from _task_api import MOTION_FAILED, MOTION_RUNNING, MOTION_SUCCESS, first_output_url, get_motion_task


def main():
    parser = argparse.ArgumentParser(description="轮询图生视频任务直到完成")
    parser.add_argument("--unique-id", required=True, help="图生视频任务 unique_id")
    parser.add_argument("--interval", type=int, default=10, help="轮询间隔秒数，默认 10")
    parser.add_argument("--timeout", type=int, default=1800, help="轮询超时秒数，默认 1800")
    parser.add_argument("--json", action="store_true", help="成功时输出完整 JSON")
    args = parser.parse_args()

    token, err = get_token()
    if err:
        print(err, file=sys.stderr)
        sys.exit(1)

    deadline = time.monotonic() + args.timeout
    while time.monotonic() < deadline:
        try:
            data = get_motion_task(token, args.unique_id)
        except Exception as exc:
            print(str(exc), file=sys.stderr)
            sys.exit(1)

        status = data.get("progress_desc")
        if status in MOTION_SUCCESS:
            if args.json:
                print(json.dumps(data, ensure_ascii=False))
                return
            first_url = first_output_url(data)
            if not first_url:
                print("任务成功但无 output_url", file=sys.stderr)
                sys.exit(1)
            print(first_url)
            return

        if status in MOTION_FAILED:
            print(f"任务失败: {data.get('err_msg') or status}", file=sys.stderr)
            sys.exit(1)

        if status not in MOTION_RUNNING:
            print(f"未知任务状态: {status}", file=sys.stderr)
            sys.exit(1)

        time.sleep(args.interval)

    print("轮询超时", file=sys.stderr)
    sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
轮询文生图任务直到完成。
默认输出第一张图片地址；可用 --json 输出完整详情。
"""
import argparse
import json
import sys
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from _auth import get_token
from _task_api import PHOTO_FAILED, PHOTO_RUNNING, PHOTO_SUCCESS, first_output_url, get_photo_task


def main():
    parser = argparse.ArgumentParser(description="轮询文生图任务直到完成")
    parser.add_argument("--unique-id", required=True, help="文生图任务 unique_id")
    parser.add_argument("--interval", type=int, default=10, help="轮询间隔秒数，默认 10")
    parser.add_argument("--timeout", type=int, default=1800, help="轮询超时秒数，默认 1800")
    parser.add_argument("--json", action="store_true", help="成功时输出完整 JSON")
    parser.add_argument("--all-urls", action="store_true", help="成功时输出全部 output_url JSON 数组")
    args = parser.parse_args()

    token, err = get_token()
    if err:
        print(err, file=sys.stderr)
        sys.exit(1)

    deadline = time.monotonic() + args.timeout
    while time.monotonic() < deadline:
        try:
            data = get_photo_task(token, args.unique_id)
        except Exception as exc:
            print(str(exc), file=sys.stderr)
            sys.exit(1)

        status = data.get("progress_desc")
        if status in PHOTO_SUCCESS:
            if args.json:
                print(json.dumps(data, ensure_ascii=False))
                return
            urls = data.get("output_url") or []
            if args.all_urls:
                print(json.dumps(urls, ensure_ascii=False))
                return
            first_url = first_output_url(data)
            if not first_url:
                print("任务成功但无 output_url", file=sys.stderr)
                sys.exit(1)
            print(first_url)
            return

        if status in PHOTO_FAILED:
            print(f"任务失败: {data.get('err_msg') or status}", file=sys.stderr)
            sys.exit(1)

        if status not in PHOTO_RUNNING:
            print(f"未知任务状态: {status}", file=sys.stderr)
            sys.exit(1)

        time.sleep(args.interval)

    print("轮询超时", file=sys.stderr)
    sys.exit(1)


if __name__ == "__main__":
    main()