"""Generate media with Snowpixel. Python 3.12+, no dependencies. Set SNOWPIXEL_API_KEY; optionally set SNOWPIXEL_BASE_URL. """ import argparse import json import math import os import re import sys import time import urllib.error import urllib.parse import urllib.request import uuid DEFAULT_BASE_URL = "https://api.snowpixel.app" class ApiError(Exception): pass class NoRedirects(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None # Never forward the key to a redirected destination. def request_json(base, key, path, method="GET", body=None, timeout=30, idempotency_key=None): # Preserve any base URL path prefix when appending the relative /v1 polling path. request = urllib.request.Request( base + path, method=method, data=json.dumps(body).encode() if body is not None else None, headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json", "User-Agent": "Snowpixel-Example/1.0 (Python)", **({'Idempotency-Key': idempotency_key} if idempotency_key is not None else {})}, ) try: response = urllib.request.build_opener(NoRedirects).open(request, timeout=timeout) except urllib.error.HTTPError as error: response = error except (OSError, ValueError) as error: raise ApiError("Request failed or timed out. No automatic retry was made.") from error with response: status = response.code try: data = json.load(response) except (ValueError, UnicodeError) as error: raise ApiError(f"HTTP {status}: expected a JSON response.") from error if not isinstance(data, dict): raise ApiError("Expected a JSON object.") if status != (202 if method == "POST" else 200): detail = data.get("error", {}) code = detail.get("code", "unknown_error") if isinstance(detail, dict) else "unknown_error" raise ApiError(f"HTTP {status}: {code}") return data def positive_number(value): number = float(value) if not math.isfinite(number) or number <= 0: raise argparse.ArgumentTypeError("must be a positive finite number") return number def run(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--model", default="snowpixel-image-v1") parser.add_argument("--prompt", default="A snowy pixel cabin") parser.add_argument("--job-id", help="Poll an existing job without submitting again") parser.add_argument("--idempotency-key", help="Reuse this key and the same request to recover an interrupted submission") parser.add_argument("--timeout", type=positive_number, default=600, help="Polling limit in seconds") parser.add_argument("--poll-interval", type=positive_number, default=5) parser.add_argument("--request-timeout", type=positive_number, default=30) args = parser.parse_args() key = os.environ.get("SNOWPIXEL_API_KEY", "").strip() if not key: raise ApiError("Set SNOWPIXEL_API_KEY before running this script.") base = os.environ.get("SNOWPIXEL_BASE_URL", DEFAULT_BASE_URL).rstrip("/") url = urllib.parse.urlsplit(base) if (url.scheme not in ("http", "https") or not url.hostname or url.username or url.password or url.query or url.fragment or (url.scheme == "http" and url.hostname not in ("localhost", "127.0.0.1", "::1"))): raise ApiError("Use HTTPS, or HTTP on localhost for offline tests.") job_id = args.job_id if job_id is not None and not re.fullmatch(r"[A-Za-z0-9]{20}", job_id): raise ApiError("Job ID must contain exactly 20 ASCII letters or digits.") if job_id is None: submission_key = args.idempotency_key if args.idempotency_key is not None else str(uuid.uuid4()) if not re.fullmatch(r'[\x21-\x7e]{1,255}', submission_key): raise ApiError('Idempotency key must contain 1 to 255 visible ASCII characters without whitespace.') print(f'Idempotency-Key: {submission_key}; retry with --idempotency-key {submission_key} and the same request.', file=sys.stderr, flush=True) accepted = request_json(base, key, "/v1/generations", "POST", { "model": args.model, "prompt": args.prompt, }, timeout=args.request_timeout, idempotency_key=submission_key) job_id = accepted.get("id") if not isinstance(job_id, str) or not re.fullmatch(r"[A-Za-z0-9]{20}", job_id): raise ApiError("The API returned an invalid job ID.") if accepted.get("pollUrl") != f"/v1/generations/{job_id}": raise ApiError("The API returned an unexpected polling path.") print(f"Job: {job_id}", flush=True) deadline = time.monotonic() + args.timeout recovery_announced = False while True: remaining = deadline - time.monotonic() if remaining <= 0: raise ApiError(f"Polling timed out. Resume with --job-id {job_id}; do not resubmit.") job = request_json(base, key, f"/v1/generations/{job_id}", timeout=min(args.request_timeout, remaining)) if job.get("status") == "cancelled": raise ApiError(f"Job {job_id} was cancelled; inspect the job for retained outputs and billing.") if job.get("status") == "failed": raise ApiError(f"Job {job_id} failed: {(job.get('error') or {}).get('code', 'unknown_error')}") if job.get("status") in ("completed", "partial_failed"): outputs = job.get("outputs") if not isinstance(outputs, list): raise ApiError("The API returned invalid outputs.") urls = [item.get("url") or item.get("originalUrl") for item in outputs if isinstance(item, dict)] urls = [url for url in urls if isinstance(url, str) and url] for output_url in urls: print(output_url) if job.get("status") == "partial_failed": raise ApiError(f"Job {job_id} partially failed: partial_generation_failed") if not urls: print("Completed with no output URLs.") return if job.get("status") != "pending": raise ApiError("The API returned an unknown job status.") if job.get('recoveryStatus') == 'outcome_unknown' and not recovery_announced: print('Checking generation status; continuing to poll the same job.', file=sys.stderr) recovery_announced = True time.sleep(min(args.poll_interval, max(0, deadline - time.monotonic()))) if __name__ == "__main__": try: run() except (ApiError, OSError, ValueError) as error: message = str(error).replace(os.environ.get("SNOWPIXEL_API_KEY", "").strip() or "\0", "[redacted]") print(message, file=sys.stderr) sys.exit(1)