Genarrative/.codex/skills/genarrative-external-editor-api/scripts/genarrative_external_api.py

#!/usr/bin/env python3
"""Tiny stdlib client for Genarrative external editor APIs."""

from __future__ import annotations

import argparse
import json
import mimetypes
import os
import re
import struct
import sys
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
from pathlib import Path
from typing import Any


BASE_URL = "https://www.genarrative.world/"
DEFAULT_CREDENTIALS_FILE = Path.home() / ".config/genarrative/external-editor-api.json"
DEFAULT_REQUEST_TIMEOUT_SECONDS = 60
GENERATION_WAIT_TIMEOUT_SECONDS = 1800


class GenarrativeApiError(RuntimeError):
    pass


def load_api_key(credentials_file: str | os.PathLike[str] = DEFAULT_CREDENTIALS_FILE) -> str:
    path = Path(credentials_file).expanduser()
    try:
        value = json.loads(path.read_text(encoding="utf-8")).get("apiKey", "")
    except FileNotFoundError as error:
        raise GenarrativeApiError(
            f"Missing API key file: {path}. Create JSON like {{\"apiKey\":\"tnr_sk_...\"}}."
        ) from error
    except json.JSONDecodeError as error:
        raise GenarrativeApiError(f"Invalid JSON in API key file: {path}.") from error
    if not isinstance(value, str) or not value.strip():
        raise GenarrativeApiError(f"Missing apiKey string in API key file: {path}.")
    return value.strip()


def unwrap_envelope(body: Any) -> Any:
    if isinstance(body, dict) and body.get("ok") is True and "data" in body:
        return body["data"]
    return body


def guess_content_type(file_path: str | os.PathLike[str]) -> str:
    return mimetypes.guess_type(str(file_path))[0] or "application/octet-stream"


def source_layer_id_from_path(file_path: str | os.PathLike[str]) -> str:
    stem = Path(file_path).stem.lower()
    slug = re.sub(r"[^a-z0-9]+", "-", stem).strip("-") or "image"
    return f"external-reference-{slug}"


def normalize_optional_text(value: Any) -> str | None:
    if not isinstance(value, str):
        return None
    stripped = value.strip()
    return stripped or None


def art_spec_prompt(prompt: str, art_spec: dict[str, Any] | None) -> str:
    if not art_spec:
        return prompt
    spec_json = json.dumps(art_spec, ensure_ascii=False, sort_keys=True)
    return f"{prompt}\n\n美术规范(JSON): {spec_json}"


def image_dimensions(file_path: str | os.PathLike[str]) -> tuple[int, int] | None:
    path = Path(file_path)
    with path.open("rb") as fh:
        header = fh.read(32)
        if header.startswith(b"\x89PNG\r\n\x1a\n") and header[12:16] == b"IHDR":
            return struct.unpack(">II", header[16:24])
        if not header.startswith(b"\xff\xd8"):
            return None
        fh.seek(2)
        while True:
            marker_prefix = fh.read(1)
            if not marker_prefix:
                return None
            if marker_prefix != b"\xff":
                continue
            marker = fh.read(1)
            while marker == b"\xff":
                marker = fh.read(1)
            if marker in {b"\xd8", b"\xd9"}:
                continue
            length_bytes = fh.read(2)
            if len(length_bytes) != 2:
                return None
            length = struct.unpack(">H", length_bytes)[0]
            if marker and marker[0] in {0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7, 0xC9, 0xCA, 0xCB, 0xCD, 0xCE, 0xCF}:
                data = fh.read(5)
                if len(data) != 5:
                    return None
                height, width = struct.unpack(">HH", data[1:5])
                return width, height
            fh.seek(max(length - 2, 0), os.SEEK_CUR)


class GenarrativeExternalClient:
    def __init__(
        self,
        api_key: str | None = None,
        credentials_file: str | os.PathLike[str] = DEFAULT_CREDENTIALS_FILE,
        base_url: str = BASE_URL,
    ) -> None:
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key if api_key is not None else load_api_key(credentials_file)

    def request_json(
        self,
        method: str,
        path: str,
        body: dict[str, Any] | None = None,
        query: dict[str, Any] | None = None,
        auth: bool = True,
        timeout: int = DEFAULT_REQUEST_TIMEOUT_SECONDS,
        headers: dict[str, str] | None = None,
    ) -> Any:
        url = f"{self.base_url}{path}"
        if query:
            url = f"{url}?{urllib.parse.urlencode({k: v for k, v in query.items() if v is not None})}"
        data = None if body is None else json.dumps(body).encode("utf-8")
        request_headers = {"Accept": "application/json", **(headers or {})}
        if data is not None:
            request_headers["Content-Type"] = "application/json"
        if auth:
            request_headers["Authorization"] = f"Bearer {self.api_key}"
        request = urllib.request.Request(url, data=data, headers=request_headers, method=method.upper())
        try:
            with urllib.request.urlopen(request, timeout=timeout) as response:
                payload = response.read()
        except urllib.error.HTTPError as error:
            detail = error.read().decode("utf-8", errors="replace")
            raise GenarrativeApiError(f"{method.upper()} {path} failed: HTTP {error.code}: {detail}") from error
        if not payload:
            return None
        return unwrap_envelope(json.loads(payload.decode("utf-8")))

    def openapi(self) -> Any:
        return self.request_json("GET", "/api/external/v1/openapi.json", auth=False)

    def list_projects(self) -> Any:
        return self.request_json("GET", "/api/external/v1/editor/projects")

    def create_project(self, title: str | None = None) -> Any:
        body = {} if title is None else {"title": title}
        return self.request_json("POST", "/api/external/v1/editor/projects", body)

    def list_asset_library(self) -> Any:
        return self.request_json("GET", "/api/external/v1/editor/assets/library")

    def create_asset_folder(self, label: str, sort_order: int = 100) -> Any:
        return self.request_json(
            "POST",
            "/api/external/v1/editor/assets/folders",
            {"label": label, "sortOrder": sort_order},
        )

    def ensure_asset_folder(self, label: str, sort_order: int = 100) -> dict[str, Any]:
        normalized_label = normalize_optional_text(label) or "新画板"
        library = self.list_asset_library()
        folders = unwrap_envelope(library).get("library", {}).get("folders", [])
        if isinstance(folders, list):
            for folder in folders:
                if isinstance(folder, dict) and normalize_optional_text(folder.get("label")) == normalized_label:
                    return folder
        created = self.create_asset_folder(normalized_label, sort_order=sort_order)
        folder = unwrap_envelope(created).get("folder")
        if not isinstance(folder, dict):
            raise GenarrativeApiError("Create asset folder response missing folder payload.")
        return folder

    def create_asset(
        self,
        folder_id: str,
        label: str,
        image_src: str,
        width: int,
        height: int,
        **fields: Any,
    ) -> Any:
        body = {
            "folderId": folder_id,
            "label": label,
            "imageSrc": image_src,
            "width": width,
            "height": height,
            "sourceType": fields.pop("sourceType", "generated"),
            **fields,
        }
        return self.request_json("POST", "/api/external/v1/editor/assets", body)

    def prepare_canvas_session(self, canvas_name: str) -> dict[str, Any]:
        normalized_name = normalize_optional_text(canvas_name) or "新画板"
        project = unwrap_envelope(self.create_project(normalized_name)).get("project")
        if not isinstance(project, dict) or not normalize_optional_text(project.get("projectId")):
            raise GenarrativeApiError("Create project response missing projectId.")
        folder = self.ensure_asset_folder(normalized_name)
        folder_id = normalize_optional_text(folder.get("folderId"))
        if not folder_id:
            raise GenarrativeApiError("Asset folder payload missing folderId.")
        return {
            "canvasName": normalized_name,
            "projectId": project["projectId"],
            "assetFolderId": folder_id,
            "project": project,
            "folder": folder,
        }

    def build_canvas_completion(
        self,
        title: str,
        width: int,
        height: int,
        x: float = 0,
        y: float = 0,
        dialog_id: str | None = None,
    ) -> dict[str, Any]:
        payload: dict[str, Any] = {
            "title": normalize_optional_text(title) or "生成素材",
            "placeholder": {
                "x": x,
                "y": y,
                "width": width,
                "height": height,
                "originalWidth": width,
                "originalHeight": height,
            },
        }
        if normalize_optional_text(dialog_id):
            payload["dialogId"] = dialog_id
        return payload

    def canvas_generation_fields(
        self,
        session: dict[str, Any],
        asset_label: str,
        width: int = 1024,
        height: int = 1024,
        x: float = 0,
        y: float = 0,
        dialog_id: str | None = None,
        asset_label_field: str | None = "assetLabel",
    ) -> dict[str, Any]:
        fields: dict[str, Any] = {
            "projectId": session["projectId"],
            "canvasCompletion": self.build_canvas_completion(
                asset_label,
                width=width,
                height=height,
                x=x,
                y=y,
                dialog_id=dialog_id,
            ),
        }
        folder_id = normalize_optional_text(session.get("assetFolderId"))
        if folder_id:
            fields["assetFolderId"] = folder_id
        if asset_label_field:
            fields[asset_label_field] = normalize_optional_text(asset_label) or "生成素材"
        return fields

    def save_canvas(
        self,
        project_id: str,
        viewport: dict[str, Any],
        layers: dict[str, Any],
        expected_revision: int,
    ) -> Any:
        return self.request_json(
            "PATCH",
            f"/api/external/v1/editor/projects/{urllib.parse.quote(project_id, safe='')}/canvas",
            {
                "viewport": viewport,
                "layers": layers,
                "expectedRevision": expected_revision,
            },
        )

    def _apply_art_spec(self, fields: dict[str, Any], prompt: str) -> str:
        art_spec = fields.pop("artSpec", None)
        if art_spec is None:
            art_spec = fields.pop("art_spec", None)
        if not isinstance(art_spec, dict):
            return prompt
        generation_inputs = fields.get("generationInputs")
        if not isinstance(generation_inputs, dict):
            generation_inputs = {}
        generation_inputs.setdefault("artSpec", art_spec)
        fields["generationInputs"] = generation_inputs
        return art_spec_prompt(prompt, art_spec)

    def _apply_canvas_session_fields(
        self,
        fields: dict[str, Any],
        default_label: str,
        default_width: int,
        default_height: int,
        asset_label_field: str | None = "assetLabel",
    ) -> tuple[dict[str, Any] | None, str]:
        session = fields.pop("canvasSession", None)
        if session is None:
            session = fields.pop("canvas_session", None)
        if isinstance(session, str):
            session = self.prepare_canvas_session(session)
        label = (
            normalize_optional_text(fields.get(asset_label_field)) if asset_label_field else None
        ) or normalize_optional_text(fields.pop("canvasTitle", None)) or normalize_optional_text(default_label) or "生成素材"
        if asset_label_field and not normalize_optional_text(fields.get(asset_label_field)):
            fields.pop(asset_label_field, None)
        if not isinstance(session, dict):
            return None, label
        width = int(fields.pop("canvasWidth", default_width))
        height = int(fields.pop("canvasHeight", default_height))
        x = float(fields.pop("canvasX", 0))
        y = float(fields.pop("canvasY", 0))
        dialog_id = normalize_optional_text(fields.pop("dialogId", None))
        fields.update(
            {
                key: value
                for key, value in self.canvas_generation_fields(
                    session,
                    label,
                    width=width,
                    height=height,
                    x=x,
                    y=y,
                    dialog_id=dialog_id,
                    asset_label_field=asset_label_field,
                ).items()
                if key not in fields or fields[key] is None
            }
        )
        return session, label

    def create_upload_ticket(self, file_path: str | os.PathLike[str], access: str = "private") -> Any:
        path = Path(file_path)
        return self.request_json(
            "POST",
            "/api/external/v1/assets/direct-upload-tickets",
            {
                "legacyPrefix": "generated-character-drafts",
                "pathSegments": ["editor", "external-editor-references"],
                "fileName": path.name,
                "contentType": guess_content_type(path),
                "access": access,
            },
        )

    def upload_to_oss(self, ticket_response: Any, file_path: str | os.PathLike[str]) -> None:
        payload = unwrap_envelope(ticket_response)
        ticket = payload.get("upload") if isinstance(payload, dict) else None
        if not isinstance(ticket, dict):
            raise GenarrativeApiError("Upload ticket response missing upload payload.")
        fields = ticket.get("formFields")
        if not isinstance(fields, dict):
            raise GenarrativeApiError("Upload ticket response missing formFields.")
        boundary = f"----genarrative-{uuid.uuid4().hex}"
        path = Path(file_path)
        content_type = ticket.get("contentType") or guess_content_type(path)
        chunks: list[bytes] = []
        for key, value in fields.items():
            if value is None:
                continue
            chunks.extend(
                [
                    f"--{boundary}\r\n".encode(),
                    f'Content-Disposition: form-data; name="{key}"\r\n\r\n'.encode(),
                    str(value).encode(),
                    b"\r\n",
                ]
            )
        chunks.extend(
            [
                f"--{boundary}\r\n".encode(),
                f'Content-Disposition: form-data; name="file"; filename="{path.name}"\r\n'.encode(),
                f"Content-Type: {content_type}\r\n\r\n".encode(),
                path.read_bytes(),
                b"\r\n",
                f"--{boundary}--\r\n".encode(),
            ]
        )
        request = urllib.request.Request(
            ticket["host"],
            data=b"".join(chunks),
            headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
            method="POST",
        )
        try:
            with urllib.request.urlopen(request, timeout=300) as response:
                if response.status not in {200, 201, 204}:
                    raise GenarrativeApiError(f"OSS upload failed: HTTP {response.status}")
        except urllib.error.HTTPError as error:
            detail = error.read().decode("utf-8", errors="replace")
            raise GenarrativeApiError(f"OSS upload failed: HTTP {error.code}: {detail}") from error

    def confirm_asset_object(
        self,
        object_key: str,
        file_path: str | os.PathLike[str],
        asset_kind: str = "editor_reference_image",
        access_policy: str = "private",
    ) -> Any:
        path = Path(file_path)
        return self.request_json(
            "POST",
            "/api/external/v1/assets/objects/confirm",
            {
                "objectKey": object_key,
                "contentType": guess_content_type(path),
                "contentLength": path.stat().st_size,
                "assetKind": asset_kind,
                "accessPolicy": access_policy,
            },
        )

    def upload_reference_image(self, file_path: str | os.PathLike[str]) -> dict[str, Any]:
        ticket = self.create_upload_ticket(file_path)
        upload = unwrap_envelope(ticket)["upload"]
        self.upload_to_oss(ticket, file_path)
        confirmed = self.confirm_asset_object(upload["objectKey"], file_path)
        return {
            "objectKey": upload["objectKey"],
            "ticket": ticket,
            "assetObject": unwrap_envelope(confirmed).get("assetObject"),
            "dimensions": image_dimensions(file_path),
            "sourceLayerId": source_layer_id_from_path(file_path),
        }

    def read_url(self, object_key: str) -> Any:
        return self.request_json("GET", "/api/external/v1/assets/read-url", query={"objectKey": object_key})

    def submit_generation(
        self,
        path: str,
        body: dict[str, Any],
        idempotency_key: str | None = None,
    ) -> dict[str, Any]:
        key = normalize_optional_text(idempotency_key) or str(uuid.uuid4())
        submission = None
        for attempt in range(2):
            try:
                submission = self.request_json(
                    "POST",
                    path,
                    body,
                    headers={"Idempotency-Key": key},
                )
                break
            except (urllib.error.URLError, TimeoutError) as error:
                if attempt == 0:
                    time.sleep(0.5)
                    continue
                raise GenarrativeApiError(
                    "Generation submission transport outcome is unknown. "
                    f"Retry the same body with Idempotency-Key {key}; do not create a new key."
                ) from error
        if not isinstance(submission, dict) or not normalize_optional_text(submission.get("operationId")):
            raise GenarrativeApiError("Generation submission response missing operationId.")
        submission.setdefault("idempotencyKey", key)
        return submission

    def get_generation(self, operation_id: str) -> dict[str, Any]:
        result = self.request_json(
            "GET",
            f"/api/external/v1/generations/{urllib.parse.quote(operation_id, safe='')}",
        )
        if not isinstance(result, dict):
            raise GenarrativeApiError("Generation status response must be an object.")
        return result

    def wait_for_generation(
        self,
        submission_or_operation_id: dict[str, Any] | str,
        timeout_seconds: int = GENERATION_WAIT_TIMEOUT_SECONDS,
    ) -> dict[str, Any]:
        operation_id = (
            submission_or_operation_id.get("operationId")
            if isinstance(submission_or_operation_id, dict)
            else submission_or_operation_id
        )
        operation_id = normalize_optional_text(operation_id)
        if not operation_id:
            raise GenarrativeApiError("Generation operationId is required.")
        deadline = time.monotonic() + max(1, timeout_seconds)
        while True:
            try:
                job = self.get_generation(operation_id)
            except GenarrativeApiError as error:
                if any(f"HTTP {status}" in str(error) for status in (429, 502, 503, 504)):
                    if time.monotonic() >= deadline:
                        raise GenarrativeApiError(
                            f"Generation {operation_id} is still running; keep this operationId and continue polling."
                        ) from error
                    time.sleep(1.5)
                    continue
                raise
            status = job.get("status")
            if status == "completed":
                result = job.get("result")
                if not isinstance(result, dict):
                    raise GenarrativeApiError(
                        f"Generation {operation_id} completed without a result payload."
                    )
                return result
            if status == "failed":
                raise GenarrativeApiError(
                    f"Generation {operation_id} failed: {job.get('error') or 'unknown error'}"
                )
            if time.monotonic() >= deadline:
                raise GenarrativeApiError(
                    f"Generation {operation_id} is still running; keep this operationId and continue polling."
                )
            poll_after_ms = job.get("pollAfterMs", 1500)
            if not isinstance(poll_after_ms, (int, float)):
                poll_after_ms = 1500
            time.sleep(max(0.25, min(float(poll_after_ms) / 1000.0, 5.0)))

    def submit_and_wait_generation(
        self,
        path: str,
        body: dict[str, Any],
        idempotency_key: str | None = None,
        timeout_seconds: int = GENERATION_WAIT_TIMEOUT_SECONDS,
    ) -> dict[str, Any]:
        submission = self.submit_generation(path, body, idempotency_key=idempotency_key)
        return self.wait_for_generation(submission, timeout_seconds=timeout_seconds)

    def generate_image(self, prompt: str, **fields: Any) -> Any:
        self._apply_canvas_session_fields(fields, prompt, 1024, 1024)
        prompt = self._apply_art_spec(fields, prompt)
        idempotency_key = fields.pop("idempotencyKey", None)
        return self.submit_and_wait_generation(
            "/api/external/v1/editor/images/generations",
            {"prompt": prompt, **fields},
            idempotency_key=idempotency_key,
        )

    def edit_image(self, prompt: str, source_reference_id: str, **fields: Any) -> Any:
        source_reference_id = source_reference_id.strip()
        if not source_reference_id:
            raise GenarrativeApiError("source_reference_id must be a registered resource or asset ID")
        self._apply_canvas_session_fields(fields, prompt, 1024, 1024)
        prompt = self._apply_art_spec(fields, prompt)
        idempotency_key = fields.pop("idempotencyKey", None)
        return self.submit_and_wait_generation(
            "/api/external/v1/editor/images/edits",
            {"prompt": prompt, "sourceReferenceId": source_reference_id, **fields},
            idempotency_key=idempotency_key,
        )

    def remove_background(
        self,
        source_image_src: str,
        source_width: int | None = None,
        source_height: int | None = None,
        **fields: Any,
    ) -> Any:
        source_image_src = normalize_optional_text(source_image_src)
        if not source_image_src:
            raise GenarrativeApiError(
                "source_image_src must be an owner-scoped object key, resource ID, or asset ID"
            )
        if (source_width is None) != (source_height is None):
            raise GenarrativeApiError("source_width and source_height must be provided together")
        if source_width is not None and (
            source_width <= 0 or source_height is None or source_height <= 0
        ):
            raise GenarrativeApiError("source_width and source_height must be positive integers")
        session = fields.get("canvasSession")
        if session is None:
            session = fields.get("canvas_session")
        target_layer_id = normalize_optional_text(fields.get("targetLayerId"))
        if target_layer_id and fields.get("canvasCompletion") is not None:
            raise GenarrativeApiError(
                "targetLayerId and canvasCompletion are mutually exclusive for background removal"
            )
        canvas_width = fields.get("canvasWidth")
        canvas_height = fields.get("canvasHeight")
        if (canvas_width is None) != (canvas_height is None):
            raise GenarrativeApiError("canvasWidth and canvasHeight must be provided together")
        if session is not None and canvas_width is None and not target_layer_id:
            if source_width is None or source_height is None:
                raise GenarrativeApiError(
                    "remove_background requires source_width and source_height when canvasSession is used without canvasWidth/canvasHeight"
                )
            fields["canvasWidth"] = source_width
            fields["canvasHeight"] = source_height
        self._apply_canvas_session_fields(
            fields,
            fields.get("assetLabel", "去背景结果"),
            source_width or 1,
            source_height or 1,
        )
        if target_layer_id:
            fields.pop("canvasCompletion", None)
        idempotency_key = fields.pop("idempotencyKey", None)
        return self.submit_and_wait_generation(
            "/api/external/v1/editor/images/background-removals",
            {"sourceImageSrc": source_image_src, **fields},
            idempotency_key=idempotency_key,
        )

    def generate_icon_spritesheet(
        self,
        reference_id: str,
        icon_descriptions: list[str],
        slice_mode: str,
        **fields: Any,
    ) -> Any:
        reference_id = normalize_optional_text(reference_id)
        if not reference_id:
            raise GenarrativeApiError("reference_id must be a registered icon-spec resource or asset ID")
        descriptions = [item.strip() for item in icon_descriptions if item.strip()]
        if not descriptions:
            raise GenarrativeApiError("icon_descriptions must contain at least one non-empty item")
        slice_mode = normalize_optional_text(slice_mode)
        if slice_mode not in ("connected-components", "grid"):
            raise GenarrativeApiError(
                "slice_mode must be declared explicitly as 'connected-components' or 'grid'; the API has no default"
            )
        grid_x = fields.get("gridX")
        grid_y = fields.get("gridY")
        if slice_mode == "grid":
            if grid_x is None or grid_y is None:
                raise GenarrativeApiError("slice_mode='grid' requires both gridX and gridY")
        elif grid_x is not None or grid_y is not None:
            raise GenarrativeApiError(
                "slice_mode='connected-components' must not carry gridX/gridY"
            )
        label = fields.get("assetLabel", "图标图集")
        self._apply_canvas_session_fields(fields, label, 1024, 1024)
        fields.setdefault("screenColor", "auto")
        idempotency_key = fields.pop("idempotencyKey", None)
        return self.submit_and_wait_generation(
            "/api/external/v1/editor/icon-spritesheets/generations",
            {
                **fields,
                "referenceId": reference_id,
                "iconDescriptions": descriptions,
                "sliceMode": slice_mode,
            },
            idempotency_key=idempotency_key,
        )

    def extract_ui_assets(self, source_image_src: str, image_size: str = "1K", **fields: Any) -> Any:
        fields.pop("aspectRatio", None)
        self._apply_canvas_session_fields(fields, fields.get("spritesheetLabel", "UI 素材拆分"), 1024, 1024, "spritesheetLabel")
        idempotency_key = fields.pop("idempotencyKey", None)
        return self.submit_and_wait_generation(
            "/api/external/v1/editor/ui-designs/assets/extractions",
            {"sourceImageSrc": source_image_src, "imageSize": image_size, **fields, "aspectRatio": "1:1"},
            idempotency_key=idempotency_key,
        )

    def animate_character(
        self,
        source_image_src: str,
        source_width: int,
        source_height: int,
        prompt_text: str,
        source_layer_id: str,
        **fields: Any,
    ) -> Any:
        self._apply_canvas_session_fields(
            fields,
            fields.get("canvasTitle", "角色动画"),
            source_width,
            source_height,
            asset_label_field="assetLabel",
        )
        prompt_text = self._apply_art_spec(fields, prompt_text)
        idempotency_key = fields.pop("idempotencyKey", None)
        body = {
            "sourceLayerId": source_layer_id,
            "sourceImageSrc": source_image_src,
            "sourceWidth": source_width,
            "sourceHeight": source_height,
            "promptText": prompt_text,
            "resolution": fields.pop("resolution", "720p"),
            "ratio": fields.pop("ratio", "same"),
            "frameCount": fields.pop("frameCount", 40),
            "durationSeconds": fields.pop("durationSeconds", 5),
            **fields,
            "model": "seedance2.0-fast",
        }
        return self.submit_and_wait_generation(
            "/api/external/v1/editor/character-animations/generations",
            body,
            idempotency_key=idempotency_key,
        )

    def generate_video(self, prompt: str, **fields: Any) -> Any:
        fields.pop("mode", None)
        self._apply_canvas_session_fields(fields, prompt, 1280, 720)
        prompt = self._apply_art_spec(fields, prompt)
        idempotency_key = fields.pop("idempotencyKey", None)
        body = {
            "prompt": prompt,
            "model": fields.pop("model", "seedance2.0-fast"),
            "aspectRatio": fields.pop("aspectRatio", "16:9"),
            "durationSeconds": fields.pop("durationSeconds", 5),
            "resolution": fields.pop("resolution", "720p"),
            "sound": fields.pop("sound", "off"),
            **fields,
            "mode": "std",
        }
        return self.submit_and_wait_generation(
            "/api/external/v1/editor/videos/generations",
            body,
            idempotency_key=idempotency_key,
        )

    def generate_sound_effect(
        self,
        prompt: str,
        duration: float | None = None,
        loop: bool = False,
        **fields: Any,
    ) -> Any:
        self._apply_canvas_session_fields(fields, prompt, 360, 120)
        prompt = self._apply_art_spec(fields, prompt)
        idempotency_key = fields.pop("idempotencyKey", None)
        return self.submit_and_wait_generation(
            "/api/external/v1/editor/audios/sound-effects/generations",
            {"prompt": prompt, "duration": duration, "loop": loop, **fields},
            idempotency_key=idempotency_key,
        )

    def generate_background_music(self, description: str, **fields: Any) -> Any:
        self._apply_canvas_session_fields(fields, description, 360, 120)
        description = self._apply_art_spec(fields, description)
        idempotency_key = fields.pop("idempotencyKey", None)
        return self.submit_and_wait_generation(
            "/api/external/v1/editor/audios/background-music/generations",
            {"gptDescriptionPrompt": description, **fields, "makeInstrumental": True},
            idempotency_key=idempotency_key,
        )


def _self_test() -> None:
    png = (
        b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR"
        b"\x00\x00\x00\x02\x00\x00\x00\x03\x08\x06\x00\x00\x00"
    )
    with tempfile.NamedTemporaryFile(suffix="Hero Image.png", delete=False) as fh:
        fh.write(png)
        temp_path = fh.name
    try:
        assert image_dimensions(temp_path) == (2, 3)
        assert source_layer_id_from_path(temp_path).startswith("external-reference-")
    finally:
        Path(temp_path).unlink(missing_ok=True)
    assert unwrap_envelope({"ok": True, "data": {"upload": 1}}) == {"upload": 1}
    client = GenarrativeExternalClient(api_key="test")
    session = {"projectId": "proj-demo", "assetFolderId": "editor-asset-folder-demo"}
    fields = client.canvas_generation_fields(session, "英雄角色", width=512, height=768)
    assert fields["projectId"] == "proj-demo"
    assert fields["assetFolderId"] == "editor-asset-folder-demo"
    assert fields["assetLabel"] == "英雄角色"
    assert fields["canvasCompletion"]["placeholder"]["height"] == 768
    assert "美术规范" in art_spec_prompt("生成角色", {"style": "水彩"})
    calls: list[dict[str, Any]] = []

    def fake_request_json(
        method: str,
        path: str,
        body: dict[str, Any] | None = None,
        query: dict[str, Any] | None = None,
        auth: bool = True,
        timeout: int = DEFAULT_REQUEST_TIMEOUT_SECONDS,
        headers: dict[str, str] | None = None,
    ) -> Any:
        calls.append({
            "method": method,
            "path": path,
            "body": body,
            "timeout": timeout,
            "headers": headers,
        })
        generated = {
            "taskId": "task-demo",
            "model": "seedance2.0-fast",
            "prompt": "角色呼吸",
            "previewVideoPath": "/generated/preview.mp4",
            "frames": [{"imageSrc": "/generated/frame01.png", "width": 512, "height": 768}],
            "resource": {
                "resourceId": "editor-resource-demo",
                "assetKind": "character-animation",
                "sourceResourceId": "editor-resource-preview-demo",
                "imageSequenceFrames": [
                    {"imageSrc": "/generated/frame01.png", "width": 512, "height": 768},
                    {"imageSrc": "/generated/frame02.png", "width": 512, "height": 768},
                ],
                "imageSequenceDurationMs": 4000,
            },
            "asset": {
                "assetId": "editor-asset-demo",
                "assetKind": "character-animation",
                "imageSequenceFrames": [
                    {"imageSrc": "/generated/frame01.png", "width": 512, "height": 768},
                    {"imageSrc": "/generated/frame02.png", "width": 512, "height": 768},
                ],
                "imageSequenceDurationMs": 4000,
            },
        }
        if method == "POST":
            return {"operationId": "task-operation-demo", "status": "queued", "pollAfterMs": 1}
        if path == "/api/external/v1/generations/task-operation-demo":
            return {"operationId": "task-operation-demo", "status": "completed", "result": generated}
        return generated

    client.request_json = fake_request_json  # type: ignore[method-assign]
    result = client.animate_character(
        "/generated/source.png",
        512,
        768,
        "角色呼吸",
        "layer-hero",
        canvasSession=session,
        canvasTitle="角色呼吸动画",
    )
    assert calls[0]["timeout"] == DEFAULT_REQUEST_TIMEOUT_SECONDS
    assert calls[0]["headers"]["Idempotency-Key"]
    assert calls[0]["body"]["projectId"] == "proj-demo"
    assert calls[0]["body"]["assetFolderId"] == "editor-asset-folder-demo"
    assert calls[0]["body"]["assetLabel"] == "角色呼吸动画"
    assert calls[0]["body"]["canvasCompletion"]["title"] == "角色呼吸动画"
    assert calls[1]["path"] == "/api/external/v1/generations/task-operation-demo"
    assert result["asset"]["assetId"] == "editor-asset-demo"
    assert result["asset"]["assetKind"] == "character-animation"
    assert len(result["asset"]["imageSequenceFrames"]) == 2
    assert result["asset"]["imageSequenceDurationMs"] == 4000
    calls.clear()
    background_result = client.remove_background(
        "uploads/source.png",
        720,
        1280,
        canvasSession=session,
        assetLabel="去背景结果",
    )
    assert background_result["taskId"] == "task-demo"
    assert calls[0]["path"] == "/api/external/v1/editor/images/background-removals"
    assert calls[0]["body"]["sourceImageSrc"] == "uploads/source.png"
    assert calls[0]["body"]["projectId"] == "proj-demo"
    assert calls[0]["body"]["assetFolderId"] == "editor-asset-folder-demo"
    assert calls[0]["body"]["assetLabel"] == "去背景结果"
    assert calls[0]["body"]["canvasCompletion"]["title"] == "去背景结果"
    assert calls[0]["body"]["canvasCompletion"]["placeholder"]["width"] == 720
    assert calls[0]["body"]["canvasCompletion"]["placeholder"]["height"] == 1280
    calls.clear()
    client.remove_background(
        "uploads/source.png",
        canvasSession=session,
        targetLayerId="layer-1",
        assetLabel="原位去背景结果",
    )
    assert calls[0]["body"]["projectId"] == "proj-demo"
    assert calls[0]["body"]["assetFolderId"] == "editor-asset-folder-demo"
    assert calls[0]["body"]["assetLabel"] == "原位去背景结果"
    assert calls[0]["body"]["targetLayerId"] == "layer-1"
    assert "canvasCompletion" not in calls[0]["body"]
    calls.clear()
    try:
        client.remove_background(
            "uploads/source.png",
            canvasSession=session,
            targetLayerId="layer-1",
            canvasCompletion={"title": "冲突完成指令"},
        )
    except GenarrativeApiError as error:
        assert "targetLayerId and canvasCompletion are mutually exclusive" in str(error)
    else:
        raise AssertionError("background removal must reject conflicting canvas placement modes")
    assert calls == []
    try:
        client.remove_background("uploads/source.png", canvasSession=session)
    except GenarrativeApiError as error:
        assert "source_width and source_height" in str(error)
    else:
        raise AssertionError("canvas background removal must not guess source dimensions")
    assert calls == []
    client.generate_icon_spritesheet(
        "editor-resource-spec",
        ["蛇头向上", "蛇身直线", "转角", "尾部", "四类食物"],
        "connected-components",
        canvasSession=session,
        assetLabel="贪吃蛇透明图集",
        referenceId="must-not-override-explicit-reference",
        iconDescriptions=["不得覆盖显式图标描述"],
    )
    assert calls[0]["path"] == "/api/external/v1/editor/icon-spritesheets/generations"
    assert calls[0]["body"]["referenceId"] == "editor-resource-spec"
    assert calls[0]["body"]["screenColor"] == "auto"
    assert calls[0]["body"]["sliceMode"] == "connected-components"
    assert calls[0]["body"]["iconDescriptions"][0] == "蛇头向上"
    assert calls[1]["path"] == "/api/external/v1/generations/task-operation-demo"
    print("self-test ok")


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--self-test", action="store_true")
    parser.add_argument("--credentials-file", default=str(DEFAULT_CREDENTIALS_FILE))
    parser.add_argument("command", nargs="?", choices=["openapi", "list-projects"])
    args = parser.parse_args(argv)
    if args.self_test:
        _self_test()
        return 0
    if args.command is None:
        parser.print_help()
        return 0
    if args.command == "openapi":
        client = GenarrativeExternalClient(api_key="", credentials_file=args.credentials_file)
        print(json.dumps(client.openapi(), ensure_ascii=False, indent=2))
    elif args.command == "list-projects":
        client = GenarrativeExternalClient(credentials_file=args.credentials_file)
        print(json.dumps(client.list_projects(), ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())