"""CloudFilm — the official Python library (standard library only; Python 3.8+). Turn an idea or a whole script into a finished film. Docs: https://cloudfilm.ai/developers from cloudfilm import CloudFilm cloudfilm = CloudFilm() # reads CLOUDFILM_KEY film = cloudfilm.films.create(idea="A paper boat reaches the sea", length_seconds=30, shape="9:16") done = cloudfilm.films.wait(film["id"]) print(done["film"]["download_url"]) """ import hashlib import hmac import json import os import time import urllib.error import urllib.parse import urllib.request __version__ = "1.0.0" __all__ = ["CloudFilm", "CloudFilmError", "verify_webhook"] DEFAULT_BASE = "https://cloudfilm.ai/api/v1" DONE = {"ready", "failed", "cancelled"} WAITING = {"script_ready", "pictures_ready", "needs_credits", "limit_reached"} class CloudFilmError(Exception): """Every error carries CloudFilm's plain-words message, a stable code and the HTTP status.""" def __init__(self, status, body): err = (body or {}).get("error") or {} super().__init__(err.get("message") or "Request failed ({})".format(status)) self.status = status self.code = err.get("code", "UNKNOWN") self.details = err.get("details") self.docs = err.get("docs") class _Films: def __init__(self, t): self._t = t def create(self, idempotency_key=None, **film): """Make a film: idea=... or script=..., plus any choices (mode, shape, resolution, language, captions, music...).""" headers = {"Idempotency-Key": idempotency_key} if idempotency_key else {} return self._t.request("POST", "/films", film, headers) def quote(self, **film): """The price before anything is made - free.""" return self._t.request("POST", "/films/quote", film) def get(self, film_id, include_script=False): return self._t.request("GET", "/films/{}{}".format(urllib.parse.quote(film_id), "?include=script" if include_script else "")) def list(self, limit=None, state=None, before=None): q = {k: v for k, v in {"limit": limit, "state": state, "before": before}.items() if v is not None} return self._t.request("GET", "/films?" + urllib.parse.urlencode(q)) def approve_script(self, film_id): """Go: the waiting script and its price are approved.""" return self._t.request("POST", "/films/{}/approve".format(urllib.parse.quote(film_id)), {"what": "script"}) def approve_pictures(self, film_id, shots=None): """Approve pictures - all of them, or some shots ([1, 3]).""" body = {"what": "pictures"} if shots: body["shots"] = shots return self._t.request("POST", "/films/{}/approve".format(urllib.parse.quote(film_id)), body) def change_script(self, film_id, change): return self._t.request("POST", "/films/{}/script".format(urllib.parse.quote(film_id)), {"change": change}) def redo(self, film_id, shots): """Remake pictures: [{"shot": 3, "moment": "new words for the moment"}].""" return self._t.request("POST", "/films/{}/redo".format(urllib.parse.quote(film_id)), {"shots": shots}) def change(self, film_id, **changes): """Change a ready film: lines=[{"line": 2, "text": "..."}], captions=..., music=..., resolution="4k".""" return self._t.request("POST", "/films/{}/change".format(urllib.parse.quote(film_id)), changes) def languages(self, film_id, languages): return self._t.request("POST", "/films/{}/languages".format(urllib.parse.quote(film_id)), {"languages": languages}) def share(self, film_id): return self._t.request("POST", "/films/{}/share".format(urllib.parse.quote(film_id)), {}) def cancel(self, film_id): return self._t.request("DELETE", "/films/{}".format(urllib.parse.quote(film_id))) def retry(self, film_id): """Try a stopped film again from where it stopped.""" return self._t.request("POST", "/films/{}/retry".format(urllib.parse.quote(film_id)), {}) def wait(self, film_id, interval=5.0, timeout=None, on_update=None, until_ready=False): """Follow a film until it's ready, stopped, or waiting for you (script_ready, pictures_ready...).""" start = time.time() last = None while True: f = self.get(film_id) key = (f.get("state"), f.get("stage"), f.get("progress")) if key != last and on_update: on_update(f) last = key if f.get("state") in DONE or (not until_ready and f.get("state") in WAITING): return f if timeout and time.time() - start > timeout: return f time.sleep(interval) class _Uploads: def __init__(self, t): self._t = t def create(self, kind, name=None, what=None, description=None): """A secure upload link: kind is "photo", "music" or "clip".""" body = {k: v for k, v in {"kind": kind, "name": name, "what": what, "description": description}.items() if v is not None} return self._t.request("POST", "/uploads", body) def get(self, upload_id): return self._t.request("GET", "/uploads/{}".format(urllib.parse.quote(upload_id))) def put(self, upload_id, data, content_type, file_name=None): """Send the file's bytes with its type, e.g. "image/jpeg" (up to 20 MB for a photo, 30 MB otherwise).""" headers = {"Content-Type": content_type} if file_name: headers["X-File-Name"] = file_name return self._t.request("PUT", "/uploads/{}".format(urllib.parse.quote(upload_id)), data, headers, raw=True) class CloudFilm: def __init__(self, api_key=None, base_url=DEFAULT_BASE, timeout=60): self.api_key = api_key or os.environ.get("CLOUDFILM_KEY") or os.environ.get("TRIVEN_KEY") if not self.api_key: raise ValueError("Pass your CloudFilm API key, or set CLOUDFILM_KEY.") self.base_url = base_url.rstrip("/") self.timeout = timeout self.films = _Films(self) self.uploads = _Uploads(self) def characters(self): return self.request("GET", "/characters") def voices(self, language=None, search=None, gender=None): q = {k: v for k, v in {"language": language, "search": search, "gender": gender}.items() if v is not None} return self.request("GET", "/voices?" + urllib.parse.urlencode(q)) def languages(self): return self.request("GET", "/languages") def credits(self): return self.request("GET", "/credits") def status(self): return self.request("GET", "/status") def request(self, method, path, body=None, headers=None, raw=False): h = {"Authorization": "Bearer " + self.api_key, "User-Agent": "cloudfilm-python/" + __version__} data = None if body is not None: if raw: data = body else: data = json.dumps(body).encode("utf-8") h["Content-Type"] = "application/json" h.update(headers or {}) for attempt in range(4): req = urllib.request.Request(self.base_url + path, data=data, headers=h, method=method) try: with urllib.request.urlopen(req, timeout=self.timeout) as res: text = res.read().decode("utf-8") return json.loads(text) if text else None except urllib.error.HTTPError as e: text = e.read().decode("utf-8", "replace") try: payload = json.loads(text) if text else None except ValueError: payload = None if e.code == 429 and attempt < 3: time.sleep(float(e.headers.get("Retry-After") or 2)) continue raise CloudFilmError(e.code, payload) def verify_webhook(raw_body, header, secret, tolerance=300): """Check a webhook's CloudFilm-Signature header: True when it's really from CloudFilm and fresh.""" if not header: return False try: parts = dict(p.split("=", 1) for p in header.split(",")) t = int(parts["t"]) v1 = parts["v1"] except (KeyError, ValueError): return False if abs(time.time() - t) > tolerance: return False body = raw_body if isinstance(raw_body, bytes) else raw_body.encode("utf-8") expected = hmac.new(secret.encode("utf-8"), str(t).encode() + b"." + body, hashlib.sha256).hexdigest() return hmac.compare_digest(expected, v1)