레시피

PopNow API 바로 쓰는 레시피

오늘 바로 복사해 쓸 수 있는 작은 자동화입니다. 모두 적힌 그대로 동작합니다. 모델, 프롬프트, 파일 저장 위치만 바꾸세요.

스프레드시트로 이미지 일괄 생성

CSV 파일에서 프롬프트를 읽어 행마다 이미지를 한 장씩 만들고 옆에 저장합니다. 상품 사진, 썸네일, 광고 변형에 좋습니다.

동작 방식

prompts.csv에 prompt라는 머리글 아래 한 줄에 프롬프트 하나씩 적습니다.
POPNOW_API_KEY를 설정하고 스크립트를 실행합니다.
이미지는 out 폴더에 행 번호 이름으로 저장됩니다.

batch_images.py

# 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()

사진을 짧은 영상으로

로컬 사진을 업로드하고 이미지로 영상을 만드는 모델로 움직이게 한 뒤 영상을 내려받습니다. 이미지 필드가 있는 모델이면 어떤 모델이든 같은 순서입니다.

동작 방식

스크립트에 컴퓨터에 있는 JPG나 PNG를 지정합니다.
사진을 업로드하고, 작업을 만들고, 기다립니다.
MP4가 clip.mp4로 저장됩니다.

사용 모델Wan 3.0 API

photo_to_video.py

# 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()

대사 목록을 음성으로

대본을 한 줄씩 별도의 오디오 파일로 만듭니다. 목소리는 모두 같습니다. 튜토리얼, 게임, 숏폼 영상에 편리합니다.

동작 방식

lines.txt에 한 줄에 한 대사씩 적습니다.
모델의 목소리 목록에서 목소리를 고릅니다.
한 줄마다 MP3가 하나씩 만들어집니다.

사용 모델Gemini 3.8 Flash TTS API

voice_lines.py

# 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()

영상이 완성되면 알림 받기

폴링은 필요 없습니다. 작업마다 callBackUrl을 주고, 서명을 확인한 뒤, 결과를 Slack 등으로 보내세요.

동작 방식

인터넷에서 접속할 수 있는 곳에 작은 서버를 실행합니다.
작업을 만들 때 그 URL을 callBackUrl로 넘깁니다.
영상이 완성되면 링크를 Slack 웹훅으로 보냅니다.

사용 모델Wan 3.0 API

notify.mjs

// 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);