Chanjing Content Creation Skill

"""将蝉镜结果 URL 下载到本地 output 子目录。"""
from __future__ import annotations

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


def infer_download_filename(url: str, default_basename: str) -> str:
    parsed = urllib.parse.urlparse(url)
    name = Path(parsed.path).name or default_basename
    if "." not in name:
        name += ".bin"
    return name


def download_chanjing_result_url(
    url: str,
    *,
    outputs_subdir: str,
    user_agent: str,
    default_basename: str,
    output_path: Path | None = None,
    timeout: int = 120,
) -> Path:
    """
    下载到 output_path；若未指定则写入
    skills/chanjing-content-creation-skill/output/<outputs_subdir>/<推断文件名>。
    失败时抛出 RuntimeError（与原有脚本语义一致，便于 CLI 捕获）。
    """
    skill_root = Path(__file__).resolve().parents[1]
    default_dir = skill_root / "output" / outputs_subdir
    dest = output_path if output_path is not None else default_dir / infer_download_filename(url, default_basename)
    dest.parent.mkdir(parents=True, exist_ok=True)

    req = urllib.request.Request(url, headers={"User-Agent": user_agent}, method="GET")
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp, open(dest, "wb") as handle:
            handle.write(resp.read())
    except Exception as exc:
        raise RuntimeError(f"下载失败: {exc}") from exc

    if not dest.exists() or dest.stat().st_size == 0:
        raise RuntimeError("下载失败: 输出文件为空")

    return dest


def print_downloaded_path(path: Path) -> None:
    print(os.fspath(path))

from __future__ import annotations

import json
import os
import subprocess
import sys
import time
import urllib.parse
import urllib.request
from pathlib import Path
from typing import Any, Callable

from exceptions import SkillExecutionError, SkillHTTPError, SkillTimeoutError


# --- HTTP（原 http_client.py，合并入 L1 base） ---


def request_json(
    method: str,
    url: str,
    *,
    headers: dict[str, str] | None = None,
    query: dict[str, Any] | None = None,
    body: dict[str, Any] | None = None,
    timeout: int = 30,
) -> dict[str, Any]:
    if query:
        url = f"{url}?{urllib.parse.urlencode(query)}"
    data = None
    request_headers = dict(headers or {})
    if body is not None:
        data = json.dumps(body).encode("utf-8")
        request_headers.setdefault("Content-Type", "application/json")
    req = urllib.request.Request(url, data=data, headers=request_headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except Exception as exc:
        raise SkillHTTPError(str(exc)) from exc


# --- 轮询（原 polling.py） ---


def poll_until(
    fetch: Callable[[], Any],
    *,
    is_done: Callable[[Any], bool],
    is_failed: Callable[[Any], bool] | None = None,
    interval_seconds: float = 3.0,
    timeout_seconds: float = 300.0,
) -> Any:
    deadline = time.time() + timeout_seconds
    while True:
        value = fetch()
        if is_done(value):
            return value
        if is_failed and is_failed(value):
            return value
        if time.time() >= deadline:
            raise SkillTimeoutError(f"polling timed out after {timeout_seconds}s")
        time.sleep(interval_seconds)


# --- 鉴权配置（原 auth.py，依赖 request_json） ---

SKILL_DIR = Path(__file__).resolve().parent.parent
ENV_FILE = Path(os.environ.get("CHANJING_ENV_FILE", str(SKILL_DIR / ".env")))
API_BASE = os.environ.get("CHANJING_API_BASE", "https://open-api.chanjing.cc")
LOGIN_URL = "https://www.chanjing.cc/openapi/login"
BUFFER_SECONDS = 300

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


def _run_open_login_page() -> None:
    try:
        skills_dir = Path(__file__).resolve().parent.parent
        script = skills_dir / "products" / "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 _strip_quotes(value: str) -> str:
    text = value.strip()
    if len(text) >= 2 and ((text[0] == '"' and text[-1] == '"') or (text[0] == "'" and text[-1] == "'")):
        return text[1:-1]
    return text


def _read_env_file(path: Path) -> dict[str, str]:
    if not path.exists():
        return {}
    result: dict[str, str] = {}
    for raw in path.read_text(encoding="utf-8").splitlines():
        line = raw.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        key = key.strip()
        if key:
            result[key] = _strip_quotes(value)
    return result


def _write_env_file(path: Path, values: dict[str, str]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    lines = [f"{k}={values[k]}" for k in sorted(values.keys())]
    path.write_text("\n".join(lines) + ("\n" if lines else ""), encoding="utf-8")
    try:
        os.chmod(path, 0o600)
    except OSError:
        pass


def _resolve_app_and_secret() -> tuple[str, str]:
    # 优先级：进程环境变量 -> 项目 .env
    env_map = {
        "app_id": (
            os.environ.get("CHANJING_APP_ID")
            or os.environ.get("APP_ID")
            or os.environ.get("CHANJING_USER_ID")
            or os.environ.get("USER_ID")
            or ""
        ).strip(),
        "secret_key": (
            os.environ.get("CHANJING_SECRET_KEY")
            or os.environ.get("SECRET_KEY")
            or ""
        ).strip(),
    }
    if env_map["app_id"] and env_map["secret_key"]:
        return env_map["app_id"], env_map["secret_key"]

    file_map = _read_env_file(ENV_FILE)
    file_user = (
        file_map.get("CHANJING_APP_ID")
        or file_map.get("APP_ID")
        or file_map.get("CHANJING_USER_ID")
        or file_map.get("USER_ID")
        or ""
    ).strip()
    file_secret = (file_map.get("CHANJING_SECRET_KEY") or file_map.get("SECRET_KEY") or "").strip()
    if file_user and file_secret:
        return file_user, file_secret

    return "", ""


def refresh_chanjing_access_token(app_id: str, secret_key: str) -> tuple[str | None, str | None]:
    body = request_json(
        "POST",
        f"{API_BASE}/open/v1/access_token",
        body={"app_id": app_id, "secret_key": secret_key},
    )
    if body.get("code") != 0:
        return None, body.get("msg", "获取 Token 失败")
    payload = body.get("data") or {}
    token = payload.get("access_token")
    if not token:
        return None, "API 返回无 access_token"
    env_data = _read_env_file(ENV_FILE)
    env_data["CHANJING_APP_ID"] = app_id
    env_data["CHANJING_SECRET_KEY"] = secret_key
    env_data["CHANJING_ACCESS_TOKEN"] = str(token)
    env_data["CHANJING_TOKEN_EXPIRE_IN"] = str(payload.get("expire_in") or "")
    _write_env_file(ENV_FILE, env_data)
    return token, None


def resolve_chanjing_access_token() -> tuple[str | None, str | None]:
    app_id, secret_key = _resolve_app_and_secret()
    if not app_id or not secret_key:
        _run_open_login_page()
        return None, NO_CREDENTIALS_MSG

    env_data = _read_env_file(ENV_FILE)
    token = (env_data.get("CHANJING_ACCESS_TOKEN") or "").strip()
    expire_in = (env_data.get("CHANJING_TOKEN_EXPIRE_IN") or "").strip()
    try:
        expire_at = int(expire_in) if expire_in else 0
    except ValueError:
        expire_at = 0
    now = int(time.time())
    if token and expire_at > now + BUFFER_SECONDS:
        return token, None
    return refresh_chanjing_access_token(app_id, secret_key)


def print_chanjing_access_token_cli() -> None:
    token, err = resolve_chanjing_access_token()
    if err:
        raise SystemExit(err)
    print(token)


# --- 脚本与能力目录（原 base 能力函数） ---


def repo_root() -> Path:
    return Path(__file__).resolve().parents[3]


def skills_root() -> Path:
    return repo_root() / "skills"


def skill_dir(skill_name: str) -> Path:
    return skills_root() / "chanjing-content-creation-skill" / "products" / skill_name


def script_path(skill_name: str, script_name: str) -> Path:
    """L2：`products/<skill>/scripts/`；仅存在于 L3 的技能（如一键成片）在 `orchestration/<skill>/scripts/`。"""
    under_products = skill_dir(skill_name) / "scripts" / script_name
    if under_products.exists():
        return under_products
    under_orch = (
        skills_root()
        / "chanjing-content-creation-skill"
        / "orchestration"
        / skill_name
        / "scripts"
        / script_name
    )
    if under_orch.exists():
        return under_orch
    return under_products


def run_skill_script(
    skill_name: str,
    script_name: str,
    *,
    args: list[str] | None = None,
    parse_json: bool = False,
    cwd: Path | None = None,
) -> dict[str, Any]:
    cmd = [sys.executable, str(script_path(skill_name, script_name)), *(args or [])]
    # 默认沿用调用方当前工作目录，确保相对输出路径落在用户当前路径。
    effective_cwd = cwd or Path.cwd()
    proc = subprocess.run(
        cmd,
        cwd=str(effective_cwd),
        text=True,
        capture_output=True,
        check=False,
    )
    result: dict[str, Any] = {
        "ok": proc.returncode == 0,
        "exit_code": proc.returncode,
        "stdout": proc.stdout.strip(),
        "stderr": proc.stderr.strip(),
        "command": cmd,
    }
    if parse_json and proc.stdout.strip():
        try:
            result["json"] = json.loads(proc.stdout)
        except json.JSONDecodeError as exc:
            raise SkillExecutionError(
                f"{skill_name}/{script_name} did not return valid JSON"
            ) from exc
    return result


def capability_catalog(
    skill_name: str,
    *,
    manual: str,
    operations: list[dict[str, Any]],
) -> dict[str, Any]:
    return {
        "skill": skill_name,
        "manual": manual,
        "operations": operations,
    }


def config_contract(
    *,
    preconditions: list[str],
    required: list[str],
    optional: list[str] | None = None,
) -> dict[str, Any]:
    return {
        "preconditions": preconditions,
        "required": required,
        "optional": optional or [],
    }


def usage_contract(*, examples: list[str], outputs: list[str]) -> dict[str, Any]:
    return {
        "examples": examples,
        "outputs": outputs,
    }


if __name__ == "__main__":
    print_chanjing_access_token_cli()

from __future__ import annotations

class SkillError(Exception):
    """Base error for shared skill infrastructure."""


class SkillConfigError(SkillError):
    """Raised when local configuration is missing or invalid."""


class SkillExecutionError(SkillError):
    """Raised when a script invocation fails."""


class SkillHTTPError(SkillError):
    """Raised when an HTTP request fails."""


class SkillTimeoutError(SkillError):
    """Raised when polling or script execution exceeds timeout."""

"""蝉镜通用文件上传：create_upload_url、PUT、file_detail 轮询。"""
from __future__ import annotations

import json
import os
import time
import urllib.parse
import urllib.request
from pathlib import Path
from typing import Any

API_BASE = os.environ.get("CHANJING_API_BASE", "https://open-api.chanjing.cc")


def get_chanjing_file_detail(token: str, file_id: str) -> tuple[Any | None, str | None]:
    """GET file_detail；API 业务错误返回 (None, msg)，网络异常向上抛出。"""
    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, str(body.get("msg", "file_detail failed"))
    return body.get("data"), None


def poll_chanjing_file_until_ready(
    token: str,
    file_id: str,
    *,
    interval: float = 5.0,
    timeout: float = 300.0,
    ready_statuses: set[Any],
    failed_statuses: set[Any] | None = None,
    pending_statuses: set[Any] | None = None,
) -> tuple[bool, str | None]:
    """
    pending_statuses 非 None 时：仅这些状态视为处理中；其余非 ready 视为失败（定制数字人）。
    pending_statuses 为 None 时：非 ready 且非 failed_statuses 则继续等待（视频合成上传）。
    """
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        data, err = get_chanjing_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 ready_statuses:
            return True, None
        if failed_statuses is not None and status in failed_statuses:
            return False, data.get("msg") or f"file status={status}"
        if pending_statuses is not None:
            if status not in pending_statuses:
                return False, data.get("msg") or f"unexpected file status={status}"
        time.sleep(interval)
    return False, "poll timeout"


def fetch_create_upload_url_payload(
    token: str, service: str, name: str
) -> tuple[dict[str, Any] | None, str | None]:
    """GET create_upload_url；成功返回 (data 字段内容, None)。"""
    qs = urllib.parse.urlencode({"service": service, "name": 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:
        return None, str(body.get("msg", body))
    data = body.get("data", {})
    if not isinstance(data, dict):
        return None, "invalid create_upload_url data"
    return data, None


def put_file_to_sign_url(sign_url: str, content: bytes, mime_type: str) -> tuple[bool, str | None]:
    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):
                return False, f"上传返回状态: {put_resp.status}"
    except Exception as exc:
        return False, f"上传失败: {exc}"
    return True, None


def upload_local_file_via_chanjing(
    token: str,
    local_path: Path,
    service: str,
    *,
    poll_interval: float = 5.0,
    poll_timeout: float = 300.0,
    ready_statuses: set[Any],
    failed_statuses: set[Any] | None = None,
    pending_statuses: set[Any] | None = None,
) -> tuple[str | None, str | None]:
    """
    完整上传链路。成功返回 (file_id, None)，失败返回 (None, error_message)。
    """
    if not local_path.is_file():
        return None, f"文件不存在: {local_path}"

    payload, err = fetch_create_upload_url_payload(token, service, local_path.name)
    if err:
        return None, err

    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:
        return None, "响应缺少 sign_url 或 file_id"

    content = local_path.read_bytes()
    ok, put_err = put_file_to_sign_url(sign_url, content, mime_type)
    if not ok:
        return None, put_err or "上传失败"

    ready, poll_err = poll_chanjing_file_until_ready(
        token,
        file_id,
        interval=poll_interval,
        timeout=poll_timeout,
        ready_statuses=ready_statuses,
        failed_statuses=failed_statuses,
        pending_statuses=pending_statuses,
    )
    if not ready:
        return None, poll_err or "文件未就绪"

    return str(file_id), None

from __future__ import annotations

import json
import logging
import sys
from typing import Any


def get_logger(name: str) -> logging.Logger:
    logger = logging.getLogger(name)
    if logger.handlers:
        return logger
    handler = logging.StreamHandler(sys.stderr)
    handler.setFormatter(logging.Formatter("%(message)s"))
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)
    logger.propagate = False
    return logger


def log_event(logger: logging.Logger, event: str, **fields: Any) -> None:
    payload = {"event": event, **fields}
    logger.info(json.dumps(payload, ensure_ascii=False, sort_keys=True))

"""AI 创作（ai_creation）Open API 封装。"""
from __future__ import annotations

from typing import Any

from open_api_client import api_get, api_post

RUNNING_STATUSES = {"Queued", "Ready", "Generating"}
SUCCESS_STATUSES = {"Success"}
FAILED_STATUSES = {"Error", "Fail"}


def get_ai_creation_task(token: str, unique_id: str) -> Any:
    return api_get(token, "/open/v1/ai_creation/task", {"unique_id": unique_id}, query_doseq=True)


def list_ai_creation_tasks(
    token: str,
    creation_type: Any,
    page: int = 1,
    page_size: int = 50,
    unique_ids: list | None = None,
    is_success: bool | None = None,
) -> Any:
    payload: dict[str, Any] = {
        "unique_ids": unique_ids or [],
        "type": creation_type,
        "page": page,
        "page_size": page_size,
    }
    if is_success is not None:
        payload["is_success"] = bool(is_success)
    return api_post(token, "/open/v1/ai_creation/task/page", payload)


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

"""文生数字人（aigc photo/motion/lora）Open API 封装。"""
from __future__ import annotations

from typing import Any

from open_api_client import api_get, api_post

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 get_photo_task(token: str, unique_id: str) -> Any:
    return api_get(token, "/open/v1/aigc/photo/task", {"unique_id": unique_id})


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


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


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


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

"""蝉镜 Open API：带 access_token 的 JSON GET/POST，与 base.API_BASE 环境约定一致。"""
from __future__ import annotations

import json
import os
import urllib.parse
import urllib.request
from typing import Any

from exceptions import SkillHTTPError

API_BASE = os.environ.get("CHANJING_API_BASE", "https://open-api.chanjing.cc")


def api_get(
    token: str,
    path: str,
    query: dict[str, Any] | None = None,
    *,
    query_doseq: bool = False,
    timeout: int = 30,
) -> Any:
    query = query or {}
    suffix = ""
    if query:
        suffix = "?" + urllib.parse.urlencode(query, doseq=query_doseq)
    url = f"{API_BASE}{path}{suffix}"
    req = urllib.request.Request(url, headers={"access_token": token}, method="GET")
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            body = json.loads(resp.read().decode("utf-8"))
    except Exception as exc:
        raise SkillHTTPError(str(exc)) from exc
    if body.get("code") != 0:
        raise SkillHTTPError(str(body.get("msg", body)))
    return body.get("data")


def api_post(token: str, path: str, payload: dict[str, Any], *, timeout: int = 30) -> Any:
    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",
    )
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            body = json.loads(resp.read().decode("utf-8"))
    except Exception as exc:
        raise SkillHTTPError(str(exc)) from exc
    if body.get("code") != 0:
        raise SkillHTTPError(str(body.get("msg", body)))
    return body.get("data")

{"id":"route-001","prompt":"帮我做一个60秒的品牌口播短视频","expected_route":"orchestration/chanjing-one-click-video-creation/chanjing-one-click-video-creation-SKILL.md","expected_outcome":"ok"}
{"id":"route-002","prompt":"我只要配音，把这段文案转成女声","expected_route":"products/chanjing-tts/chanjing-tts-SKILL.md","expected_outcome":"ok"}
{"id":"route-003","prompt":"用我的声音做克隆，再读这段稿子","expected_route":"products/chanjing-tts-voice-clone/chanjing-tts-voice-clone-SKILL.md","expected_outcome":"ok"}
{"id":"route-004","prompt":"我有一段真人视频，帮我做对口型","expected_route":"products/chanjing-avatar/chanjing-avatar-SKILL.md","expected_outcome":"ok"}
{"id":"route-005","prompt":"帮我做卡通风格12镜短剧","expected_route":"orchestration/cartoon-video-creation/cartoon-video-creation-SKILL.md","expected_outcome":"ok"}
{"id":"route-006","prompt":"上传这个人物视频，训练一个定制数字人","expected_route":"products/chanjing-customised-person/chanjing-customised-person-SKILL.md","expected_outcome":"ok"}
{"id":"route-007","prompt":"先给我生成人设图，再做一个会说话的视频","expected_route":"products/chanjing-text-to-digital-person/chanjing-text-to-digital-person-SKILL.md","expected_outcome":"ok"}
{"id":"route-008","prompt":"用蝉镜生成一张海报，再生成5秒视频","expected_route":"products/chanjing-ai-creation/chanjing-ai-creation-SKILL.md","expected_outcome":"ok"}
{"id":"route-009","prompt":"我没配过秘钥，先帮我配置蝉镜AKSK","expected_route":"products/chanjing-credentials-guard/chanjing-credentials-guard-SKILL.md","expected_outcome":"auth_required"}
{"id":"route-010","prompt":"我就要写标题和脚本，不要生成视频","expected_route":"none","expected_outcome":"need_param"}
{"id":"route-011","prompt":"帮我做视频，但我不给任何主题和素材","expected_route":"orchestration/chanjing-one-click-video-creation/chanjing-one-click-video-creation-SKILL.md","expected_outcome":"need_param"}
{"id":"route-012","prompt":"系统一直超时，先别下载，告诉我怎么重试","expected_route":"current_skill","expected_outcome":"timeout"}

# skill文档规范 §1–§2 合规
name: chanjing-content-creation-skill
version: 0.1.0
vendor: chanjing
runtime:
  interpreter: python3
  dependencies:
    - ffmpeg
    - ffprobe
env:
  required: []
  optional:
    - CHANJING_APP_ID
    - CHANJING_SECRET_KEY
    - CHANJING_ACCESS_TOKEN
    - CHANJING_TOKEN_EXPIRE_IN
    - CHANJING_API_BASE
    - CHAN_SKILLS_DIR
permissions:
  network_mode: allowlist
  allowed_hosts:
    - open-api.chanjing.cc
    - www.chanjing.cc
  filesystem:
    read_roots:
      - "${WORKSPACE_ROOT}"
      - "${SKILL_DIR}"
    write_roots:
      - "${WORKSPACE_ROOT}"
  allowed_commands:
    - python3
    - ffmpeg
    - ffprobe

schemaVersion: 1
skill:
  id: chanjing-content-creation-skill
  author: chan-skills
  category: 媒体处理
  tags:
    - 蝉镜
    - Chanjing
    - TTS
    - 数字人
    - 对口型
    - 文生视频
    - 一键成片
    - 卡通视频
  summary: >-
    蝉镜内容创作聚合技能包：凭据管理、TTS、数字人口播、对口型、文生视频、声音克隆、
    定制数字人、一键成片编排等。本 Skill 是路由入口，按意图分流到子产品/编排。
  skillDoc: SKILL.md

environment:
  variables:
    - name: CHANJING_API_BASE
      required: false
      description: Open API 基址，默认 https://open-api.chanjing.cc
    - name: CHAN_SKILLS_DIR
      required: false
      description: 一键成片脚本仓库根路径（当前实现实际读取）

credentials:
  model: environment
  persistAccessTokenOnDisk: false
  sensitiveFields: []

clientPermissions:
  network:
    httpsOutbound: true
    documentedHosts:
      - open-api.chanjing.cc
  filesystem:
    read:
      - "${SKILL_DIR}/"
      - user_workflow_and_output_paths
    write:
      - "${SKILL_DIR}/.env"
      - user_output_directory
  browser:
    mayOpenForAuth: true
  subprocess:
    allowedPatterns:
      - python3
      - ffmpeg
      - ffprobe
    mayInvokeSiblingSkillScripts: true
  userContent:
    mayDownloadFromApiResponseUrls: true

metadata:
  openclaw:
    homepage: https://doc.chanjing.cc
    os:
      - darwin
      - linux
    requires:
      bins:
        - python3
      anyBins:
        - ffmpeg
        - ffprobe
      env:
        - CHANJING_APP_ID
        - CHANJING_SECRET_KEY

agentPolicy:
  alwaysSkill: false
  modifiesOtherSkillsOrGlobalAgent: false

{
  "topic": "可选：内容层追溯用；run_render 不读取。例如：防晒",
  "video_plan": {
    "scene_count": 5,
    "core_angle": "可选：策划元数据；run_render 不读取。例如：SPF/PA 认知 + 用量 + 补涂与硬防晒",
    "style": "科普种草",
    "platform": "抖音/视频号等竖屏信息流",
    "target_audience": "18–35 岁都市通勤族",
    "tone": "亲切、克制、可执行",
    "success_metrics_note": "完播与收藏可后续在宿主侧打点，本 JSON 不落指标"
  },
  "full_script": "（开场）防晒做不对，晒黑又伤皮。选对指数与用量，紫外线才挡得住。出门前挤满一枚硬币，别心疼。户外两小时记得补涂，帽子也不能少。坚持防晒，皮肤会给你回报。",
  "audio_man": "每次任务据语义选型：list_voices 的 id，宜与同条形象的 audio_man_id 一致",
  "person_id": "每次任务据语义与 video_plan：list_figures 所选来源行的 person_id",
  "figure_type": "whole_body",
  "speed": 1,
  "pitch": 1,
  "ai_video_duration_sec": 10,
  "model_code": "Doubao-Seedance-1.0-pro",
  "max_retry_per_step": 2,
  "subtitle_required": false,
  "subtitle_color": null,
  "subtitle_stroke_color": null,
  "subtitle_stroke_width": null,
  "scenes": [
    {
      "scene_id": 1,
      "beat": "开场钩子",
      "voiceover": "（开场）防晒做不对，晒黑又伤皮。",
      "use_avatar": true,
      "subtitle": "防晒做不对，晒黑又伤皮。",
      "planning": {}
    },
    {
      "scene_id": 2,
      "beat": "知识点｜选品",
      "voiceover": "选对指数与用量，紫外线才挡得住。",
      "use_avatar": false,
      "ref_prompt": "Contemporary urban China, bright bathroom vanity in morning light, locked tripod medium shot. Generic sunscreen tubes and pump bottles on a tray, caps closed, labels blurred unreadable, no logos. Soft sidelight, clean tiles, vertical 9:16. East Asian young adult woman's hands only at frame edge adjusting a bottle, slow subtle motion. Focus on objects and environment. No readable text, no on-screen captions, no before-after split. Calm documentary product B-roll.",
      "subtitle": "选对指数与用量，紫外线才挡得住。",
      "planning": {
        "shot_type": "静物 + 手部点缀",
        "negative_prompt_hint": "避免可读商标、说明书特写、分屏教程 UI"
      }
    },
    {
      "scene_id": 3,
      "beat": "用量示范（口播）",
      "voiceover": "出门前挤满一枚硬币，别心疼。",
      "use_avatar": true,
      "subtitle": "出门前挤满一枚硬币，别心疼。",
      "planning": {}
    },
    {
      "scene_id": 4,
      "beat": "场景｜户外补涂",
      "voiceover": "户外两小时记得补涂，帽子也不能少。",
      "use_avatar": false,
      "ref_prompt": "Contemporary Chinese city park promenade, sunny summer, wide tripod shot. East Asian young woman from behind wearing wide-brim hat and light long sleeves, holding a small plain sunscreen tube without readable label, walking slowly in tree shade. Dappled light, relaxed pace, vertical 9:16. No readable text, no logos, no split-screen tutorial. Calm lifestyle B-roll, very subtle slow push-in.",
      "subtitle": "户外两小时记得补涂，帽子也不能少。",
      "planning": {
        "shot_type": "环境人像（背影/中景）",
        "negative_prompt_hint": "避免人脸大特写、夸张表情、文字贴纸"
      }
    },
    {
      "scene_id": 5,
      "beat": "收尾强化",
      "voiceover": "坚持防晒，皮肤会给你回报。",
      "use_avatar": true,
      "subtitle": "（结尾）坚持防晒，皮肤会给你回报。",
      "planning": {}
    }
  ]
}

# 合规：根目录 skill开发规范.md §1–§2
name: chanjing-one-click-video-creation
version: 0.2.0
vendor: chanjing
runtime:
  interpreter: python3
  dependencies:
    - ffmpeg
    - ffprobe
    - chan-skill
env:
  required: []
  optional:
    - CHANJING_API_BASE
    - CHANJING_APP_ID
    - CHANJING_SECRET_KEY
    - CHANJING_ACCESS_TOKEN
    - CHANJING_TOKEN_EXPIRE_IN
    - CHAN_SKILLS_DIR
    - AI_VIDEO_PROMPT_MAX_CHARS
    - AI_VIDEO_MODEL
    - FIRST_DIGITAL_HUMAN_MAX_CHARS
permissions:
  network_mode: allowlist
  allowed_hosts:
    - open-api.chanjing.cc
    - www.chanjing.cc
  filesystem:
    read_roots:
      - "${WORKSPACE_ROOT}"
      - "${SKILL_DIR}"
    write_roots:
      - "${WORKSPACE_ROOT}"
  allowed_commands:
    - python3
    - ffmpeg
    - ffprobe
    - chan-skill

schemaVersion: 1
skill:
  id: chanjing-one-click-video-creation
  author: chan-skills
  category: 媒体处理
  tags:
    - 视频渲染
    - ffmpeg
    - ChanjingAPI
    - 蝉镜
    - Chanjing
    - 一键成片
  summary: >-
    选题/workflow 驱动一键短视频成片：编排 TTS、数字人合成、文生视频与本地 ffmpeg 封装。
  skillDoc: SKILL.md
  supportCommands:
    - run_render
  siblingSkills:
    - chanjing-tts
    - chanjing-video-compose
    - chanjing-ai-creation

environment:
  variables:
    - name: CHANJING_API_BASE
      required: false
      description: Open API 基址，默认 https://open-api.chanjing.cc
    - name: CHANJING_APP_ID
      required: false
      description: 鉴权应用标识（从项目 .env 读取）
    - name: CHANJING_SECRET_KEY
      required: false
      description: 鉴权密钥（从项目 .env 读取）
    - name: CHAN_SKILLS_DIR
      required: false
      description: 一键成片脚本仓库根路径
    - name: AI_VIDEO_PROMPT_MAX_CHARS
      required: false
      description: ref_prompt 总长上限，默认 8000
    - name: AI_VIDEO_MODEL
      required: false
      description: 文生视频 model_code 缺省覆盖
    - name: FIRST_DIGITAL_HUMAN_MAX_CHARS
      required: false
      description: 首个数字人分镜字数上限，默认 20

credentials:
  model: environment
  persistAccessTokenOnDisk: false
  primaryEnvIntentionallyOmitted: false
  sensitiveFields: []

clientPermissions:
  network:
    httpsOutbound: true
    documentedHosts:
      - open-api.chanjing.cc
  filesystem:
    read:
      - "${CHAN_SKILLS_DIR or repo_root}/skills/"
      - user_workflow_and_output_paths
    write:
      - user_output_directory
  browser:
    mayOpenForAuth: false
  subprocess:
    allowedPatterns:
      - python3
      - ffmpeg
      - ffprobe
      - chan-skill
    mayInvokeSiblingSkillScripts: true
  userContent:
    mayDownloadFromApiResponseUrls: true

metadata:
  openclaw:
    homepage: https://doc.chanjing.cc
    requires:
      bins:
        - ffmpeg
        - ffprobe
      env:
        - CHANJING_APP_ID
        - CHANJING_SECRET_KEY

agentPolicy:
  alwaysSkill: false
  modifiesOtherSkillsOrGlobalAgent: false

from __future__ import annotations

from pathlib import Path
import sys

sys.path.insert(0, str(Path(__file__).resolve().parents[3] / "common"))

from base import print_chanjing_access_token_cli, resolve_chanjing_access_token  # noqa: E402


if __name__ == "__main__":
    print_chanjing_access_token_cli()

from __future__ import annotations

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[3] / "common"))

from base import capability_catalog, config_contract, run_skill_script, usage_contract  # noqa: E402

SKILL_NAME = "chanjing-one-click-video-creation"


def list():
    return capability_catalog(
        SKILL_NAME,
        manual="chanjing-one-click-video-creation-SKILL.md",
        operations=[
            {"name": "run_render", "script": "run_render.py"},
            {
                "name": "validate_ai_resolution",
                "script": "validate_ai_resolution.py",
            },
            {"name": "workflow_contract", "script": "workflow.json / workflow_result.json"},
        ],
    )


def config():
    return config_contract(
        preconditions=["本机存在 ffmpeg 与 ffprobe"],
        required=["topic 或 workflow.json"],
        optional=["output_dir", "CHAN_SKILLS_DIR", "max_retry_per_step"],
    )


def usage():
    return usage_contract(
        examples=[
            "python skills/chanjing-content-creation-skill/orchestration/chanjing-one-click-video-creation/scripts/run_render.py --input workflow.json --output-dir output/run1",
            "python .../scripts/validate_ai_resolution.py --input workflow.json --check-ref-prompt",
        ],
        outputs=["final_one_click.mp4 / workflow_result.json / work/"],
    )


def run_render(input_file: str, output_dir: str = ""):
    args = ["--input", input_file]
    if output_dir:
        args.extend(["--output-dir", output_dir])
    return run_skill_script(
        SKILL_NAME,
        "run_render.py",
        args=args,
    )

{
  "audio_man": "",
  "person_id": "",
  "figure_type": "sit_body",
  "_comment": "复制为 chanjing_test_defaults.json（同目录，勿提交真实 ID）。非空则单元测试直接用；留空则每次跑测试时现场 list_voices / list_figures，不读写仓库缓存。"
}

#!/usr/bin/env python3
from __future__ import annotations

from pathlib import Path
import sys

sys.path.insert(0, str(Path(__file__).resolve().parents[3] / "common"))

from base import print_chanjing_access_token_cli, resolve_chanjing_access_token  # noqa: E402


if __name__ == "__main__":
    print_chanjing_access_token_cli()

#!/usr/bin/env python3
"""兼容层：实现已迁至 common/open_ai_creation.py 与 common/open_api_client.py。"""
from __future__ import annotations

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[3] / "common"))

from open_ai_creation import (  # noqa: F401
    FAILED_STATUSES,
    RUNNING_STATUSES,
    SUCCESS_STATUSES,
    first_output_url,
    get_ai_creation_task,
    list_ai_creation_tasks,
)
from open_api_client import api_get, api_post

get_task = get_ai_creation_task
list_tasks = list_ai_creation_tasks

__all__ = [
    "api_get",
    "api_post",
    "FAILED_STATUSES",
    "RUNNING_STATUSES",
    "SUCCESS_STATUSES",
    "first_output_url",
    "get_task",
    "list_tasks",
]