batch_client.py

"""Submit consented text rows to an OpenVoice service and record output URLs."""

from __future__ import annotations

import argparse
import csv
from pathlib import Path

import requests


def run(input_csv: Path, output_csv: Path, service_url: str, timeout: int) -> None:
    with input_csv.open(encoding="utf-8", newline="") as source:
        rows = list(csv.DictReader(source))
    if not rows or not {"text", "language"}.issubset(rows[0]):
        raise ValueError("input CSV must contain 'text' and 'language' columns")

    results: list[dict[str, str]] = []
    for row in rows:
        try:
            response = requests.post(
                f"{service_url.rstrip('/')}/synthesize",
                json={"text": row["text"], "language": row["language"]},
                timeout=timeout,
            )
            response.raise_for_status()
            body = response.json()
            results.append({**row, "audio_url": body["audio_url"], "error": ""})
        except (requests.RequestException, KeyError, ValueError) as exc:
            results.append({**row, "audio_url": "", "error": str(exc)})

    output_csv.parent.mkdir(parents=True, exist_ok=True)
    fieldnames = list(rows[0]) + ["audio_url", "error"]
    with output_csv.open("w", encoding="utf-8", newline="") as destination:
        writer = csv.DictWriter(destination, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(results)


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("input_csv", type=Path)
    parser.add_argument("output_csv", type=Path)
    parser.add_argument("--service-url", default="http://localhost:5002")
    parser.add_argument("--timeout", type=int, default=300)
    args = parser.parse_args()
    run(args.input_csv, args.output_csv, args.service_url, args.timeout)


if __name__ == "__main__":
    main()