Automate the whole flow

import { randomUUID } from 'node:crypto';
import { mkdir, open, readdir, stat, writeFile } from 'node:fs/promises';
import { dirname, join } from 'node:path';

const API = process.env.SW_VOD_API ?? 'https://api-vod-express.syncwords.com/v1';
const PROJECT_NAME = 'Dub a folder — Spanish';
const [inputDir, outputDir] = process.argv.slice(2);

const CONFIG = {
  sourceMedia: { '@type': 'S3FolderSource' },
  transcription: { '@type': 'Speechmatics', language: 'en' },
  translations: [
    { language: 'es', engine: { '@type': 'Gemini' }, ttsOptions: { provider: { '@type': 'Deepdub', voice: 'auto' } } },
  ],
  outputs: [
    {
      '@type': 'S3FolderOutput',
      items: [
        {
          '@type': 'VideoFile',
          ext: 'mp4',
          nameTemplate: '${name}.es.${ext}',
          useOriginalTracks: true,
          dubTracks: [{ language: 'es', title: 'Spanish' }],
        },
      ],
    },
  ],
};

// --- Authentication: a token from the key pair, and a new one on 401 ---------------------------
let token = process.env.SW_TOKEN;

async function newToken() {
  const response = await fetch(`${API}/token/create`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json', Accept: 'application/json' },
    body: JSON.stringify({ username: process.env.SW_PUBLIC_KEY, password: process.env.SW_SECRET_KEY }),
  });
  if (!response.ok) throw new Error(`token: ${response.status} ${await response.text()}`);
  token = (await response.json()).access_token;
}

async function api(method, path, body, headers = {}, retried = false) {
  if (!token) await newToken();
  const response = await fetch(`${API}${path}`, {
    method,
    headers: {
      Authorization: `Bearer ${token}`,
      Accept: 'application/json',
      ...(body ? { 'Content-Type': 'application/json' } : {}),
      ...headers,
    },
    body: body ? JSON.stringify(body) : undefined,
  });
  if (response.status === 401 && !retried) {
    await newToken();
    return api(method, path, body, headers, true);
  }
  if (response.status === 204) return null;
  const json = await response.json();
  if (!response.ok) {
    const message = `${method} ${path}: ${response.status} ${JSON.stringify(json)}`;
    throw Object.assign(new Error(message), { status: response.status });
  }
  return json.data;
}

// A create repeated with the same Idempotency-Key answers the first one's project, so a dropped
// connection, a 5xx or a 409 (the first call still running) can be retried without making two.
async function createProject(body) {
  const key = randomUUID();
  for (let attempt = 1; ; attempt++) {
    try {
      return await api('POST', '/projects', body, { 'Idempotency-Key': key });
    } catch (error) {
      const retryable = error.status === undefined || error.status === 409 || error.status >= 500;
      if (!retryable || attempt === 3) throw error;
      await new Promise((resolve) => setTimeout(resolve, 2000 * attempt));
    }
  }
}

// --- Upload one file through the API ------------------------------------------------------------
async function upload(projectId, file, name) {
  const { size } = await stat(file);
  const started = await api('POST', `/projects/${projectId}/uploads`, { name, size });
  const base = `/projects/${projectId}/uploads/${encodeURIComponent(started.uploadId)}`;
  const key = encodeURIComponent(started.key);
  const handle = await open(file);
  const parts = [];
  try {
    for (let n = 1; n <= started.partCount; n++) {
      const buffer = Buffer.alloc(Math.min(started.partSize, size - (n - 1) * started.partSize));
      await handle.read(buffer, 0, buffer.length, (n - 1) * started.partSize);
      const { url } = await api('GET', `${base}/parts/${n}?key=${key}`);
      const put = await fetch(url, { method: 'PUT', body: buffer });
      if (!put.ok) throw new Error(`${name} part ${n}: S3 answered ${put.status}`);
      parts.push({ partNumber: n, etag: put.headers.get('etag') });
    }
  } catch (error) {
    await api('DELETE', `${base}?key=${key}`);
    throw error;
  } finally {
    await handle.close();
  }
  await api('POST', `${base}/complete`, { key: started.key, parts });
}

// --- The flow -----------------------------------------------------------------------------------
// One project per configuration: found by its exact name, so a later run reuses it, and created
// only the first time.
const [found] = await api('GET', `/projects?name=${encodeURIComponent(PROJECT_NAME)}`);
const project =
  found ?? (await createProject({ name: PROJECT_NAME, config: CONFIG, options: { autoStart: false } }));
console.log(`project ${project.id}`);

// Files a previous run uploaded are in the project already.
const uploaded = new Set((await api('GET', `/projects/${project.id}/files`)).map((f) => f.name));
const videos = (await readdir(inputDir)).filter((f) => /\.(mp4|mov|mkv)$/i.test(f) && !uploaded.has(f));
for (const video of videos) {
  await upload(project.id, join(inputDir, video), video);
  console.log(`uploaded ${video}`);
}

await api('POST', `/projects/${project.id}/start`);

let files;
do {
  await new Promise((resolve) => setTimeout(resolve, 30_000));
  files = await api('GET', `/projects/${project.id}/files`);
  console.log(files.map((f) => `${f.name}: ${f.status}`).join(', '));
} while (files.some((f) => f.status === 'pending' || f.status === 'processing'));

for (const file of files) {
  if (file.status === 'failed') {
    console.error(`${file.name} failed: ${file.failureReason}`);
    continue;
  }
  for (const output of file.outputs.filter((o) => o.available)) {
    const { url } = await api('GET', `/projects/${project.id}/outputs/download?path=${encodeURIComponent(output.path)}`);
    const target = join(outputDir, output.path);
    await mkdir(dirname(target), { recursive: true });
    await writeFile(target, Buffer.from(await (await fetch(url)).arrayBuffer()));
    console.log(`downloaded ${target}`);
  }
}

import os
import re
import sys
import time
import uuid
from urllib.parse import quote

import requests

API = os.environ.get("SW_VOD_API", "https://api-vod-express.syncwords.com/v1")
PROJECT_NAME = "Dub a folder — Spanish"
input_dir, output_dir = sys.argv[1], sys.argv[2]

CONFIG = {
    "sourceMedia": {"@type": "S3FolderSource"},
    "transcription": {"@type": "Speechmatics", "language": "en"},
    "translations": [
        {"language": "es", "engine": {"@type": "Gemini"}, "ttsOptions": {"provider": {"@type": "Deepdub", "voice": "auto"}}}
    ],
    "outputs": [
        {
            "@type": "S3FolderOutput",
            "items": [
                {
                    "@type": "VideoFile",
                    "ext": "mp4",
                    "nameTemplate": "${name}.es.${ext}",
                    "useOriginalTracks": True,
                    "dubTracks": [{"language": "es", "title": "Spanish"}],
                }
            ],
        }
    ],
}

# --- Authentication: a token from the key pair, and a new one on 401 ----------------------------
token = os.environ.get("SW_TOKEN")


def new_token():
    global token
    response = requests.post(
        API + "/token/create",
        json={"username": os.environ["SW_PUBLIC_KEY"], "password": os.environ["SW_SECRET_KEY"]},
        headers={"Accept": "application/json"},
    )
    response.raise_for_status()
    token = response.json()["access_token"]


class ApiError(RuntimeError):
    def __init__(self, status, message):
        super().__init__(message)
        self.status = status


def api(method, path, body=None, headers=None, retried=False):
    if not token:
        new_token()
    response = requests.request(
        method, API + path, json=body,
        headers={"Authorization": f"Bearer {token}", "Accept": "application/json", **(headers or {})},
    )
    if response.status_code == 401 and not retried:
        new_token()
        return api(method, path, body, headers, retried=True)
    if not response.ok:
        raise ApiError(response.status_code, f"{method} {path}: {response.status_code} {response.text}")
    return response.json().get("data") if response.content else None


# A create repeated with the same Idempotency-Key answers the first one's project, so a dropped
# connection, a 5xx or a 409 (the first call still running) can be retried without making two.
def create_project(body):
    key = str(uuid.uuid4())
    for attempt in range(1, 4):
        try:
            return api("POST", "/projects", body, headers={"Idempotency-Key": key})
        except (requests.ConnectionError, requests.Timeout, ApiError) as error:
            status = getattr(error, "status", None)
            retryable = status is None or status == 409 or status >= 500
            if not retryable or attempt == 3:
                raise
            time.sleep(2 * attempt)


# --- Upload one file through the API -------------------------------------------------------------
def upload(project_id, file, name):
    started = api("POST", f"/projects/{project_id}/uploads", {"name": name, "size": os.path.getsize(file)})
    base = f"/projects/{project_id}/uploads/{quote(started['uploadId'], safe='')}"
    key = quote(started["key"], safe="")
    parts = []
    try:
        with open(file, "rb") as handle:
            for n in range(1, started["partCount"] + 1):
                url = api("GET", f"{base}/parts/{n}?key={key}")["url"]
                put = requests.put(url, data=handle.read(started["partSize"]))
                put.raise_for_status()
                parts.append({"partNumber": n, "etag": put.headers["ETag"]})
    except Exception:
        api("DELETE", f"{base}?key={key}")
        raise
    api("POST", f"{base}/complete", {"key": started["key"], "parts": parts})


# --- The flow ------------------------------------------------------------------------------------
# One project per configuration: found by its exact name, so a later run reuses it, and created
# only the first time.
found = api("GET", f"/projects?name={quote(PROJECT_NAME, safe='')}")
project = found[0] if found else create_project({
    "name": PROJECT_NAME,
    "config": CONFIG,
    "options": {"autoStart": False},
})
print("project", project["id"])

# Files a previous run uploaded are in the project already.
uploaded = {f["name"] for f in api("GET", f"/projects/{project['id']}/files")}
for video in sorted(os.listdir(input_dir)):
    if re.search(r"\.(mp4|mov|mkv)$", video, re.I) and video not in uploaded:
        upload(project["id"], os.path.join(input_dir, video), video)
        print("uploaded", video)

api("POST", f"/projects/{project['id']}/start")

while True:
    time.sleep(30)
    files = api("GET", f"/projects/{project['id']}/files")
    print(", ".join(f"{f['name']}: {f['status']}" for f in files))
    if all(f["status"] in ("done", "failed") for f in files):
        break

for file in files:
    if file["status"] == "failed":
        print(f"{file['name']} failed: {file['failureReason']}", file=sys.stderr)
        continue
    for output in (o for o in file["outputs"] if o["available"]):
        url = api("GET", f"/projects/{project['id']}/outputs/download?path={quote(output['path'], safe='')}")["url"]
        target = os.path.join(output_dir, output["path"])
        os.makedirs(os.path.dirname(target), exist_ok=True)
        with requests.get(url, stream=True) as response, open(target, "wb") as out:
            for chunk in response.iter_content(1 << 20):
                out.write(chunk)
        print("downloaded", target)