Personalized Video Messaging

name,pronunciation
Maya,
Siobhan,shivawn
Liam,

#!/usr/bin/env python3
"""Personalize one video for every name in a CSV with dialogue edits and one batch.
Usage:
  export SYNC_API_KEY=your-api-key
  python3 personalize_batch.py --video-url URL --csv names.csv --word SaaS > results.tsv
The CSV needs a `name` column and 20 to 500 rows. An optional `pronunciation`
column holds a respelling the voice speaks instead of the name. Set
SYNC_API_BASE to use another API host. Python 3.8+, standard library only.
"""
import argparse
import csv
import json
import os
import re
import sys
import time
import urllib.error
import urllib.request
BASE = os.environ.get("SYNC_API_BASE", "https://api.sync.so").rstrip("/")
API_KEY = os.environ.get("SYNC_API_KEY", "")
PREVIEW_GAP_SECONDS = 2.1  # previews allow 30 requests per minute
MIN_ROWS, MAX_ROWS = 20, 500
STARTED = time.time()
class Timeout(Exception):
    pass
class ApiError(Exception):
    def __init__(self, status, body):
        self.status = status
        self.body = body
        super().__init__(
            f"HTTP {status} {body.get('errorCode', '')}: {body.get('message', body)}"
        )
def log(message):
    print(f"[{time.time() - STARTED:4.0f}s] {message}", file=sys.stderr, flush=True)
def request(method, path, body=None, retry_statuses=(429,), retry_network=False):
    """Sends one API call and returns its JSON, waiting out the given statuses."""
    for attempt in range(6):
        req = urllib.request.Request(
            BASE + path,
            method=method,
            data=None if body is None else json.dumps(body).encode(),
            headers={
                "x-api-key": API_KEY,
                "Content-Type": "application/json",
                "User-Agent": "sync-personalize-batch-example/1.0",
            },
        )
        try:
            with urllib.request.urlopen(req, timeout=120) as res:
                return json.load(res)
        except urllib.error.HTTPError as err:
            raw = err.read().decode(errors="replace")
            try:
                detail = json.loads(raw)
            except ValueError:
                detail = {"message": raw[:300]}
            if err.code not in retry_statuses or attempt == 5:
                raise ApiError(err.code, detail)
            retry_after = err.headers.get("Retry-After", "")
            wait = float(retry_after) if retry_after.isdigit() else 5 * 2**attempt
        except OSError as err:
            # A POST that timed out may still have been accepted, so only reads retry.
            if not retry_network or attempt == 5:
                raise
            wait = 5 * 2**attempt
            log(f"  network error ({err}), retrying in {wait:.0f}s")
        time.sleep(wait)
def get(path):
    return request("GET", path, retry_statuses=(429, 500, 502, 503, 504), retry_network=True)
def post(path, body):
    return request("POST", path, body)
def wait_for(path, done, deadline, interval=5):
    while True:
        job = get(path)
        if job["status"] in done:
            return job
        if time.monotonic() > deadline:
            raise Timeout(f"{job.get('id')} is still {job['status']}")
        time.sleep(interval)
def read_recipients(path):
    with open(path, newline="", encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)
        if "name" not in (reader.fieldnames or []):
            sys.exit(f"{path} needs a 'name' column")
        recipients = []
        for row in reader:
            name = (row.get("name") or "").strip()
            if not name:
                sys.exit(f"{path} line {reader.line_num}: 'name' is empty")
            pronunciation = (row.get("pronunciation") or "").strip()
            recipients.append(
                {"line": reader.line_num, "name": name, "pronunciation": pronunciation}
            )
    if not MIN_ROWS <= len(recipients) <= MAX_ROWS:
        sys.exit(f"A batch needs {MIN_ROWS} to {MAX_ROWS} rows; {path} has {len(recipients)}")
    return recipients
def normalize(text):
    return re.sub(r"\W", "", text).lower()
def transcribe(video_url, deadline):
    log("Transcribing the video")
    job = post("/v2/transcriptions", {"sourceVideoUrl": video_url})
    try:
        job = wait_for(f"/v2/transcriptions/{job['id']}", {"COMPLETED", "FAILED"}, deadline)
    except Timeout as err:
        sys.exit(f"Transcription {err}; rerun later, the same request reuses this job")
    if job["status"] == "FAILED":
        sys.exit(f"Transcription failed: {job.get('error')}")
    transcript = job["transcript"]
    speakers = transcript.get("speakerCount") or job.get("speakerCount") or 1
    if speakers > 1:
        sys.exit("Dialogue edits support single-speaker videos only")
    return transcript
def find_word(transcript, word, occurrence):
    words = [w for segment in transcript["segments"] for w in segment["words"]]
    target = normalize(word)
    if not target or " " in word.strip():
        sys.exit("--word must be a single word from the transcript")
    matches = [w for w in words if normalize(w["text"]) == target]
    if len(matches) < occurrence:
        text = " ".join(w["text"] for w in words)
        sys.exit(f"Found {len(matches)} occurrence(s) of '{word}' in: {text}")
    match = matches[occurrence - 1]
    context = " ".join(f"[{w['text']}]" if w is match else w["text"] for w in words)
    log(f"Replacing word {match['id']}: {context}")
    return match["id"]
def create_preview(video_url, transcript, word_id, recipient, voice_id):
    edit = {"kind": "change", "wordId": word_id, "replacement": recipient["name"]}
    if recipient["pronunciation"]:
        edit["pronunciation"] = recipient["pronunciation"]
    body = {"sourceVideoUrl": video_url, "transcript": transcript, "edits": [edit]}
    if voice_id:
        body["voiceId"] = voice_id
    return post("/v2/dialogue-edits", body)["id"]
def finish_preview(recipient, deadline):
    try:
        job = wait_for(
            f"/v2/dialogue-edits/{recipient['preview_id']}",
            {"COMPLETED", "COMPLETED_PARTIAL", "FAILED"},
            deadline,
        )
    except Timeout as err:
        recipient["error"] = f"preview {err} (check GET /v2/dialogue-edits/{recipient['preview_id']})"
        return {}
    if job["status"] == "FAILED":
        error = job.get("error") or {}
        recipient["error"] = f"preview {job['id']} FAILED {error.get('code', '')}".strip()
        return job
    recipient["preview"] = job["id"]
    log(f"  CSV line {recipient['line']} ({recipient['name']}): {job.get('previewAudioUrl')}")
    if job["status"] == "COMPLETED_PARTIAL":
        log(f"  note: CSV line {recipient['line']} is COMPLETED_PARTIAL (not every edit applied); listen first")
    return job
def make_previews(recipients, video_url, transcript, word_id, deadline):
    """Creates one preview per name; the first clones the voice, the rest reuse it."""
    first, rest = recipients[0], recipients[1:]
    log(f"Preview 1/{len(recipients)}: cloning the voice")
    first["preview_id"] = create_preview(video_url, transcript, word_id, first, None)
    voice_id = finish_preview(first, deadline).get("voiceId")
    if "preview" not in first:
        sys.exit(f"The first preview did not complete ({first['error']}); fix it before continuing")
    for i, recipient in enumerate(rest, start=2):
        time.sleep(PREVIEW_GAP_SECONDS)
        try:
            recipient["preview_id"] = create_preview(
                video_url, transcript, word_id, recipient, voice_id
            )
        except ApiError as err:
            recipient["error"] = str(err)
        if i % 10 == 0 or i == len(recipients):
            log(f"Submitted {i}/{len(recipients)} previews")
    for recipient in rest:
        if "preview_id" in recipient:
            finish_preview(recipient, deadline)
    ready = sum("preview" in r for r in recipients)
    log(f"{ready}/{len(recipients)} previews ready")
def refusal(err, rows, by_request_id):
    """Describes a refused batch and maps the failing line back to its CSV row."""
    body = err.body
    # Dialogue edit refusals carry lineNumber; other record errors name the line in message.
    line = body.get("lineNumber")
    if not isinstance(line, int):
        found = re.search(r"line (\d+)", str(body.get("message", "")))
        line = int(found.group(1)) if found else None
    text = f"HTTP {err.status} {body.get('errorCode', '')} lineNumber={line}: {body.get('message', body)}"
    if line and 1 <= line <= len(rows):
        recipient = by_request_id[rows[line - 1]["request_id"]]
        text += f" (CSV line {recipient['line']}, {recipient['name']})"
    return text
def build_rows(recipients, video_url, model):
    return [
        {
            "request_id": f"line-{r['line']}",
            "endpoint": "/v2/generate",
            "payload": {
                "model": model,
                "input": [{"type": "video", "url": video_url}],
                "dialogueEdit": {"id": r["preview"]},
            },
        }
        for r in recipients
        if "preview" in r
    ]
def download_results(output_url):
    # The result URL redirects to a signed file, so send no API key with it.
    req = urllib.request.Request(output_url, headers={"User-Agent": "sync-personalize-batch-example/1.0"})
    with urllib.request.urlopen(req, timeout=120) as res:
        return [json.loads(line) for line in res.read().decode().splitlines() if line.strip()]
def main():
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--video-url", help="Sync-hosted URL of the source video")
    parser.add_argument("--csv", required=True, help="CSV with a 'name' column")
    parser.add_argument("--word", help="Word in the video to replace with each name")
    parser.add_argument("--occurrence", type=int, default=1, help="Which match of --word to replace")
    parser.add_argument("--model", default="sync-3")
    parser.add_argument("--webhook-url", help="Optional batch completion webhook")
    parser.add_argument("--dry-run", action="store_true", help="Validate the batch without rendering")
    parser.add_argument("--batch-id", help="Resume: poll this batch and print its results")
    parser.add_argument("--timeout-min", type=float, default=60, help="Minutes to wait for each step (default 60)")
    args = parser.parse_args()
    if not API_KEY:
        sys.exit("Set SYNC_API_KEY")
    if not args.batch_id and not (args.video_url and args.word):
        sys.exit("--video-url and --word are required unless you pass --batch-id")
    if args.occurrence < 1 or args.timeout_min <= 0:
        sys.exit("--occurrence must be 1 or more and --timeout-min above 0")
    timeout = args.timeout_min * 60
    recipients = read_recipients(args.csv)
    by_request_id = {f"line-{r['line']}": r for r in recipients}
    batch_id = args.batch_id
    if not batch_id:
        transcript = transcribe(args.video_url, time.monotonic() + timeout)
        word_id = find_word(transcript, args.word, args.occurrence)
        make_previews(recipients, args.video_url, transcript, word_id, time.monotonic() + timeout)
        for r in recipients:
            if "error" in r:
                log(f"  skipping CSV line {r['line']} ({r['name']}): {r['error']}")
        rows = build_rows(recipients, args.video_url, args.model)
        if len(rows) < MIN_ROWS:
            sys.exit(f"Only {len(rows)} previews completed; a batch needs at least {MIN_ROWS}")
        try:
            post("/v2/batch", {"rows": rows, "dry_run": True})
        except ApiError as err:
            sys.exit(f"Dry run failed: {refusal(err, rows, by_request_id)}")
        log(f"Dry run validated {len(rows)} rows")
        if args.dry_run:
            return
        body = {"rows": rows}
        if args.webhook_url:
            body["webhook_url"] = args.webhook_url
        try:
            # Never retried: if the first attempt was accepted, a retry is a second batch.
            batch_id = request("POST", "/v2/batch", body, retry_statuses=())["id"]
        except (ApiError, OSError) as err:
            if isinstance(err, ApiError) and err.status < 500:
                sys.exit(f"Batch refused: {refusal(err, rows, by_request_id)}")
            sys.exit(
                f"Batch submit outcome unknown ({err}). The batch may have been created: "
                "list your batches with GET /v2/batch and resume with --batch-id <id> "
                "instead of submitting again."
            )
        log(f"Batch {batch_id} submitted; resume with --batch-id {batch_id}")
    progress, deadline = None, time.monotonic() + timeout
    while True:
        batch = get(f"/v2/batch/{batch_id}")
        metrics = batch.get("metrics") or {}
        if (batch["status"], metrics) != progress:
            progress = (batch["status"], metrics)
            log(f"Batch {batch['status']}: {json.dumps(metrics)}")
        if batch["status"] in ("COMPLETED", "FAILED"):
            break
        if time.monotonic() > deadline:
            sys.exit(f"Batch {batch_id} is still {batch['status']}; resume later with --batch-id {batch_id}")
        time.sleep(20)
    if not batch.get("outputUrl"):
        sys.exit(f"Batch {batch['status']} without a results file")
    # Results go to stdout as tab-separated lines; progress goes to stderr.
    print("csv_line\tname\tstatus\toutput_url_or_error")
    for line in download_results(batch["outputUrl"]):
        recipient = by_request_id.get(line["request_id"], {})
        response = line.get("response") or {}
        error = line.get("error") or ""
        if isinstance(error, dict):
            error = f"{error.get('code', '')} {error.get('message', '')}".strip()
        output = response.get("output_url") or response.get("outputUrl") or error
        print(f"{recipient.get('line', '')}\t{recipient.get('name', '')}\t{line['status']}\t{output}")
if __name__ == "__main__":
    main()

csv_line	name	status	output_url_or_error
2	Olivia	COMPLETED	<output URL>
3	James	COMPLETED	<output URL>