diff --git a/skills/README.md b/skills/README.md index ff483c83..b699aed1 100644 --- a/skills/README.md +++ b/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 继承 | diff --git a/skills/atlascloud-img-gen/SKILL.md b/skills/atlascloud-img-gen/SKILL.md new file mode 100644 index 00000000..5a84a623 --- /dev/null +++ b/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-` unless `--out-dir` is set. The directory contains downloaded images and `result.json` with the prediction metadata. diff --git a/skills/atlascloud-img-gen/atlascloud-img-gen.sh b/skills/atlascloud-img-gen/atlascloud-img-gen.sh new file mode 100755 index 00000000..5a08a829 --- /dev/null +++ b/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" "$@" diff --git a/skills/atlascloud-img-gen/scripts/gen.py b/skills/atlascloud-img-gen/scripts/gen.py new file mode 100755 index 00000000..6c383e00 --- /dev/null +++ b/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() diff --git a/skills/atlascloud-img-gen/scripts/tests/test_gen.py b/skills/atlascloud-img-gen/scripts/tests/test_gen.py new file mode 100644 index 00000000..782b082c --- /dev/null +++ b/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()