Des recettes prêtes à l’emploi pour l’API PopNow

# pip install requests
import csv, hashlib, os, pathlib, time
import requests

MODEL = "google/nano-banana-2.1/text-to-image"
OPTIONS = {
    "aspect_ratio": "16:9",
    "resolution": "1k",
    "output_format": "png"
}


def main():
    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")]

    # Submit every row first, then collect: the images render in parallel.
    tasks = []
    for i, prompt in enumerate(rows, start=1):
        # Same row, same key: rerunning after a crash never pays twice.
        key = "batch-" + hashlib.sha256(f"{i}:{prompt}".encode()).hexdigest()[:32]
        payload = {"model": MODEL, "input": {"prompt": prompt, **OPTIONS}}
        tasks.append((i, create(payload, key)))

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


# ---- helpers ----

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)


if __name__ == "__main__":
    main()

# pip install requests
import mimetypes, os, sys, time, uuid
import requests

MODEL = "alibaba/wan-3.0/image-to-video"
OPTIONS = {
    "prompt": "A slow aerial shot drifting over a misty pine forest at sunrise",
    "resolution": "720p",
    "duration_seconds": 5
}


def main():
    ref = upload(sys.argv[1])  # python photo_to_video.py photo.jpg
    # Describe the motion in "prompt"; the photo sets the look.
    task_input = {**OPTIONS, "image_url": ref}
    task_id = create({"model": MODEL, "input": task_input}, str(uuid.uuid4()))
    save(wait(task_id), "clip.mp4")
    print("saved clip.mp4")


# ---- helpers ----

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)


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


if __name__ == "__main__":
    main()

# pip install requests
import hashlib, os, pathlib, time
import requests

MODEL = "google/gemini-3.8-flash-tts/text-to-speech"
# voice_id options include Achernar, Achird, Algenib, Algieba, Alnilam, Aoede. The model's API page lists them all.
OPTIONS = {
    "voice_id": "Kore"
}


def main():
    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, **OPTIONS}}, key)))

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


# ---- helpers ----

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)


if __name__ == "__main__":
    main()

// npm install express   (Node 18+: fetch and crypto are built in)
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": "alibaba/wan-3.0/text-to-video",
      "input": {
        "prompt": "A slow aerial shot drifting over a misty pine forest at sunrise",
        "aspect_ratio": "16:9",
        "resolution": "720p",
        "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);