Batch processing with Python

# python batch_reels.py ./lessons ./reels
import hashlib
import json
import os
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

import requests

API = "https://api.slicer.media/v1"
SETTINGS = {"templateId": "clean", "targetDurationSec": 60, "direction": {"mode": "talking", "pace": "balanced"}}
PARALLEL = 3

source_dir, output_dir = Path(sys.argv[1]), Path(sys.argv[2])
output_dir.mkdir(parents=True, exist_ok=True)
state_file = output_dir / "state.json"
state = json.loads(state_file.read_text()) if state_file.exists() else {}

session = requests.Session()
session.headers["Authorization"] = f"Bearer {os.environ['SLICER_TOKEN']}"


class ApiError(Exception):
    def __init__(self, status, error):
        super().__init__(f"{status} {error['code']}: {error['message']}")
        self.code = error["code"]


def api(method, path, json_body=None, key=None):
    headers = {"Idempotency-Key": key} if key else {}
    for attempt in range(1, 6):
        response = session.request(method, API + path, json=json_body, headers=headers, timeout=120)
        body = response.json()
        if body["ok"]:
            return body["data"]
        if response.status_code in (429, 500, 502, 503, 504) and attempt < 5:
            time.sleep(float(response.headers.get("Retry-After") or 2 ** attempt))
            continue
        raise ApiError(response.status_code, body["error"])


def key_for(path, step):
    """The same key on a restart: a retry won't create a second montage (keys live 24 h)."""
    stat = path.stat()
    return hashlib.sha256(f"{path.resolve()}:{stat.st_size}:{stat.st_mtime_ns}:{step}".encode()).hexdigest()[:64]


def save_state():
    state_file.write_text(json.dumps(state, ensure_ascii=False, indent=2))


def upload(path):
    created = api("POST", "/uploads", {"fileName": path.name, "size": path.stat().st_size})
    with path.open("rb") as file:
        body = requests.put(created["uploadUrl"], data=file, timeout=3600).json()
    if not body["ok"]:
        raise RuntimeError(body["error"]["message"])
    item = body["data"]
    while item["status"] in ("uploading", "processing"):
        time.sleep(3)
        item = api("GET", f"/uploads/{item['id']}")
    if item["status"] != "ready":
        raise RuntimeError(item["error"] or item["status"])
    return item["id"]


def process(path):
    entry = state.setdefault(path.name, {})
    if entry.get("done"):
        return f"{path.name}: already done"
    if "project" not in entry:
        upload_id = upload(path)
        request = {"uploadIds": [upload_id], "title": path.stem, "settings": SETTINGS}
        quote = api("POST", "/projects", request)
        project = api("POST", "/projects", {**request, "confirmCost": quote["cost"]}, key_for(path, "montage"))
        entry["project"] = project["id"]
        save_state()
    project_id = entry["project"]
    while True:
        project = api("GET", f"/projects/{project_id}")
        if project["status"] not in ("queued", "processing"):
            break
        time.sleep(5)
    if project["status"] != "succeeded":
        return f"{path.name}: didn't work out — {(project['error'] or {}).get('message', project['status'])}"
    target = output_dir / f"{path.stem}.{project['result']['container']}"
    with session.get(project["result"]["url"], stream=True, timeout=600) as video:
        video.raise_for_status()
        with target.open("wb") as out:
            for chunk in video.iter_content(1 << 20):
                out.write(chunk)
    entry["done"] = True
    save_state()
    return f"{path.name}: done → {target.name}"


# 1. Files and budget
extensions = {f".{ext}" for ext in api("GET", "/styles")["materialLimits"]["videoExtensions"]}
files = sorted(p for p in source_dir.iterdir() if p.suffix.lower() in extensions)
todo = [p for p in files if not state.get(p.name, {}).get("project")]
price = api("POST", "/estimate", {"kind": "montage"})
me = api("GET", "/me")
budget = price["cost"] * len(todo)
limit_left = price["spendLimitLeft"]
print(f"Videos: {len(files)}, new montages: {len(todo)}, budget: ~{budget} tokens, available: {me['wallet']['available']}")
if budget > me["wallet"]["available"] or (limit_left is not None and budget > limit_left):
    sys.exit("Not enough tokens or spend limit — shrink the batch or top up")
if input("Start? [y/N] ").strip().lower() != "y":
    sys.exit(0)

# 2. Parallel processing
with ThreadPoolExecutor(max_workers=PARALLEL) as pool:
    futures = {pool.submit(process, path): path for path in files}
    for future in as_completed(futures):
        try:
            print(future.result())
        except Exception as error:  # one failed file doesn't stop the others
            print(f"{futures[future].name}: error — {error}")