Chanjing Tts Voice Clone

# 合规：根目录 合规规则.md §1–§2
name: chanjing-tts-voice-clone
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-tts-voice-clone
  author: chan-skills
  category: 媒体处理
  tags:
    - TTS
    - 声音克隆
    - ChanjingAPI
    - 蝉镜
  summary: >-
    参考音频公网 URL 创建音色并 TTS 合成；轮询任务并从接口 URL 下载结果。
  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"
    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
    userSuppliedPublicReferenceAudioUrlFetchedByChanjingServers: 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
"""
使用定制声音创建 TTS 语音合成任务。voice_id 需先通过 create_voice + poll_voice 获得。
与 chanjing-credentials-guard 使用同一配置文件；无 AK/SK 时执行 open_login_page.py 打开登录页。
用法: create_task.py --audio-man <voice_id> --text "要合成的文本" [--speed 1] [--pitch 1] [--aigc-watermark]
输出: task_id（一行）或错误到 stderr
"""
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="使用定制声音创建 TTS 任务")
    parser.add_argument("--audio-man", required=True, help="定制声音 ID（来自 poll_voice）")
    parser.add_argument("--text", required=True, help="要合成的文本，不超过 4000 字")
    parser.add_argument("--speed", type=float, default=1, help="语速 0.5-2，默认 1")
    parser.add_argument("--pitch", type=float, default=1, help="语调，默认 1")
    parser.add_argument("--aigc-watermark", action="store_true", help="是否添加 AIGC 水印")
    args = parser.parse_args()

    if len(args.text) > 4000:
        print("文本长度不能超过 4000 字符", file=sys.stderr)
        sys.exit(1)

    body = {
        "audio_man": args.audio_man.strip(),
        "speed": args.speed,
        "pitch": args.pitch,
        "text": {"text": args.text},
        "aigc_watermark": args.aigc_watermark,
    }

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

    url = f"{API_BASE}/open/v1/create_audio_task"
    req = urllib.request.Request(
        url,
        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)

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


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
创建定制声音（声音克隆）。需提供参考音频的公开 URL。
与 chanjing-credentials-guard 使用同一配置文件；无 AK/SK 时执行 open_login_page.py 打开登录页。
用法: create_voice.py --name "我的声音" --url "https://example.com/ref.mp3" [--model-type Cicada3.0-turbo] [--language cn]
输出: voice_id（一行）或错误到 stderr
"""
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="创建定制声音（参考音频 URL）")
    parser.add_argument("--name", required=True, help="声音名称")
    parser.add_argument("--url", required=True, help="参考音频的公开 URL（mp3/wav/m4a，建议 30s–5min）")
    parser.add_argument("--model-type", default="Cicada3.0-turbo", help="模型类型，默认 Cicada3.0-turbo")
    parser.add_argument("--language", default="cn", help="cn 或 en，默认 cn")
    args = parser.parse_args()

    body = {
        "name": args.name,
        "url": args.url,
        "model_type": args.model_type,
        "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_audio"
    req = urllib.request.Request(
        url,
        data=json.dumps(body).encode("utf-8"),
        headers={"access_token": token, "Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=60) 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)

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


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
轮询 TTS 任务状态，直到完成或失败。status: 1-生成中 9-完成
用法: poll_task.py --task-id <task_id> [--interval 3]
输出: 成功时打印音频下载 URL（data.full.url），失败时打印错误到 stderr 并 exit 1
"""
import argparse
import json
import sys
import time
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="轮询 TTS 任务直到完成")
    parser.add_argument("--task-id", required=True, help="任务 ID（来自 create_task）")
    parser.add_argument("--interval", type=int, default=3, help="轮询间隔秒数，默认 3")
    args = parser.parse_args()

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

    url = f"{API_BASE}/open/v1/audio_task_state"
    body = json.dumps({"task_id": args.task_id}).encode("utf-8")

    while True:
        req = urllib.request.Request(
            url,
            data=body,
            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)

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

        if status == 9:
            full = data.get("full") or {}
            audio_url = full.get("url")
            if audio_url:
                print(audio_url)
                return 0
            print("任务完成但无 full.url", file=sys.stderr)
            sys.exit(1)

        if status not in (1, None):
            err_msg = data.get("errMsg") or data.get("errReason") 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
"""
轮询定制声音任务状态，直到完成或失败。status: 0-排队 1-进行中 2-完成 3-过期 4-失败 99-已删除
用法: poll_voice.py --voice-id <voice_id> [--interval 15]
输出: 成功时打印 voice_id（与输入一致），失败时打印错误到 stderr 并 exit 1
"""
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("--voice-id", required=True, help="定制声音 ID（来自 create_voice）")
    parser.add_argument("--interval", type=int, default=15, help="轮询间隔秒数，默认 15")
    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_audio?id={urllib.parse.quote(args.voice_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 == 2:
            print(args.voice_id)
            return 0

        if status in (3, 4, 99):
            err_msg = data.get("err_msg") or data.get("errReason") or f"status={status}"
            print(f"声音任务失败: {err_msg}", file=sys.stderr)
            sys.exit(1)

        time.sleep(args.interval)


if __name__ == "__main__":
    main()