Real-time Translation & Multilingual AI

# Full Speech-to-Speech Translation Pipeline
Audio (Source Language)
  → STT Model (Whisper / Azure Speech)
    → Source Text
      → Translation (LLM / NLLB / DeepL API)
        → Target Text
          → TTS Model (Edge-TTS / ElevenLabs / Bark)
            → Audio (Target Language)

# End-to-End Alternative
Audio (Source Language)
  → SeamlessM4T (Meta)
    → Audio (Target Language)
# Single model handles STT + Translation + TTS

import whisper
import anthropic
import edge_tts
import asyncio
import sounddevice as sd
import numpy as np
import tempfile

class RealTimeSpeechTranslator:
    def __init__(self, source_lang="en", target_lang="es"):
        self.whisper_model = whisper.load_model("small")  # Balance speed/accuracy
        self.llm = anthropic.Anthropic()
        self.source_lang = source_lang
        self.target_lang = target_lang
        self.tts_voices = {
            "es": "es-ES-AlvaroNeural",
            "fr": "fr-FR-HenriNeural",
            "de": "de-DE-ConradNeural",
            "ja": "ja-JP-KeitaNeural",
            "zh": "zh-CN-YunxiNeural",
            "pt": "pt-BR-AntonioNeural",
            "hi": "hi-IN-MadhurNeural",
        }

    def transcribe(self, audio_array: np.ndarray) -> str:
        """Transcribe audio using Whisper."""
        result = self.whisper_model.transcribe(
            audio_array.astype(np.float32),
            language=self.source_lang,
            fp16=False
        )
        return result["text"].strip()

    def translate(self, text: str) -> str:
        """Translate text using Claude for nuanced, context-aware translation."""
        lang_names = {"es": "Spanish", "fr": "French", "de": "German",
                      "ja": "Japanese", "zh": "Chinese", "pt": "Portuguese",
                      "hi": "Hindi", "en": "English"}

        response = self.llm.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=500,
            messages=[{
                "role": "user",
                "content": f"""Translate the following text to {lang_names[self.target_lang]}.
Preserve tone, idioms, and cultural context. Return ONLY the translation.

Text: {text}"""
            }]
        )
        return response.content[0].text.strip()

    async def synthesize_speech(self, text: str, output_path: str):
        """Convert translated text to speech using Edge-TTS."""
        voice = self.tts_voices.get(self.target_lang, "en-US-GuyNeural")
        communicate = edge_tts.Communicate(text, voice)
        await communicate.save(output_path)

    async def translate_audio_chunk(self, audio_chunk: np.ndarray) -> str:
        """Full pipeline: transcribe → translate → speak."""
        # Step 1: Speech to text
        source_text = self.transcribe(audio_chunk)
        if not source_text:
            return None

        # Step 2: Translate
        translated_text = self.translate(source_text)

        # Step 3: Text to speech
        output_path = tempfile.mktemp(suffix=".mp3")
        await self.synthesize_speech(translated_text, output_path)

        print(f"[{self.source_lang}] {source_text}")
        print(f"[{self.target_lang}] {translated_text}")
        return output_path

# Usage
translator = RealTimeSpeechTranslator(source_lang="en", target_lang="es")
# Record 5 seconds of audio
audio = sd.rec(int(5 * 16000), samplerate=16000, channels=1, dtype="float32")
sd.wait()
result = asyncio.run(translator.translate_audio_chunk(audio.flatten()))

import anthropic
from docx import Document
from copy import deepcopy

class DocumentTranslator:
    def __init__(self, target_lang: str):
        self.client = anthropic.Anthropic()
        self.target_lang = target_lang
        self.translation_cache = {}

    def translate_text_batch(self, texts: list[str]) -> list[str]:
        """Translate multiple text segments in a single LLM call for efficiency."""
        # Filter out empty strings and already-cached translations
        to_translate = []
        indices = []
        for i, text in enumerate(texts):
            if text.strip() and text not in self.translation_cache:
                to_translate.append(text)
                indices.append(i)

        if not to_translate:
            return [self.translation_cache.get(t, t) for t in texts]

        # Batch translate with numbered segments
        numbered = "\n".join(f"[{i}] {t}" for i, t in enumerate(to_translate))

        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            messages=[{
                "role": "user",
                "content": f"""Translate each numbered segment to {self.target_lang}.
Preserve formatting, line breaks, and special characters.
Return translations in the same numbered format.

{numbered}"""
            }]
        )

        # Parse results and cache
        translated = self._parse_numbered_response(response.content[0].text)
        for orig, trans in zip(to_translate, translated):
            self.translation_cache[orig] = trans

        return [self.translation_cache.get(t, t) for t in texts]

    def translate_docx(self, input_path: str, output_path: str):
        """Translate a DOCX file preserving all formatting."""
        doc = Document(input_path)
        new_doc = deepcopy(doc)

        # Collect all text runs from paragraphs
        all_texts = []
        text_locations = []  # (paragraph_idx, run_idx)

        for p_idx, paragraph in enumerate(new_doc.paragraphs):
            for r_idx, run in enumerate(paragraph.runs):
                if run.text.strip():
                    all_texts.append(run.text)
                    text_locations.append((p_idx, r_idx))

        # Translate in batches of 50 segments
        batch_size = 50
        for i in range(0, len(all_texts), batch_size):
            batch = all_texts[i:i + batch_size]
            translated = self.translate_text_batch(batch)

            for j, trans in enumerate(translated):
                p_idx, r_idx = text_locations[i + j]
                new_doc.paragraphs[p_idx].runs[r_idx].text = trans

        new_doc.save(output_path)
        print(f"Translated document saved to {output_path}")

# Usage
translator = DocumentTranslator(target_lang="French")
translator.translate_docx("report_en.docx", "report_fr.docx")

import anthropic

class MultilingualChatbot:
    def __init__(self):
        self.client = anthropic.Anthropic()
        self.conversation_history = []

    def detect_and_respond(self, user_message: str) -> dict:
        """Detect language, process, and respond in the user's language."""
        # Single LLM call handles detection + translation + response
        system_prompt = """You are a multilingual AI assistant. For each user message:
1. Detect the language of the input
2. Understand the intent regardless of language
3. Respond in the SAME language the user wrote in
4. If the user switches languages mid-conversation, follow their lead

Always respond naturally in the detected language. Include a JSON header
on the first line: {"detected_lang": "xx", "confidence": 0.99}
Then provide your response on subsequent lines."""

        self.conversation_history.append({
            "role": "user",
            "content": user_message
        })

        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            system=system_prompt,
            messages=self.conversation_history
        )

        reply = response.content[0].text
        self.conversation_history.append({
            "role": "assistant",
            "content": reply
        })

        # Parse the language detection header
        lines = reply.split("\n", 1)
        import json
        meta = json.loads(lines[0])
        return {
            "language": meta["detected_lang"],
            "confidence": meta["confidence"],
            "response": lines[1].strip() if len(lines) > 1 else ""
        }

# Usage
bot = MultilingualChatbot()
print(bot.detect_and_respond("Bonjour, comment puis-je suivre ma commande?"))
# Responds in French with order tracking help
print(bot.detect_and_respond("Hola, quiero cambiar mi dirección de envío"))
# Responds in Spanish with shipping address help

import whisper
import anthropic

def generate_translated_subtitles(video_path: str, target_lang: str) -> str:
    """Generate translated SRT subtitles from a video file."""
    # Step 1: Transcribe with timestamps
    model = whisper.load_model("medium")
    result = model.transcribe(video_path, word_timestamps=True)

    # Step 2: Extract segments with timing
    segments = [{
        "start": seg["start"],
        "end": seg["end"],
        "text": seg["text"].strip()
    } for seg in result["segments"]]

    # Step 3: Translate all segments (batch for efficiency)
    client = anthropic.Anthropic()
    texts = [s["text"] for s in segments]
    numbered = "\n".join(f"[{i}] {t}" for i, t in enumerate(texts))

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        messages=[{
            "role": "user",
            "content": f"""Translate these subtitle segments to {target_lang}.
Keep translations concise (subtitles must be readable).
Preserve the numbered format. Return ONLY translations.

{numbered}"""
        }]
    )

    # Step 4: Generate SRT format
    srt_lines = []
    translated = response.content[0].text.strip().split("\n")
    for i, seg in enumerate(segments):
        start = _format_srt_time(seg["start"])
        end = _format_srt_time(seg["end"])
        text = translated[i].split("] ", 1)[-1] if i < len(translated) else seg["text"]
        srt_lines.append(f"{i+1}\n{start} --> {end}\n{text}\n")

    return "\n".join(srt_lines)

def _format_srt_time(seconds: float) -> str:
    h = int(seconds // 3600)
    m = int((seconds % 3600) // 60)
    s = int(seconds % 60)
    ms = int((seconds % 1) * 1000)
    return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"