Chanjing Customised Person

# 合规：根目录 合规规则.md §1–§2
# 网络：create_upload_url 返回的 sign_url 常指向独立对象存储域，主机名随蝉镜配置变化，
# 无法用固定 allowed_hosts 穷举，故 network_mode 为 open（见 clientPermissions.network 说明）。
name: chanjing-customised-person
version: 0.1.0
vendor: chanjing
runtime:
  interpreter: python3
  dependencies: []
env:
  required: []
  optional:
    - CHANJING_OPENAPI_CREDENTIALS_DIR
    - CHANJING_OPENAPI_BASE_URL
    - CHANJING_CONFIG_DIR
    - CHANJING_API_BASE
permissions:
  network_mode: open
  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-customised-person
  author: chan-skills
  category: 媒体处理
  tags:
    - 定制数字人
    - 数字人
    - ChanjingAPI
    - 蝉镜
  summary: >-
    蝉镜定制数字人：get_upload_url（create_upload_url）→ 对返回的 sign_url 执行 PUT 上传本地视频
    → 创建/列出/获取/轮询/删除定制形象；不引入其它云厂商凭证，仅 Open API 与接口给出的上传端点。
  skillDoc: SKILL.md
  siblingSkills:
    - chanjing-credentials-guard

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
  persistNote: >-
    access_token 与 expire_in 刷新后写回 credentials.json 为预期行为；须限制文件权限（如 0600/目录 0700）
    并确保永不提交版本库。

clientPermissions:
  network:
    httpsOutbound: true
    documentedHosts:
      - open-api.chanjing.cc
    signUrlPutNote: >-
      upload_file.py 在 GET create_upload_url 后，向响应中的 sign_url 发起 HTTPS PUT 上传视频正文；
      该 URL 的主机通常为蝉镜侧对象存储或 CDN，域名可能不同于 open-api.chanjing.cc，且可能随环境变化，
      故 permissions.network_mode 使用 open 而非固定主机白名单。
    userCallbackNote: >-
      create_person.py 可选 --callback：若提供 URL，远程服务可能将任务结果 POST 到用户指定端点，
      载荷可能含任务状态与资源引用等敏感信息；须自行信任该端点并承担出站与数据暴露风险。
  filesystem:
    read:
      - "${CHANJING_OPENAPI_CREDENTIALS_DIR or CHANJING_CONFIG_DIR or ~/.chanjing}/credentials.json"
      - user_provided_video_paths
    write:
      - "${CHANJING_OPENAPI_CREDENTIALS_DIR or CHANJING_CONFIG_DIR or ~/.chanjing}/credentials.json"
  browser:
    mayOpenForAuth: true
    documentedHosts:
      - www.chanjing.cc
  subprocess:
    allowedPatterns:
      - python3
  userContent:
    uploadsLocalVideoToApi: 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
"""
创建定制数字人任务。
用法:
  create_person --name "我的数字人" --file-id <file_id> [--train-type figure]
输出: 定制数字人 id（一行）
"""
import argparse
import json
import sys
import urllib.request
from pathlib import Path

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

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("/")


def main():
    parser = argparse.ArgumentParser(description="创建定制数字人任务")
    parser.add_argument("--name", required=True, help="定制数字人名称")
    parser.add_argument("--file-id", required=True, help="上传后的 file_id")
    parser.add_argument(
        "--callback",
        help="任务回调 URL（若提供，服务端可能向该地址 POST 任务结果，载荷可能含敏感信息）",
    )
    parser.add_argument(
        "--train-type",
        help="训练类型。文档表格明确写了 figure；错误码文案还出现 voice/both，若未确认能力，建议留空或使用 figure",
    )
    parser.add_argument("--language", choices=["cn", "en"], help="语种，默认由服务端决定")
    parser.add_argument(
        "--error-skip",
        action="store_true",
        help="跳过部分失败，目前主要用于跳过声音克隆失败",
    )
    parser.add_argument(
        "--resolution-rate",
        type=int,
        choices=[0, 1],
        default=0,
        help="数字人分辨率，0=1080p，1=4K",
    )
    args = parser.parse_args()

    body = {
        "name": args.name,
        "file_id": args.file_id,
        "error_skip": args.error_skip,
        "resolution_rate": args.resolution_rate,
    }
    if args.callback:
        body["callback"] = args.callback
    if args.train_type:
        body["train_type"] = args.train_type
    if args.language:
        body["language"] = args.language

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

    url = f"{API_BASE}/open/v1/create_customised_person"
    req = urllib.request.Request(
        url,
        data=json.dumps(body).encode("utf-8"),
        headers={"access_token": token, "Content-Type": "application/json; charset=utf-8"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        result = json.loads(resp.read().decode("utf-8"))

    if result.get("code") != 0:
        print(result.get("msg", result), file=sys.stderr)
        sys.exit(1)

    person_id = result.get("data")
    if not person_id:
        print("响应无 data", file=sys.stderr)
        sys.exit(1)

    print(person_id)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
删除定制数字人。
用法: delete_person.py --id <person_id>
输出: 被删除的 person_id
"""
import argparse
import json
import sys
import urllib.request
from pathlib import Path

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

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("/")


def main():
    parser = argparse.ArgumentParser(description="删除定制数字人")
    parser.add_argument("--id", required=True, help="定制数字人 id")
    args = parser.parse_args()

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

    req = urllib.request.Request(
        f"{API_BASE}/open/v1/delete_customised_person",
        data=json.dumps({"id": args.id}).encode("utf-8"),
        headers={"access_token": token, "Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        result = json.loads(resp.read().decode("utf-8"))

    if result.get("code") != 0:
        print(result.get("msg", result), file=sys.stderr)
        sys.exit(1)

    print(args.id)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
获取定制数字人详情。
用法: get_person.py --id <person_id> [--field preview_url]
默认输出: 详情 JSON
"""
import argparse
import json
import sys
import urllib.parse
import urllib.request
from pathlib import Path

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

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("/")


def main():
    parser = argparse.ArgumentParser(description="获取定制数字人详情")
    parser.add_argument("--id", required=True, help="定制数字人 id")
    parser.add_argument("--field", help="仅输出 data 下的某个字段")
    args = parser.parse_args()

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

    url = f"{API_BASE}/open/v1/customised_person?id={urllib.parse.quote(args.id)}"
    req = urllib.request.Request(url, headers={"access_token": token}, method="GET")
    with urllib.request.urlopen(req, timeout=30) as resp:
        result = json.loads(resp.read().decode("utf-8"))

    if result.get("code") != 0:
        print(result.get("msg", result), file=sys.stderr)
        sys.exit(1)

    data = result.get("data", {})
    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
"""
获取定制数字人素材上传链接（create_upload_url）。
客户端需自行对返回的 sign_url 执行 PUT 上传；sign_url 主机由接口决定，可能与 Open API 域不同。
用法: get_upload_url.py --name source.mp4 [--service customised_person]
输出: JSON，包含 sign_url / mime_type / file_id
"""
import argparse
import json
import sys
import urllib.parse
import urllib.request
from pathlib import Path

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

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("/")


def main():
    parser = argparse.ArgumentParser(description="获取定制数字人素材上传链接")
    parser.add_argument("--name", required=True, help="原始文件名，如 source.mp4")
    parser.add_argument(
        "--service",
        default="customised_person",
        help="文件用途，默认 customised_person",
    )
    args = parser.parse_args()

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

    query = urllib.parse.urlencode({"service": args.service, "name": args.name})
    url = f"{API_BASE}/open/v1/common/create_upload_url?{query}"
    req = urllib.request.Request(url, 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:
        print(body.get("msg", body), file=sys.stderr)
        sys.exit(1)

    print(json.dumps(body.get("data", {}), ensure_ascii=False))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
列出定制数字人形象。
用法: list_persons.py [--page 1] [--page-size 20] [--json]
默认输出: 制表分隔的摘要表
"""
import argparse
import json
import sys
import urllib.request
from pathlib import Path

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

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("/")


def main():
    parser = argparse.ArgumentParser(description="列出定制数字人形象")
    parser.add_argument("--page", type=int, default=1, help="页码，默认 1")
    parser.add_argument("--page-size", type=int, default=20, help="每页数量，默认 20")
    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)

    body = {"page": args.page, "page_size": args.page_size}
    req = urllib.request.Request(
        f"{API_BASE}/open/v1/list_customised_person",
        data=json.dumps(body).encode("utf-8"),
        headers={"access_token": token, "Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        result = json.loads(resp.read().decode("utf-8"))

    if result.get("code") != 0:
        print(result.get("msg", result), file=sys.stderr)
        sys.exit(1)

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

    items = result.get("data", {}).get("list", [])
    if not items:
        return

    for item in items:
        columns = [
            item.get("id", ""),
            item.get("name", ""),
            str(item.get("status", "")),
            str(item.get("progress", "")),
            item.get("audio_man_id", "") or "",
            str(item.get("support_4k", False)).lower(),
            item.get("preview_url", "") or "",
        ]
        print("\t".join(columns))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
轮询定制数字人状态，直到成功或失败。
用法: poll_person.py --id <person_id> [--interval 10] [--timeout 1800] [--json]
状态: 1=制作中, 2=成功, 4=失败, 5=系统错误
"""
import argparse
import json
import sys
import time
import urllib.parse
import urllib.request
from pathlib import Path

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

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("/")
SUCCESS_STATUS = 2
RUNNING_STATUS = 1
FAILED_STATUSES = {4, 5}


def fetch_detail(token, person_id):
    url = f"{API_BASE}/open/v1/customised_person?id={urllib.parse.quote(person_id)}"
    req = urllib.request.Request(url, 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:
        return None, body.get("msg", body)
    return body.get("data", {}), None


def main():
    parser = argparse.ArgumentParser(description="轮询定制数字人状态直到完成")
    parser.add_argument("--id", required=True, help="定制数字人 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:
        data, err = fetch_detail(token, args.id)
        if err:
            print(err, file=sys.stderr)
            sys.exit(1)

        status = data.get("status")
        if status == SUCCESS_STATUS:
            if args.json:
                print(json.dumps(data, ensure_ascii=False))
                return
            preview_url = data.get("preview_url")
            print(preview_url or data.get("id", ""))
            return

        if status in FAILED_STATUSES:
            message = data.get("err_reason") or data.get("reason") or "unknown"
            print(f"任务失败: {message}", file=sys.stderr)
            sys.exit(1)

        if status != RUNNING_STATUS:
            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
"""
上传定制数字人素材，并轮询到文件可用。
流程: GET create_upload_url（Open API）→ HTTPS PUT 至响应中的 sign_url（主机常为蝉镜侧存储域，非固定）→ 轮询 file_detail。
用法: upload_file.py --file /path/to/source.mp4 [--service customised_person]
输出: file_id（一行）
"""
import argparse
import json
import sys
import time
import urllib.parse
import urllib.request
from pathlib import Path

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

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("/")
FILE_READY_STATUSES = {1}
FILE_PENDING_STATUSES = {0, None}
FILE_FAILURE_STATUSES = {98, 99, 100}
POLL_INTERVAL_DEFAULT = 5
POLL_TIMEOUT_DEFAULT = 300


def get_file_detail(token, file_id):
    url = f"{API_BASE}/open/v1/common/file_detail?id={urllib.parse.quote(file_id)}"
    req = urllib.request.Request(url, 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:
        return None, body.get("msg", "file_detail failed")
    return body.get("data"), None


def poll_file_ready(token, file_id, interval=5, timeout=300):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        data, err = get_file_detail(token, file_id)
        if err:
            return False, err
        if data is None:
            return False, "no data"

        status = data.get("status")
        if status in FILE_READY_STATUSES:
            return True, None
        if status in FILE_FAILURE_STATUSES:
            return False, data.get("msg") or f"file status={status}"
        if status not in FILE_PENDING_STATUSES:
            return False, data.get("msg") or f"unexpected file status={status}"

        time.sleep(interval)

    return False, "poll timeout"


def main():
    parser = argparse.ArgumentParser(description="上传定制数字人素材并轮询到文件可用")
    parser.add_argument("--file", required=True, help="本地文件路径")
    parser.add_argument(
        "--service",
        default="customised_person",
        help="文件用途，默认 customised_person",
    )
    parser.add_argument("--poll-interval", type=int, default=POLL_INTERVAL_DEFAULT, help="轮询间隔秒数")
    parser.add_argument("--poll-timeout", type=int, default=POLL_TIMEOUT_DEFAULT, help="轮询超时秒数")
    args = parser.parse_args()

    path = Path(args.file)
    if not path.is_file():
        print(f"文件不存在: {path}", file=sys.stderr)
        sys.exit(1)

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

    query = urllib.parse.urlencode({"service": args.service, "name": path.name})
    url = f"{API_BASE}/open/v1/common/create_upload_url?{query}"
    req = urllib.request.Request(url, 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:
        print(body.get("msg", body), file=sys.stderr)
        sys.exit(1)

    payload = body.get("data", {})
    sign_url = payload.get("sign_url")
    mime_type = payload.get("mime_type", "application/octet-stream")
    file_id = payload.get("file_id")

    if not sign_url or not file_id:
        print("响应缺少 sign_url 或 file_id", file=sys.stderr)
        sys.exit(1)

    with open(path, "rb") as f:
        content = f.read()
    put_req = urllib.request.Request(
        sign_url,
        data=content,
        headers={"Content-Type": mime_type},
        method="PUT",
    )
    try:
        with urllib.request.urlopen(put_req, timeout=120) as put_resp:
            if put_resp.status not in (200, 204):
                print(f"上传返回状态异常: {put_resp.status}", file=sys.stderr)
                sys.exit(1)
    except Exception as e:
        print(f"上传失败: {e}", file=sys.stderr)
        sys.exit(1)

    ready, err = poll_file_ready(
        token,
        file_id,
        interval=args.poll_interval,
        timeout=args.poll_timeout,
    )
    if not ready:
        print(f"文件未就绪: {err}", file=sys.stderr)
        sys.exit(1)

    print(file_id)


if __name__ == "__main__":
    main()