#!/usr/bin/env python3
import argparse
import json
import mimetypes
import os
import pathlib
import sys
import time
import hashlib
import traceback
from datetime import datetime, UTC
import boto3
import requests
from boto3.s3.transfer import TransferConfig
ALLOWED_VIDEO_MIMES = {
"video/mp4",
"video/webm",
"video/quicktime", # mov
}
def now():
return time.strftime("%Y-%m-%d %H:%M:%S")
def log(msg: str):
print(f"[{now()}] {msg}", flush=True)
def parse_args():
p = argparse.ArgumentParser(description="Dub videos with Synthesia API (resumable, file or folder; S3 multipart).")
p.add_argument("--api-key", required=True, help="Synthesia API key (paste directly).")
p.add_argument("--path", dest="input_path", required=True,
help="Absolute path to a source video file OR a directory containing videos.")
p.add_argument("--targets", required=True, help="Comma-separated target languages (e.g. fr-FR,es-ES).")
p.add_argument("--source-language", default="en", help="Original language code (default: en).")
p.add_argument("--title", default=None, help="Optional title prefix for the dubbed videos. Defaults to file name.")
p.add_argument("--state-path", default=os.path.expanduser("~/.synthesia_dub_state.json"),
help="Where to store resumable job state (default: ~/.synthesia_dub_state.json).")
p.add_argument("--force-new", action="store_true",
help="Ignore any saved state for jobs and start fresh (creates new asset(s)).")
p.add_argument("--poll-interval", type=int, default=10, help="Seconds between polling attempts (default: 10).")
p.add_argument("--recursive", dest="recursive", action="store_true", default=True,
help="Recurse into subfolders when PATH is a directory (default: on).")
p.add_argument("--no-recursive", dest="recursive", action="store_false",
help="Do NOT recurse when PATH is a directory.")
return p.parse_args()
# --------------------------
# State helpers (resumable)
# --------------------------
def load_state(state_path: str) -> dict:
# Ensure parent exists; create empty file if missing so users can see it
state_path = os.path.abspath(os.path.expanduser(state_path))
pathlib.Path(state_path).parent.mkdir(parents=True, exist_ok=True)
if not os.path.exists(state_path):
with open(state_path, "w", encoding="utf-8") as f:
json.dump({}, f)
log(f"State file created: {state_path}")
return {}
try:
with open(state_path, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
# Corrupted state → back it up and start clean
bak = state_path + ".bak"
try:
os.replace(state_path, bak)
log(f"State file unreadable; backed up to {bak} and reinitialized.")
except Exception:
log("State file unreadable and backup failed; reinitializing.")
with open(state_path, "w", encoding="utf-8") as f:
json.dump({}, f)
return {}
def save_state(state_path: str, state: dict):
state_path = os.path.abspath(os.path.expanduser(state_path))
pathlib.Path(state_path).parent.mkdir(parents=True, exist_ok=True)
tmp = state_path + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(state, f, indent=2)
os.replace(tmp, state_path)
def job_key(source_path: str, src_lang: str, targets: list, title: str | None) -> str:
# Stable identity for this dubbing request
h = hashlib.sha256()
h.update(os.path.abspath(source_path).encode("utf-8"))
h.update(b"|")
h.update(src_lang.encode("utf-8"))
h.update(b"|")
h.update(",".join(sorted(targets)).encode("utf-8"))
h.update(b"|")
h.update((title or "").encode("utf-8"))
return h.hexdigest()
# --------------------------
# API helpers
# --------------------------
def get_asset(api_key: str, asset_id: str) -> dict:
r = requests.get(
f"https://api.synthesia.io/v2/assets/{asset_id}",
headers={"Authorization": api_key, "accept": "application/json"},
timeout=30,
)
if r.status_code != 200:
raise RuntimeError(f"[Get asset] {r.status_code}: {r.text}")
return r.json()
def create_asset(api_key: str, source_path: str, title: str) -> dict:
content_type = mimetypes.guess_type(source_path)[0] or "video/mp4"
payload = {
"contentType": content_type,
"configuration": {"name": "dubbing", "detectLanguage": True},
"title": title or pathlib.Path(source_path).name,
}
log(f"Creating Synthesia asset for {os.path.basename(source_path)} …")
r = requests.post(
"https://api.synthesia.io/v2/assets",
headers={"Authorization": api_key, "accept": "application/json"},
json=payload,
timeout=60,
)
if r.status_code != 201:
raise RuntimeError(f"[Create asset] {r.status_code}: {r.text}")
data = r.json()
log(f"Asset created: id={data.get('id')}")
return data
def upload_to_s3(upload_credentials: dict, source_path: str, content_type: str):
s3 = boto3.client(
"s3",
aws_access_key_id=upload_credentials["accessKeyId"],
aws_secret_access_key=upload_credentials["secretAccessKey"],
aws_session_token=upload_credentials["sessionToken"],
region_name="eu-west-1",
)
cfg = TransferConfig(
multipart_threshold=8 * 1024 * 1024,
multipart_chunksize=32 * 1024 * 1024,
max_concurrency=6,
use_threads=True,
)
bucket, key = upload_credentials["bucket"], upload_credentials["key"]
log(f"Uploading {source_path} → s3://{bucket}/{key} …")
s3.upload_file(source_path, bucket, key, ExtraArgs={"ContentType": content_type}, Config=cfg)
log("Upload complete ✅")
def start_dubbing(api_key: str, asset_id: str, src_lang: str, targets: list, title: str) -> str:
payload = {
"sourceLanguage": src_lang,
"targetLanguages": targets,
"lipsyncEnabled": False,
"burnedInCaptionsEnabled": False,
"videoDuration": "adaptive",
"visibility": "private",
"title": f"{title} (dubbed)" if title else "Dubbed video",
"sourceAssetId": asset_id,
}
log(f"Starting dubbing job for {len(targets)} target(s)…")
r = requests.post(
"https://api.synthesia.io/v2/dubbing",
headers={
"Authorization": api_key,
"accept": "application/json",
"content-type": "application/json",
},
json=payload,
timeout=60,
)
if r.status_code != 202:
raise RuntimeError(f"[Start dubbing] {r.status_code}: {r.text}")
data = r.json()
imported_id = data["createdImportedAsset"]["id"]
log(f"Dubbing started: imported_id={imported_id}")
return imported_id
def poll_and_download(api_key: str, imported_id: str, targets: list, source_path: str, poll_interval: int = 10):
def fetch_status():
rr = requests.get(
f"https://api.synthesia.io/v2/dubbing/{imported_id}",
params={"targetLanguages": ",".join(targets)},
headers={"Authorization": api_key, "accept": "application/json"},
timeout=30,
)
rr.raise_for_status()
return rr.json()
def iter_assets(data):
# Normalize payload → list of {language, downloadUrl}
entries = data.get("dubbedAssets") or data.get("dubbedVideos") or data.get("dubbed") or []
norm = []
for dv in entries:
lang = dv.get("language") or dv.get("locale") or "xx"
url = dv.get("download") or dv.get("downloadUrl") or dv.get("url")
if url:
norm.append((lang, url))
return norm
log(f"Polling dubbing status every {poll_interval}s… (Ctrl-C to stop; you can resume later)")
while True:
data = fetch_status()
status_raw = data.get("status", "")
status = str(status_raw).strip().lower()
log(f"Status: {status_raw}")
if status in ("complete", "completed"):
assets = iter_assets(data)
if not assets:
log(f"Complete but no downloadable entries. Keys: {list(data.keys())}")
return
out_dir = os.path.dirname(source_path)
base_name = pathlib.Path(source_path).stem
# Per-language download with refresh-on-403
for lang, url in assets:
out_path = os.path.join(out_dir, f"{base_name}.dubbed-{lang}.mp4")
attempt = 0
while True:
attempt += 1
try:
log(f"Downloading {lang} (attempt {attempt}) → {out_path}")
with requests.get(url, stream=True, timeout=60) as r:
if r.status_code in (403, 404) and attempt <= 3:
log(f"Received {r.status_code} for {lang}; refreshing signed URL…")
time.sleep(0.8)
# Refresh URLs and pick the matching language again
new_data = fetch_status()
new_assets = dict(iter_assets(new_data))
if lang in new_assets:
url = new_assets[lang]
continue
r.raise_for_status()
with open(out_path, "wb") as f:
for chunk in r.iter_content(chunk_size=1024 * 128):
if chunk:
f.write(chunk)
log(f"Saved: {out_path}")
break
except requests.HTTPError as e:
if attempt >= 3:
raise
log(f"HTTP error on {lang}: {e}. Retrying…")
time.sleep(1.2)
log("All downloads complete ✅")
return
if status == "error":
raise RuntimeError(json.dumps(data, indent=2))
time.sleep(poll_interval)
# --------------------------
# File discovery
# --------------------------
def is_allowed_video(path: str) -> bool:
mime, _ = mimetypes.guess_type(path)
return (mime in ALLOWED_VIDEO_MIMES)
def discover_inputs(input_path: str, recursive: bool) -> list[str]:
p = pathlib.Path(input_path)
if p.is_file():
return [str(p)]
if not p.is_dir():
raise FileNotFoundError(f"Path not found: {input_path}")
files = []
if recursive:
it = p.rglob("*")
else:
it = p.glob("*")
for f in it:
if f.is_file() and is_allowed_video(str(f)):
files.append(str(f))
return files
# --------------------------
# Orchestrate a single file
# --------------------------
def process_one_file(
api_key: str,
source_path: str,
src_lang: str,
targets: list[str],
title_prefix: str | None,
state: dict,
state_path: str,
force_new: bool,
poll_interval: int,
):
if not os.path.isfile(source_path):
raise FileNotFoundError(f"File not found: {source_path}")
content_type = mimetypes.guess_type(source_path)[0] or "video/mp4"
base_title = pathlib.Path(source_path).stem
title = (title_prefix + " - " + base_title) if title_prefix else base_title
key = job_key(source_path, src_lang, targets, title)
job = None if force_new else state.get(key)
# Resume path
if job:
log(f"[{os.path.basename(source_path)}] Found existing job state; attempting to resume.")
asset_id = job.get("asset_id")
imported_id = job.get("imported_id")
if imported_id:
log(f"[{os.path.basename(source_path)}] Resuming from imported_id={imported_id}")
poll_and_download(api_key, imported_id, targets, source_path, poll_interval=poll_interval)
state.pop(key, None)
save_state(state_path, state)
return
if asset_id:
log(f"[{os.path.basename(source_path)}] Validating existing asset_id={asset_id}")
try:
_ = get_asset(api_key, asset_id)
except Exception as e:
log(f"[{os.path.basename(source_path)}] Existing asset check failed: {e}. Creating a new asset.")
job = None
else:
job = None
else:
log(f"[{os.path.basename(source_path)}] No prior job state found or --force-new; starting fresh.")
# Fresh path
if not job:
asset = create_asset(api_key, source_path, title)
asset_id = asset["id"]
state[key] = {
"asset_id": asset_id,
"imported_id": None,
"targets": targets,
"title": title,
"source_path": os.path.abspath(source_path),
"created_at": datetime.now(UTC).isoformat(),
}
save_state(state_path, state)
upload_credentials = asset.get("uploadCredentials")
if not upload_credentials:
raise RuntimeError("Missing uploadCredentials in asset response (required for S3 multipart upload).")
try:
upload_to_s3(upload_credentials, source_path, content_type)
except Exception:
log(f"[{os.path.basename(source_path)}] Upload failed. If credentials expired, re-run to create a fresh asset.")
raise
# Start dubbing if needed
job = state.get(key)
asset_id = job["asset_id"]
if not job.get("imported_id"):
imported_id = start_dubbing(api_key, asset_id, src_lang, targets, title)
job["imported_id"] = imported_id
job["started_dubbing_at"] = datetime.now(UTC).isoformat()
state[key] = job
save_state(state_path, state)
else:
imported_id = job["imported_id"]
log(f"[{os.path.basename(source_path)}] Resuming polling for imported_id={imported_id}")
# Poll & download
poll_and_download(api_key, imported_id, targets, source_path, poll_interval=poll_interval)
# Cleanup
state.pop(key, None)
save_state(state_path, state)
# --------------------------
# Main
# --------------------------
def main():
args = parse_args()
try:
# Parse + normalize
targets = [t.strip() for t in args.targets.split(",") if t.strip()]
if not targets:
raise ValueError("No valid target languages provided.")
# Discover input files
files = discover_inputs(args.input_path, recursive=args.recursive)
if not files:
raise FileNotFoundError("No matching video files found. "
"Allowed types: video/mp4, video/webm, video/quicktime.")
log(f"Discovered {len(files)} file(s).")
# Load state once
state = load_state(args.state_path)
successes, failures = 0, 0
# Process sequentially
for idx, f in enumerate(files, 1):
log(f"----- [{idx}/{len(files)}] {f} -----")
try:
process_one_file(
api_key=args.api_key,
source_path=f,
src_lang=args.source_language,
targets=targets,
title_prefix=args.title,
state=state,
state_path=args.state_path,
force_new=args.force_new,
poll_interval=args.poll_interval,
)
successes += 1
except KeyboardInterrupt:
log("Interrupted by user. Progress saved; re-run to resume remaining files.")
sys.exit(130)
except Exception as e:
failures += 1
log(f"❌ ERROR on {f}: {e}")
traceback.print_exc()
# Summary
log(f"Done. Success: {successes}, Failed: {failures}")
# Exit code reflects failures (useful for CI)
sys.exit(0 if failures == 0 else 1)
except Exception as e:
log(f"❌ ERROR: {e}")
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
main()
