Makelore 2.0 initial clean snapshot

.opencode/skills/game-assets/meowart_api.py

@@ -0,0 +1,392 @@
+#!/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())