393 lines
15 KiB
Python
393 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
"""NianCode's fixed, keyless adapter for the Meowa game-assets Skill.
|
||
|
||
The Meowa credential stays in Electron Main. This CLI talks only to the local
|
||
NianCode Host API and writes generated output plus provenance metadata into the
|
||
active project.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import base64
|
||
from datetime import datetime, timezone
|
||
import json
|
||
import mimetypes
|
||
import os
|
||
from pathlib import Path
|
||
import re
|
||
import sys
|
||
import time
|
||
from typing import Any
|
||
from urllib.error import HTTPError, URLError
|
||
from urllib.parse import urlencode
|
||
from urllib.request import Request, urlopen
|
||
|
||
|
||
SKILL_VERSION = "2026.06.19.1"
|
||
DEFAULT_HOST_BASE = "http://127.0.0.1:13210"
|
||
DEFAULT_OUTPUT_DIR = "assets/generated/meowa"
|
||
TERMINAL_STATUSES = {"success", "completed", "failure", "failed", "cancelled", "canceled"}
|
||
|
||
|
||
class AdapterError(RuntimeError):
|
||
pass
|
||
|
||
|
||
def host_base() -> str:
|
||
value = os.environ.get("NIANCODE_HOST_API_BASE_URL", DEFAULT_HOST_BASE).strip()
|
||
return value.rstrip("/")
|
||
|
||
|
||
def host_token() -> str:
|
||
token = os.environ.get("NIANCODE_HOST_API_TOKEN", "").strip()
|
||
if not token:
|
||
raise AdapterError("NianCode Host API token is unavailable; run this command from a NianCode Agent")
|
||
return token
|
||
|
||
|
||
def parse_payload(raw: bytes) -> Any:
|
||
if not raw:
|
||
return None
|
||
try:
|
||
return json.loads(raw.decode("utf-8"))
|
||
except (UnicodeDecodeError, json.JSONDecodeError):
|
||
return raw.decode("utf-8", errors="replace")[:500]
|
||
|
||
|
||
def payload_message(payload: Any, fallback: str) -> str:
|
||
if isinstance(payload, dict):
|
||
for key in ("error", "message", "detail", "msg"):
|
||
value = payload.get(key)
|
||
if isinstance(value, str) and value.strip():
|
||
return value.strip()[:500]
|
||
if isinstance(payload, str) and payload.strip():
|
||
return payload.strip()[:500]
|
||
return fallback
|
||
|
||
|
||
def request_json(path: str, method: str = "GET", payload: dict[str, Any] | None = None, timeout: int = 240) -> Any:
|
||
url = path if path.startswith("http://") or path.startswith("https://") else f"{host_base()}{path}"
|
||
body = None
|
||
headers = {
|
||
"Accept": "application/json",
|
||
"Authorization": f"Bearer {host_token()}",
|
||
}
|
||
if payload is not None:
|
||
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
||
headers["Content-Type"] = "application/json"
|
||
request = Request(url, data=body, headers=headers, method=method)
|
||
try:
|
||
with urlopen(request, timeout=timeout) as response:
|
||
return parse_payload(response.read())
|
||
except HTTPError as error:
|
||
payload_value = parse_payload(error.read())
|
||
raise AdapterError(payload_message(payload_value, f"Host API returned HTTP {error.code}")) from error
|
||
except URLError as error:
|
||
raise AdapterError(f"无法连接 NianCode Host API：{error.reason}") from error
|
||
|
||
|
||
def json_dump(path: Path, value: Any) -> None:
|
||
path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
|
||
|
||
def print_json(value: Any) -> None:
|
||
print(json.dumps(value, ensure_ascii=False, indent=2))
|
||
|
||
|
||
def cli_path() -> Path:
|
||
return Path(__file__).resolve()
|
||
|
||
|
||
def command_skill_doc(args: argparse.Namespace) -> int:
|
||
query = urlencode({key: value for key, value in {"task": args.task, "topic": args.topic}.items() if value})
|
||
path = "/api/meowa/game-assets/skill-doc"
|
||
if query:
|
||
path = f"{path}?{query}"
|
||
try:
|
||
payload = request_json(path)
|
||
content = payload.get("content") if isinstance(payload, dict) else None
|
||
if isinstance(content, str) and content.strip():
|
||
print(content)
|
||
return 0
|
||
raise AdapterError("Skill 文档响应没有 content")
|
||
except AdapterError as error:
|
||
fallback = cli_path().with_name("meowart_api.md")
|
||
if fallback.is_file():
|
||
print(f"[WARN] online Meowa guide unavailable; using bundled guide: {error}", file=sys.stderr)
|
||
print(fallback.read_text(encoding="utf-8"))
|
||
return 0
|
||
raise
|
||
|
||
|
||
def command_skill_doc_status(_args: argparse.Namespace) -> int:
|
||
guide = cli_path().with_name("meowart_api.md")
|
||
print_json({
|
||
"skillVersion": SKILL_VERSION,
|
||
"bundledGuide": str(guide),
|
||
"bundledGuideExists": guide.is_file(),
|
||
"remoteGuide": "通过 NianCode Main 代理读取",
|
||
})
|
||
return 0
|
||
|
||
|
||
def command_config_status(_args: argparse.Namespace) -> int:
|
||
print_json(request_json("/api/meowa/game-assets/config"))
|
||
return 0
|
||
|
||
|
||
def command_template_info(args: argparse.Namespace) -> int:
|
||
print_json(request_json(f"/api/meowa/game-assets/template-info?kind={args.kind}"))
|
||
return 0
|
||
|
||
|
||
def command_credits(_args: argparse.Namespace) -> int:
|
||
print_json(request_json("/api/meowa/game-assets/credits"))
|
||
return 0
|
||
|
||
|
||
def slug(value: str) -> str:
|
||
value = re.sub(r"[^A-Za-z0-9_-]+", "-", value.strip()).strip("-_")
|
||
return value[:72] or "asset"
|
||
|
||
|
||
def now_iso() -> str:
|
||
return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")
|
||
|
||
|
||
def read_reference_files(paths: list[str]) -> list[dict[str, str]]:
|
||
references: list[dict[str, str]] = []
|
||
total = 0
|
||
for raw_path in paths:
|
||
path = Path(raw_path).expanduser()
|
||
if not path.is_file():
|
||
raise AdapterError(f"参考文件不存在：{path}")
|
||
data = path.read_bytes()
|
||
if len(data) > 8 * 1024 * 1024:
|
||
raise AdapterError(f"参考文件过大：{path}")
|
||
total += len(data)
|
||
if total > 16 * 1024 * 1024:
|
||
raise AdapterError("参考文件总大小超过限制")
|
||
references.append({
|
||
"name": path.name,
|
||
"mimeType": mimetypes.guess_type(path.name)[0] or "application/octet-stream",
|
||
"dataBase64": base64.b64encode(data).decode("ascii"),
|
||
})
|
||
return references
|
||
|
||
|
||
def safe_request_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
result = dict(payload)
|
||
references = result.get("referenceFiles")
|
||
if isinstance(references, list):
|
||
result["referenceFiles"] = [
|
||
{"name": item.get("name"), "mimeType": item.get("mimeType")}
|
||
for item in references
|
||
if isinstance(item, dict)
|
||
]
|
||
return result
|
||
|
||
|
||
def extension_for(mime_type: str, file_name: str) -> str:
|
||
suffix = Path(file_name).suffix.lower()
|
||
if suffix and len(suffix) <= 10:
|
||
return suffix
|
||
return {
|
||
"image/png": ".png",
|
||
"image/jpeg": ".jpg",
|
||
"image/webp": ".webp",
|
||
"application/zip": ".zip",
|
||
"audio/mpeg": ".mp3",
|
||
"audio/wav": ".wav",
|
||
}.get(mime_type.lower(), ".bin")
|
||
|
||
|
||
def download_job(kind: str, job_id: str, output_index: int | None = None) -> dict[str, Any]:
|
||
params: dict[str, str] = {"kind": kind, "id": job_id}
|
||
if output_index is not None:
|
||
params["outputIndex"] = str(output_index)
|
||
payload = request_json(f"/api/meowa/game-assets/download?{urlencode(params)}")
|
||
if not isinstance(payload, dict) or not payload.get("success"):
|
||
raise AdapterError(payload_message(payload, "Meowa 下载响应无效"))
|
||
return payload
|
||
|
||
|
||
def generation_payload(args: argparse.Namespace, kind: str) -> dict[str, Any]:
|
||
template_config: Any = {}
|
||
if args.template_config:
|
||
try:
|
||
template_config = json.loads(args.template_config)
|
||
except json.JSONDecodeError as error:
|
||
raise AdapterError(f"template-config 不是有效 JSON：{error}") from error
|
||
if not isinstance(template_config, dict):
|
||
raise AdapterError("template-config 必须是 JSON 对象")
|
||
|
||
payload: dict[str, Any] = {
|
||
"kind": kind,
|
||
"templateName": args.template_name,
|
||
"templateConfig": template_config,
|
||
"requirement": args.requirement,
|
||
"jobName": args.job_name,
|
||
"modelName": args.model_name,
|
||
"resolution": args.resolution,
|
||
"aspectRatio": args.aspect_ratio,
|
||
"temperature": args.temperature,
|
||
"includeBase64": args.include_base64,
|
||
"referenceFiles": read_reference_files(args.reference_file),
|
||
}
|
||
if kind == "hd":
|
||
payload.update({
|
||
"hdRemoveBgMode": args.remove_bg_mode,
|
||
"projectId": args.project_id,
|
||
"threadId": args.thread_id,
|
||
})
|
||
return payload
|
||
|
||
|
||
def poll_job(kind: str, job_id: str, max_wait: int, poll_interval: float) -> dict[str, Any]:
|
||
deadline = time.time() + max(max_wait, 1)
|
||
last: dict[str, Any] = {}
|
||
while time.time() <= deadline:
|
||
payload = request_json(f"/api/meowa/game-assets/jobs?{urlencode({'kind': kind, 'id': job_id})}")
|
||
if not isinstance(payload, dict):
|
||
raise AdapterError("Meowa 任务响应不是 JSON 对象")
|
||
last = payload
|
||
status = str(payload.get("status") or "").strip().lower()
|
||
print(f"[INFO] meowa job={job_id} status={status or 'unknown'}", file=sys.stderr)
|
||
if status in TERMINAL_STATUSES:
|
||
return payload
|
||
time.sleep(max(poll_interval, 0.2))
|
||
raise AdapterError(f"Meowa 任务等待超时：{job_id}，最后状态 {last.get('status', 'unknown')}")
|
||
|
||
|
||
def command_generate(args: argparse.Namespace, kind: str) -> int:
|
||
payload = generation_payload(args, kind)
|
||
output_root = Path(args.output_dir).expanduser()
|
||
run_dir = output_root / f"{slug(args.template_name)}-{int(time.time())}"
|
||
run_dir.mkdir(parents=True, exist_ok=True)
|
||
started_at = now_iso()
|
||
|
||
json_dump(run_dir / "request.meta.json", {
|
||
"provider": "meowa",
|
||
"skillVersion": SKILL_VERSION,
|
||
"kind": kind,
|
||
"startedAt": started_at,
|
||
"request": safe_request_payload(payload),
|
||
})
|
||
|
||
try:
|
||
submit = request_json("/api/meowa/game-assets/generate", method="POST", payload=payload)
|
||
json_dump(run_dir / "submit_response.json", submit)
|
||
if not isinstance(submit, dict):
|
||
raise AdapterError("Meowa 提交响应不是 JSON 对象")
|
||
job_id = str(submit.get("api_job_id") or submit.get("job_id") or "").strip()
|
||
if not job_id:
|
||
raise AdapterError("Meowa 提交响应缺少任务 ID")
|
||
print(f"[INFO] submitted meowa job={job_id}", file=sys.stderr)
|
||
|
||
final = poll_job(kind, job_id, args.max_wait, args.poll_interval)
|
||
json_dump(run_dir / "final_response.json", final)
|
||
status = str(final.get("status") or "").strip().lower()
|
||
if status not in {"success", "completed"}:
|
||
raise AdapterError(f"Meowa 任务未成功结束：{status or 'unknown'}")
|
||
|
||
download = download_job(kind, job_id)
|
||
file_name = str(download.get("fileName") or f"{job_id}.bin")
|
||
mime_type = str(download.get("mimeType") or "application/octet-stream")
|
||
encoded = download.get("dataBase64")
|
||
if not isinstance(encoded, str) or not encoded:
|
||
raise AdapterError("Meowa 下载响应缺少文件内容")
|
||
file_path = run_dir / f"asset{extension_for(mime_type, file_name)}"
|
||
file_path.write_bytes(base64.b64decode(encoded))
|
||
|
||
metadata = {
|
||
"provider": "meowa",
|
||
"integration": "niancode-meowa-game-assets",
|
||
"skillVersion": SKILL_VERSION,
|
||
"kind": kind,
|
||
"templateName": args.template_name,
|
||
"requirement": args.requirement,
|
||
"jobId": job_id,
|
||
"generatedAt": now_iso(),
|
||
"localPath": str(file_path.as_posix()),
|
||
"mimeType": mime_type,
|
||
"sourceType": "ai-generated",
|
||
"licenseStatus": "verify-meowa-plan-before-shipping",
|
||
"commercialUse": "依 Meowa 当前账户方案和条款确认；不得默认视为 CC0。",
|
||
"termsUrl": "https://meowa.ai/tos",
|
||
"upstreamSkill": "https://github.com/Meowa-AI/meowa-skills/tree/main/skills/game-assets",
|
||
}
|
||
json_dump(run_dir / "generation.meta.json", metadata)
|
||
print(f"[INFO] saved_dir={run_dir}", file=sys.stderr)
|
||
print_json({"success": True, "jobId": job_id, "file": str(file_path.as_posix()), "metadata": str((run_dir / 'generation.meta.json').as_posix())})
|
||
return 0
|
||
except Exception as error:
|
||
json_dump(run_dir / "error.json", {"error": str(error), "failedAt": now_iso()})
|
||
raise
|
||
|
||
|
||
def add_generation_args(parser: argparse.ArgumentParser, kind: str) -> None:
|
||
parser.add_argument("--template-name", required=True)
|
||
parser.add_argument("--requirement", required=True)
|
||
parser.add_argument("--template-config", default="")
|
||
parser.add_argument("--job-name", default="")
|
||
parser.add_argument("--model-name", default="gemini-3.1-flash-image-preview" if kind == "hd" else "")
|
||
parser.add_argument("--resolution", default="")
|
||
parser.add_argument("--aspect-ratio", default="1:1")
|
||
parser.add_argument("--temperature", type=float, default=0.0)
|
||
parser.add_argument("--include-base64", action="store_true")
|
||
parser.add_argument("--reference-file", action="append", default=[])
|
||
parser.add_argument("--output-dir", default=DEFAULT_OUTPUT_DIR)
|
||
parser.add_argument("--max-wait", type=int, default=900)
|
||
parser.add_argument("--poll-interval", type=float, default=3.0)
|
||
if kind == "hd":
|
||
parser.add_argument("--remove-bg-mode", default="")
|
||
parser.add_argument("--project-id", default=None)
|
||
parser.add_argument("--thread-id", default=None)
|
||
|
||
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
parser = argparse.ArgumentParser(description="NianCode keyless adapter for Meowa game assets")
|
||
subparsers = parser.add_subparsers(dest="command", required=True)
|
||
|
||
skill_doc = subparsers.add_parser("skill-doc")
|
||
skill_doc.add_argument("--task", default="")
|
||
skill_doc.add_argument("--topic", default="")
|
||
skill_doc.set_defaults(handler=command_skill_doc)
|
||
|
||
status = subparsers.add_parser("skill-doc-status")
|
||
status.set_defaults(handler=command_skill_doc_status)
|
||
|
||
config = subparsers.add_parser("config-status")
|
||
config.set_defaults(handler=command_config_status)
|
||
|
||
template = subparsers.add_parser("template-info")
|
||
template.add_argument("--kind", choices=("pixel", "hd"), required=True)
|
||
template.set_defaults(handler=command_template_info)
|
||
|
||
credits = subparsers.add_parser("credits")
|
||
credits.set_defaults(handler=command_credits)
|
||
|
||
for kind in ("pixel", "hd"):
|
||
generator = subparsers.add_parser(f"{kind}-gen")
|
||
add_generation_args(generator, kind)
|
||
generator.set_defaults(handler=lambda args, selected_kind=kind: command_generate(args, selected_kind))
|
||
|
||
return parser
|
||
|
||
|
||
def main() -> int:
|
||
args = build_parser().parse_args()
|
||
try:
|
||
return int(args.handler(args))
|
||
except AdapterError as error:
|
||
print(f"[ERROR] {error}", file=sys.stderr)
|
||
return 1
|
||
except (OSError, ValueError, json.JSONDecodeError) as error:
|
||
print(f"[ERROR] {error}", file=sys.stderr)
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|