Recipes

import os, time
import requests

BASE = "https://api.popnow.ai/api/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['POPNOW_API_KEY']}"}


def call(method, path, **kwargs):
    headers = {**HEADERS, **kwargs.pop("headers", {})}
    body = requests.request(method, BASE + path, headers=headers, timeout=60, **kwargs).json()
    if body["code"] != 200:
        raise RuntimeError(f'{body["code"]} {body["msg"]}')
    return body["data"]


def create(payload, key):
    return call("POST", "/jobs/createTask", json=payload, headers={"Idempotency-Key": key})["taskId"]


def wait(task_id):
    delay = 2.0
    while True:
        time.sleep(delay)
        task = call("GET", "/jobs/recordInfo", params={"taskId": task_id})
        assets = (task.get("output") or {}).get("assets") or []
        copying = any(a.get("pending") and not a.get("unavailable") for a in assets)
        if task["state"] in ("succeeded", "failed", "canceled", "expired") and not copying:
            return task
        delay = min(delay * 1.5, 15.0)


def save(task, path):
    if task["state"] != "succeeded":
        raise RuntimeError(f'{task["state"]}: {task.get("errorMessage", "no detail")}')
    url = task["output"]["assets"][0]["url"]
    with open(path, "wb") as f:
        f.write(requests.get(url, timeout=300).content)

import csv, hashlib, pathlib
from helpers import create, wait, save

MODEL = "alibaba/z-image-turbo/text-to-image"

out = pathlib.Path("out")
out.mkdir(exist_ok=True)
with open("prompts.csv", newline="", encoding="utf-8") as f:
    rows = [r["prompt"] for r in csv.DictReader(f) if r.get("prompt")]

tasks = []
for i, prompt in enumerate(rows, start=1):
    key = "batch-" + hashlib.sha256(f"{i}:{prompt}".encode()).hexdigest()[:32]
    tasks.append((i, create({"model": MODEL, "input": {"prompt": prompt, "aspect_ratio": "1:1"}}, key)))

for i, task_id in tasks:
    save(wait(task_id), out / f"{i:03d}.png")
    print("saved", i)

import { createHash } from 'node:crypto';
import { readFile, writeFile, mkdir } from 'node:fs/promises';

const API = 'https://api.popnow.ai/api/v1';
const auth = { Authorization: `Bearer ${process.env.POPNOW_API_KEY}` };
const call = async (path, init = {}) => {
  const body = await fetch(API + path, { ...init, headers: { ...auth, ...init.headers } }).then((r) => r.json());
  if (body.code !== 200) throw new Error(`${body.code} ${body.msg}`);
  return body.data;
};

const prompts = (await readFile('prompts.txt', 'utf8')).split('\n').map((l) => l.trim()).filter(Boolean);
await mkdir('out', { recursive: true });

const ids = await Promise.all(prompts.map((prompt, i) => call('/jobs/createTask', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json', 'Idempotency-Key': 'batch-' + createHash('sha256').update(`${i}:${prompt}`).digest('hex').slice(0, 32) },
  body: JSON.stringify({ model: 'alibaba/z-image-turbo/text-to-image', input: { prompt, aspect_ratio: '1:1' } }),
}).then((d) => d.taskId)));

for (const [i, id] of ids.entries()) {
  let task;
  do {
    await new Promise((r) => setTimeout(r, 3000));
    task = await call(`/jobs/recordInfo?taskId=${id}`);
  } while (task.state === 'queued' || task.state === 'running');
  if (task.state !== 'succeeded') { console.error(i, task.state, task.errorMessage); continue; }
  const file = await fetch(task.output.assets[0].url).then((r) => r.arrayBuffer());
  await writeFile(`out/${String(i + 1).padStart(3, '0')}.png`, Buffer.from(file));
}

import mimetypes, sys, uuid
import requests
from helpers import call, create, wait, save

MODEL = "minimax/h3/image-to-video"


def upload(path):
    """Local file -> popnow:// reference you can put in any image, video or audio field."""
    data = open(path, "rb").read()
    kind = mimetypes.guess_type(path)[0] or "image/jpeg"
    ticket = call("POST", "/common/upload-url", json={"contentType": kind, "bytes": len(data)})
    requests.put(ticket["uploadUrl"], data=data, headers=ticket.get("headers", {}), timeout=120).raise_for_status()
    return call("POST", f'/files/{ticket["fileId"]}/commit')["uri"]


ref = upload(sys.argv[1])
task_input = {"image_url": ref, "prompt": "Gentle camera push-in, hair moving in the wind", "duration_seconds": 5, "resolution": "768p"}
task_id = create({"model": MODEL, "input": task_input}, str(uuid.uuid4()))
save(wait(task_id), "clip.mp4")
print("saved clip.mp4")

curl -X POST "https://api.popnow.ai/api/v1/jobs/createTask" \
  -H "Authorization: Bearer $POPNOW_API_KEY" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: $(uuidgen)" \
  -d '{"model": "minimax/h3/image-to-video", "input": {"image_url": "https://media.popnow.ai/movie/lighthouse.webp", "prompt": "The waves roll in slowly", "duration_seconds": 5}}'

import hashlib, pathlib
from helpers import create, wait, save

MODEL = "google/gemini-3.8-flash-tts/text-to-speech"

out = pathlib.Path("voice")
out.mkdir(exist_ok=True)
lines = [l.strip() for l in open("lines.txt", encoding="utf-8") if l.strip()]

tasks = []
for i, line in enumerate(lines, start=1):
    key = "voice-" + hashlib.sha256(f"{i}:{line}".encode()).hexdigest()[:32]
    tasks.append((i, create({"model": MODEL, "input": {"text": line, "voice_id": "Kore", "style": "empathetic"}}, key)))

for i, task_id in tasks:
    task = wait(task_id)
    mime = ((task.get("output") or {}).get("assets") or [{}])[0].get("mime", "")
    ext = {"audio/mpeg": ".mp3", "audio/wav": ".wav", "audio/ogg": ".ogg"}.get(mime, ".audio")
    save(task, out / f"{i:03d}{ext}")
    print("saved", i)

import { createHash, createHmac, randomUUID, timingSafeEqual } from 'node:crypto';
import express from 'express';

const BASE = 'https://api.popnow.ai/api/v1';

// 1. Create the task with a callBackUrl that points at this server.
export async function createVideo() {
  const res = await fetch(BASE + '/jobs/createTask', {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${process.env.POPNOW_API_KEY}`,
      'Content-Type': 'application/json',
      'Idempotency-Key': randomUUID(),
    },
    body: JSON.stringify({
      model: 'minimax/h3/image-to-video',
      input: { image_url: 'https://media.popnow.ai/movie/lighthouse.webp', prompt: 'The waves roll in slowly', duration_seconds: 5 },
      callBackUrl: 'https://your-server.example.com/hooks/popnow',
    }),
  });
  return (await res.json()).data.taskId;
}

// 2. Receive the finished task, check the signature on the raw body, then post to Slack.
const app = express();
app.post('/hooks/popnow', express.raw({ type: 'application/json' }), async (req, res) => {
  const timestamp = req.header('X-Webhook-Timestamp');
  const signature = req.header('X-Webhook-Signature') ?? '';
  const event = JSON.parse(req.body.toString('utf8'));
  const task = event.data;
  const digest = createHash('sha256').update(req.body).digest('hex');
  const expected = createHmac('sha256', process.env.POPNOW_WEBHOOK_SECRET)
    .update(`${task.taskId}.${timestamp}.${digest}`)
    .digest('base64');
  const ok = expected.length === signature.length && timingSafeEqual(Buffer.from(expected), Buffer.from(signature));
  if (!ok || Math.abs(Date.now() / 1000 - Number(timestamp)) > 300) return res.status(401).end();
  res.status(200).end(); // answer first; deliveries are retried until you return 2xx

  const text = task.state === 'succeeded'
    ? `Video ready: ${task.output.assets[0].url}`
    : `Task ${task.taskId} ${task.state}: ${task.errorMessage ?? 'no detail'}`;
  await fetch(process.env.SLACK_WEBHOOK_URL, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ text }) });
});
app.listen(3000);