Files
Genarrative/.codex/skills/genarrative-external-editor-api/scripts/genarrative_external_api.py
T
k88936 8abeb333bb
Project CI / Repository checks (push) Successful in 56s
Project CI / Frontend tests (push) Successful in 4m1s
Project CI / Backend tests (push) Successful in 4m26s
Project CI / Native shell tests (push) Successful in 15m12s
统一画布快速编辑白名单与入口行为 (#140)
- 使用白名单控制快速编辑功能的显示

- 同步浮动工具栏、右键菜单、打开入口和提交门禁
- 禁用角色动作序列帧(整体)、音频和单个图标(explain: backend reject)的快速编辑

- 把快速编辑api改为只接受resource Id | asset Id,删除 image_src 参数, 以保证校验

其中包含生产数据迁移。迁移用于清理历史普通图片错误持久化的 assetKind="image",覆盖 asset → project-resource → showcase → canvas 四个作用域,并包含 operator 鉴权、分批 dry-run、批次 SHA-256 绑定 apply、active 结构化画
  > 布迁移摘要同步及最终零残留复核。

---------

Co-authored-by: kdletters <kdletters@qq.com>
Reviewed-on: http://192.168.35.82/git/GenarrativeAI/Genarrative/pulls/140
Co-authored-by: 王德宇 <kvtodev@outlook.com>
Co-committed-by: 王德宇 <kvtodev@outlook.com>
2026-08-12 14:46:24 +08:00

815 lines
32 KiB
Python

#!/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 generate_icon_spritesheet(
self,
reference_id: str,
icon_descriptions: list[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")
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,
},
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") as fh:
fh.write(png)
fh.flush()
assert image_dimensions(fh.name) == (2, 3)
assert source_layer_id_from_path(fh.name).startswith("external-reference-")
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()
client.generate_icon_spritesheet(
"editor-resource-spec",
["蛇头向上", "蛇身直线", "转角", "尾部", "四类食物"],
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"]["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())