From 7f30313005d72b9f8922ccf0393e3c5bf5f178f7 Mon Sep 17 00:00:00 2001 From: Musab Date: Tue, 13 Oct 2026 01:43:52 +0500 Subject: [PATCH 01/11] Add localized listing videos example Renders one property listing video in English, Spanish, German and Japanese from one saved template, with generated voice-overs, captions and merge-field text. Includes an optional translate.py that drafts missing translations with Gemini. Companion code for the video localization API guide. --- README.md | 1 + .../localized-listing-videos/.env.example | 11 + examples/localized-listing-videos/.gitignore | 3 + examples/localized-listing-videos/README.md | 95 +++++++ .../localized-listing-videos/languages.json | 34 +++ examples/localized-listing-videos/render.py | 239 ++++++++++++++++++ .../localized-listing-videos/save_template.py | 56 ++++ .../localized-listing-videos/template.json | 139 ++++++++++ .../localized-listing-videos/translate.py | 123 +++++++++ .../upload_footage.py | 82 ++++++ 10 files changed, 783 insertions(+) create mode 100644 examples/localized-listing-videos/.env.example create mode 100644 examples/localized-listing-videos/.gitignore create mode 100644 examples/localized-listing-videos/README.md create mode 100644 examples/localized-listing-videos/languages.json create mode 100644 examples/localized-listing-videos/render.py create mode 100644 examples/localized-listing-videos/save_template.py create mode 100644 examples/localized-listing-videos/template.json create mode 100644 examples/localized-listing-videos/translate.py create mode 100644 examples/localized-listing-videos/upload_footage.py diff --git a/README.md b/README.md index b2e7272..13b7c99 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,7 @@ Clone this repository, or open the directory of the example you want. Each examp - [hedra-spokesperson-ad](examples/hedra-spokesperson-ad) turns a portrait and a voice recording into a lip-synced Hedra clip, then renders an ad with product cutaways from a CSV, captions and an end card. - [in-app-video-creation](examples/in-app-video-creation) lets a user create the same promo video three ways: an embedded Studio SDK editor, a quick form, and a one-click headless render, all through one render proxy that keeps the API key server-side. Companion code for [Add video creation to your app without building an editor](https://shotstack.io/learn/add-video-creation-to-your-app/). - [instagram-ai-video](examples/instagram-ai-video) generates a script, voiceover and background image with AI, renders a 1080x1920 video, and publishes it as an Instagram Reel. Companion code for [How to automate Instagram posts with AI video](https://shotstack.io/learn/automate-instagram-posts-with-ai-video/). +- [localized-listing-videos](examples/localized-listing-videos) renders one property listing video in English, Spanish, German and Japanese from one saved template, each with an AI voice-over, captions and on-screen text, with an optional step where a language model drafts the translations. Companion code for [Video localization API: render one template in multiple languages](https://shotstack.io/learn/video-localization-api/). - [multi-client-video-automation](examples/multi-client-video-automation) renders branded promo videos for three clients in three aspect ratios from one master template, and records which render belongs to which client. Companion code for [A guide to automating video content production for multiple clients](https://shotstack.io/learn/automating-video-production-multiple-clients/). - [rapidreels](examples/rapidreels) creates faceless short-form videos using generative AI. [View demo](https://shotstack.io/demos/social-media-video-maker/). - [reelestate](examples/reelestate) turns static real estate images into fully edited video slideshows. [View demo](https://shotstack.io/demos/real-estate-video-listing-maker/). diff --git a/examples/localized-listing-videos/.env.example b/examples/localized-listing-videos/.env.example new file mode 100644 index 0000000..07724fd --- /dev/null +++ b/examples/localized-listing-videos/.env.example @@ -0,0 +1,11 @@ +# Your Shotstack production API key: https://dashboard.shotstack.io +SHOTSTACK_API_KEY= + +# The ID that save_template.py prints +SHOTSTACK_TEMPLATE_ID= + +# Optional, for translate.py. Your Gemini API key: https://aistudio.google.com/apikey +GEMINI_API_KEY= + +# Optional, for translate.py. Another Gemini model: https://ai.google.dev/gemini-api/docs/models +GEMINI_MODEL= diff --git a/examples/localized-listing-videos/.gitignore b/examples/localized-listing-videos/.gitignore new file mode 100644 index 0000000..896128d --- /dev/null +++ b/examples/localized-listing-videos/.gitignore @@ -0,0 +1,3 @@ +.env +output/ +__pycache__/ diff --git a/examples/localized-listing-videos/README.md b/examples/localized-listing-videos/README.md new file mode 100644 index 0000000..cc8ee61 --- /dev/null +++ b/examples/localized-listing-videos/README.md @@ -0,0 +1,95 @@ +# Localized listing videos + +Render one property listing video in English, Spanish, German and Japanese +from one saved template. Each language gets its own AI voice-over, captions, +headline and call to action. You get one MP4 file for each language in +`output/`. + +Related guide: [Video localization API: render one template in multiple languages](https://shotstack.io/learn/video-localization-api/). + +## Requirements + +- A [Shotstack account](https://dashboard.shotstack.io/register) and your + **production** API key +- Python 3.8 or later, with `requests` 2 or later +- A video that is longer than your longest voice-over, as a file or at a + public URL +- Optional: a [Gemini API key](https://aistudio.google.com/apikey) for + `translate.py` + +The scripts use the production API. Each render generates a voice-over and +captions, and uses credits. + +## Setup + +```bash +git clone https://github.com/shotstack/shotstack-cookbook.git +cd shotstack-cookbook/examples/localized-listing-videos +python3 -m pip install requests +cp .env.example .env +``` + +Set `SHOTSTACK_API_KEY` in `.env`. Then load the file. Do this in each new +terminal, and each time you change `.env`: + +```bash +set -a +source .env +set +a +``` + +If your video is a file on your computer, upload it. The script prints the +URL of the uploaded video: + +```bash +python3 upload_footage.py path/to/footage.mp4 +``` + +In `template.json`, replace `PASTE_FOOTAGE_URL_HERE` with the URL of your +video. Then save the template to your account: + +```bash +python3 save_template.py +``` + +Set `SHOTSTACK_TEMPLATE_ID` in `.env` to the ID that the script prints. Then +load `.env` again. + +## Run + +Render Japanese first to test the setup. Then render the other languages: + +```bash +python3 render.py ja +python3 render.py +``` + +To translate with a language model, use `translate.py`. It reads the first +entry in `languages.json` as the source language. For each other language, +keep `code`, `model` and `voice`, and make `script`, `headline` and `cta` +empty. Set `GEMINI_API_KEY` in `.env`, load `.env` again, and run: + +```bash +python3 translate.py +python3 render.py +``` + +`translate.py` fills only empty fields. Ask a speaker of each language to read +the translations before you render them. + +## What happens + +`render.py` starts one render for each language in `languages.json`. Then it +prints the status of each render until it is done. It saves each video as +`output/.mp4`. + +The script writes each render ID to `output/renders.json`. When you run it +again, it skips the videos that are done and not changed. It continues the +renders that did not finish. It starts new renders only for new or changed +languages. + +If you save the template again, the template gets a new ID. Then all the +languages render again. + +The template loads the Noto Sans JP font from the Noto CJK repository. This +font shows Japanese and the accented letters of Spanish and German. diff --git a/examples/localized-listing-videos/languages.json b/examples/localized-listing-videos/languages.json new file mode 100644 index 0000000..dd1a572 --- /dev/null +++ b/examples/localized-listing-videos/languages.json @@ -0,0 +1,34 @@ +[ + { + "code": "en", + "model": "elevenlabs-multilingual-v2", + "voice": "Rachel", + "script": "Welcome to Aurelia Residences, set among landscaped gardens in the heart of the city. Tall windows and crystal chandeliers fill the rooms with light, and every detail, from the marble floors to the furnishings, is finished with care. There's space to gather, space to unwind, and a home that's ready for you. Book your private viewing today.", + "headline": "A quiet retreat in the city", + "cta": "Book a private viewing" + }, + { + "code": "es", + "model": "elevenlabs-multilingual-v2", + "voice": "Rachel", + "script": "Le damos la bienvenida a Aurelia Residences, entre jardines cuidados en el corazón de la ciudad. Grandes ventanales y lámparas de cristal llenan de luz cada espacio, y cada detalle, desde el mármol hasta el mobiliario, está pensado con esmero. Hay lugar para reunirse, lugar para descansar y un hogar listo para usted. Reserve hoy su visita privada.", + "headline": "Un refugio tranquilo en la ciudad", + "cta": "Reserve una visita privada" + }, + { + "code": "de", + "model": "elevenlabs-multilingual-v2", + "voice": "Rachel", + "script": "Willkommen in den Aurelia Residences, umgeben von gepflegten Gärten im Herzen der Stadt. Hohe Fenster und Kristallleuchter tauchen die Räume in helles Licht, und jedes Detail, vom Marmorboden bis zur Einrichtung, ist mit Sorgfalt gestaltet. Hier ist Raum zum Zusammenkommen, Raum zum Entspannen und ein Zuhause, das auf Sie wartet. Vereinbaren Sie noch heute Ihre private Besichtigung.", + "headline": "Ein ruhiger Rückzugsort in der Stadt", + "cta": "Vereinbaren Sie eine private Besichtigung" + }, + { + "code": "ja", + "model": "polly-neural", + "voice": "Kazuha", + "script": "都心の緑豊かな庭園に囲まれた、アウレリア・レジデンスへようこそ。大きな窓とクリスタルのシャンデリアが室内を光で満たし、大理石の床から家具まで、細部にこだわって仕上げられています。みんなで集える空間も、ゆったりくつろげる空間もそろい、新しい暮らしをすぐに始められます。ぜひ本日、プライベート内覧をご予約ください。", + "headline": "都心の静かな隠れ家", + "cta": "プライベート内覧をご予約ください" + } +] diff --git a/examples/localized-listing-videos/render.py b/examples/localized-listing-videos/render.py new file mode 100644 index 0000000..6b3793f --- /dev/null +++ b/examples/localized-listing-videos/render.py @@ -0,0 +1,239 @@ +"""Render one localized video per language from a saved Shotstack template. + +Reads languages.json, submits one template render per language, waits for each +render to finish, and saves the videos to output/.mp4. Each render is +recorded in output/renders.json together with a fingerprint of what it was made +from: the template ID and that language's merge fields. + +Running the script again skips videos already finished with the same template +ID and text, picks up renders that were still running (or finished after the +script stopped), and starts new renders for new, changed or failed languages. + +Pass language codes to render only those, e.g. `python render.py ja`. +""" + +import hashlib +import json +import os +import sys +import time +from pathlib import Path + +import requests + +API = "https://api.shotstack.io/edit/v1" +API_KEY = os.environ.get("SHOTSTACK_API_KEY") +TEMPLATE_ID = os.environ.get("SHOTSTACK_TEMPLATE_ID") +HEADERS = {"x-api-key": API_KEY or ""} +HERE = Path(__file__).resolve().parent +OUTPUT_DIR = HERE / "output" +LOG_PATH = OUTPUT_DIR / "renders.json" +POLL_SECONDS = 5 +TIMEOUT_SECONDS = 15 * 60 + + +def error_detail(resp): + """Return the API's error message as one line.""" + try: + errors = resp.json().get("errors") or [{}] + return errors[0].get("detail") or resp.text + except (ValueError, AttributeError): + return resp.text.strip()[:200] + + +def merge_fields(language): + """Map one languages.json entry to the template's merge fields.""" + return [ + {"find": "SCRIPT", "replace": language["script"]}, + {"find": "MODEL", "replace": language["model"]}, + {"find": "VOICE", "replace": language["voice"]}, + {"find": "HEADLINE", "replace": language["headline"]}, + {"find": "CTA", "replace": language["cta"]}, + ] + + +def fingerprint(language): + """Identify a video by the template ID and text it is rendered from.""" + payload = json.dumps( + {"template": TEMPLATE_ID, "merge": merge_fields(language)}, + sort_keys=True, + ensure_ascii=False, + ) + return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16] + + +def submit(language): + resp = requests.post( + f"{API}/templates/render", + headers=HEADERS, + json={"id": TEMPLATE_ID, "merge": merge_fields(language)}, + timeout=30, + ) + if resp.status_code in (401, 403): + # A rejected key fails every language the same way, so stop now. + raise SystemExit(f"Shotstack rejected the API key: {error_detail(resp)}") + if not resp.ok: + raise RuntimeError(f"submit failed ({resp.status_code}): {error_detail(resp)}") + return resp.json()["response"]["id"] + + +def get_render(render_id): + resp = requests.get(f"{API}/render/{render_id}", headers=HEADERS, timeout=30) + if resp.status_code >= 500 or resp.status_code == 429: + # A server error or rate limit is retried like a connection problem. + resp.raise_for_status() + if not resp.ok: + raise RuntimeError( + f"status check failed ({resp.status_code}): {error_detail(resp)}" + ) + return resp.json()["response"] + + +def wait_for(code, render_id): + """Poll until the render is done or failed, printing each new status. + + The render keeps running on Shotstack's side if this script loses its + connection, so connection problems are retried until the timeout. + """ + deadline = time.monotonic() + TIMEOUT_SECONDS + last_status = None + while time.monotonic() < deadline: + try: + render = get_render(render_id) + except requests.RequestException as err: + print(f"[{code}] connection problem, retrying ({type(err).__name__})") + time.sleep(POLL_SECONDS) + continue + if render["status"] != last_status: + last_status = render["status"] + print(f"[{code}] {last_status}") + if last_status == "done": + return render["url"] + if last_status == "failed": + raise RuntimeError(f"render failed: {render.get('error')}") + time.sleep(POLL_SECONDS) + raise RuntimeError( + f"render {render_id} did not finish within {TIMEOUT_SECONDS} seconds" + ) + + +def can_resume(render_id): + """True if a render from an earlier run is still running or has finished.""" + try: + return get_render(render_id)["status"] != "failed" + except requests.RequestException: + return True # no answer yet; wait_for() keeps trying + except RuntimeError: + return False # for example, Shotstack no longer knows this render ID + + +def download(url, path): + # The render URL is temporary, so save the file as soon as it is ready. + # Write to a .part file first so an interrupted download is never mistaken + # for a finished video on the next run. + partial = path.with_name(path.name + ".part") + with requests.get(url, stream=True, timeout=60) as resp: + resp.raise_for_status() + with open(partial, "wb") as out: + for chunk in resp.iter_content(chunk_size=1024 * 1024): + out.write(chunk) + partial.replace(path) + + +def failed_entry(render_id, fp): + return {"render_id": render_id, "fingerprint": fp, "status": "failed"} + + +def save_log(log): + # Write to a temporary file first, so a stopped script cannot leave half a log. + temporary = LOG_PATH.with_name(LOG_PATH.name + ".tmp") + temporary.write_text(json.dumps(log, indent=2), encoding="utf-8") + temporary.replace(LOG_PATH) + + +def main(): + if not API_KEY: + raise RuntimeError("Set the SHOTSTACK_API_KEY environment variable first.") + if not TEMPLATE_ID: + raise RuntimeError( + "Set the SHOTSTACK_TEMPLATE_ID environment variable first. " + "Run save_template.py to get the ID." + ) + languages = json.loads((HERE / "languages.json").read_text(encoding="utf-8")) + only = set(sys.argv[1:]) + if only: + unknown = only - {language["code"] for language in languages} + if unknown: + raise SystemExit(f"Not in languages.json: {', '.join(sorted(unknown))}") + languages = [language for language in languages if language["code"] in only] + OUTPUT_DIR.mkdir(exist_ok=True) + log = {} + if LOG_PATH.exists(): + log = json.loads(LOG_PATH.read_text(encoding="utf-8")) + + # Start (or pick up) every render first so they run at the same time. + pending = {} + failed = [] + for language in languages: + code = language["code"] + fp = fingerprint(language) + entry = log.get(code, {}) + unchanged = entry.get("fingerprint") == fp + video_exists = (OUTPUT_DIR / f"{code}.mp4").exists() + if unchanged and entry.get("status") == "done" and video_exists: + print(f"[{code}] already rendered, skipping") + continue + render_id = entry.get("render_id") + if unchanged and render_id and can_resume(render_id): + print(f"[{code}] picking up render {entry['render_id']}") + pending[code] = (entry["render_id"], fp) + continue + try: + render_id = submit(language) + except (RuntimeError, requests.RequestException) as err: + print(f"[{code}] {err}") + failed.append(code) + continue + pending[code] = (render_id, fp) + log[code] = {"render_id": render_id, "fingerprint": fp, "status": "submitted"} + save_log(log) + print(f"[{code}] queued as {render_id}") + + for code, (render_id, fp) in pending.items(): + try: + url = wait_for(code, render_id) + download(url, OUTPUT_DIR / f"{code}.mp4") + log[code] = {"render_id": render_id, "fingerprint": fp, "status": "done"} + print(f"[{code}] saved to output/{code}.mp4") + except requests.HTTPError as err: + failed.append(code) + print(f"[{code}] download failed: {err}") + if err.response is not None and err.response.status_code < 500: + # The temporary URL has probably expired, so forget this render + # and start a new one next run. + log.pop(code, None) + else: + # A server error is temporary: keep the render, so the next run + # downloads it again without paying for a new render. + log[code] = failed_entry(render_id, fp) + except (RuntimeError, requests.RequestException) as err: + log[code] = failed_entry(render_id, fp) + failed.append(code) + print(f"[{code}] {err}") + save_log(log) + + if failed: + raise SystemExit( + f"Not rendered: {', '.join(failed)}. Run the script again to retry them." + ) + + +if __name__ == "__main__": + try: + main() + except KeyboardInterrupt: + # Renders already started keep running. The next run picks them up. + raise SystemExit(1) from None + except Exception as error: + print(error, file=sys.stderr) + raise SystemExit(1) from error diff --git a/examples/localized-listing-videos/save_template.py b/examples/localized-listing-videos/save_template.py new file mode 100644 index 0000000..40d51ea --- /dev/null +++ b/examples/localized-listing-videos/save_template.py @@ -0,0 +1,56 @@ +"""Save template.json as a Shotstack template and print the template ID.""" + +import json +import os +import sys +from pathlib import Path + +import requests + +API = "https://api.shotstack.io/edit/v1" +HERE = Path(__file__).resolve().parent + + +def error_detail(resp): + """Return the API's error message as one line.""" + try: + errors = resp.json().get("errors") or [{}] + return errors[0].get("detail") or resp.text + except (ValueError, AttributeError): + return resp.text.strip()[:200] + + +def main(): + api_key = os.environ.get("SHOTSTACK_API_KEY") + if not api_key: + raise RuntimeError("Set the SHOTSTACK_API_KEY environment variable first.") + + template_text = (HERE / "template.json").read_text(encoding="utf-8") + if "PASTE_FOOTAGE_URL_HERE" in template_text: + raise RuntimeError( + "Replace PASTE_FOOTAGE_URL_HERE in template.json " + "with your footage URL first." + ) + + resp = requests.post( + f"{API}/templates", + headers={"x-api-key": api_key}, + json={"name": "Localized listing video", "template": json.loads(template_text)}, + timeout=30, + ) + if not resp.ok: + raise RuntimeError( + f"Could not save the template ({resp.status_code}): {error_detail(resp)}" + ) + + print(resp.json()["response"]["id"]) + + +if __name__ == "__main__": + try: + main() + except KeyboardInterrupt: + raise SystemExit(1) from None + except Exception as error: + print(error, file=sys.stderr) + raise SystemExit(1) from error diff --git a/examples/localized-listing-videos/template.json b/examples/localized-listing-videos/template.json new file mode 100644 index 0000000..52be04f --- /dev/null +++ b/examples/localized-listing-videos/template.json @@ -0,0 +1,139 @@ +{ + "timeline": { + "background": "#000000", + "fonts": [ + { + "src": "https://raw.githubusercontent.com/notofonts/noto-cjk/f8d157532fbfaeda587e826d4cd5b21a49186f7c/Sans/SubsetOTF/JP/NotoSansJP-Regular.otf" + }, + { + "src": "https://raw.githubusercontent.com/notofonts/noto-cjk/f8d157532fbfaeda587e826d4cd5b21a49186f7c/Sans/SubsetOTF/JP/NotoSansJP-Bold.otf" + } + ], + "tracks": [ + { + "clips": [ + { + "asset": { + "type": "rich-text", + "text": "{{HEADLINE}}", + "font": { + "family": "Noto Sans JP", + "size": 64, + "weight": 700, + "color": "#FFFFFF" + }, + "background": { + "color": "#000000", + "opacity": 0.55, + "borderRadius": 16, + "wrap": true + }, + "align": { "horizontal": "left", "vertical": "top" }, + "padding": { "top": 12, "right": 28, "bottom": 14, "left": 28 } + }, + "start": 0, + "length": "alias://voice", + "width": 1200, + "height": 200, + "position": "topLeft", + "offset": { "x": 0.04, "y": -0.06 }, + "transition": { "in": "fade" } + } + ] + }, + { + "clips": [ + { + "asset": { + "type": "rich-text", + "text": "{{CTA}}", + "font": { + "family": "Noto Sans JP", + "size": 38, + "weight": 700, + "color": "#1A1A1A" + }, + "background": { + "color": "#FFFFFF", + "opacity": 0.92, + "borderRadius": 40, + "wrap": true + }, + "align": { "horizontal": "left", "vertical": "top" }, + "padding": { "top": 10, "right": 26, "bottom": 12, "left": 26 } + }, + "start": 0, + "length": "alias://voice", + "width": 1200, + "height": 110, + "position": "topLeft", + "offset": { "x": 0.04, "y": -0.25 }, + "transition": { "in": "fade" } + } + ] + }, + { + "clips": [ + { + "asset": { + "type": "rich-caption", + "src": "alias://voice", + "font": { + "family": "Noto Sans JP", + "size": 46, + "weight": 700, + "color": "#FFFFFF" + }, + "background": { + "color": "#000000", + "opacity": 0.55, + "borderRadius": 12, + "wrap": true, + "padding": 14 + }, + "active": { "font": { "color": "#FFD166" } }, + "animation": { "style": "highlight" }, + "align": { "horizontal": "center", "vertical": "bottom" } + }, + "start": 0, + "length": "end", + "width": 1600, + "height": 240, + "position": "bottom", + "offset": { "y": 0.06 } + } + ] + }, + { + "clips": [ + { + "alias": "voice", + "asset": { + "type": "audio", + "prompt": "{{SCRIPT}}", + "model": "{{MODEL}}", + "options": { "voice": "{{VOICE}}" } + }, + "start": 0, + "length": "auto" + } + ] + }, + { + "clips": [ + { + "asset": { + "type": "video", + "src": "PASTE_FOOTAGE_URL_HERE", + "volume": 0 + }, + "start": 0, + "length": "alias://voice", + "transition": { "out": "fade" } + } + ] + } + ] + }, + "output": { "format": "mp4", "resolution": "1080", "fps": 30 } +} diff --git a/examples/localized-listing-videos/translate.py b/examples/localized-listing-videos/translate.py new file mode 100644 index 0000000..e721fb4 --- /dev/null +++ b/examples/localized-listing-videos/translate.py @@ -0,0 +1,123 @@ +"""Fill in missing translations in languages.json with a language model. + +The first entry in languages.json is the source language. For every other +entry with an empty script, headline or cta, this script asks Gemini to +translate the source entry and writes the result back to languages.json. +Fields that already have text are never changed, so a translation you have +reviewed stays as it is. To translate a field again, empty it and rerun. + +Pass language codes to translate only those, e.g. `python translate.py de`. +""" + +import json +import os +import sys +from pathlib import Path + +import requests + +MODEL = os.environ.get("GEMINI_MODEL") or "gemini-flash-latest" +API_URL = ( + f"https://generativelanguage.googleapis.com/v1beta/models/{MODEL}:generateContent" +) +API_KEY_NAME = "GEMINI_API_KEY" +API_KEY = os.environ.get(API_KEY_NAME) +LANGUAGES_PATH = Path(__file__).resolve().parent / "languages.json" +FIELDS = ("script", "headline", "cta") + +INSTRUCTIONS = """\ +Translate the copy for a narrated video from the language with code "{source}" +into the language with code "{target}". + +- "script" is read aloud by a text-to-speech voice. Write it to be spoken: + spell out numbers, prices, dates and abbreviations the way a speaker says them. +- Keep the script close to the length of the source. The video lasts as long as + the voice-over. +- "headline" and "cta" (the call to action) appear on screen. Keep them as short + as the source. +- Keep brand and product names as they are, unless the target language normally + writes them in its own script, such as katakana in Japanese. +- Address the viewer politely and formally. + +Return the translated "script", "headline" and "cta" as JSON. + +Source copy: +""" + + +def translate(source, target_code): + """Translate the source entry's script, headline and cta into one language.""" + prompt = INSTRUCTIONS.format(source=source["code"], target=target_code) + copy = {field: source[field] for field in FIELDS} + prompt += json.dumps(copy, ensure_ascii=False, indent=2) + resp = requests.post( + API_URL, + headers={"x-goog-api-key": API_KEY}, + json={ + "contents": [{"parts": [{"text": prompt}]}], + "generationConfig": { + "responseMimeType": "application/json", + "responseSchema": { + "type": "OBJECT", + "properties": {field: {"type": "STRING"} for field in FIELDS}, + "required": list(FIELDS), + }, + }, + }, + timeout=60, + ) + if not resp.ok: + try: + detail = resp.json()["error"]["message"] + except (ValueError, KeyError, TypeError): + detail = resp.text.strip()[:200] + raise RuntimeError( + f"[{target_code}] translation failed ({resp.status_code}): {detail}" + ) + parts = resp.json()["candidates"][0]["content"]["parts"] + text = "".join(part.get("text", "") for part in parts if not part.get("thought")) + result = json.loads(text) + return {field: result[field].strip() for field in FIELDS} + + +def main(): + if not API_KEY: + raise RuntimeError(f"Set the {API_KEY_NAME} environment variable first.") + languages = json.loads(LANGUAGES_PATH.read_text(encoding="utf-8")) + source, targets = languages[0], languages[1:] + if not all(source.get(field) for field in FIELDS): + raise RuntimeError( + f"The first entry ({source['code']}) needs a script, headline and cta." + ) + only = set(sys.argv[1:]) + unknown = only - {language["code"] for language in targets} + if unknown: + raise RuntimeError( + f"Not a translation target in languages.json: {', '.join(sorted(unknown))}" + ) + + for language in targets: + code = language["code"] + if only and code not in only: + continue + missing = [field for field in FIELDS if not language.get(field)] + if not missing: + print(f"[{code}] already translated, skipping") + continue + translated = translate(source, code) + for field in missing: + language[field] = translated[field] + # Save after each language, so a later failure keeps the earlier results. + text = json.dumps(languages, indent=2, ensure_ascii=False) + "\n" + LANGUAGES_PATH.write_text(text, encoding="utf-8") + print(f"[{code}] translated {', '.join(missing)}") + + +if __name__ == "__main__": + try: + main() + except KeyboardInterrupt: + raise SystemExit(1) from None + except Exception as error: + print(error, file=sys.stderr) + raise SystemExit(1) from error diff --git a/examples/localized-listing-videos/upload_footage.py b/examples/localized-listing-videos/upload_footage.py new file mode 100644 index 0000000..ffc7b29 --- /dev/null +++ b/examples/localized-listing-videos/upload_footage.py @@ -0,0 +1,82 @@ +"""Upload a local video file to Shotstack's Ingest API and print its hosted URL. + +Usage: python upload_footage.py path/to/footage.mp4 +""" + +import os +import sys +import time + +import requests + +API = "https://api.shotstack.io/ingest/v1" +POLL_SECONDS = 5 +TIMEOUT_SECONDS = 15 * 60 + + +def error_detail(resp): + """Return the API's error message as one line.""" + try: + errors = resp.json().get("errors") or [{}] + return errors[0].get("detail") or resp.text + except (ValueError, AttributeError): + return resp.text.strip()[:200] + + +def check(resp, action): + if not resp.ok: + raise RuntimeError( + f"Could not {action} ({resp.status_code}): {error_detail(resp)}" + ) + return resp + + +def main(): + api_key = os.environ.get("SHOTSTACK_API_KEY") + if not api_key: + raise RuntimeError("Set the SHOTSTACK_API_KEY environment variable first.") + if len(sys.argv) != 2: + raise RuntimeError("Usage: python upload_footage.py ") + path = sys.argv[1] + if not os.path.isfile(path): + raise RuntimeError(f"File not found: {path}") + headers = {"x-api-key": api_key} + + # 1. Ask for a signed upload URL (valid for one hour). + response = requests.post(f"{API}/upload", headers=headers, timeout=30) + upload = check(response, "request an upload URL") + data = upload.json()["data"] + source_id, signed_url = data["id"], data["attributes"]["url"] + + # 2. Send the file to the signed URL with a plain PUT request. + with open(path, "rb") as video: + check(requests.put(signed_url, data=video, timeout=900), "upload the file") + + # 3. Wait until Shotstack has stored the file, then print its URL. + deadline = time.monotonic() + TIMEOUT_SECONDS + while time.monotonic() < deadline: + response = requests.get( + f"{API}/sources/{source_id}", headers=headers, timeout=30 + ) + attributes = check(response, "check the upload").json()["data"]["attributes"] + status = attributes["status"] + if status == "ready": + print(attributes["source"]) + return + if status in ("failed", "deleted"): + raise RuntimeError(f"Ingest {status} for source {source_id}") + time.sleep(POLL_SECONDS) + raise RuntimeError( + f"Source {source_id} was not ready after {TIMEOUT_SECONDS} seconds. " + "Run the script again." + ) + + +if __name__ == "__main__": + try: + main() + except KeyboardInterrupt: + raise SystemExit(1) from None + except Exception as error: + print(error, file=sys.stderr) + raise SystemExit(1) from error From 93ddca7dda8629e094c11769fe82dea5a61b00e9 Mon Sep 17 00:00:00 2001 From: Musab Date: Tue, 13 Oct 2026 01:45:08 +0500 Subject: [PATCH 02/11] Print one actionable line when the network is down --- examples/localized-listing-videos/render.py | 5 +++++ examples/localized-listing-videos/save_template.py | 7 +++++++ examples/localized-listing-videos/translate.py | 7 +++++++ examples/localized-listing-videos/upload_footage.py | 7 +++++++ 4 files changed, 26 insertions(+) diff --git a/examples/localized-listing-videos/render.py b/examples/localized-listing-videos/render.py index 6b3793f..897c1a1 100644 --- a/examples/localized-listing-videos/render.py +++ b/examples/localized-listing-videos/render.py @@ -190,6 +190,11 @@ def main(): continue try: render_id = submit(language) + except requests.ConnectionError: + raise SystemExit( + "Could not reach Shotstack. Check your network connection " + "and run the script again." + ) from None except (RuntimeError, requests.RequestException) as err: print(f"[{code}] {err}") failed.append(code) diff --git a/examples/localized-listing-videos/save_template.py b/examples/localized-listing-videos/save_template.py index 40d51ea..0ef53c7 100644 --- a/examples/localized-listing-videos/save_template.py +++ b/examples/localized-listing-videos/save_template.py @@ -51,6 +51,13 @@ def main(): main() except KeyboardInterrupt: raise SystemExit(1) from None + except requests.ConnectionError: + print( + "Could not reach Shotstack. Check your network connection " + "and run the script again.", + file=sys.stderr, + ) + raise SystemExit(1) from None except Exception as error: print(error, file=sys.stderr) raise SystemExit(1) from error diff --git a/examples/localized-listing-videos/translate.py b/examples/localized-listing-videos/translate.py index e721fb4..8cddc80 100644 --- a/examples/localized-listing-videos/translate.py +++ b/examples/localized-listing-videos/translate.py @@ -118,6 +118,13 @@ def main(): main() except KeyboardInterrupt: raise SystemExit(1) from None + except requests.ConnectionError: + print( + "Could not reach Gemini. Check your network connection " + "and run the script again.", + file=sys.stderr, + ) + raise SystemExit(1) from None except Exception as error: print(error, file=sys.stderr) raise SystemExit(1) from error diff --git a/examples/localized-listing-videos/upload_footage.py b/examples/localized-listing-videos/upload_footage.py index ffc7b29..2ab26b9 100644 --- a/examples/localized-listing-videos/upload_footage.py +++ b/examples/localized-listing-videos/upload_footage.py @@ -77,6 +77,13 @@ def main(): main() except KeyboardInterrupt: raise SystemExit(1) from None + except requests.ConnectionError: + print( + "Could not reach Shotstack. Check your network connection " + "and run the script again.", + file=sys.stderr, + ) + raise SystemExit(1) from None except Exception as error: print(error, file=sys.stderr) raise SystemExit(1) from error From 98a6646205280ed582c7cfe846c3decc7f943316 Mon Sep 17 00:00:00 2001 From: Musab Date: Tue, 13 Oct 2026 18:39:56 +0500 Subject: [PATCH 03/11] Stop with one line when the network drops while waiting for a render render.py retried a lost connection for up to 15 minutes when it picked up a render from an earlier run. Now it stops with the network message after six failed attempts and keeps the log entry, so the next run picks the render up. Use python3 in docstrings, as the README does. --- examples/localized-listing-videos/render.py | 30 +++++++++++++++---- .../localized-listing-videos/translate.py | 2 +- .../upload_footage.py | 4 +-- 3 files changed, 27 insertions(+), 9 deletions(-) diff --git a/examples/localized-listing-videos/render.py b/examples/localized-listing-videos/render.py index 897c1a1..1873f09 100644 --- a/examples/localized-listing-videos/render.py +++ b/examples/localized-listing-videos/render.py @@ -9,7 +9,7 @@ ID and text, picks up renders that were still running (or finished after the script stopped), and starts new renders for new, changed or failed languages. -Pass language codes to render only those, e.g. `python render.py ja`. +Pass language codes to render only those, e.g. `python3 render.py ja`. """ import hashlib @@ -30,6 +30,7 @@ LOG_PATH = OUTPUT_DIR / "renders.json" POLL_SECONDS = 5 TIMEOUT_SECONDS = 15 * 60 +CONNECTION_RETRIES = 6 def error_detail(resp): @@ -93,17 +94,25 @@ def wait_for(code, render_id): """Poll until the render is done or failed, printing each new status. The render keeps running on Shotstack's side if this script loses its - connection, so connection problems are retried until the timeout. + connection. Server errors are retried until the timeout. A lost network is + retried CONNECTION_RETRIES times, then the script stops and the next run + picks the render up. """ deadline = time.monotonic() + TIMEOUT_SECONDS last_status = None + connection_failures = 0 while time.monotonic() < deadline: try: render = get_render(render_id) except requests.RequestException as err: + if isinstance(err, requests.ConnectionError): + connection_failures += 1 + if connection_failures > CONNECTION_RETRIES: + raise print(f"[{code}] connection problem, retrying ({type(err).__name__})") time.sleep(POLL_SECONDS) continue + connection_failures = 0 if render["status"] != last_status: last_status = render["status"] print(f"[{code}] {last_status}") @@ -121,6 +130,8 @@ def can_resume(render_id): """True if a render from an earlier run is still running or has finished.""" try: return get_render(render_id)["status"] != "failed" + except requests.ConnectionError: + raise # the network is down; main() stops with one line except requests.RequestException: return True # no answer yet; wait_for() keeps trying except RuntimeError: @@ -191,10 +202,7 @@ def main(): try: render_id = submit(language) except requests.ConnectionError: - raise SystemExit( - "Could not reach Shotstack. Check your network connection " - "and run the script again." - ) from None + raise except (RuntimeError, requests.RequestException) as err: print(f"[{code}] {err}") failed.append(code) @@ -210,6 +218,9 @@ def main(): download(url, OUTPUT_DIR / f"{code}.mp4") log[code] = {"render_id": render_id, "fingerprint": fp, "status": "done"} print(f"[{code}] saved to output/{code}.mp4") + except requests.ConnectionError: + # Keep the "submitted" entry, so the next run picks the render up. + raise except requests.HTTPError as err: failed.append(code) print(f"[{code}] download failed: {err}") @@ -239,6 +250,13 @@ def main(): except KeyboardInterrupt: # Renders already started keep running. The next run picks them up. raise SystemExit(1) from None + except requests.ConnectionError: + print( + "Could not reach Shotstack. Check your network connection " + "and run the script again.", + file=sys.stderr, + ) + raise SystemExit(1) from None except Exception as error: print(error, file=sys.stderr) raise SystemExit(1) from error diff --git a/examples/localized-listing-videos/translate.py b/examples/localized-listing-videos/translate.py index 8cddc80..c56ed37 100644 --- a/examples/localized-listing-videos/translate.py +++ b/examples/localized-listing-videos/translate.py @@ -6,7 +6,7 @@ Fields that already have text are never changed, so a translation you have reviewed stays as it is. To translate a field again, empty it and rerun. -Pass language codes to translate only those, e.g. `python translate.py de`. +Pass language codes to translate only those, e.g. `python3 translate.py de`. """ import json diff --git a/examples/localized-listing-videos/upload_footage.py b/examples/localized-listing-videos/upload_footage.py index 2ab26b9..3ba24b9 100644 --- a/examples/localized-listing-videos/upload_footage.py +++ b/examples/localized-listing-videos/upload_footage.py @@ -1,6 +1,6 @@ """Upload a local video file to Shotstack's Ingest API and print its hosted URL. -Usage: python upload_footage.py path/to/footage.mp4 +Usage: python3 upload_footage.py path/to/footage.mp4 """ import os @@ -36,7 +36,7 @@ def main(): if not api_key: raise RuntimeError("Set the SHOTSTACK_API_KEY environment variable first.") if len(sys.argv) != 2: - raise RuntimeError("Usage: python upload_footage.py ") + raise RuntimeError("Usage: python3 upload_footage.py ") path = sys.argv[1] if not os.path.isfile(path): raise RuntimeError(f"File not found: {path}") From 024431b297bbd0307a353061a774141086dad1b9 Mon Sep 17 00:00:00 2001 From: Musab Date: Tue, 13 Oct 2026 18:43:35 +0500 Subject: [PATCH 04/11] Fix code-review findings in the localized listing example - Read Edit API errors ({message, response}) as well as gateway errors ({errors}), so a failed render or template save prints one line. - Stop before any paid render if a language has an empty field. - Stop on a submit timeout, because the render may have started. - Render again when a finished video was deleted, instead of downloading an expired URL. - Name the language and Gemini's reason when a translation is blocked or cut off. - Stop polling on the overwritten Ingest status, and do not tell the user to upload again after a timeout. --- examples/localized-listing-videos/render.py | 38 +++++++++++++++++-- .../localized-listing-videos/save_template.py | 15 ++++++-- .../localized-listing-videos/translate.py | 21 ++++++++-- .../upload_footage.py | 4 +- 4 files changed, 65 insertions(+), 13 deletions(-) diff --git a/examples/localized-listing-videos/render.py b/examples/localized-listing-videos/render.py index 1873f09..e0e66cc 100644 --- a/examples/localized-listing-videos/render.py +++ b/examples/localized-listing-videos/render.py @@ -36,10 +36,19 @@ def error_detail(resp): """Return the API's error message as one line.""" try: - errors = resp.json().get("errors") or [{}] - return errors[0].get("detail") or resp.text - except (ValueError, AttributeError): + body = resp.json() + except ValueError: + body = None + if not isinstance(body, dict): return resp.text.strip()[:200] + # A rejected key comes back as {"errors": [...]}. Other Edit API errors + # come back as {"message": ..., "response": ...}. + if body.get("errors"): + return body["errors"][0].get("detail") or resp.text.strip()[:200] + response = body.get("response") + if isinstance(response, dict): + response = response.get("error") or response.get("message") + return str(response or body.get("message") or resp.text.strip()[:200]) def merge_fields(language): @@ -177,6 +186,18 @@ def main(): if unknown: raise SystemExit(f"Not in languages.json: {', '.join(sorted(unknown))}") languages = [language for language in languages if language["code"] in only] + # Each render uses credits, so stop before any render if a field is empty. + fields = ("model", "voice", "script", "headline", "cta") + incomplete = [ + language["code"] + for language in languages + if not all(language.get(field) for field in fields) + ] + if incomplete: + raise RuntimeError( + f"Fill in {', '.join(fields)} in languages.json for: " + f"{', '.join(incomplete)}. Or run translate.py to fill them." + ) OUTPUT_DIR.mkdir(exist_ok=True) log = {} if LOG_PATH.exists(): @@ -195,7 +216,10 @@ def main(): print(f"[{code}] already rendered, skipping") continue render_id = entry.get("render_id") - if unchanged and render_id and can_resume(render_id): + # A "done" entry with no video means the file was deleted. Its temporary + # URL may have expired, so render it again. + resumable = unchanged and render_id and entry.get("status") != "done" + if resumable and can_resume(render_id): print(f"[{code}] picking up render {entry['render_id']}") pending[code] = (entry["render_id"], fp) continue @@ -203,6 +227,12 @@ def main(): render_id = submit(language) except requests.ConnectionError: raise + except requests.Timeout: + # Shotstack may have accepted the render. A new run would pay twice. + raise SystemExit( + f"[{code}] Shotstack did not answer in time, so the render may have " + "started. Check your renders in the dashboard before you run again." + ) from None except (RuntimeError, requests.RequestException) as err: print(f"[{code}] {err}") failed.append(code) diff --git a/examples/localized-listing-videos/save_template.py b/examples/localized-listing-videos/save_template.py index 0ef53c7..ace5217 100644 --- a/examples/localized-listing-videos/save_template.py +++ b/examples/localized-listing-videos/save_template.py @@ -14,10 +14,19 @@ def error_detail(resp): """Return the API's error message as one line.""" try: - errors = resp.json().get("errors") or [{}] - return errors[0].get("detail") or resp.text - except (ValueError, AttributeError): + body = resp.json() + except ValueError: + body = None + if not isinstance(body, dict): return resp.text.strip()[:200] + # A rejected key comes back as {"errors": [...]}. Other Edit API errors + # come back as {"message": ..., "response": ...}. + if body.get("errors"): + return body["errors"][0].get("detail") or resp.text.strip()[:200] + response = body.get("response") + if isinstance(response, dict): + response = response.get("error") or response.get("message") + return str(response or body.get("message") or resp.text.strip()[:200]) def main(): diff --git a/examples/localized-listing-videos/translate.py b/examples/localized-listing-videos/translate.py index c56ed37..3c6574e 100644 --- a/examples/localized-listing-videos/translate.py +++ b/examples/localized-listing-videos/translate.py @@ -74,10 +74,23 @@ def translate(source, target_code): raise RuntimeError( f"[{target_code}] translation failed ({resp.status_code}): {detail}" ) - parts = resp.json()["candidates"][0]["content"]["parts"] - text = "".join(part.get("text", "") for part in parts if not part.get("thought")) - result = json.loads(text) - return {field: result[field].strip() for field in FIELDS} + body = resp.json() + try: + parts = body["candidates"][0]["content"]["parts"] + text = "".join( + part.get("text", "") for part in parts if not part.get("thought") + ) + result = json.loads(text) + return {field: result[field].strip() for field in FIELDS} + except Exception: + # A blocked prompt has no candidates. A cut-off reply is not valid JSON. + reason = body.get("promptFeedback", {}).get("blockReason") or ( + body.get("candidates") or [{}] + )[0].get("finishReason") + raise RuntimeError( + f"[{target_code}] Gemini returned no usable translation " + f"({reason or 'unknown reason'}). Run the script again." + ) from None def main(): diff --git a/examples/localized-listing-videos/upload_footage.py b/examples/localized-listing-videos/upload_footage.py index 3ba24b9..214b081 100644 --- a/examples/localized-listing-videos/upload_footage.py +++ b/examples/localized-listing-videos/upload_footage.py @@ -63,12 +63,12 @@ def main(): if status == "ready": print(attributes["source"]) return - if status in ("failed", "deleted"): + if status in ("failed", "deleted", "overwritten"): raise RuntimeError(f"Ingest {status} for source {source_id}") time.sleep(POLL_SECONDS) raise RuntimeError( f"Source {source_id} was not ready after {TIMEOUT_SECONDS} seconds. " - "Run the script again." + f"Do not upload it again. Check it later at {API}/sources/{source_id}" ) From f6fbf73a81caa32d9b3b9adec351893f535d87df Mon Sep 17 00:00:00 2001 From: Musab Date: Tue, 13 Oct 2026 18:45:57 +0500 Subject: [PATCH 05/11] Use one error_detail in all three scripts and fix two messages upload_footage.py now reads the error the same way as render.py and save_template.py. The empty-field message says what translate.py fills. The submit-timeout message no longer names a dashboard page. --- examples/localized-listing-videos/render.py | 10 +++++----- .../localized-listing-videos/save_template.py | 2 +- .../localized-listing-videos/upload_footage.py | 15 ++++++++++++--- 3 files changed, 18 insertions(+), 9 deletions(-) diff --git a/examples/localized-listing-videos/render.py b/examples/localized-listing-videos/render.py index e0e66cc..9b3fa54 100644 --- a/examples/localized-listing-videos/render.py +++ b/examples/localized-listing-videos/render.py @@ -41,7 +41,7 @@ def error_detail(resp): body = None if not isinstance(body, dict): return resp.text.strip()[:200] - # A rejected key comes back as {"errors": [...]}. Other Edit API errors + # A rejected key comes back as {"errors": [...]}. Other Shotstack errors # come back as {"message": ..., "response": ...}. if body.get("errors"): return body["errors"][0].get("detail") or resp.text.strip()[:200] @@ -195,8 +195,8 @@ def main(): ] if incomplete: raise RuntimeError( - f"Fill in {', '.join(fields)} in languages.json for: " - f"{', '.join(incomplete)}. Or run translate.py to fill them." + f"Empty fields in languages.json for: {', '.join(incomplete)}. " + "Set model and voice. Run translate.py to fill script, headline and cta." ) OUTPUT_DIR.mkdir(exist_ok=True) log = {} @@ -230,8 +230,8 @@ def main(): except requests.Timeout: # Shotstack may have accepted the render. A new run would pay twice. raise SystemExit( - f"[{code}] Shotstack did not answer in time, so the render may have " - "started. Check your renders in the dashboard before you run again." + f"[{code}] Shotstack did not answer in time. The render may have " + "started. If you run the script again, it starts another render." ) from None except (RuntimeError, requests.RequestException) as err: print(f"[{code}] {err}") diff --git a/examples/localized-listing-videos/save_template.py b/examples/localized-listing-videos/save_template.py index ace5217..0e500f0 100644 --- a/examples/localized-listing-videos/save_template.py +++ b/examples/localized-listing-videos/save_template.py @@ -19,7 +19,7 @@ def error_detail(resp): body = None if not isinstance(body, dict): return resp.text.strip()[:200] - # A rejected key comes back as {"errors": [...]}. Other Edit API errors + # A rejected key comes back as {"errors": [...]}. Other Shotstack errors # come back as {"message": ..., "response": ...}. if body.get("errors"): return body["errors"][0].get("detail") or resp.text.strip()[:200] diff --git a/examples/localized-listing-videos/upload_footage.py b/examples/localized-listing-videos/upload_footage.py index 214b081..29f9f24 100644 --- a/examples/localized-listing-videos/upload_footage.py +++ b/examples/localized-listing-videos/upload_footage.py @@ -17,10 +17,19 @@ def error_detail(resp): """Return the API's error message as one line.""" try: - errors = resp.json().get("errors") or [{}] - return errors[0].get("detail") or resp.text - except (ValueError, AttributeError): + body = resp.json() + except ValueError: + body = None + if not isinstance(body, dict): return resp.text.strip()[:200] + # A rejected key comes back as {"errors": [...]}. Other Shotstack errors + # come back as {"message": ..., "response": ...}. + if body.get("errors"): + return body["errors"][0].get("detail") or resp.text.strip()[:200] + response = body.get("response") + if isinstance(response, dict): + response = response.get("error") or response.get("message") + return str(response or body.get("message") or resp.text.strip()[:200]) def check(resp, action): From 7552bd5fe7423064c950452c0a09b96cebbfac2b Mon Sep 17 00:00:00 2001 From: Musab Date: Tue, 13 Oct 2026 18:47:48 +0500 Subject: [PATCH 06/11] Fix the second review's findings in the localized listing example - error_detail falls back to the raw body for any error shape it cannot read. - A finished video that was deleted downloads again from the old render, as before. A render again happens only when the URL has expired. - A submit timeout no longer stops the run. The other languages still render and download. - translate.py handles a non-JSON reply, and no longer says to run again after a blocked prompt. - The empty-field check also rejects fields with only spaces. - upload_footage.py names the source ID when a status check fails, and says how to get its status after a timeout. --- examples/localized-listing-videos/render.py | 35 +++++++++---------- .../localized-listing-videos/save_template.py | 20 +++++------ .../localized-listing-videos/translate.py | 17 ++++++--- .../upload_footage.py | 26 +++++++------- 4 files changed, 50 insertions(+), 48 deletions(-) diff --git a/examples/localized-listing-videos/render.py b/examples/localized-listing-videos/render.py index 9b3fa54..bdfb525 100644 --- a/examples/localized-listing-videos/render.py +++ b/examples/localized-listing-videos/render.py @@ -35,20 +35,18 @@ def error_detail(resp): """Return the API's error message as one line.""" + # A rejected key comes back as {"errors": [...]}. Other Shotstack errors + # come back as {"message": ..., "response": ...}. try: body = resp.json() - except ValueError: - body = None - if not isinstance(body, dict): + if body.get("errors"): + return body["errors"][0]["detail"] + response = body.get("response") + if isinstance(response, dict): + response = response.get("error") or response.get("message") + return str(response or body["message"]) + except Exception: return resp.text.strip()[:200] - # A rejected key comes back as {"errors": [...]}. Other Shotstack errors - # come back as {"message": ..., "response": ...}. - if body.get("errors"): - return body["errors"][0].get("detail") or resp.text.strip()[:200] - response = body.get("response") - if isinstance(response, dict): - response = response.get("error") or response.get("message") - return str(response or body.get("message") or resp.text.strip()[:200]) def merge_fields(language): @@ -191,7 +189,7 @@ def main(): incomplete = [ language["code"] for language in languages - if not all(language.get(field) for field in fields) + if not all(str(language.get(field, "")).strip() for field in fields) ] if incomplete: raise RuntimeError( @@ -216,10 +214,7 @@ def main(): print(f"[{code}] already rendered, skipping") continue render_id = entry.get("render_id") - # A "done" entry with no video means the file was deleted. Its temporary - # URL may have expired, so render it again. - resumable = unchanged and render_id and entry.get("status") != "done" - if resumable and can_resume(render_id): + if unchanged and render_id and can_resume(render_id): print(f"[{code}] picking up render {entry['render_id']}") pending[code] = (entry["render_id"], fp) continue @@ -228,11 +223,13 @@ def main(): except requests.ConnectionError: raise except requests.Timeout: - # Shotstack may have accepted the render. A new run would pay twice. - raise SystemExit( + # Shotstack may have accepted the render, but there is no ID to log. + print( f"[{code}] Shotstack did not answer in time. The render may have " "started. If you run the script again, it starts another render." - ) from None + ) + failed.append(code) + continue except (RuntimeError, requests.RequestException) as err: print(f"[{code}] {err}") failed.append(code) diff --git a/examples/localized-listing-videos/save_template.py b/examples/localized-listing-videos/save_template.py index 0e500f0..e779bbb 100644 --- a/examples/localized-listing-videos/save_template.py +++ b/examples/localized-listing-videos/save_template.py @@ -13,20 +13,18 @@ def error_detail(resp): """Return the API's error message as one line.""" + # A rejected key comes back as {"errors": [...]}. Other Shotstack errors + # come back as {"message": ..., "response": ...}. try: body = resp.json() - except ValueError: - body = None - if not isinstance(body, dict): + if body.get("errors"): + return body["errors"][0]["detail"] + response = body.get("response") + if isinstance(response, dict): + response = response.get("error") or response.get("message") + return str(response or body["message"]) + except Exception: return resp.text.strip()[:200] - # A rejected key comes back as {"errors": [...]}. Other Shotstack errors - # come back as {"message": ..., "response": ...}. - if body.get("errors"): - return body["errors"][0].get("detail") or resp.text.strip()[:200] - response = body.get("response") - if isinstance(response, dict): - response = response.get("error") or response.get("message") - return str(response or body.get("message") or resp.text.strip()[:200]) def main(): diff --git a/examples/localized-listing-videos/translate.py b/examples/localized-listing-videos/translate.py index 3c6574e..e3e9004 100644 --- a/examples/localized-listing-videos/translate.py +++ b/examples/localized-listing-videos/translate.py @@ -74,8 +74,8 @@ def translate(source, target_code): raise RuntimeError( f"[{target_code}] translation failed ({resp.status_code}): {detail}" ) - body = resp.json() try: + body = resp.json() parts = body["candidates"][0]["content"]["parts"] text = "".join( part.get("text", "") for part in parts if not part.get("thought") @@ -84,15 +84,22 @@ def translate(source, target_code): return {field: result[field].strip() for field in FIELDS} except Exception: # A blocked prompt has no candidates. A cut-off reply is not valid JSON. - reason = body.get("promptFeedback", {}).get("blockReason") or ( - body.get("candidates") or [{}] - )[0].get("finishReason") raise RuntimeError( f"[{target_code}] Gemini returned no usable translation " - f"({reason or 'unknown reason'}). Run the script again." + f"({finish_reason(resp)})." ) from None +def finish_reason(resp): + """Return why Gemini gave no usable reply, such as SAFETY or MAX_TOKENS.""" + try: + body = resp.json() + feedback = body.get("promptFeedback") or {} + return feedback.get("blockReason") or body["candidates"][0]["finishReason"] + except Exception: + return "unknown reason" + + def main(): if not API_KEY: raise RuntimeError(f"Set the {API_KEY_NAME} environment variable first.") diff --git a/examples/localized-listing-videos/upload_footage.py b/examples/localized-listing-videos/upload_footage.py index 29f9f24..a204361 100644 --- a/examples/localized-listing-videos/upload_footage.py +++ b/examples/localized-listing-videos/upload_footage.py @@ -16,20 +16,18 @@ def error_detail(resp): """Return the API's error message as one line.""" + # A rejected key comes back as {"errors": [...]}. Other Shotstack errors + # come back as {"message": ..., "response": ...}. try: body = resp.json() - except ValueError: - body = None - if not isinstance(body, dict): + if body.get("errors"): + return body["errors"][0]["detail"] + response = body.get("response") + if isinstance(response, dict): + response = response.get("error") or response.get("message") + return str(response or body["message"]) + except Exception: return resp.text.strip()[:200] - # A rejected key comes back as {"errors": [...]}. Other Shotstack errors - # come back as {"message": ..., "response": ...}. - if body.get("errors"): - return body["errors"][0].get("detail") or resp.text.strip()[:200] - response = body.get("response") - if isinstance(response, dict): - response = response.get("error") or response.get("message") - return str(response or body.get("message") or resp.text.strip()[:200]) def check(resp, action): @@ -67,7 +65,8 @@ def main(): response = requests.get( f"{API}/sources/{source_id}", headers=headers, timeout=30 ) - attributes = check(response, "check the upload").json()["data"]["attributes"] + action = f"check source {source_id}" + attributes = check(response, action).json()["data"]["attributes"] status = attributes["status"] if status == "ready": print(attributes["source"]) @@ -77,7 +76,8 @@ def main(): time.sleep(POLL_SECONDS) raise RuntimeError( f"Source {source_id} was not ready after {TIMEOUT_SECONDS} seconds. " - f"Do not upload it again. Check it later at {API}/sources/{source_id}" + "Do not upload the file again. " + f"Get its status later with GET {API}/sources/{source_id}" ) From b6a8d30d9500679b7231aec13cafcd04fcf048bb Mon Sep 17 00:00:00 2001 From: Musab Date: Tue, 13 Oct 2026 19:27:31 +0500 Subject: [PATCH 07/11] Say who the localized listing example is for --- examples/localized-listing-videos/README.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/examples/localized-listing-videos/README.md b/examples/localized-listing-videos/README.md index cc8ee61..ccc27e4 100644 --- a/examples/localized-listing-videos/README.md +++ b/examples/localized-listing-videos/README.md @@ -1,7 +1,8 @@ # Localized listing videos Render one property listing video in English, Spanish, German and Japanese -from one saved template. Each language gets its own AI voice-over, captions, +from one saved template. Use it for listings that you sell to buyers in other +countries. Each language gets its own AI voice-over, captions, headline and call to action. You get one MP4 file for each language in `output/`. From 07b872890eea72b61fd4ab512d1086cb082f72a0 Mon Sep 17 00:00:00 2001 From: Musab Date: Tue, 13 Oct 2026 19:40:18 +0500 Subject: [PATCH 08/11] Apply points from earlier cookbook reviews - Stop with one line when languages.json or renders.json is not valid JSON, or when an entry has no code. - Stop with one line when a finished render has no video URL. - Require requests 2.18, which made Response a context manager. - Write one instruction in each README sentence. --- examples/localized-listing-videos/README.md | 7 +++--- examples/localized-listing-videos/render.py | 23 +++++++++++++++++-- .../localized-listing-videos/translate.py | 19 ++++++++++++++- 3 files changed, 43 insertions(+), 6 deletions(-) diff --git a/examples/localized-listing-videos/README.md b/examples/localized-listing-videos/README.md index ccc27e4..3ee4fd8 100644 --- a/examples/localized-listing-videos/README.md +++ b/examples/localized-listing-videos/README.md @@ -12,7 +12,7 @@ Related guide: [Video localization API: render one template in multiple language - A [Shotstack account](https://dashboard.shotstack.io/register) and your **production** API key -- Python 3.8 or later, with `requests` 2 or later +- Python 3.8 or later, with `requests` 2.18 or later - A video that is longer than your longest voice-over, as a file or at a public URL - Optional: a [Gemini API key](https://aistudio.google.com/apikey) for @@ -67,8 +67,9 @@ python3 render.py To translate with a language model, use `translate.py`. It reads the first entry in `languages.json` as the source language. For each other language, -keep `code`, `model` and `voice`, and make `script`, `headline` and `cta` -empty. Set `GEMINI_API_KEY` in `.env`, load `.env` again, and run: +keep `code`, `model` and `voice`. Make `script`, `headline` and `cta` empty. + +Set `GEMINI_API_KEY` in `.env`. Load `.env` again. Then run: ```bash python3 translate.py diff --git a/examples/localized-listing-videos/render.py b/examples/localized-listing-videos/render.py index bdfb525..6c67fb1 100644 --- a/examples/localized-listing-videos/render.py +++ b/examples/localized-listing-videos/render.py @@ -49,6 +49,23 @@ def error_detail(resp): return resp.text.strip()[:200] +def read_json(path): + """Read a JSON file, or stop with one line that names the file.""" + try: + return json.loads(path.read_text(encoding="utf-8")) + except ValueError as err: + raise RuntimeError(f"{path.name} is not valid JSON: {err}") from None + + +def read_languages(path): + """Read languages.json and check that each entry has a language code.""" + languages = read_json(path) + for number, language in enumerate(languages, start=1): + if not isinstance(language, dict) or not language.get("code"): + raise RuntimeError(f"Entry {number} in {path.name} has no code.") + return languages + + def merge_fields(language): """Map one languages.json entry to the template's merge fields.""" return [ @@ -124,6 +141,8 @@ def wait_for(code, render_id): last_status = render["status"] print(f"[{code}] {last_status}") if last_status == "done": + if not render.get("url"): + raise RuntimeError("render is done but has no video URL") return render["url"] if last_status == "failed": raise RuntimeError(f"render failed: {render.get('error')}") @@ -177,7 +196,7 @@ def main(): "Set the SHOTSTACK_TEMPLATE_ID environment variable first. " "Run save_template.py to get the ID." ) - languages = json.loads((HERE / "languages.json").read_text(encoding="utf-8")) + languages = read_languages(HERE / "languages.json") only = set(sys.argv[1:]) if only: unknown = only - {language["code"] for language in languages} @@ -199,7 +218,7 @@ def main(): OUTPUT_DIR.mkdir(exist_ok=True) log = {} if LOG_PATH.exists(): - log = json.loads(LOG_PATH.read_text(encoding="utf-8")) + log = read_json(LOG_PATH) # Start (or pick up) every render first so they run at the same time. pending = {} diff --git a/examples/localized-listing-videos/translate.py b/examples/localized-listing-videos/translate.py index e3e9004..504bfb5 100644 --- a/examples/localized-listing-videos/translate.py +++ b/examples/localized-listing-videos/translate.py @@ -45,6 +45,23 @@ """ +def read_json(path): + """Read a JSON file, or stop with one line that names the file.""" + try: + return json.loads(path.read_text(encoding="utf-8")) + except ValueError as err: + raise RuntimeError(f"{path.name} is not valid JSON: {err}") from None + + +def read_languages(path): + """Read languages.json and check that each entry has a language code.""" + languages = read_json(path) + for number, language in enumerate(languages, start=1): + if not isinstance(language, dict) or not language.get("code"): + raise RuntimeError(f"Entry {number} in {path.name} has no code.") + return languages + + def translate(source, target_code): """Translate the source entry's script, headline and cta into one language.""" prompt = INSTRUCTIONS.format(source=source["code"], target=target_code) @@ -103,7 +120,7 @@ def finish_reason(resp): def main(): if not API_KEY: raise RuntimeError(f"Set the {API_KEY_NAME} environment variable first.") - languages = json.loads(LANGUAGES_PATH.read_text(encoding="utf-8")) + languages = read_languages(LANGUAGES_PATH) source, targets = languages[0], languages[1:] if not all(source.get(field) for field in FIELDS): raise RuntimeError( From fd61b538d8b3ceb4058430485ed8aff0678237c6 Mon Sep 17 00:00:00 2001 From: Musab Date: Tue, 13 Oct 2026 19:44:56 +0500 Subject: [PATCH 09/11] Fix the third review's findings in the localized listing example - A submit with no answer is reported apart from failed renders. The final line says a new run renders it again and uses credits again. - A null field counts as empty. translate.py uses the same empty test as render.py, so a field with only spaces gets translated. - A done render with no URL is rendered again on the next run. - error_detail falls back to the raw body when the message is empty. - languages.json must be a list with at least one language. - A Gemini reply with no usable translation says what to do next. --- examples/localized-listing-videos/render.py | 42 ++++++++++++------- .../localized-listing-videos/save_template.py | 14 ++++--- .../localized-listing-videos/translate.py | 14 +++++-- .../upload_footage.py | 14 ++++--- 4 files changed, 55 insertions(+), 29 deletions(-) diff --git a/examples/localized-listing-videos/render.py b/examples/localized-listing-videos/render.py index 6c67fb1..9c194eb 100644 --- a/examples/localized-listing-videos/render.py +++ b/examples/localized-listing-videos/render.py @@ -40,13 +40,15 @@ def error_detail(resp): try: body = resp.json() if body.get("errors"): - return body["errors"][0]["detail"] - response = body.get("response") - if isinstance(response, dict): - response = response.get("error") or response.get("message") - return str(response or body["message"]) + detail = body["errors"][0].get("detail") + else: + detail = body.get("response") + if isinstance(detail, dict): + detail = detail.get("error") or detail.get("message") + detail = detail or body.get("message") except Exception: - return resp.text.strip()[:200] + detail = None + return str(detail) if detail else resp.text.strip()[:200] def read_json(path): @@ -60,6 +62,8 @@ def read_json(path): def read_languages(path): """Read languages.json and check that each entry has a language code.""" languages = read_json(path) + if not isinstance(languages, list) or not languages: + raise RuntimeError(f"{path.name} must be a list with at least one language.") for number, language in enumerate(languages, start=1): if not isinstance(language, dict) or not language.get("code"): raise RuntimeError(f"Entry {number} in {path.name} has no code.") @@ -155,7 +159,11 @@ def wait_for(code, render_id): def can_resume(render_id): """True if a render from an earlier run is still running or has finished.""" try: - return get_render(render_id)["status"] != "failed" + render = get_render(render_id) + # A "done" render with no URL cannot be downloaded, so render it again. + return render["status"] != "failed" and ( + render["status"] != "done" or bool(render.get("url")) + ) except requests.ConnectionError: raise # the network is down; main() stops with one line except requests.RequestException: @@ -208,7 +216,7 @@ def main(): incomplete = [ language["code"] for language in languages - if not all(str(language.get(field, "")).strip() for field in fields) + if not all(str(language.get(field) or "").strip() for field in fields) ] if incomplete: raise RuntimeError( @@ -223,6 +231,7 @@ def main(): # Start (or pick up) every render first so they run at the same time. pending = {} failed = [] + no_answer = [] for language in languages: code = language["code"] fp = fingerprint(language) @@ -243,11 +252,8 @@ def main(): raise except requests.Timeout: # Shotstack may have accepted the render, but there is no ID to log. - print( - f"[{code}] Shotstack did not answer in time. The render may have " - "started. If you run the script again, it starts another render." - ) - failed.append(code) + print(f"[{code}] Shotstack did not answer in time.") + no_answer.append(code) continue except (RuntimeError, requests.RequestException) as err: print(f"[{code}] {err}") @@ -284,10 +290,18 @@ def main(): print(f"[{code}] {err}") save_log(log) + messages = [] if failed: - raise SystemExit( + messages.append( f"Not rendered: {', '.join(failed)}. Run the script again to retry them." ) + if no_answer: + messages.append( + f"No answer for: {', '.join(no_answer)}. These renders may have started. " + "If you run the script again, they render again and use credits again." + ) + if messages: + raise SystemExit(" ".join(messages)) if __name__ == "__main__": diff --git a/examples/localized-listing-videos/save_template.py b/examples/localized-listing-videos/save_template.py index e779bbb..3bd621b 100644 --- a/examples/localized-listing-videos/save_template.py +++ b/examples/localized-listing-videos/save_template.py @@ -18,13 +18,15 @@ def error_detail(resp): try: body = resp.json() if body.get("errors"): - return body["errors"][0]["detail"] - response = body.get("response") - if isinstance(response, dict): - response = response.get("error") or response.get("message") - return str(response or body["message"]) + detail = body["errors"][0].get("detail") + else: + detail = body.get("response") + if isinstance(detail, dict): + detail = detail.get("error") or detail.get("message") + detail = detail or body.get("message") except Exception: - return resp.text.strip()[:200] + detail = None + return str(detail) if detail else resp.text.strip()[:200] def main(): diff --git a/examples/localized-listing-videos/translate.py b/examples/localized-listing-videos/translate.py index 504bfb5..3a7b9cb 100644 --- a/examples/localized-listing-videos/translate.py +++ b/examples/localized-listing-videos/translate.py @@ -56,12 +56,19 @@ def read_json(path): def read_languages(path): """Read languages.json and check that each entry has a language code.""" languages = read_json(path) + if not isinstance(languages, list) or not languages: + raise RuntimeError(f"{path.name} must be a list with at least one language.") for number, language in enumerate(languages, start=1): if not isinstance(language, dict) or not language.get("code"): raise RuntimeError(f"Entry {number} in {path.name} has no code.") return languages +def filled(language, field): + """True if the field has text. render.py uses the same test.""" + return bool(str(language.get(field) or "").strip()) + + def translate(source, target_code): """Translate the source entry's script, headline and cta into one language.""" prompt = INSTRUCTIONS.format(source=source["code"], target=target_code) @@ -103,7 +110,8 @@ def translate(source, target_code): # A blocked prompt has no candidates. A cut-off reply is not valid JSON. raise RuntimeError( f"[{target_code}] Gemini returned no usable translation " - f"({finish_reason(resp)})." + f"({finish_reason(resp)}). Run the script again. If it fails again, " + "change the source text." ) from None @@ -122,7 +130,7 @@ def main(): raise RuntimeError(f"Set the {API_KEY_NAME} environment variable first.") languages = read_languages(LANGUAGES_PATH) source, targets = languages[0], languages[1:] - if not all(source.get(field) for field in FIELDS): + if not all(filled(source, field) for field in FIELDS): raise RuntimeError( f"The first entry ({source['code']}) needs a script, headline and cta." ) @@ -137,7 +145,7 @@ def main(): code = language["code"] if only and code not in only: continue - missing = [field for field in FIELDS if not language.get(field)] + missing = [field for field in FIELDS if not filled(language, field)] if not missing: print(f"[{code}] already translated, skipping") continue diff --git a/examples/localized-listing-videos/upload_footage.py b/examples/localized-listing-videos/upload_footage.py index a204361..d658118 100644 --- a/examples/localized-listing-videos/upload_footage.py +++ b/examples/localized-listing-videos/upload_footage.py @@ -21,13 +21,15 @@ def error_detail(resp): try: body = resp.json() if body.get("errors"): - return body["errors"][0]["detail"] - response = body.get("response") - if isinstance(response, dict): - response = response.get("error") or response.get("message") - return str(response or body["message"]) + detail = body["errors"][0].get("detail") + else: + detail = body.get("response") + if isinstance(detail, dict): + detail = detail.get("error") or detail.get("message") + detail = detail or body.get("message") except Exception: - return resp.text.strip()[:200] + detail = None + return str(detail) if detail else resp.text.strip()[:200] def check(resp, action): From fec853a6f6e8de00776bfdbf980315815bb1b527 Mon Sep 17 00:00:00 2001 From: Musab Date: Tue, 13 Oct 2026 19:46:56 +0500 Subject: [PATCH 10/11] Reject repeated language codes and catch every error per language - languages.json may not use one code twice. - An unexpected reply for one language no longer stops the other languages. - An empty translation from Gemini is an error, not a saved translation. - error_detail falls back to the error title, then the HTTP reason. - The README says what happens after a request gets no answer. --- examples/localized-listing-videos/README.md | 6 +++++- examples/localized-listing-videos/render.py | 14 ++++++++++---- examples/localized-listing-videos/save_template.py | 4 ++-- examples/localized-listing-videos/translate.py | 8 ++++++++ .../localized-listing-videos/upload_footage.py | 4 ++-- 5 files changed, 27 insertions(+), 9 deletions(-) diff --git a/examples/localized-listing-videos/README.md b/examples/localized-listing-videos/README.md index 3ee4fd8..69a78c5 100644 --- a/examples/localized-listing-videos/README.md +++ b/examples/localized-listing-videos/README.md @@ -87,9 +87,13 @@ prints the status of each render until it is done. It saves each video as The script writes each render ID to `output/renders.json`. When you run it again, it skips the videos that are done and not changed. It continues the -renders that did not finish. It starts new renders only for new or changed +renders that did not finish. It starts new renders for new, changed or failed languages. +Sometimes Shotstack does not answer a render request in time. The script then +names the language. The render can still start. If you run the script again, +that language renders again and uses credits again. + If you save the template again, the template gets a new ID. Then all the languages render again. diff --git a/examples/localized-listing-videos/render.py b/examples/localized-listing-videos/render.py index 9c194eb..fe04728 100644 --- a/examples/localized-listing-videos/render.py +++ b/examples/localized-listing-videos/render.py @@ -40,7 +40,7 @@ def error_detail(resp): try: body = resp.json() if body.get("errors"): - detail = body["errors"][0].get("detail") + detail = body["errors"][0].get("detail") or body["errors"][0].get("title") else: detail = body.get("response") if isinstance(detail, dict): @@ -48,7 +48,7 @@ def error_detail(resp): detail = detail or body.get("message") except Exception: detail = None - return str(detail) if detail else resp.text.strip()[:200] + return str(detail) if detail else resp.text.strip()[:200] or resp.reason def read_json(path): @@ -67,6 +67,12 @@ def read_languages(path): for number, language in enumerate(languages, start=1): if not isinstance(language, dict) or not language.get("code"): raise RuntimeError(f"Entry {number} in {path.name} has no code.") + codes = [language["code"] for language in languages] + repeated = sorted({code for code in codes if codes.count(code) > 1}) + if repeated: + raise RuntimeError( + f"Codes used more than once in {path.name}: {', '.join(repeated)}" + ) return languages @@ -255,7 +261,7 @@ def main(): print(f"[{code}] Shotstack did not answer in time.") no_answer.append(code) continue - except (RuntimeError, requests.RequestException) as err: + except Exception as err: print(f"[{code}] {err}") failed.append(code) continue @@ -284,7 +290,7 @@ def main(): # A server error is temporary: keep the render, so the next run # downloads it again without paying for a new render. log[code] = failed_entry(render_id, fp) - except (RuntimeError, requests.RequestException) as err: + except Exception as err: log[code] = failed_entry(render_id, fp) failed.append(code) print(f"[{code}] {err}") diff --git a/examples/localized-listing-videos/save_template.py b/examples/localized-listing-videos/save_template.py index 3bd621b..d0b77be 100644 --- a/examples/localized-listing-videos/save_template.py +++ b/examples/localized-listing-videos/save_template.py @@ -18,7 +18,7 @@ def error_detail(resp): try: body = resp.json() if body.get("errors"): - detail = body["errors"][0].get("detail") + detail = body["errors"][0].get("detail") or body["errors"][0].get("title") else: detail = body.get("response") if isinstance(detail, dict): @@ -26,7 +26,7 @@ def error_detail(resp): detail = detail or body.get("message") except Exception: detail = None - return str(detail) if detail else resp.text.strip()[:200] + return str(detail) if detail else resp.text.strip()[:200] or resp.reason def main(): diff --git a/examples/localized-listing-videos/translate.py b/examples/localized-listing-videos/translate.py index 3a7b9cb..10a1fca 100644 --- a/examples/localized-listing-videos/translate.py +++ b/examples/localized-listing-videos/translate.py @@ -61,6 +61,12 @@ def read_languages(path): for number, language in enumerate(languages, start=1): if not isinstance(language, dict) or not language.get("code"): raise RuntimeError(f"Entry {number} in {path.name} has no code.") + codes = [language["code"] for language in languages] + repeated = sorted({code for code in codes if codes.count(code) > 1}) + if repeated: + raise RuntimeError( + f"Codes used more than once in {path.name}: {', '.join(repeated)}" + ) return languages @@ -105,6 +111,8 @@ def translate(source, target_code): part.get("text", "") for part in parts if not part.get("thought") ) result = json.loads(text) + if not all(filled(result, field) for field in FIELDS): + raise ValueError("empty field") return {field: result[field].strip() for field in FIELDS} except Exception: # A blocked prompt has no candidates. A cut-off reply is not valid JSON. diff --git a/examples/localized-listing-videos/upload_footage.py b/examples/localized-listing-videos/upload_footage.py index d658118..f9a9d92 100644 --- a/examples/localized-listing-videos/upload_footage.py +++ b/examples/localized-listing-videos/upload_footage.py @@ -21,7 +21,7 @@ def error_detail(resp): try: body = resp.json() if body.get("errors"): - detail = body["errors"][0].get("detail") + detail = body["errors"][0].get("detail") or body["errors"][0].get("title") else: detail = body.get("response") if isinstance(detail, dict): @@ -29,7 +29,7 @@ def error_detail(resp): detail = detail or body.get("message") except Exception: detail = None - return str(detail) if detail else resp.text.strip()[:200] + return str(detail) if detail else resp.text.strip()[:200] or resp.reason def check(resp, action): From c97202d86f5617df3b76137e2774040cb3a4a3bd Mon Sep 17 00:00:00 2001 From: Musab Date: Thu, 15 Oct 2026 09:29:07 +0500 Subject: [PATCH 11/11] Keep the render when a download is rate limited A download that got HTTP 429 deleted the saved render ID, so the next run paid for a new render instead of downloading the finished one. A rate limit, a request timeout or a server error now keeps the render, and only another 4xx (an expired URL) starts a new render. The same rule is used when the render status is checked. test_render.py covers the retry: after a failed download the next run downloads the same render and does not submit again. CI runs the tests of each Python example that has a test file. --- .github/workflows/check.yml | 7 + examples/localized-listing-videos/render.py | 20 ++- .../localized-listing-videos/test_render.py | 140 ++++++++++++++++++ 3 files changed, 160 insertions(+), 7 deletions(-) create mode 100644 examples/localized-listing-videos/test_render.py diff --git a/.github/workflows/check.yml b/.github/workflows/check.yml index 13a2bc0..fd975c6 100644 --- a/.github/workflows/check.yml +++ b/.github/workflows/check.yml @@ -23,3 +23,10 @@ jobs: # Passes trivially until an example ships Python. - run: pip install ruff==0.16.1 && ruff check . + + # Runs the unit tests of each Python example that has a test_*.py file. + - run: | + pip install requests==2.34.2 + for dir in $(git ls-files 'examples/*/test_*.py' | xargs -r -n1 dirname | sort -u); do + python -m unittest discover -s "$dir" -v + done diff --git a/examples/localized-listing-videos/render.py b/examples/localized-listing-videos/render.py index fe04728..0fa6cf9 100644 --- a/examples/localized-listing-videos/render.py +++ b/examples/localized-listing-videos/render.py @@ -112,10 +112,15 @@ def submit(language): return resp.json()["response"]["id"] +def is_temporary(status_code): + """True for a rate limit, a timeout or a server error: try again later.""" + return status_code is None or status_code >= 500 or status_code in (408, 429) + + def get_render(render_id): resp = requests.get(f"{API}/render/{render_id}", headers=HEADERS, timeout=30) - if resp.status_code >= 500 or resp.status_code == 429: - # A server error or rate limit is retried like a connection problem. + if is_temporary(resp.status_code): + # A temporary error is retried like a connection problem. resp.raise_for_status() if not resp.ok: raise RuntimeError( @@ -282,14 +287,15 @@ def main(): except requests.HTTPError as err: failed.append(code) print(f"[{code}] download failed: {err}") - if err.response is not None and err.response.status_code < 500: + status = err.response.status_code if err.response is not None else None + if is_temporary(status): + # Keep the render, so the next run downloads it again without + # paying for a new render. + log[code] = failed_entry(render_id, fp) + else: # The temporary URL has probably expired, so forget this render # and start a new one next run. log.pop(code, None) - else: - # A server error is temporary: keep the render, so the next run - # downloads it again without paying for a new render. - log[code] = failed_entry(render_id, fp) except Exception as err: log[code] = failed_entry(render_id, fp) failed.append(code) diff --git a/examples/localized-listing-videos/test_render.py b/examples/localized-listing-videos/test_render.py new file mode 100644 index 0000000..77f0f75 --- /dev/null +++ b/examples/localized-listing-videos/test_render.py @@ -0,0 +1,140 @@ +"""Tests for the resume logic of render.py. They do not call Shotstack. + +Run them from this directory: python3 -m unittest +""" + +import contextlib +import io +import json +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +import requests + +import render + +LANGUAGE = { + "code": "ja", + "model": "elevenlabs-multilingual-v2", + "voice": "Rachel", + "script": "script", + "headline": "headline", + "cta": "cta", +} +URL = "https://cdn.shotstack.io/au/v1/render/ja.mp4" + + +def http_error(status_code): + """Return the error that requests raises for a response with this status. + + None makes an error with no response, as a broken connection can raise. + """ + if status_code is None: + return requests.HTTPError(f"no response for {URL}") + response = requests.Response() + response.status_code = status_code + return requests.HTTPError(f"{status_code} for {URL}", response=response) + + +def download_stub(errors): + """Fail the first downloads with the given errors, then save the video.""" + + def download(url, path): + if errors: + raise errors.pop(0) + path.write_bytes(b"video") + + return mock.Mock(side_effect=download) + + +class ResumeAfterDownloadFailure(unittest.TestCase): + """A render that is done must not be paid for twice.""" + + def setUp(self): + temp = tempfile.TemporaryDirectory() + self.addCleanup(temp.cleanup) + self.output_dir = Path(temp.name) + self.submit = mock.Mock(side_effect=["render-1", "render-2"]) + replacements = { + "API_KEY": "test-key", + "TEMPLATE_ID": "test-template", + "OUTPUT_DIR": self.output_dir, + "LOG_PATH": self.output_dir / "renders.json", + "read_languages": lambda path: [dict(LANGUAGE)], + "submit": self.submit, + "get_render": lambda render_id: {"status": "done", "url": URL}, + } + for name, value in replacements.items(): + patcher = mock.patch.object(render, name, value) + patcher.start() + self.addCleanup(patcher.stop) + argv = mock.patch.object(sys, "argv", ["render.py"]) + argv.start() + self.addCleanup(argv.stop) + + def run_main(self): + """Run main() and return its exit message, or None if it finished.""" + with contextlib.redirect_stdout(io.StringIO()): + try: + render.main() + except SystemExit as stop: + return str(stop) + return None + + def read_log(self): + return json.loads((self.output_dir / "renders.json").read_text("utf-8")) + + def assert_download_is_retried(self, status_code): + download = download_stub([http_error(status_code)]) + with mock.patch.object(render, "download", download): + first = self.run_main() + log_after_failure = self.read_log() + second = self.run_main() + + self.assertIn("Not rendered: ja", first) + self.assertEqual(log_after_failure["ja"]["render_id"], "render-1") + self.assertEqual(log_after_failure["ja"]["status"], "failed") + self.assertIsNone(second) + self.assertEqual(self.submit.call_count, 1) + video = self.output_dir / "ja.mp4" + self.assertEqual(download.call_args_list, [mock.call(URL, video)] * 2) + self.assertTrue(video.exists()) + self.assertEqual( + self.read_log()["ja"], + { + "render_id": "render-1", + "fingerprint": render.fingerprint(LANGUAGE), + "status": "done", + }, + ) + + def test_rate_limited_download_keeps_the_render(self): + self.assert_download_is_retried(429) + + def test_server_error_keeps_the_render(self): + self.assert_download_is_retried(503) + + def test_request_timeout_keeps_the_render(self): + self.assert_download_is_retried(408) + + def test_error_without_response_keeps_the_render(self): + self.assert_download_is_retried(None) + + def test_expired_url_starts_a_new_render(self): + download = download_stub([http_error(404)]) + with mock.patch.object(render, "download", download): + first = self.run_main() + self.assertNotIn("ja", self.read_log()) + second = self.run_main() + + self.assertIn("Not rendered: ja", first) + self.assertIsNone(second) + self.assertEqual(self.submit.call_count, 2) + self.assertEqual(self.read_log()["ja"]["render_id"], "render-2") + + +if __name__ == "__main__": + unittest.main()