translation_examples.py

#!/usr/bin/env python3
"""
Translation Service Examples

This script demonstrates how to use the AI Content Processing API
translation endpoints for single text translation, batch translation,
and language detection.
"""

import requests
import json
import time
from typing import List, Dict, Any


class TranslationClient:
    """Client for the AI Content Processing API translation endpoints."""

    def __init__(self, base_url: str = "http://localhost:8000"):
        """Initialize the translation client."""
        self.base_url = base_url.rstrip('/')

    def translate_text(
        self,
        text: str,
        source_language: str,
        target_language: str,
        use_openai: bool = True
    ) -> Dict[str, Any]:
        """
        Translate a single text.

        Args:
            text: Text to translate
            source_language: Source language (e.g., 'English', 'Spanish', 'auto-detect')
            target_language: Target language (e.g., 'English', 'Spanish')
            use_openai: Whether to use OpenAI (True) or Gemini (False)

        Returns:
            Translation result dictionary
        """
        url = f"{self.base_url}/translate"
        payload = {
            "text": text,
            "source_language": source_language,
            "target_language": target_language,
            "use_openai": use_openai
        }

        try:
            response = requests.post(url, json=payload)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            return {
                "success": False,
                "error": f"Request failed: {str(e)}"
            }

    def translate_batch(
        self,
        texts: List[str],
        source_language: str,
        target_language: str,
        use_openai: bool = True
    ) -> Dict[str, Any]:
        """
        Translate multiple texts.

        Args:
            texts: List of texts to translate
            source_language: Source language (e.g., 'English', 'Spanish', 'auto-detect')
            target_language: Target language (e.g., 'English', 'Spanish')
            use_openai: Whether to use OpenAI (True) or Gemini (False)

        Returns:
            Batch translation result dictionary
        """
        url = f"{self.base_url}/translate-batch"
        payload = {
            "texts": texts,
            "source_language": source_language,
            "target_language": target_language,
            "use_openai": use_openai
        }

        try:
            response = requests.post(url, json=payload)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            return {
                "success": False,
                "error": f"Request failed: {str(e)}"
            }

    def detect_language(self, text: str, use_openai: bool = True) -> Dict[str, Any]:
        """
        Detect the language of a text.

        Args:
            text: Text to analyze
            use_openai: Whether to use OpenAI (True) or Gemini (False)

        Returns:
            Language detection result dictionary
        """
        url = f"{self.base_url}/detect-language"
        payload = {
            "text": text,
            "use_openai": use_openai
        }

        try:
            response = requests.post(url, json=payload)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            return {
                "success": False,
                "error": f"Request failed: {str(e)}"
            }

    def get_supported_languages(self) -> Dict[str, Any]:
        """
        Get list of supported languages.

        Returns:
            Dictionary with supported languages and API status
        """
        url = f"{self.base_url}/supported-languages"

        try:
            response = requests.get(url)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            return {
                "error": f"Request failed: {str(e)}"
            }

    def check_health(self) -> Dict[str, Any]:
        """
        Check API health status.

        Returns:
            Health check result
        """
        url = f"{self.base_url}/health"

        try:
            response = requests.get(url)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            return {
                "error": f"Request failed: {str(e)}"
            }


def example_single_translation():
    """Example: Single text translation."""
    print("=== Single Text Translation Example ===")

    client = TranslationClient()

    # Example texts to translate
    examples = [
        {
            "text": "Hello, how are you today?",
            "source": "en",
            "target": "es"
        },
        {
            "text": "Bonjour, comment allez-vous?",
            "source": "fr",
            "target": "en"
        },
        {
            "text": "¿Cómo está el clima hoy?",
            "source": "auto-detect",  # Auto-detect source language
            "target": "en"
        }
    ]

    for i, example in enumerate(examples, 1):
        print(f"\n--- Example {i} ---")
        print(f"Original: {example['text']}")
        print(f"Source: {example['source']} → Target: {example['target']}")

        # Translate using OpenAI
        print("\n🔄 Using OpenAI:")
        result = client.translate_text(
            text=example["text"],
            source_language=example["source"],
            target_language=example["target"],
            use_openai=True
        )

        if result.get("success"):
            print(f"✅ Translation: {result['translated_text']}")
            print(f"   API: {result.get('api_used', 'N/A')}")
            print(f"   Time: {result.get('processing_time', 0):.2f}s")
            if result.get("source_language") != example["source"]:
                print(f"   Detected language: {result.get('source_language')}")
        else:
            print(f"❌ Error: {result.get('error', 'Unknown error')}")

        # Translate using Gemini (if available)
        print("\n🔄 Using Gemini:")
        result = client.translate_text(
            text=example["text"],
            source_language=example["source"],
            target_language=example["target"],
            use_openai=False
        )

        if result.get("success"):
            print(f"✅ Translation: {result['translated_text']}")
            print(f"   API: {result.get('api_used', 'N/A')}")
            print(f"   Time: {result.get('processing_time', 0):.2f}s")
        else:
            print(f"❌ Error: {result.get('error', 'Unknown error')}")


def example_batch_translation():
    """Example: Batch text translation."""
    print("\n\n=== Batch Translation Example ===")

    client = TranslationClient()

    # Multiple texts to translate
    texts = [
        "Good morning!",
        "How can I help you?",
        "Thank you very much.",
        "Have a great day!",
        "See you tomorrow."
    ]

    print(f"Translating {len(texts)} texts from en to es:")
    for i, text in enumerate(texts, 1):
        print(f"  {i}. {text}")

    result = client.translate_batch(
        texts=texts,
        source_language="en",
        target_language="es",
        use_openai=True
    )

    if result.get("success"):
        print(f"\n✅ Batch translation completed!")
        print(f"   Successful: {result['successful']}/{result['total_texts']}")
        print(f"   Total time: {result.get('processing_time', 0):.2f}s")

        print("\n📝 Results:")
        for item in result.get("results", []):
            if item.get("success"):
                print(f"   ✅ '{item['original_text']}' → '{item['translated_text']}'")
            else:
                print(f"   ❌ '{item['original_text']}' → Error: {item.get('error', 'Unknown')}")
    else:
        print(f"❌ Batch translation failed: {result.get('error', 'Unknown error')}")


def example_language_detection():
    """Example: Language detection."""
    print("\n\n=== Language Detection Example ===")

    client = TranslationClient()

    # Texts in various languages
    test_texts = [
        "Hello, this is a test in English.",
        "Hola, esto es una prueba en español.",
        "Bonjour, ceci est un test en français.",
        "Hallo, das ist ein Test auf Deutsch.",
        "こんにちは、これは日本語のテストです。",
        "Привет, это тест на русском языке."
    ]

    print("Detecting languages for various texts:\n")

    for i, text in enumerate(test_texts, 1):
        print(f"{i}. Text: {text}")

        result = client.detect_language(text, use_openai=True)

        if result.get("success"):
            print(f"   🔍 Detected: {result['detected_language']}")
            print(f"   📊 Confidence: {result.get('confidence', 0):.1%}")
            print(f"   🔧 API: {result.get('api_used', 'N/A')}")
            print(f"   ⏱️  Time: {result.get('processing_time', 0):.2f}s")
        else:
            print(f"   ❌ Error: {result.get('error', 'Unknown error')}")

        print()


def example_supported_languages():
    """Example: Get supported languages."""
    print("\n\n=== Supported Languages Example ===")

    client = TranslationClient()

    result = client.get_supported_languages()

    if "error" not in result:
        print(f"✅ Total supported languages: {result.get('total_languages', 0)}")
        print(f"🔧 Auto-detection supported: {result.get('auto_detection_supported', False)}")

        apis = result.get("available_apis", {})
        print(f"📡 Available APIs:")
        print(f"   OpenAI: {'✅' if apis.get('openai') else '❌'}")
        print(f"   Gemini: {'✅' if apis.get('gemini') else '❌'}")

        languages = result.get("supported_languages", [])
        print(f"\n🌐 Supported languages ({len(languages)}):")

        # Display languages in columns
        cols = 4
        for i in range(0, len(languages), cols):
            row = languages[i:i+cols]
            formatted_row = [f"{lang:<15}" for lang in row]
            print("   " + "".join(formatted_row))
    else:
        print(f"❌ Error getting supported languages: {result.get('error')}")


def example_comprehensive_workflow():
    """Example: Complete translation workflow."""
    print("\n\n=== Comprehensive Translation Workflow ===")

    client = TranslationClient()

    # Step 1: Check API health
    print("1. Checking API health...")
    health = client.check_health()
    if health.get("status") == "healthy":
        print("   ✅ API is healthy")
    else:
        print("   ❌ API health check failed")
        return

    # Step 2: Test text with unknown language
    unknown_text = "Ciao, come stai? Questo è un testo in italiano."
    print(f"\n2. Working with text: '{unknown_text}'")

    # Step 2a: Detect language
    print("   🔍 Detecting language...")
    detection = client.detect_language(unknown_text)
    if detection.get("success"):
        detected_lang = detection["detected_language"]
        print(f"   ✅ Detected: {detected_lang}")
    else:
        print(f"   ❌ Detection failed: {detection.get('error')}")
        return

    # Step 2b: Translate to multiple languages
    target_languages = ["en", "es", "fr"]
    print(f"\n   🔄 Translating to {len(target_languages)} languages...")

    for target_lang in target_languages:
        result = client.translate_text(
            text=unknown_text,
            source_language=detected_lang,
            target_language=target_lang
        )

        if result.get("success"):
            print(f"   ✅ {target_lang}: {result['translated_text']}")
        else:
            print(f"   ❌ {target_lang}: {result.get('error')}")

    print("\n🎉 Workflow completed!")


def main():
    """Run all translation examples."""
    print("🌐 AI Content Processing API - Translation Service Examples")
    print("=" * 60)

    try:
        # Check if API is running
        client = TranslationClient()
        health = client.check_health()

        if "error" in health:
            print("❌ API server is not running or not accessible.")
            print("   Please start the API server with: python api_server.py")
            return

        # Run all examples
        example_single_translation()
        example_batch_translation()
        example_language_detection()
        example_supported_languages()
        example_comprehensive_workflow()

        print("\n" + "=" * 60)
        print("✅ All translation examples completed successfully!")

    except KeyboardInterrupt:
        print("\n\n⏹️  Examples interrupted by user.")
    except Exception as e:
        print(f"\n❌ Unexpected error: {e}")


if __name__ == "__main__":
    main()