Translate a Pre-Recorded Audio File

import os
import time
import requests

AUTH_KEY = "YOUR_AUTH_KEY"
BASE_URL = "https://api.deepl.com"
AUDIO_FILE = "podcast-episode-42.mp3"

def create_job(file_path: str) -> dict:
    file_size = os.path.getsize(file_path)

    response = requests.post(
        f"{BASE_URL}/v1/jobs/voice/translate",
        headers={"Authorization": f"DeepL-Auth-Key {AUTH_KEY}"},
        json={
            "source_file": {
                "name": file_path,
                "content_type": "audio/mpeg",
                "content_length": file_size,
            },
            "parameters": {"source_language": "en"},
            "targets": [
                {"language": "de", "type": "text/plain"},
                {"language": "es", "type": "audio/pcm;encoding=s16le;rate=16000"},
            ],
        },
    )
    response.raise_for_status()
    return response.json()


def upload_file(upload_url: str, file_path: str) -> None:
    with open(file_path, "rb") as f:
        response = requests.put(
            upload_url,
            headers={"Content-Type": "audio/mpeg"},
            data=f,
        )
    response.raise_for_status()


def poll_until_done(job_id: str, poll_interval: int = 15) -> list:
    while True:
        response = requests.get(
            f"{BASE_URL}/v1/jobs/voice/translate/{job_id}",
            headers={"Authorization": f"DeepL-Auth-Key {AUTH_KEY}"},
        )
        response.raise_for_status()
        data = response.json()
        results = data["results"]

        # Check whether all targets have reached a terminal state
        if all(r["status"] in ("complete", "failed") for r in results):
            return results

        print(f"Status: {[r['status'] for r in results]}, polling again in {poll_interval}s")
        time.sleep(poll_interval)


def download_results(results: list, targets: list) -> None:
    extensions = {"text/plain": "txt", "audio/pcm;encoding=s16le;rate=16000": "pcm"}

    for i, result in enumerate(results):
        if result["status"] != "complete":
            print(f"Target {i} failed: {result.get('error', {}).get('message')}")
            continue

        target = targets[i]
        ext = extensions.get(target["type"], "bin")
        filename = f"translation-{target['language']}.{ext}"

        url = result.get("download_url")
        if not url:
            print(f"Target {i} missing download_url")
            continue
        content = requests.get(url).content
        with open(filename, "wb") as f:
            f.write(content)
        print(f"Saved {filename}")


job_response = create_job(AUDIO_FILE)
job_id = job_response["job_id"]
upload_url = job_response["upload_url"]
print(f"Job created: {job_id}")

upload_file(upload_url, AUDIO_FILE)
print("Upload complete")

results = poll_until_done(job_id)
download_results(results, job_response["targets"])