Chanjing Video Compose

# 合规：根目录 合规规则.md §1–§2
name: chanjing-video-compose
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-video-compose
  author: chan-skills
  category: 媒体处理
  tags:
    - 视频合成
    - 数字人
    - ChanjingAPI
    - 蝉镜
  summary: >-
    从文本或音频合成数字人视频；可选背景上传、轮询；可按需下载成片。
  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

clientPermissions:
  network:
    httpsOutbound: true
    documentedHosts:
      - open-api.chanjing.cc
  filesystem:
    read:
      - "${CHANJING_OPENAPI_CREDENTIALS_DIR or CHANJING_CONFIG_DIR or ~/.chanjing}/credentials.json"
      - user_provided_background_media_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:
    mayDownloadFromApiResponseUrls: true
    optionalCallbackUrlForTaskResults: true

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

agentPolicy:
  alwaysSkill: false
  modifiesOtherSkillsOrGlobalAgent: false

# 鉴权：与 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 失败")

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

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

#!/usr/bin/env python3
"""
创建蝉镜视频合成任务。
用法:
  公共数字人文本驱动:
    create_task --person-id <id> --figure-type sit_body --text "台词" --audio-man <声音id> --subtitle show
  定制数字人文本驱动:
    create_task --person-id <id> --text "台词" --audio-man <声音id> --subtitle hide
  音频驱动:
    create_task --person-id <id> --audio-file-id <file_id>
    create_task --person-id <id> --wav-url https://example.com/demo.wav
输出: video_id（一行）或错误到 stderr
"""
import argparse
import json
import re
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 validate_hex_color(value, arg_name):
    if value is None:
        return
    if not re.fullmatch(r"#[0-9A-Fa-f]{6}", value):
        raise ValueError(f"{arg_name} 格式不正确，应为 #RRGGBB")


def get_default_subtitle_fields(args):
    # 官方文档给出了 1080p 和 4K 两组推荐值；只有在画布已放大到 4K 尺寸时才使用 4K 组。
    use_4k_defaults = args.resolution_rate == 1 and args.screen_width >= 2160 and args.screen_height >= 3840
    if use_4k_defaults:
        return {
            "x": 80,
            "y": 2840,
            "width": 2000,
            "height": 1000,
            "font_size": 150,
            "color": "#FFFFFF",
            "stroke_width": 7,
            "asr_type": 0,
        }
    return {
        "x": 31,
        "y": 1521,
        "width": 1000,
        "height": 200,
        "font_size": 64,
        "color": "#FFFFFF",
        "stroke_width": 7,
        "asr_type": 0,
    }


def build_subtitle_config(args):
    subtitle_fields = {
        "x": args.subtitle_x,
        "y": args.subtitle_y,
        "width": args.subtitle_width,
        "height": args.subtitle_height,
        "font_size": args.subtitle_font_size,
        "color": args.subtitle_color,
        "stroke_color": args.subtitle_stroke_color,
        "stroke_width": args.subtitle_stroke_width,
        "font_id": args.subtitle_font_id,
        "asr_type": args.subtitle_asr_type,
    }
    has_style_fields = any(value is not None for value in subtitle_fields.values())

    if args.hide_subtitle:
        if has_style_fields:
            raise ValueError("隐藏字幕时，不能同时传字幕位置或样式参数")
        return {"show": False}

    if args.subtitle is None:
        if has_style_fields:
            raise ValueError("传字幕位置或样式参数时，必须同时传 --subtitle show")
        return None

    config = {"show": args.subtitle == "show"}
    if not config["show"]:
        if has_style_fields:
            raise ValueError("使用 --subtitle hide 时，不能同时传字幕位置或样式参数")
        return config

    validate_hex_color(args.subtitle_color, "--subtitle-color")
    validate_hex_color(args.subtitle_stroke_color, "--subtitle-stroke-color")

    config.update(get_default_subtitle_fields(args))

    for key, value in subtitle_fields.items():
        if value is not None:
            config[key] = value

    if config["width"] > args.screen_width:
        raise ValueError("字幕宽度不能超过屏幕宽度")
    if config["height"] > args.screen_height:
        raise ValueError("字幕高度不能超过屏幕高度")
    if config["x"] + config["width"] > args.screen_width:
        raise ValueError("字幕区域超出屏幕宽度，请检查 --subtitle-x 和 --subtitle-width")
    if config["y"] + config["height"] > args.screen_height:
        raise ValueError("字幕区域超出屏幕高度，请检查 --subtitle-y 和 --subtitle-height")
    return config


def build_body(args):
    if args.audio_file_id and args.wav_url:
        raise ValueError("--audio-file-id 和 --wav-url 只能二选一")

    person = {
        "id": args.person_id,
        "x": args.person_x,
        "y": args.person_y,
        "width": args.person_width,
        "height": args.person_height,
    }
    if args.figure_type:
        person["figure_type"] = args.figure_type

    if args.audio_file_id or args.wav_url:
        audio = {
            "type": "audio",
            "volume": args.volume,
            "language": args.language,
        }
        if args.audio_file_id:
            audio["file_id"] = args.audio_file_id
        if args.wav_url:
            audio["wav_url"] = args.wav_url
    else:
        if not args.text or not args.audio_man:
            raise ValueError("文本驱动需同时提供 --text 和 --audio-man")
        if len(args.text) > 4000:
            raise ValueError("文本长度不能超过 4000 字符")
        audio = {
            "type": "tts",
            "volume": args.volume,
            "language": args.language,
            "tts": {
                "text": [args.text],
                "speed": args.speed,
                "audio_man": args.audio_man,
                "pitch": args.pitch,
            },
        }

    body = {
        "person": person,
        "audio": audio,
        "bg_color": args.bg_color,
        "screen_width": args.screen_width,
        "screen_height": args.screen_height,
        "model": args.model,
        "backway": args.backway,
        "add_compliance_watermark": args.add_compliance_watermark,
        "compliance_watermark_position": args.compliance_watermark_position,
        "resolution_rate": args.resolution_rate,
    }

    if args.drive_mode:
        body["drive_mode"] = args.drive_mode
    if args.callback:
        body["callback"] = args.callback
    if args.rgba_mode:
        body["is_rgba_mode"] = True
    subtitle_config = build_subtitle_config(args)
    if subtitle_config is not None:
        body["subtitle_config"] = subtitle_config

    if args.bg_file_id or args.bg_src_url:
        bg = {
            "x": args.bg_x,
            "y": args.bg_y,
            "width": args.bg_width,
            "height": args.bg_height,
        }
        if args.bg_file_id:
            bg["file_id"] = args.bg_file_id
        if args.bg_src_url:
            bg["src_url"] = args.bg_src_url
        body["bg"] = bg

    return body


def main():
    parser = argparse.ArgumentParser(description="创建蝉镜视频合成任务")
    parser.add_argument("--person-id", required=True, help="数字人形象 ID（来自 list_figures）")
    parser.add_argument("--text", help="文本驱动时的台词")
    parser.add_argument("--audio-man", help="文本驱动时的声音 ID，默认可用 list_figures 返回的 audio_man_id")
    parser.add_argument("--audio-file-id", help="音频驱动文件 ID（来自 upload_file）")
    parser.add_argument("--wav-url", help="音频驱动的远端音频链接")
    parser.add_argument("--person-x", type=int, default=0, help="人物 x 坐标")
    parser.add_argument("--person-y", type=int, default=0, help="人物 y 坐标")
    parser.add_argument("--person-width", type=int, default=1080, help="人物宽度")
    parser.add_argument("--person-height", type=int, default=1920, help="人物高度")
    parser.add_argument("--figure-type", help="公共数字人形态，如 sit_body / whole_body")
    parser.add_argument("--drive-mode", choices=["random"], help="驱动模式；不传表示正常驱动")
    parser.add_argument("--backway", type=int, choices=[1, 2], default=1, help="人物素材播放顺序")
    parser.add_argument("--rgba-mode", action="store_true", help="生成四通道 webm 视频")
    parser.add_argument("--screen-width", type=int, default=1080, help="画布宽度")
    parser.add_argument("--screen-height", type=int, default=1920, help="画布高度")
    parser.add_argument("--model", type=int, choices=[0, 1], default=0, help="0 基础版，1 高质版")
    parser.add_argument("--resolution-rate", type=int, choices=[0, 1], default=0, help="0 为 1080p，1 为 4K")
    parser.add_argument("--speed", type=float, default=1, help="TTS 语速 0.5-2")
    parser.add_argument("--pitch", type=float, default=1, help="TTS 音调")
    parser.add_argument("--volume", type=int, default=100, help="音量 1-100")
    parser.add_argument("--language", default="cn", help="语言类型，默认 cn")
    parser.add_argument("--bg-color", default="#EDEDED", help="纯色背景，默认 #EDEDED")
    parser.add_argument("--bg-file-id", help="背景素材 file_id（来自 upload_file）")
    parser.add_argument("--bg-src-url", help="背景图片远端链接，仅支持 jpg/png")
    parser.add_argument("--bg-x", type=int, default=0, help="背景 x 坐标")
    parser.add_argument("--bg-y", type=int, default=0, help="背景 y 坐标")
    parser.add_argument("--bg-width", type=int, default=1080, help="背景宽度")
    parser.add_argument("--bg-height", type=int, default=1920, help="背景高度")
    subtitle_group = parser.add_mutually_exclusive_group()
    subtitle_group.add_argument(
        "--subtitle",
        choices=["show", "hide"],
        help="显式设置字幕开关；show 时可继续配合字幕位置和样式参数",
    )
    subtitle_group.add_argument("--hide-subtitle", action="store_true", help="隐藏字幕（兼容旧用法）")
    parser.add_argument("--subtitle-x", type=int, help="字幕区域起始 x 坐标，基于左上角原点")
    parser.add_argument("--subtitle-y", type=int, help="字幕区域起始 y 坐标，基于左上角原点")
    parser.add_argument("--subtitle-width", type=int, help="字幕区域宽度")
    parser.add_argument("--subtitle-height", type=int, help="字幕区域高度")
    parser.add_argument("--subtitle-font-size", type=int, help="字幕字号")
    parser.add_argument("--subtitle-color", help="字幕颜色，格式 #RRGGBB")
    parser.add_argument("--subtitle-stroke-color", help="字幕描边颜色，格式 #RRGGBB")
    parser.add_argument("--subtitle-stroke-width", type=int, help="字幕描边宽度")
    parser.add_argument("--subtitle-font-id", help="字幕字体 ID")
    parser.add_argument(
        "--subtitle-asr-type",
        type=int,
        choices=[0, 1],
        help="字幕时间戳来源：0 自动生成，1 用户输入",
    )
    parser.add_argument("--callback", help="任务完成回调 URL")
    parser.add_argument("--add-compliance-watermark", action="store_true", help="添加 AI 合规水印")
    parser.add_argument(
        "--compliance-watermark-position",
        type=int,
        choices=[0, 1, 2, 3],
        default=0,
        help="合规水印位置：0 左上，1 右上，2 左下，3 右下",
    )
    args = parser.parse_args()

    try:
        body = build_body(args)
    except ValueError as exc:
        print(str(exc), file=sys.stderr)
        sys.exit(1)

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

    req = urllib.request.Request(
        f"{API_BASE}/open/v1/create_video",
        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:
        res = json.loads(resp.read().decode("utf-8"))

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

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


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
下载视频合成结果到本地。
用法:
  download_result --url https://example.com/output.mp4
  download_result --url https://example.com/output.mp4 --output outputs/video-compose/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 "video-compose.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="video_url")
    parser.add_argument(
        "--output",
        help="输出文件路径；默认保存到 outputs/video-compose/<文件名>",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    default_dir = Path("outputs") / "video-compose"
    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-video-compose-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
"""
列出可用于视频合成的数字人形象。
用法: list_figures.py [--source customised|common] [--page 1] [--page-size 20] [--json]
输出: 默认打印摘要表；--json 时输出完整 data
"""
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 fetch_customised(token, page, page_size):
    body = json.dumps({"page": page, "page_size": page_size}).encode("utf-8")
    req = urllib.request.Request(
        f"{API_BASE}/open/v1/list_customised_person",
        data=body,
        headers={"access_token": token, "Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        return json.loads(resp.read().decode("utf-8"))


def fetch_common(token, page, page_size):
    params = urllib.parse.urlencode({"page": page, "size": page_size})
    req = urllib.request.Request(
        f"{API_BASE}/open/v1/list_common_dp?{params}",
        headers={"access_token": token, "Content-Type": "application/json"},
        method="GET",
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        return json.loads(resp.read().decode("utf-8"))


def build_rows(source, items):
    rows = []
    if source == "customised":
        for item in items:
            width = item.get("width")
            height = item.get("height")
            size = f"{width}x{height}" if width and height else "-"
            rows.append(
                {
                    "source": "customised",
                    "person_id": item.get("id", ""),
                    "name": item.get("name", ""),
                    "figure_type": "-",
                    "size": size,
                    "audio_man_id": item.get("audio_man_id", ""),
                    "note": f"support_4k={item.get('support_4k', '')}",
                    "preview_url": item.get("preview_url", ""),
                }
            )
        return rows

    for item in items:
        for figure in item.get("figures", []):
            rows.append(
                {
                    "source": "common",
                    "person_id": item.get("id", ""),
                    "name": item.get("name", ""),
                    "figure_type": figure.get("type", ""),
                    "size": f"{figure.get('width', '')}x{figure.get('height', '')}",
                    "audio_man_id": item.get("audio_man_id", ""),
                    "note": f"audio_name={item.get('audio_name', '')}",
                    "preview_url": figure.get("preview_video_url", ""),
                }
            )
    return rows


def main():
    parser = argparse.ArgumentParser(description="列出蝉镜视频合成可用数字人形象")
    parser.add_argument(
        "--source",
        choices=["customised", "common"],
        default="customised",
        help="数字人来源：customised 为定制数字人，common 为公共数字人",
    )
    parser.add_argument("--page", type=int, default=1, help="页码")
    parser.add_argument("--page-size", type=int, default=20, help="每页数量")
    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)

    if args.source == "customised":
        res = fetch_customised(token, args.page, args.page_size)
    else:
        res = fetch_common(token, args.page, args.page_size)

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

    data = res.get("data", {})
    if args.json:
        print(json.dumps({"source": args.source, "data": data}, ensure_ascii=False, indent=2))
        return

    items = data.get("list", [])
    page_info = data.get("page_info", {})
    rows = build_rows(args.source, items)
    print(f"# source={args.source} 共 {page_info.get('total_count', len(items))} 个形象 (page={args.page})")
    print(
        f"{'source':<10}  {'person_id':<36}  {'name':<14}  {'figure_type':<12}  "
        f"{'size':<11}  {'audio_man_id':<36}  {'note':<24}  preview_url"
    )
    print("-" * 220)
    for row in rows:
        print(
            f"{row['source']:<10}  "
            f"{row['person_id']:<36}  "
            f"{row['name']:<14}  "
            f"{row['figure_type']:<12}  "
            f"{row['size']:<11}  "
            f"{row['audio_man_id']:<36}  "
            f"{row['note']:<24}  "
            f"{row['preview_url']}"
        )


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
轮询视频合成任务状态，直到完成或失败。
用法: poll_task.py --id <video_id> [--interval 10] [--json]
输出: 默认成功时打印 video_url；--json 时输出完整 data
状态: 10-生成中, 30-成功, 4X 参数异常, 5X 服务异常
"""
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("/")


def main():
    parser = argparse.ArgumentParser(description="轮询蝉镜视频合成任务直到完成")
    parser.add_argument("--id", required=True, help="视频任务 ID（来自 create_task）")
    parser.add_argument("--interval", type=int, default=10, help="轮询间隔秒数，默认 10")
    parser.add_argument("--json", action="store_true", help="成功后输出完整 JSON 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/video?id={urllib.parse.quote(args.id)}"

    while True:
        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)

        data = body.get("data", {})
        status = data.get("status")

        if status == 30:
            if args.json:
                print(json.dumps(data, ensure_ascii=False, indent=2))
                return 0
            video_url = data.get("video_url")
            if video_url:
                print(video_url)
                return 0
            print("任务成功但无 video_url", file=sys.stderr)
            sys.exit(1)

        if isinstance(status, int) and status >= 40:
            err_msg = data.get("msg") or f"status={status}"
            print(f"任务失败: {err_msg}", file=sys.stderr)
            sys.exit(1)

        time.sleep(args.interval)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
上传视频合成所需文件并轮询直到就绪，最后输出 file_id。
与 chanjing-credentials-guard 使用同一配置文件获取 Token。
用法:
  upload_file --service make_video_audio --file /path/to/input.wav
  upload_file --service make_video_background --file /path/to/bg.png
输出: file_id（一行）或错误到 stderr
"""
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_FAILED_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_FAILED_STATUSES:
            return False, data.get("msg") or f"file status={status}"

        time.sleep(interval)
    return False, "poll timeout"


def main():
    parser = argparse.ArgumentParser(description="上传视频合成文件并轮询直到就绪，返回 file_id")
    parser.add_argument(
        "--service",
        required=True,
        choices=["make_video_audio", "make_video_background", "ai_creation"],
        help="文件用途：音频、背景素材或 AI 创作",
    )
    parser.add_argument("--file", required=True, help="本地文件路径")
    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)

    qs = urllib.parse.urlencode({"service": args.service, "name": path.name})
    url = f"{API_BASE}/open/v1/common/create_upload_url?{qs}"
    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)

    data = body.get("data", {})
    sign_url = data.get("sign_url")
    mime_type = data.get("mime_type", "application/octet-stream")
    file_id = data.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 exc:
        print(f"上传失败: {exc}", 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()