translate_batch.py

"""
Azure AI Document Translation - Batch (Asynchronous) Sample
============================================================

This script submits a batch document translation job to an Azure AI Translator
(Document Translation) resource and translates every document in a source
Azure Blob Storage container into a target container.

Authentication
--------------
* Translator / Foundry endpoint  : **key-based** (AzureKeyCredential)
* Source & target storage access : **Managed Identity (RBAC)** on the
  Translator resource. NO SAS tokens are sent in the request.

Prerequisites for the managed-identity storage path
---------------------------------------------------
The Translator (or Azure AI Foundry / multi-service Cognitive Services)
resource must have a system-assigned or user-assigned managed identity, and
that identity must hold the following role assignments:

  * Source container  -> "Storage Blob Data Reader"      (or Contributor)
  * Target container  -> "Storage Blob Data Contributor"

When these roles are in place, the Document Translation service reads from
the source and writes to the target using its own identity, so the request
only needs the plain container URLs - no SAS tokens.

Required environment variables
------------------------------
Update these in your own environment (e.g. `.env`, shell `export`, or your
CI/CD secret store) BEFORE running:

  AZURE_TRANSLATOR_ENDPOINT
      Document Translation endpoint of your Translator / Foundry resource.
      Example: https://<your-resource-name>.cognitiveservices.azure.com/

  AZURE_TRANSLATOR_KEY
      One of the two access keys for the Translator resource above.
      Find it in the Azure Portal under: Resource -> Keys and Endpoint.

  AZURE_SOURCE_STORAGE_URL
      Full HTTPS URL of the SOURCE blob container (no SAS).
      Example: https://<source-account>.blob.core.windows.net/<source-container>

  AZURE_TARGET_STORAGE_URL
      Full HTTPS URL of the TARGET blob container (no SAS). The container
      should be empty or point to an unused sub-path.
      Example: https://<target-account>.blob.core.windows.net/<target-container>

  AZURE_TARGET_LANGUAGE   (optional, default: "en")
      BCP-47 code to translate INTO (e.g. "en", "fr", "zh-Hans").

  AZURE_SOURCE_LANGUAGE   (optional)
      BCP-47 code of the source documents. Omit to auto-detect per document.

How to set them
---------------
Linux / macOS:
    export AZURE_TRANSLATOR_ENDPOINT="https://<your-resource>.cognitiveservices.azure.com/"
    export AZURE_TRANSLATOR_KEY="<your-key>"
    export AZURE_SOURCE_STORAGE_URL="https://<src>.blob.core.windows.net/<src-container>"
    export AZURE_TARGET_STORAGE_URL="https://<dst>.blob.core.windows.net/<dst-container>"
    export AZURE_TARGET_LANGUAGE="en"

Windows PowerShell:
    $env:AZURE_TRANSLATOR_ENDPOINT="https://<your-resource>.cognitiveservices.azure.com/"
    $env:AZURE_TRANSLATOR_KEY="<your-key>"
    ...

Or copy `.env.sample` to `.env`, fill in the values, and the script will load
it automatically (if `python-dotenv` is installed).

Install dependencies
--------------------
    pip install -r requirements.txt

Run
---
    python translate_batch.py
"""

from __future__ import annotations

import os
import sys

try:
    from dotenv import load_dotenv  # optional convenience
    load_dotenv()
except ImportError:
    pass

from azure.core.credentials import AzureKeyCredential
from azure.core.exceptions import HttpResponseError
from azure.ai.translation.document import (
    DocumentTranslationClient,
    DocumentTranslationInput,
    TranslationTarget,
)


def _require_env(name: str) -> str:
    value = os.environ.get(name)
    if not value:
        sys.exit(
            f"ERROR: Required environment variable '{name}' is not set. "
            f"See the docstring at the top of translate_batch.py for setup instructions."
        )
    return value


def main() -> int:
    endpoint = _require_env("AZURE_TRANSLATOR_ENDPOINT")
    key = _require_env("AZURE_TRANSLATOR_KEY")

    # Plain container URLs - no SAS. The Translator resource's managed
    # identity must have Storage Blob Data Reader/Contributor on these
    # containers.
    source_url = _require_env("AZURE_SOURCE_STORAGE_URL")
    target_url = _require_env("AZURE_TARGET_STORAGE_URL")

    target_language = os.environ.get("AZURE_TARGET_LANGUAGE", "en")
    source_language = os.environ.get("AZURE_SOURCE_LANGUAGE")  # optional

    client = DocumentTranslationClient(endpoint, AzureKeyCredential(key))

    translation_input = DocumentTranslationInput(
        source_url=source_url,
        source_language=source_language,
        targets=[
            TranslationTarget(
                target_url=target_url,
                language=target_language,
            )
        ],
    )

    print("Submitting batch translation job ...")
    print(f"  Source : {source_url}")
    print(f"  Target : {target_url}  (language: {target_language})")
    if source_language:
        print(f"  Source language (forced): {source_language}")
    else:
        print("  Source language: auto-detect")
    print("  Storage auth: Managed Identity on Translator resource (no SAS)")

    try:
        poller = client.begin_translation(inputs=[translation_input])
        job_details = poller.details
        print(f"\nJob submitted. Operation ID: {job_details.id}")
        print("Waiting for completion (this may take a while for large batches) ...")

        result = poller.result()  # blocks until the batch finishes

        final = poller.details
        print("\n--- Job complete ---")
        print(f"Status       : {final.status}")
        print(f"Created      : {final.created_on}")
        print(f"Last updated : {final.last_updated_on}")
        print(f"Total docs   : {final.documents_total_count}")
        print(f"  succeeded  : {final.documents_succeeded_count}")
        print(f"  failed     : {final.documents_failed_count}")
        print(f"  cancelled  : {final.documents_cancelled_count}")
        print(f"  in progress: {final.documents_in_progress_count}")
        print(f"  not started: {final.documents_not_started_count}")

        print("\nPer-document results:")
        for doc in result:
            if doc.status == "Succeeded":
                print(f"  [OK]   {doc.source_document_url}")
                print(f"         -> {doc.translated_document_url}")
            else:
                err = doc.error.message if doc.error else "unknown error"
                print(f"  [FAIL] {doc.source_document_url}")
                print(f"         status={doc.status}, error={err}")

        return 0 if final.documents_failed_count == 0 else 2

    except HttpResponseError as ex:
        print(f"\nAzure request failed: {ex.message}", file=sys.stderr)
        if ex.error and getattr(ex.error, "code", None):
            print(f"Code: {ex.error.code}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())