Build with Snowpixel.

"""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)

/** Node.js 22+. Set SNOWPIXEL_API_KEY before running this script. */
import { parseArgs } from 'node:util';
import { randomUUID } from 'node:crypto';
import { setTimeout as sleep } from 'node:timers/promises';

const DEFAULT_BASE_URL = 'https://api.snowpixel.app';

async function requestJson(base, key, path, method = 'GET', body, timeout = 30, idempotencyKey) {
  let response;
  let data;
  try {
    // Preserve any base URL path prefix when appending the relative /v1 polling path.
    response = await fetch(base + path, {
      method,
      headers: { Authorization: `Bearer ${key}`, 'Content-Type': 'application/json',
        'User-Agent': 'Snowpixel-Example/1.0 (Node.js)',
        ...(idempotencyKey === undefined ? {} : { 'Idempotency-Key': idempotencyKey }) },
      body: body === undefined ? undefined : JSON.stringify(body),
      signal: AbortSignal.timeout(Math.max(1, Math.ceil(timeout * 1000))),
      redirect: 'error',
    });
    data = await response.json();
  } catch (error) {
    if (error instanceof SyntaxError) throw new Error(`HTTP ${response.status}: expected a JSON response.`);
    throw new Error('Request failed or timed out. No automatic retry was made.');
  }
  if (!data || typeof data !== 'object' || Array.isArray(data)) throw new Error('Expected a JSON object.');
  if (response.status !== (method === 'POST' ? 202 : 200)) {
    throw new Error(`HTTP ${response.status}: ${data.error?.code ?? 'unknown_error'}`);
  }
  return data;
}

async function run() {
  const { values } = parseArgs({ options: {
    model: { type: 'string', default: 'snowpixel-image-v1' },
    prompt: { type: 'string', default: 'A snowy pixel cabin' },
    'job-id': { type: 'string' },
    'idempotency-key': { type: 'string' },
    timeout: { type: 'string', default: '600' },
    'poll-interval': { type: 'string', default: '5' },
    'request-timeout': { type: 'string', default: '30' },
    help: { type: 'boolean', default: false },
  } });
  if (values.help) {
    console.log('node generate.mjs [--model MODEL] [--prompt PROMPT] [--job-id ID] [--idempotency-key KEY]\n' +
      '[--timeout 600] [--poll-interval 5] [--request-timeout 30] (seconds)\n' +
      'Set SNOWPIXEL_API_KEY; optionally set SNOWPIXEL_BASE_URL.');
    return;
  }
  const key = process.env.SNOWPIXEL_API_KEY?.trim();
  if (!key) throw new Error('Set SNOWPIXEL_API_KEY before running this script.');
  const base = (process.env.SNOWPIXEL_BASE_URL ?? DEFAULT_BASE_URL).replace(/\/+$/, '');
  const url = new URL(base);
  if (!['http:', 'https:'].includes(url.protocol) || url.username || url.password || url.search || url.hash ||
      (url.protocol === 'http:' && !['localhost', '127.0.0.1', '[::1]'].includes(url.hostname))) {
    throw new Error('Use HTTPS, or HTTP on localhost for offline tests.');
  }
  const timeout = Number(values.timeout);
  const interval = Number(values['poll-interval']);
  const requestTimeout = Number(values['request-timeout']);
  if (![timeout, interval, requestTimeout].every(value => Number.isFinite(value) && value > 0)) {
    throw new Error('Timeouts and polling interval must be positive finite numbers.');
  }
  let jobId = values['job-id'];
  if (jobId !== undefined && !/^[A-Za-z0-9]{20}$/.test(jobId)) throw new Error('Job ID must contain exactly 20 ASCII letters or digits.');
  if (jobId === undefined) {
    const submissionKey = values['idempotency-key'] ?? randomUUID();
    if (!submissionKey.length || submissionKey.length > 255 || /[^\x21-\x7e]/.test(submissionKey)) throw new Error('Idempotency key must contain 1 to 255 visible ASCII characters without whitespace.');
    console.error(`Idempotency-Key: ${submissionKey}; retry with --idempotency-key ${submissionKey} and the same request.`);
    const accepted = await requestJson(base, key, '/v1/generations', 'POST', {
      model: values.model, prompt: values.prompt,
    }, requestTimeout, submissionKey);
    jobId = accepted.id;
    if (typeof jobId !== 'string' || !/^[A-Za-z0-9]{20}$/.test(jobId)) throw new Error('The API returned an invalid job ID.');
    if (accepted.pollUrl !== `/v1/generations/${jobId}`) throw new Error('The API returned an unexpected polling path.');
  }
  console.log(`Job: ${jobId}`);
  const deadline = performance.now() + timeout * 1000;
  let recoveryAnnounced = false;
  while (true) {
    const remaining = (deadline - performance.now()) / 1000;
    if (remaining <= 0) throw new Error(`Polling timed out. Resume with --job-id ${jobId}; do not resubmit.`);
    const job = await requestJson(base, key, `/v1/generations/${jobId}`, 'GET', undefined, Math.min(requestTimeout, remaining));
    if (job.status === 'cancelled') throw new Error(`Job ${jobId} was cancelled; inspect the job for retained outputs and billing.`);
    if (job.status === 'failed') throw new Error(`Job ${jobId} failed: ${job.error?.code ?? 'unknown_error'}`);
    if (job.status === 'completed' || job.status === 'partial_failed') {
      if (!Array.isArray(job.outputs)) throw new Error('The API returned invalid outputs.');
      const urls = job.outputs.map(item => item?.url || item?.originalUrl).filter(url => typeof url === 'string' && url);
      for (const outputUrl of urls) console.log(outputUrl);
      if (job.status === 'partial_failed') throw new Error(`Job ${jobId} partially failed: partial_generation_failed`);
      if (!urls.length) console.log('Completed with no output URLs.');
      return;
    }
    if (job.status !== 'pending') throw new Error('The API returned an unknown job status.');
    if (job.recoveryStatus === 'outcome_unknown' && !recoveryAnnounced) {
      console.error('Checking generation status; continuing to poll the same job.');
      recoveryAnnounced = true;
    }
    await sleep(Math.min(interval * 1000, Math.max(0, deadline - performance.now())));
  }
}

run().catch(error => {
  console.error(String(error.message).replaceAll(process.env.SNOWPIXEL_API_KEY?.trim() || '\0', '[redacted]'));
  process.exitCode = 1;
});

{
  "id": "AbCdEfGhIjKlMnOpQr01",
  "model": "snowpixel-image-v1",
  "prompt": "A snowy pixel cabin",
  "status": "pending",
  "createdAt": "2026-06-25T12:00:00+00:00",
  "completedAt": null,
  "outputs": []
}

{
  "id": "AbCdEfGhIjKlMnOpQr01",
  "model": "snowpixel-image-v1",
  "prompt": "A snowy pixel cabin",
  "status": "completed",
  "createdAt": "2026-06-25T12:00:00+00:00",
  "completedAt": "2026-06-25T12:03:00+00:00",
  "outputs": [
    {
      "url": "https://cdn.example/generated.png",
      "originalUrl": "https://cdn.example/original.png",
      "metadata": {
        "width": 1024,
        "height": 1024
      }
    }
  ]
}

{
  "id": "AbCdEfGhIjKlMnOpQr01",
  "model": "snowpixel-audio-v1",
  "prompt": "A snowy pixel cabin",
  "status": "partial_failed",
  "createdAt": "2026-06-25T12:00:00+00:00",
  "completedAt": "2026-06-25T12:03:00+00:00",
  "outputs": [
    {
      "url": "https://cdn.example/generated.wav",
      "originalUrl": null,
      "metadata": {}
    }
  ],
  "error": {
    "code": "partial_generation_failed"
  }
}

{
  "id": "AbCdEfGhIjKlMnOpQr01",
  "model": "snowpixel-image-v1",
  "prompt": "A snowy pixel cabin",
  "status": "failed",
  "createdAt": "2026-06-25T12:00:00+00:00",
  "completedAt": "2026-06-25T12:00:01+00:00",
  "outputs": [],
  "error": {
    "code": "dispatch_failed"
  }
}