EchoScript : made it automatic, not exclusive: TranslationService picks anthropic when ANTHROPIC_API_KEY is set, otherwise falls back to marian (fully offline/free) - overridable via ECHOSCRIPT_TRANSLATION_BACKEND=anthropic|marian. So EchoScript still works with zero API key, but gets better translation quality and zero missing-pair errors the moment you set one.

requirements.txt

@@ -4,3 +4,4 @@ transformers>=4.40
 sentencepiece>=0.2
 sacremoses>=0.1
 torch>=2.0
+anthropic>=0.40

services/translation.py

@@ -12,14 +12,44 @@ the transcript (v1.1: Transcript Editing), every translation regenerated
 afterwards picks up the fix automatically, and every translation stays in
 sync with the same segment timings as the transcript (so subtitles still
 work for translated output).
+
+Two interchangeable backends are available:
+
+- "anthropic": sends transcript text to the Claude API for translation.
+  Used automatically when an ANTHROPIC_API_KEY is configured (the same
+  BYOK pattern Archon uses). No missing-language-pair failure mode --
+  every supported language translates directly to every other one in a
+  single call, with no local model downloads.
+- "marian": fully offline, no API key needed. Uses local Helsinki-NLP
+  MarianMT models via `transformers`. Helsinki-NLP doesn't publish a
+  direct model for every language pair (Persian in particular only has
+  reliable direct models paired with English), so this backend pivots
+  through English when no direct model exists for a pair.
+
+Set ECHOSCRIPT_TRANSLATION_BACKEND=anthropic|marian to force one; if
+unset, "anthropic" is used when ANTHROPIC_API_KEY is present, otherwise
+"marian".
 """
 
 from __future__ import annotations
 
+import os
+import re
 from functools import lru_cache
 
 from models.transcript import Segment, Transcript, Translation
 
+# Display names for every language EchoScript can act as a source or
+# target -- used both for the UI's language dropdown/checkboxes (via
+# services/transcription.py) and for prompting the Anthropic backend.
+LANGUAGE_NAMES: dict[str, str] = {
+    "fr": "French",
+    "en": "English",
+    "de": "German",
+    "fa": "Persian",
+    "es": "Spanish",
+}
+
 # Target languages exposed as the "Outputs" checkboxes in the UI.
 SUPPORTED_TARGET_LANGUAGES: dict[str, str] = {
     "en": "English",
@@ -28,27 +58,131 @@ SUPPORTED_TARGET_LANGUAGES: dict[str, str] = {
     "es": "Spanish",
 }
 
+# Anthropic model used for translation -- Haiku is fast and inexpensive,
+# which fits well for what is otherwise a mechanical translation task.
+_ANTHROPIC_MODEL = "claude-haiku-4-5-20251001"
+_ANTHROPIC_BATCH_SIZE = 40  # segments per API call, to keep prompts small
+
 
 class TranslationError(RuntimeError):
-    """Raised when no translation model/engine is available for a pair."""
+    """Raised when no translation backend/model is available for a pair."""
+
+
+def _resolve_backend() -> str:
+    forced = os.environ.get("ECHOSCRIPT_TRANSLATION_BACKEND", "").strip().lower()
+    if forced in ("anthropic", "marian"):
+        return forced
+    return "anthropic" if os.environ.get("ANTHROPIC_API_KEY") else "marian"
+
+
+# ---------------------------------------------------------------------------
+# Anthropic backend
+# ---------------------------------------------------------------------------
+
+_NUMBERED_LINE_RE = re.compile(r"^\s*(\d+)[.\)]\s?(.*)$")
+
+
+@lru_cache(maxsize=1)
+def _anthropic_client():
+    import anthropic  # heavy import, deferred until needed
+
+    return anthropic.Anthropic()
+
+
+def _translate_batch_via_anthropic(
+    texts: list[str], source_language: str, target_language: str
+) -> list[str]:
+    """Translate a batch of lines in one Claude call, preserving order/count."""
+    source_name = LANGUAGE_NAMES.get(source_language, source_language)
+    target_name = LANGUAGE_NAMES.get(target_language, target_language)
+
+    numbered_input = "\n".join(f"{i + 1}. {text}" for i, text in enumerate(texts))
+
+    try:
+        response = _anthropic_client().messages.create(
+            model=_ANTHROPIC_MODEL,
+            max_tokens=4096,
+            system=(
+                f"You translate transcript lines from {source_name} to {target_name}. "
+                "You will be given a numbered list of lines, one sentence or "
+                "fragment per line. Reply with the same numbers, translated, "
+                "one per line, in the same order. Keep the same number of "
+                "lines as the input -- never merge, split, drop, or add lines. "
+                "Output only the numbered translated lines, with no preamble, "
+                "no explanations, and no extra commentary."
+            ),
+            messages=[{"role": "user", "content": numbered_input}],
+        )
+    except Exception as exc:  # pragma: no cover - depends on network/API key
+        raise TranslationError(
+            f"Anthropic translation request failed for "
+            f"'{source_language}' -> '{target_language}': {exc}"
+        ) from exc
+
+    raw_text = "".join(
+        block.text for block in response.content if getattr(block, "type", None) == "text"
+    )
+
+    parsed: dict[int, str] = {}
+    for line in raw_text.splitlines():
+        match = _NUMBERED_LINE_RE.match(line)
+        if match:
+            parsed[int(match.group(1))] = match.group(2).strip()
+
+    if len(parsed) != len(texts) or any((i + 1) not in parsed for i in range(len(texts))):
+        raise TranslationError(
+            f"Anthropic translation response didn't match expected line count "
+            f"for '{source_language}' -> '{target_language}' "
+            f"(expected {len(texts)}, parsed {len(parsed)})."
+        )
+
+    return [parsed[i + 1] for i in range(len(texts))]
+
+
+def _translate_segments_via_anthropic(
+    segments: list[Segment], source_language: str, target_language: str
+) -> list[Segment]:
+    non_empty = [(i, seg) for i, seg in enumerate(segments) if seg.text]
+    translated_text_by_index: dict[int, str] = {}
+
+    for start in range(0, len(non_empty), _ANTHROPIC_BATCH_SIZE):
+        chunk = non_empty[start : start + _ANTHROPIC_BATCH_SIZE]
+        texts = [seg.text for _, seg in chunk]
+        translated = _translate_batch_via_anthropic(texts, source_language, target_language)
+        for (i, _), text in zip(chunk, translated):
+            translated_text_by_index[i] = text
+
+    return [
+        Segment(
+            index=seg.index,
+            start=seg.start,
+            end=seg.end,
+            text=translated_text_by_index.get(i, seg.text),
+        )
+        for i, seg in enumerate(segments)
+    ]
+
+
+# ---------------------------------------------------------------------------
+# Marian (offline) backend
+# ---------------------------------------------------------------------------
 
 
 @lru_cache(maxsize=None)
-def _load_engine(source_language: str, target_language: str):
-    """Lazily load and cache a MarianMT model+tokenizer for one language pair.
-
-    Cached so repeated translations within a session don't reload a model
-    from disk every time. Swapping the translation backend later (a
-    different model, a hosted API, an offline engine like Argos) only
-    requires changing this one function.
-
-    Note: this loads the model/tokenizer directly via AutoModelForSeq2SeqLM
-    rather than transformers.pipeline("translation", ...). As of
-    transformers v5, the generic "translation" pipeline task was removed
-    entirely (see huggingface/transformers#43825) -- pipeline("translation")
-    now raises KeyError. Loading the model directly and calling
-    `model.generate()` is unaffected by that change and is the path the
-    transformers docs now show for MarianMT.
+def _try_load_marian_engine(source_language: str, target_language: str):
+    """Try to load+cache a MarianMT model+tokenizer for one language pair.
+
+    Returns (tokenizer, model), or None if no such model exists on the Hub
+    (e.g. Helsinki-NLP doesn't publish every pair directly). Cached either
+    way so a missing pair isn't re-checked over the network on every call.
+
+    Loads the model/tokenizer directly via AutoModelForSeq2SeqLM rather
+    than transformers.pipeline("translation", ...): as of transformers v5,
+    the generic "translation" pipeline task was removed entirely (see
+    huggingface/transformers#43825) -- pipeline("translation") now raises
+    KeyError. Loading the model directly and calling `model.generate()` is
+    unaffected by that change and is the path the transformers docs now
+    show for MarianMT.
     """
     from transformers import AutoModelForSeq2SeqLM, AutoTokenizer  # heavy import, deferred
 
@@ -56,22 +190,74 @@ def _load_engine(source_language: str, target_language: str):
     try:
         tokenizer = AutoTokenizer.from_pretrained(model_name)
         model = AutoModelForSeq2SeqLM.from_pretrained(model_name)
-    except Exception as exc:  # pragma: no cover - depends on model availability
-        raise TranslationError(
-            f"No translation model available for "
-            f"'{source_language}' -> '{target_language}': {exc}"
-        ) from exc
+    except Exception:  # pragma: no cover - depends on model availability
+        return None
     return tokenizer, model
 
 
-def _run_translation(tokenizer, model, text: str) -> str:
+def _resolve_marian_engines(source_language: str, target_language: str) -> list[tuple]:
+    """Work out which model(s) to chain to get from source to target.
+
+    Prefers a single direct Helsinki-NLP model. If none exists for the
+    pair, pivots through English (source -> en -> target), since that's
+    where Helsinki-NLP's coverage is densest -- this is what makes
+    something like French -> Persian work even though no direct
+    opus-mt-fr-fa model exists.
+    """
+    direct = _try_load_marian_engine(source_language, target_language)
+    if direct is not None:
+        return [direct]
+
+    if source_language != "en" and target_language != "en":
+        hop1 = _try_load_marian_engine(source_language, "en")
+        hop2 = _try_load_marian_engine("en", target_language)
+        if hop1 is not None and hop2 is not None:
+            return [hop1, hop2]
+
+    raise TranslationError(
+        f"No direct or English-pivot translation model available for "
+        f"'{source_language}' -> '{target_language}'."
+    )
+
+
+def _run_marian_translation(tokenizer, model, text: str) -> str:
     inputs = tokenizer(text, return_tensors="pt", truncation=True)
     generated = model.generate(**inputs, max_new_tokens=512)
     return tokenizer.decode(generated[0], skip_special_tokens=True).strip()
 
 
+def _translate_segments_via_marian(
+    segments: list[Segment], source_language: str, target_language: str
+) -> list[Segment]:
+    engines = _resolve_marian_engines(source_language, target_language)
+
+    translated_segments = []
+    for seg in segments:
+        if not seg.text:
+            translated_segments.append(seg)
+            continue
+        text = seg.text
+        for tokenizer, model in engines:
+            text = _run_marian_translation(tokenizer, model, text)
+        translated_segments.append(Segment(index=seg.index, start=seg.start, end=seg.end, text=text))
+    return translated_segments
+
+
+# ---------------------------------------------------------------------------
+# Public service
+# ---------------------------------------------------------------------------
+
+
 class TranslationService:
-    """Translates a Transcript into one or more target languages."""
+    """Translates a Transcript into one or more target languages.
+
+    Backend ("anthropic" or "marian") is resolved once at construction
+    time -- pass `backend` explicitly to override the
+    ECHOSCRIPT_TRANSLATION_BACKEND / ANTHROPIC_API_KEY auto-detection.
+    """
+
+    def __init__(self, backend: str | None = None) -> None:
+        self.backend = backend or _resolve_backend()
 
     def translate(self, transcript: Transcript, target_language: str) -> Translation:
         """Translate every segment of `transcript`, preserving timing."""
@@ -83,16 +269,13 @@ class TranslationService:
                 segments=list(transcript.segments),
             )
 
-        tokenizer, model = _load_engine(transcript.language, target_language)
-
-        translated_segments = []
-        for seg in transcript.segments:
-            if not seg.text:
-                translated_segments.append(seg)
-                continue
-            result_text = _run_translation(tokenizer, model, seg.text)
-            translated_segments.append(
-                Segment(index=seg.index, start=seg.start, end=seg.end, text=result_text)
+        if self.backend == "anthropic":
+            translated_segments = _translate_segments_via_anthropic(
+                transcript.segments, transcript.language, target_language
+            )
+        else:
+            translated_segments = _translate_segments_via_marian(
+                transcript.segments, transcript.language, target_language
             )
 
         return Translation(