Quickstart: dub a video from upload to download

#!/usr/bin/env bash
# Upload a video, dub it into Spanish and download the result: bash dub_video.sh clip.mp4
#
# Progress is saved in clip.mp4.dub.env. If the script stops, run it again to continue; delete the file to dub anew.
set -euo pipefail

FILE=${1:?usage: bash dub_video.sh clip.mp4}
API=${RASK_BASE_URL:-https://api.rask.ai/v3}
STATE="$FILE.dub.env"
HEADERS=$(mktemp)
trap 'rm -f "$HEADERS"' EXIT
touch "$STATE"
. "$STATE"

# api METHOD PATH [CURL OPTION...]: one call to the API and its answer. An error answer goes to stderr, and the call fails.
api() {
  local answer
  if answer=$(curl -sS --fail-with-body -X "$1" "$API$2" -H "Authorization: Bearer $RASK_API_KEY" "${@:3}"); then
    printf '%s\n' "$answer"
  else
    printf '%s\n' "$answer" >&2
    return 1
  fi
}

# wait_for_job ID: long-poll a job until it ends, never faster than its Retry-After, and stop unless it succeeded. JOB is
# the last answer.
wait_for_job() {
  local started hint pause
  while true; do
    started=$SECONDS
    JOB=$(api GET "/jobs/$1?wait_seconds=20" -D "$HEADERS")
    case $(jq -r .status <<<"$JOB") in
      succeeded) return ;;
      failed | canceled) jq -c '{id, status, error}' <<<"$JOB" >&2; exit 1 ;;
    esac
    hint=$(awk -F': *' 'tolower($1) == "retry-after" { print $2 + 0 }' "$HEADERS")
    pause=$((${hint:-5} - (SECONDS - started)))
    if [ "$pause" -gt 0 ]; then sleep "$pause"; fi
  done
}

# Each paid request has a key of its own, saved before it is sent, so a run after a timeout or a crash sends the same one.
if [ -z "${CREATE_KEY:-}" ]; then
  printf 'CREATE_KEY=%s\nDUB_KEY=%s\n' "$(uuidgen)" "$(uuidgen)" >>"$STATE"
  . "$STATE"
fi

if [ -z "${MEDIA_ID:-}" ]; then
  BODY=$(jq -n --arg name "${FILE##*/}" --argjson size "$(wc -c <"$FILE")" \
    '{filename: $name, content_type: "video/mp4", size_bytes: $size}')
  UPLOAD=$(api POST /uploads -H "Content-Type: application/json" -d "$BODY")
  UPLOAD_ID=$(jq -er .id <<<"$UPLOAD")
  UPLOAD_URL=$(jq -er .transfer.url <<<"$UPLOAD")
  # The bytes go straight to storage, with exactly the listed headers and no Authorization header.
  jq -r '.transfer.headers | to_entries[] | "\(.key): \(.value)"' <<<"$UPLOAD" >"$HEADERS"
  curl -fsS -X PUT "$UPLOAD_URL" -H @"$HEADERS" --upload-file "$FILE"
  # Each run uploads anew, so completing this upload is new work with a key of its own.
  COMPLETE=$(api POST "/uploads/$UPLOAD_ID/complete" -H "Idempotency-Key: $(uuidgen)")
  JOB_ID=$(jq -er .id <<<"$COMPLETE")
  wait_for_job "$JOB_ID"
  MEDIA_ID=$(jq -er .result.media_id <<<"$JOB")
  echo "MEDIA_ID=$MEDIA_ID" >>"$STATE"
fi
ORDER=$(jq -n --arg media "$MEDIA_ID" '{media_id: $media, target_language: "es-es"}')

# The price shown once is the most this script spends: it is the cap of every paid request below.
if [ -z "${CAP:-}" ]; then
  ESTIMATE=$(api POST /projects/estimate -H "Content-Type: application/json" -d "$ORDER")
  if [ "$(jq -r .eligible <<<"$ESTIMATE")" != true ]; then
    echo "Cannot start: $(jq -c .reasons <<<"$ESTIMATE")" >&2
    exit 1
  fi
  CAP=$(jq -er .total_minutes <<<"$ESTIMATE")
  echo "The dub costs $CAP minutes"
  echo "CAP=$CAP" >>"$STATE"
fi

PROCESSING=$(api POST /projects -H "Content-Type: application/json" -H "Idempotency-Key: $CREATE_KEY" \
  -d "$(jq --argjson cap "$CAP" '. + {expected_cost_minutes: $cap}' <<<"$ORDER")")
PROJECT_ID=$(jq -er .project_id <<<"$PROCESSING")
JOB_ID=$(jq -er .id <<<"$PROCESSING")
wait_for_job "$JOB_ID"

if [ "$(jq -r .result.auto_dub.state <<<"$JOB")" = dispatched ]; then
  DUB_JOB_ID=$(jq -er .result.auto_dub.dub_job_id <<<"$JOB")
else
  # The auto-dub was skipped and not charged. Dub by hand, within the same cap.
  ESTIMATE=$(api POST "/projects/$PROJECT_ID/dub/estimate" -H "Content-Type: application/json" -d '{}')
  PRICE=$(jq -er .total_minutes <<<"$ESTIMATE")
  if [ "$PRICE" -gt "$CAP" ]; then
    echo "The dub now costs $PRICE minutes, more than the $CAP you accepted, and nothing was charged." \
      "Project $PROJECT_ID keeps its translation: dub it once you accept the new price." >&2
    exit 1
  fi
  DUB=$(api POST "/projects/$PROJECT_ID/dub" -H "Content-Type: application/json" -H "Idempotency-Key: $DUB_KEY" \
    -d "$(jq -n --argjson cap "$CAP" '{expected_cost_minutes: $cap}')")
  DUB_JOB_ID=$(jq -er .id <<<"$DUB")
fi
wait_for_job "$DUB_JOB_ID"

OUTPUTS=$(api GET "/projects/$PROJECT_ID/outputs")
VIDEO_URL=$(jq -er 'first(.data[] | select(.kind == "dubbed_video" and .available) | .url)' <<<"$OUTPUTS") || {
  echo "Project $PROJECT_ID has no dubbed video to download: unavailable_reason in its outputs says why." >&2
  exit 1
}
curl -fsS -o dubbed.mp4 "$VIDEO_URL"
echo "Saved dubbed.mp4"

"""A small client for the Rask API v3. Needs `pip install requests`."""
import json
import os
import time
import uuid

import requests

API = os.environ.get("RASK_BASE_URL", "https://api.rask.ai/v3")
KEY = os.environ["RASK_API_KEY"]
FINISHED = {"succeeded", "failed", "canceled"}


class RaskProblem(Exception):
    """An error answer of the API or a failed job. `problem` holds its members: `code`, `hint`, `retry`..."""

    def __init__(self, problem):
        super().__init__(f"{problem['code']}: {problem['hint'] or problem['detail']}")
        self.problem = problem


class Progress:
    """What a program did, saved in a JSON file: its Idempotency-Keys, the ids it got and the price it accepted.

    After a timeout or a crash, run the program again. It sends the same keys, so the API replays what already ran instead of
    running and charging it again, and refuses a request that changed meanwhile with idempotency_key_reused. The API keeps a
    key 24 hours, and a key replays its job even when the job failed, so delete the file to start new work.
    """

    def __init__(self, path):
        self.path = path
        try:
            with open(path) as file:
                self.saved = json.load(file)
        except FileNotFoundError:
            self.saved = {}

    def once(self, name, make):
        """The value saved under `name`. The first time, `make()` makes it, and it is saved before anything uses it."""
        if name not in self.saved:
            self.saved[name] = make()
            with open(self.path + ".tmp", "w") as file:
                json.dump(self.saved, file, indent=2)
            os.replace(self.path + ".tmp", self.path)
        return self.saved[name]

    def key(self, name):
        """The Idempotency-Key of one piece of work, such as "create project": the same on every attempt and every run."""
        return self.once(f"key: {name}", lambda: str(uuid.uuid4()))


def send(method, path, body=None, params=None, key=None, if_match=None):
    """Send a request and return the response.

    `key` is the Idempotency-Key. Every attempt sends the same one, so a request that timed out but ran is replayed, not run
    twice. After a network error or a gateway's 5xx, only a GET or a request with a key is sent again.
    """
    headers = {"Authorization": f"Bearer {KEY}"}
    if key:
        headers["Idempotency-Key"] = key
    if if_match:
        headers["If-Match"] = if_match
    repeatable = bool(key) or method == "GET"
    for attempt in range(5):
        last = attempt == 4
        try:
            response = requests.request(method, API + path, json=body, params=params, headers=headers, timeout=60)
        except (requests.ConnectionError, requests.Timeout):
            if last or not repeatable:
                raise
            time.sleep(2**attempt)
            continue
        if response.ok:
            return response
        if "problem+json" in response.headers.get("Content-Type", ""):
            problem = response.json()
            if problem["retry"] != "same_key" or last:
                raise RaskProblem(problem)
        elif response.status_code < 500 or last or not repeatable:
            response.raise_for_status()  # A proxy answered, not the API.
        time.sleep(int(response.headers.get("Retry-After", 2**attempt)))


def call(method, path, body=None, **options):
    """`send`, returning the parsed body (None for a 204)."""
    response = send(method, path, body, **options)
    return response.json() if response.content else None


def wait_for_job(job_id, check=True):
    """Long-poll a job until it ends. With `check`, raise when it did not succeed."""
    while True:
        started = time.monotonic()
        response = send("GET", f"/jobs/{job_id}", params={"wait_seconds": 20})
        job = response.json()
        if job["status"] in FINISHED:
            if check and job["status"] != "succeeded":
                raise RaskProblem(job["error"] or {"code": job["status"], "hint": None, "detail": job["id"]})
            return job
        hint = int(response.headers.get("Retry-After", 5))
        time.sleep(max(0, hint - (time.monotonic() - started)))


def upload_file(path, content_type, purpose="project_source"):
    """Upload a local file, wait for its import and return the media id."""
    upload = call(
        "POST",
        "/uploads",
        {
            "filename": os.path.basename(path),
            "content_type": content_type,
            "size_bytes": os.path.getsize(path),
            "purpose": purpose,
        },
    )
    transfer = upload["transfer"]
    with open(path, "rb") as source:
        requests.put(transfer["url"], data=source, headers=transfer["headers"], timeout=3600).raise_for_status()
    # Each run uploads anew, so completing this upload is new work with a key of its own.
    job = call("POST", f"/uploads/{upload['id']}/complete", {}, key=str(uuid.uuid4()))
    return wait_for_job(job["id"])["result"]["media_id"]


def download(url, path):
    """Save a presigned link to a file. It needs no key."""
    with requests.get(url, stream=True, timeout=3600) as response:
        response.raise_for_status()
        with open(path, "wb") as target:
            for chunk in response.iter_content(chunk_size=1024 * 1024):
                target.write(chunk)

"""Upload a video, dub it into Spanish and download the result: python dub_video.py clip.mp4

Progress is saved in clip.mp4.dub.json. If the program stops, run it again to continue; delete the file to dub anew.
"""
import sys

from rask_client import Progress, call, download, upload_file, wait_for_job

path = sys.argv[1]
progress = Progress(path + ".dub.json")

media_id = progress.once("media_id", lambda: upload_file(path, "video/mp4"))
order = {"media_id": media_id, "target_language": "es-es"}


def accept_price():
    estimate = call("POST", "/projects/estimate", order)
    if not estimate["eligible"]:
        sys.exit(f"Cannot start: {estimate['reasons']}")
    print(f"The dub costs {estimate['total_minutes']} minutes")
    return estimate["total_minutes"]


# The price shown once is the most this program spends: it is the cap of every paid request below.
max_minutes = progress.once("max_minutes", accept_price)
processing = call("POST", "/projects", {**order, "expected_cost_minutes": max_minutes}, key=progress.key("create project"))
project_id = processing["project_id"]

auto_dub = wait_for_job(processing["id"])["result"]["auto_dub"]
if auto_dub["state"] == "dispatched":
    wait_for_job(auto_dub["dub_job_id"])
else:
    # The auto-dub was skipped and not charged. Dub by hand, within the same cap.
    estimate = call("POST", f"/projects/{project_id}/dub/estimate", {})
    if estimate["total_minutes"] > max_minutes:
        sys.exit(
            f"The dub now costs {estimate['total_minutes']} minutes, more than the {max_minutes} you accepted, and nothing "
            f"was charged. Project {project_id} keeps its translation: dub it once you accept the new price."
        )
    dub = call("POST", f"/projects/{project_id}/dub", {"expected_cost_minutes": max_minutes}, key=progress.key("dub"))
    wait_for_job(dub["id"])

outputs = call("GET", f"/projects/{project_id}/outputs")["data"]
video = next((output for output in outputs if output["kind"] == "dubbed_video" and output["available"]), None)
if video is None:
    sys.exit(f"Project {project_id} has no dubbed video to download: unavailable_reason in its outputs says why.")
download(video["url"], "dubbed.mp4")
print("Saved dubbed.mp4")

"""Upload a video, then dub, lip-sync and translate its on-screen text in one run: python localize_video.py clip.mp4

Progress is saved in clip.mp4.localize.json. If the program stops, run it again to continue; delete the file to start anew.
"""
import sys

from rask_client import Progress, call, download, upload_file, wait_for_job

path = sys.argv[1]
progress = Progress(path + ".localize.json")

media_id = progress.once("media_id", lambda: upload_file(path, "video/mp4"))
order = {
    "media_id": media_id,
    "target_language": "es-es",
    "localization": {"lipsync": {"model": "legacy"}, "onscreen": {"ui_language": "en"}},
}


def accept_price():
    estimate = call("POST", "/projects/estimate", order)
    if not estimate["eligible"]:
        sys.exit(f"Cannot start: {estimate['reasons']}")
    print(f"The run costs {estimate['total_minutes']} minutes")
    return estimate["total_minutes"]


# A stage priced above this cap later is skipped, uncharged, and fails the run instead of costing more.
max_minutes = progress.once("max_minutes", accept_price)
processing = call("POST", "/projects", {**order, "expected_cost_minutes": max_minutes}, key=progress.key("create project"))
run_id = wait_for_job(processing["id"])["result"]["localization_job_id"]

run = wait_for_job(run_id)
download(run["result"]["final_output"]["url"], "localized.mp4")
print("Saved localized.mp4")