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/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..69a78c5 --- /dev/null +++ b/examples/localized-listing-videos/README.md @@ -0,0 +1,101 @@ +# Localized listing videos + +Render one property listing video in English, Spanish, German and Japanese +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/`. + +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.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 + `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`. Make `script`, `headline` and `cta` empty. + +Set `GEMINI_API_KEY` in `.env`. Load `.env` again. Then 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 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. + +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..0fa6cf9 --- /dev/null +++ b/examples/localized-listing-videos/render.py @@ -0,0 +1,334 @@ +"""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. `python3 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 +CONNECTION_RETRIES = 6 + + +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() + if body.get("errors"): + detail = body["errors"][0].get("detail") or body["errors"][0].get("title") + 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: + detail = None + return str(detail) if detail else resp.text.strip()[:200] or resp.reason + + +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) + 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.") + 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 + + +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 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 is_temporary(resp.status_code): + # A temporary error 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. 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}") + 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')}") + 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: + 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: + 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 = read_languages(HERE / "languages.json") + 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] + # 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(str(language.get(field) or "").strip() for field in fields) + ] + if incomplete: + raise RuntimeError( + 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 = {} + if LOG_PATH.exists(): + log = read_json(LOG_PATH) + + # 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) + 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 requests.ConnectionError: + 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.") + no_answer.append(code) + continue + except Exception 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.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}") + 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) + except Exception as err: + log[code] = failed_entry(render_id, fp) + failed.append(code) + print(f"[{code}] {err}") + save_log(log) + + messages = [] + if failed: + 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__": + try: + 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/save_template.py b/examples/localized-listing-videos/save_template.py new file mode 100644 index 0000000..d0b77be --- /dev/null +++ b/examples/localized-listing-videos/save_template.py @@ -0,0 +1,72 @@ +"""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.""" + # A rejected key comes back as {"errors": [...]}. Other Shotstack errors + # come back as {"message": ..., "response": ...}. + try: + body = resp.json() + if body.get("errors"): + detail = body["errors"][0].get("detail") or body["errors"][0].get("title") + 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: + detail = None + return str(detail) if detail else resp.text.strip()[:200] or resp.reason + + +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 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/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/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() diff --git a/examples/localized-listing-videos/translate.py b/examples/localized-listing-videos/translate.py new file mode 100644 index 0000000..10a1fca --- /dev/null +++ b/examples/localized-listing-videos/translate.py @@ -0,0 +1,183 @@ +"""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. `python3 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 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) + 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.") + 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 + + +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) + 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}" + ) + try: + body = resp.json() + parts = body["candidates"][0]["content"]["parts"] + text = "".join( + 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. + raise RuntimeError( + f"[{target_code}] Gemini returned no usable translation " + f"({finish_reason(resp)}). Run the script again. If it fails again, " + "change the source text." + ) 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.") + languages = read_languages(LANGUAGES_PATH) + source, targets = languages[0], languages[1:] + if not all(filled(source, 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 filled(language, 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 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 new file mode 100644 index 0000000..f9a9d92 --- /dev/null +++ b/examples/localized-listing-videos/upload_footage.py @@ -0,0 +1,100 @@ +"""Upload a local video file to Shotstack's Ingest API and print its hosted URL. + +Usage: python3 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.""" + # A rejected key comes back as {"errors": [...]}. Other Shotstack errors + # come back as {"message": ..., "response": ...}. + try: + body = resp.json() + if body.get("errors"): + detail = body["errors"][0].get("detail") or body["errors"][0].get("title") + 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: + detail = None + return str(detail) if detail else resp.text.strip()[:200] or resp.reason + + +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: python3 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 + ) + action = f"check source {source_id}" + attributes = check(response, action).json()["data"]["attributes"] + status = attributes["status"] + if status == "ready": + print(attributes["source"]) + return + 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. " + "Do not upload the file again. " + f"Get its status later with GET {API}/sources/{source_id}" + ) + + +if __name__ == "__main__": + try: + 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