feat: add Atlas Cloud image generation skill

skills/README.md

@@ -19,4 +19,5 @@
 | `pixabay-footage` | Pixabay 免费素材搜索下载 | main + content-producer 继承 |
 | `wxwork-drive` | 企业微信微盘 | main + content-producer 继承 |
 | `siliconflow-img-gen` | 硅基流动生图（Phase 5 改火山） | main + content-producer 继承 |
+| `atlascloud-img-gen` | Atlas Cloud 异步图像生成与编辑 | main + content-producer 继承 |
 | `youtube-publish` | YouTube 视频发布（Data API v3 + OAuth2） | main + content-producer 继承 |

skills/atlascloud-img-gen/SKILL.md

@@ -0,0 +1,35 @@
+---
+name: atlascloud-img-gen
+description: Generate or edit images with Atlas Cloud's asynchronous Media API. Supports text prompts, remote reference images, local image upload, polling, and downloading results.
+metadata:
+  openclaw:
+    emoji: 🎨
+    requires:
+      bins:
+      - python3
+      env:
+      - ATLASCLOUD_API_KEY
+    primaryEnv: ATLASCLOUD_API_KEY
+    homepage: https://www.atlascloud.ai/console
+---
+
+# Atlas Cloud Image Generation
+
+Use `atlascloud-img-gen` for text-to-image and image-edit jobs. The wrapper submits an asynchronous Atlas Cloud task, polls it to completion, and downloads every output.
+
+```bash
+atlascloud-img-gen --prompt "a cinematic mountain village at sunrise"
+atlascloud-img-gen --prompt "turn this into a watercolor" --image https://example.com/source.jpg
+atlascloud-img-gen --prompt "remove the background" --image ./source.png --out-dir ./output
+```
+
+The default model is `bytedance/seedream-v5.0-lite`. Override it with `--model`. Model-specific options can be supplied as JSON with `--params`, for example:
+
+```bash
+atlascloud-img-gen --prompt "editorial product photo" \
+  --params '{"image_size":"2048x2048"}'
+```
+
+Local files passed to `--image` are uploaded first. Remote `http` or `https` URLs are sent directly. Credentials come only from `ATLASCLOUD_API_KEY`; do not place keys in arguments or files.
+
+Outputs are written below `./tmp/atlascloud-img-<timestamp>` unless `--out-dir` is set. The directory contains downloaded images and `result.json` with the prediction metadata.

skills/atlascloud-img-gen/atlascloud-img-gen.sh

@@ -0,0 +1,6 @@
+#!/usr/bin/env bash
+set -euo pipefail
+SELF="${BASH_SOURCE[0]}"
+while [ -L "$SELF" ]; do SELF="$(readlink -f "$SELF")"; done
+SCRIPT_DIR="$(cd "$(dirname "$SELF")" && pwd)"
+exec python3 "$SCRIPT_DIR/scripts/gen.py" "$@"

skills/atlascloud-img-gen/scripts/gen.py

@@ -0,0 +1,162 @@
+#!/usr/bin/env python3
+"""Atlas Cloud asynchronous image generation client."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import mimetypes
+import os
+import sys
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+import uuid
+from pathlib import Path
+
+API_BASE = "https://api.atlascloud.ai/api/v1"
+GENERATE_URL = f"{API_BASE}/model/generateImage"
+UPLOAD_URL = f"{API_BASE}/model/uploadMedia"
+DEFAULT_MODEL = "bytedance/seedream-v5.0-lite"
+TERMINAL_SUCCESS = {"completed", "succeeded", "success"}
+TERMINAL_FAILURE = {"failed", "canceled", "cancelled"}
+
+
+def request_json(url: str, api_key: str, *, method: str = "GET", payload: dict | None = None) -> dict:
+    data = json.dumps(payload).encode() if payload is not None else None
+    req = urllib.request.Request(
+        url,
+        data=data,
+        method=method,
+        headers={
+            "Authorization": f"Bearer {api_key}",
+            "Content-Type": "application/json",
+            "User-Agent": "xiaobei-atlascloud-img-gen/1.0",
+        },
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=120) as response:
+            return json.loads(response.read())
+    except urllib.error.HTTPError as error:
+        body = error.read().decode(errors="replace")
+        raise RuntimeError(f"Atlas Cloud HTTP {error.code}: {body}") from error
+
+
+def upload_file(path: Path, api_key: str) -> str:
+    boundary = f"----atlascloud-{uuid.uuid4().hex}"
+    mime = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
+    body = (
+        f"--{boundary}\r\n"
+        f'Content-Disposition: form-data; name="file"; filename="{path.name}"\r\n'
+        f"Content-Type: {mime}\r\n\r\n"
+    ).encode() + path.read_bytes() + f"\r\n--{boundary}--\r\n".encode()
+    req = urllib.request.Request(
+        UPLOAD_URL,
+        data=body,
+        method="POST",
+        headers={
+            "Authorization": f"Bearer {api_key}",
+            "Content-Type": f"multipart/form-data; boundary={boundary}",
+            "User-Agent": "xiaobei-atlascloud-img-gen/1.0",
+        },
+    )
+    with urllib.request.urlopen(req, timeout=120) as response:
+        result = json.loads(response.read())
+    url = result.get("data", {}).get("download_url")
+    if not url:
+        raise RuntimeError(f"Atlas Cloud upload returned no download_url: {result}")
+    return url
+
+
+def resolve_image(value: str | None, api_key: str) -> str | None:
+    if not value:
+        return None
+    if value.startswith(("http://", "https://")):
+        return value
+    path = Path(value).expanduser()
+    if not path.is_file():
+        raise ValueError(f"Image file not found: {path}")
+    return upload_file(path, api_key)
+
+
+def prediction_id(result: dict) -> str:
+    value = result.get("data", {}).get("id") or result.get("id")
+    if not value:
+        raise RuntimeError(f"Atlas Cloud returned no prediction id: {result}")
+    return str(value)
+
+
+def poll_prediction(identifier: str, api_key: str, interval: float, timeout: float) -> dict:
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        result = request_json(f"{API_BASE}/model/prediction/{identifier}", api_key)
+        data = result.get("data", result)
+        status = str(data.get("status", "")).lower()
+        if status in TERMINAL_SUCCESS:
+            return result
+        if status in TERMINAL_FAILURE:
+            raise RuntimeError(f"Atlas Cloud prediction {status}: {result}")
+        time.sleep(interval)
+    raise TimeoutError(f"Atlas Cloud prediction timed out after {timeout:g}s")
+
+
+def output_urls(result: dict) -> list[str]:
+    data = result.get("data", result)
+    outputs = data.get("outputs") or data.get("output") or []
+    if isinstance(outputs, str):
+        outputs = [outputs]
+    return [item for item in outputs if isinstance(item, str) and item.startswith(("http://", "https://"))]
+
+
+def download(url: str, destination: Path) -> None:
+    req = urllib.request.Request(url, headers={"User-Agent": "xiaobei-atlascloud-img-gen/1.0"})
+    with urllib.request.urlopen(req, timeout=120) as response:
+        destination.write_bytes(response.read())
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Generate or edit images with Atlas Cloud")
+    parser.add_argument("--prompt", required=True)
+    parser.add_argument("--model", default=DEFAULT_MODEL)
+    parser.add_argument("--image", help="Remote URL or local reference image")
+    parser.add_argument("--params", default="{}", help="Additional model parameters as JSON")
+    parser.add_argument("--out-dir")
+    parser.add_argument("--poll-interval", type=float, default=3)
+    parser.add_argument("--timeout", type=float, default=300)
+    args = parser.parse_args()
+
+    api_key = os.environ.get("ATLASCLOUD_API_KEY", "").strip()
+    if not api_key:
+        print("[error] ATLASCLOUD_API_KEY not set", file=sys.stderr)
+        raise SystemExit(1)
+    try:
+        extra = json.loads(args.params)
+        if not isinstance(extra, dict):
+            raise ValueError("--params must decode to a JSON object")
+        image_url = resolve_image(args.image, api_key)
+        payload = {"model": args.model, "prompt": args.prompt, **extra}
+        if image_url:
+            payload["image_url"] = image_url
+        submitted = request_json(GENERATE_URL, api_key, method="POST", payload=payload)
+        identifier = prediction_id(submitted)
+        print(f"[info] prediction={identifier}", file=sys.stderr)
+        result = poll_prediction(identifier, api_key, args.poll_interval, args.timeout)
+        urls = output_urls(result)
+        if not urls:
+            raise RuntimeError(f"Completed prediction returned no output URLs: {result}")
+        out_dir = Path(args.out_dir) if args.out_dir else Path(f"./tmp/atlascloud-img-{int(time.time())}")
+        out_dir.mkdir(parents=True, exist_ok=True)
+        for index, url in enumerate(urls):
+            suffix = Path(urllib.parse.urlparse(url).path).suffix
+            destination = out_dir / f"{index:02d}{suffix or '.png'}"
+            download(url, destination)
+            print(destination)
+        (out_dir / "result.json").write_text(json.dumps(result, ensure_ascii=False, indent=2))
+    except (ValueError, RuntimeError, TimeoutError, json.JSONDecodeError) as error:
+        print(f"[error] {error}", file=sys.stderr)
+        raise SystemExit(1) from error
+
+
+if __name__ == "__main__":
+    main()

skills/atlascloud-img-gen/scripts/tests/test_gen.py

@@ -0,0 +1,55 @@
+import json
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+SCRIPTS = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(SCRIPTS))
+import gen  # noqa: E402
+
+
+class AtlasCloudImageTests(unittest.TestCase):
+    def test_submit_uses_media_endpoint_and_bearer_token(self):
+        response = mock.MagicMock()
+        response.__enter__.return_value.read.return_value = json.dumps({"data": {"id": "p1"}}).encode()
+        with mock.patch("gen.urllib.request.urlopen", return_value=response) as urlopen:
+            result = gen.request_json(gen.GENERATE_URL, "secret", method="POST", payload={"model": "m"})
+        request = urlopen.call_args.args[0]
+        self.assertEqual(request.full_url, "https://api.atlascloud.ai/api/v1/model/generateImage")
+        self.assertEqual(request.headers["Authorization"], "Bearer secret")
+        self.assertEqual(request.headers["User-agent"], "xiaobei-atlascloud-img-gen/1.0")
+        self.assertEqual(gen.prediction_id(result), "p1")
+
+    def test_local_image_is_uploaded(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "source.png"
+            path.write_bytes(b"png")
+            with mock.patch("gen.upload_file", return_value="https://cdn.example/source.png") as upload:
+                self.assertEqual(gen.resolve_image(str(path), "key"), "https://cdn.example/source.png")
+            upload.assert_called_once_with(path, "key")
+
+    def test_remote_image_skips_upload(self):
+        with mock.patch("gen.upload_file") as upload:
+            value = gen.resolve_image("https://example.com/source.png", "key")
+        self.assertEqual(value, "https://example.com/source.png")
+        upload.assert_not_called()
+
+    def test_poll_returns_completed_prediction(self):
+        responses = [
+            {"data": {"status": "processing"}},
+            {"data": {"status": "completed", "outputs": ["https://cdn.example/out.png"]}},
+        ]
+        with mock.patch("gen.request_json", side_effect=responses), mock.patch("gen.time.sleep"):
+            result = gen.poll_prediction("p1", "key", 0, 10)
+        self.assertEqual(gen.output_urls(result), ["https://cdn.example/out.png"])
+
+    def test_failed_prediction_raises(self):
+        with mock.patch("gen.request_json", return_value={"data": {"status": "failed"}}):
+            with self.assertRaises(RuntimeError):
+                gen.poll_prediction("p1", "key", 0, 10)
+
+
+if __name__ == "__main__":
+    unittest.main()