Domo AI API Integration

# domo_client.py  —  pip install requests
import os, time, base64, requests

API = os.environ.get("DOMOAI_BASE_URL", "https://api.domoai.com/v1")
HEADERS = {"Authorization": f"Bearer {os.environ['DOMOAI_API_KEY']}"}
FINAL = {"SUCCESS", "FAILED", "CANCELED"}

def _post(path, payload, retries=3):
    for attempt in range(retries):
        r = requests.post(f"{API}{path}", json=payload, headers=HEADERS, timeout=30)
        if r.status_code >= 500 or r.status_code == 429:
            time.sleep(2 ** attempt)          # 1s, 2s, 4s backoff
            continue
        r.raise_for_status()                  # 401 / 402 / 403 / 422 -> raise
        return r.json()["data"]
    r.raise_for_status()

def create_text_video(prompt, seconds=5, **opts):
    payload = {"prompt": prompt, "seconds": seconds,
               "model": opts.pop("model", "t2v-2.4-faster"), **opts}
    return _post("/video/text2video", payload)["task_id"]

def create_image_video(image_path, seconds=5, prompt="", **opts):
    with open(image_path, "rb") as f:
        img = base64.b64encode(f.read()).decode()
    payload = {"model": opts.pop("model", "animate-2.4-faster"),
               "image": {"bytes_base64_encoded": img},
               "seconds": seconds, "prompt": prompt, **opts}
    return _post("/video/image2video", payload)["task_id"]

def get_task(task_id):
    r = requests.get(f"{API}/tasks/{task_id}", headers=HEADERS, timeout=30)
    r.raise_for_status()
    return r.json()["data"]

def wait_for(task_id, timeout=600):
    delay, waited = 3, 0
    while waited < timeout:
        task = get_task(task_id)
        if task["status"] in FINAL:
            return task
        time.sleep(delay); waited += delay
        delay = min(delay * 1.5, 15)          # gentle backoff, max 15s
    raise TimeoutError(f"task {task_id} still running after {timeout}s")

def download(task, folder="videos"):
    os.makedirs(folder, exist_ok=True)
    paths = []
    for i, v in enumerate(task.get("output_videos") or []):
        path = os.path.join(folder, f"{task['task_id']}_{i}.mp4")
        with requests.get(v["url"], stream=True, timeout=120) as r:
            r.raise_for_status()
            with open(path, "wb") as f:
                for chunk in r.iter_content(1 << 20):
                    f.write(chunk)
        paths.append(path)
    return paths

if __name__ == "__main__":
    tid = create_text_video("A cat DJ in a neon club, anime style",
                            seconds=5, aspect_ratio="9:16", style="japanese_anime")
    task = wait_for(tid)
    print(task["status"], "credits:", task.get("credits"))
    if task["status"] == "SUCCESS":
        print(download(task))

// domoClient.mjs  —  Node 18+ (built-in fetch)
import fs from "node:fs/promises";

const API = process.env.DOMOAI_BASE_URL ?? "https://api.domoai.com/v1";
const HEADERS = {
  Authorization: `Bearer ${process.env.DOMOAI_API_KEY}`,
  "Content-Type": "application/json",
};
const FINAL = new Set(["SUCCESS", "FAILED", "CANCELED"]);
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

async function post(path, payload, retries = 3) {
  for (let attempt = 0; attempt < retries; attempt++) {
    const res = await fetch(`${API}${path}`, {
      method: "POST", headers: HEADERS, body: JSON.stringify(payload),
    });
    if (res.status >= 500 || res.status === 429) {
      await sleep(2 ** attempt * 1000);           // 1s, 2s, 4s
      continue;
    }
    if (!res.ok) throw new Error(`${res.status}: ${await res.text()}`);
    return (await res.json()).data;
  }
  throw new Error("DomoAI unavailable after retries");
}

export async function createTextVideo(prompt, seconds = 5, opts = {}) {
  const { task_id } = await post("/video/text2video",
    { prompt, seconds, model: "t2v-2.4-faster", ...opts });
  return task_id;
}

export async function createImageVideo(imagePath, seconds = 5, opts = {}) {
  const b64 = (await fs.readFile(imagePath)).toString("base64");
  const { task_id } = await post("/video/image2video", {
    model: "animate-2.4-faster",
    image: { bytes_base64_encoded: b64 },
    seconds, ...opts,
  });
  return task_id;
}

export async function getTask(taskId) {
  const res = await fetch(`${API}/tasks/${taskId}`, { headers: HEADERS });
  if (!res.ok) throw new Error(`${res.status}: ${await res.text()}`);
  return (await res.json()).data;
}

export async function waitFor(taskId, timeoutMs = 600_000) {
  let delay = 3000; const start = Date.now();
  while (Date.now() - start < timeoutMs) {
    const task = await getTask(taskId);
    if (FINAL.has(task.status)) return task;
    await sleep(delay);
    delay = Math.min(delay * 1.5, 15_000);
  }
  throw new Error(`task ${taskId} timed out`);
}

export async function download(task, folder = "videos") {
  await fs.mkdir(folder, { recursive: true });
  const out = [];
  for (const [i, v] of (task.output_videos ?? []).entries()) {
    const buf = Buffer.from(await (await fetch(v.url)).arrayBuffer());
    const path = `${folder}/${task.task_id}_${i}.mp4`;
    await fs.writeFile(path, buf);
    out.push(path);
  }
  return out;
}

// Example
const id = await createTextVideo("A cat DJ in a neon club, anime style", 5,
  { aspect_ratio: "9:16", style: "japanese_anime" });
const task = await waitFor(id);
console.log(task.status, "credits:", task.credits);
if (task.status === "SUCCESS") console.log(await download(task));

# webhook_app.py  —  pip install flask requests
import os, hmac
from flask import Flask, request, abort
from domo_client import download

app = Flask(__name__)
SECRET = os.environ["WEBHOOK_SECRET"]
seen = set()                     # use your database in production

@app.post("/hooks/domoai")
def domoai_hook():
    # 1. Verify: we put a secret token in the callback_url we registered
    if not hmac.compare_digest(request.args.get("token", ""), SECRET):
        abort(401)

    body = request.get_json(force=True)
    task = body.get("data", body)            # accept wrapped or bare task
    task_id, status = task["task_id"], task["status"]

    # 2. Idempotency: the same update may arrive more than once
    if (task_id, status) in seen:
        return "", 200
    seen.add((task_id, status))

    # 3. Respond fast; do heavy work in a background job in production
    if status == "SUCCESS":
        download(task)                       # URLs expire after 8 hours
    elif status in ("FAILED", "CANCELED"):
        app.logger.warning("task %s ended: %s", task_id, status)
    return "", 200

# When creating a task, pass:
#   callback_url = f"https://your-app.com/hooks/domoai?token={SECRET}"

// webhookServer.mjs  —  npm i express
import express from "express";
import crypto from "node:crypto";
import { download } from "./domoClient.mjs";

const app = express();
app.use(express.json({ limit: "1mb" }));
const SECRET = process.env.WEBHOOK_SECRET;
const seen = new Set();                    // use your database in production

const safeEqual = (a, b) =>
  a.length === b.length && crypto.timingSafeEqual(Buffer.from(a), Buffer.from(b));

app.post("/hooks/domoai", async (req, res) => {
  // 1. Verify the secret token from the callback_url
  if (!safeEqual(String(req.query.token ?? ""), SECRET)) return res.sendStatus(401);

  const task = req.body.data ?? req.body;  // accept wrapped or bare task
  const key = `${task.task_id}:${task.status}`;

  // 2. Idempotency
  if (seen.has(key)) return res.sendStatus(200);
  seen.add(key);

  // 3. Acknowledge quickly, then process
  res.sendStatus(200);
  if (task.status === "SUCCESS") await download(task);   // URLs expire in 8 h
  else if (["FAILED", "CANCELED"].includes(task.status))
    console.warn(`task ${task.task_id} ended: ${task.status}`);
});

app.listen(3000, () => console.log("Webhook listening on :3000"));
// callback_url: `https://your-app.com/hooks/domoai?token=${SECRET}`