Upload 2 files

models/translation/__init__.py

@@ -0,0 +1,2 @@
+# Translation Model Package
+

models/translation/translation_utils.py

@@ -0,0 +1,578 @@
+# models/translation/translation_utils.py
+
+"""
+Translation Model Utilities for PENNY Project
+Handles multilingual translation using NLLB-200 for civic engagement accessibility.
+Provides async translation with structured error handling and language code normalization.
+"""
+
+import asyncio
+import time
+import os
+import httpx
+from typing import Dict, Any, Optional, List
+
+# --- Logging Imports ---
+from app.logging_utils import log_interaction, sanitize_for_logging
+
+# --- Hugging Face API Configuration ---
+HF_API_URL = "https://api-inference.huggingface.co/models/facebook/nllb-200-distilled-600M"
+HF_TOKEN = os.getenv("HF_TOKEN")
+
+AGENT_NAME = "penny-translate-agent"
+SERVICE_AVAILABLE = True  # Assume available since we're using API
+
+# NLLB-200 Language Code Mapping (Common languages for civic engagement)
+LANGUAGE_CODES = {
+    # English variants
+    "english": "eng_Latn",
+    "en": "eng_Latn",
+    
+    # Spanish variants
+    "spanish": "spa_Latn",
+    "es": "spa_Latn",
+    "español": "spa_Latn",
+    
+    # French
+    "french": "fra_Latn",
+    "fr": "fra_Latn",
+    "français": "fra_Latn",
+    
+    # Mandarin Chinese
+    "chinese": "zho_Hans",
+    "mandarin": "zho_Hans",
+    "zh": "zho_Hans",
+    
+    # Arabic
+    "arabic": "arb_Arab",
+    "ar": "arb_Arab",
+    
+    # Hindi
+    "hindi": "hin_Deva",
+    "hi": "hin_Deva",
+    
+    # Portuguese
+    "portuguese": "por_Latn",
+    "pt": "por_Latn",
+    
+    # Russian
+    "russian": "rus_Cyrl",
+    "ru": "rus_Cyrl",
+    
+    # German
+    "german": "deu_Latn",
+    "de": "deu_Latn",
+    
+    # Vietnamese
+    "vietnamese": "vie_Latn",
+    "vi": "vie_Latn",
+    
+    # Tagalog
+    "tagalog": "tgl_Latn",
+    "tl": "tgl_Latn",
+    
+    # Urdu
+    "urdu": "urd_Arab",
+    "ur": "urd_Arab",
+    
+    # Swahili
+    "swahili": "swh_Latn",
+    "sw": "swh_Latn",
+}
+
+# Pre-translated civic phrases for common queries
+CIVIC_PHRASES = {
+    "eng_Latn": {
+        "voting_location": "Where is my polling place?",
+        "voter_registration": "How do I register to vote?",
+        "city_services": "What city services are available?",
+        "report_issue": "I want to report a problem.",
+        "contact_city": "How do I contact city hall?",
+    },
+    "spa_Latn": {
+        "voting_location": "¿Dónde está mi lugar de votación?",
+        "voter_registration": "¿Cómo me registro para votar?",
+        "city_services": "¿Qué servicios de la ciudad están disponibles?",
+        "report_issue": "Quiero reportar un problema.",
+        "contact_city": "¿Cómo contacto al ayuntamiento?",
+    }
+}
+
+
+def is_translation_available() -> bool:
+    """
+    Check if translation service is available.
+    
+    Returns:
+        bool: True if translation API is configured and ready.
+    """
+    return HF_TOKEN is not None and len(HF_TOKEN) > 0
+
+
+def normalize_language_code(lang: str) -> str:
+    """
+    Converts common language names/codes to NLLB-200 format.
+    
+    Args:
+        lang: Language name or code (e.g., "spanish", "es", "español")
+        
+    Returns:
+        NLLB-200 language code (e.g., "spa_Latn")
+    """
+    if not lang or not isinstance(lang, str):
+        return "eng_Latn"  # Default to English
+    
+    lang_lower = lang.lower().strip()
+    
+    # Check if it's already in NLLB format (contains underscore)
+    if "_" in lang_lower:
+        return lang_lower
+    
+    # Look up in mapping
+    return LANGUAGE_CODES.get(lang_lower, lang_lower)
+
+
+def get_supported_languages() -> List[str]:
+    """
+    Get list of supported language codes.
+    
+    Returns:
+        List of NLLB-200 language codes supported by PENNY.
+    """
+    return list(set(LANGUAGE_CODES.values()))
+
+
+async def translate_text(
+    text: str,
+    source_language: str = "eng_Latn",
+    target_language: str = "spa_Latn",
+    tenant_id: Optional[str] = None
+) -> Dict[str, Any]:
+    """
+    Translates text from source language to target language using NLLB-200.
+
+    Args:
+        text: The text to translate.
+        source_language: Source language code (e.g., "eng_Latn", "spanish", "es")
+        target_language: Target language code (e.g., "spa_Latn", "french", "fr")
+        tenant_id: Optional tenant identifier for logging.
+
+    Returns:
+        A dictionary containing:
+            - translated_text (str): The translated text
+            - source_lang (str): Normalized source language code
+            - target_lang (str): Normalized target language code
+            - original_text (str): The input text
+            - available (bool): Whether the service was available
+            - error (str, optional): Error message if translation failed
+            - response_time_ms (int, optional): Translation time in milliseconds
+    """
+    start_time = time.time()
+
+    # Check availability
+    if not is_translation_available():
+        log_interaction(
+            intent="translation",
+            tenant_id=tenant_id,
+            success=False,
+            error="Translation API not configured (missing HF_TOKEN)",
+            fallback_used=True
+        )
+        return {
+            "translated_text": text,  # Return original text as fallback
+            "source_lang": source_language,
+            "target_lang": target_language,
+            "original_text": text,
+            "available": False,
+            "error": "Translation service is temporarily unavailable."
+        }
+
+    # Validate input
+    if not text or not isinstance(text, str):
+        log_interaction(
+            intent="translation",
+            tenant_id=tenant_id,
+            success=False,
+            error="Invalid text input"
+        )
+        return {
+            "translated_text": "",
+            "source_lang": source_language,
+            "target_lang": target_language,
+            "original_text": text if isinstance(text, str) else "",
+            "available": True,
+            "error": "Invalid text input provided."
+        }
+    
+    # Check text length (prevent processing extremely long texts)
+    if len(text) > 5000:  # 5k character limit for translation
+        log_interaction(
+            intent="translation",
+            tenant_id=tenant_id,
+            success=False,
+            error=f"Text too long: {len(text)} characters",
+            text_preview=sanitize_for_logging(text[:100])
+        )
+        return {
+            "translated_text": text,
+            "source_lang": source_language,
+            "target_lang": target_language,
+            "original_text": text,
+            "available": True,
+            "error": "Text is too long for translation (max 5,000 characters)."
+        }
+
+    # Normalize language codes
+    src_lang = normalize_language_code(source_language)
+    tgt_lang = normalize_language_code(target_language)
+    
+    # Skip translation if source and target are the same
+    if src_lang == tgt_lang:
+        log_interaction(
+            intent="translation_skipped",
+            tenant_id=tenant_id,
+            success=True,
+            details="Source and target languages are identical"
+        )
+        return {
+            "translated_text": text,
+            "source_lang": src_lang,
+            "target_lang": tgt_lang,
+            "original_text": text,
+            "available": True,
+            "skipped": True
+        }
+
+    try:
+        # Prepare API request
+        headers = {"Authorization": f"Bearer {HF_TOKEN}"}
+        payload = {
+            "inputs": text,
+            "parameters": {
+                "src_lang": src_lang,
+                "tgt_lang": tgt_lang
+            }
+        }
+        
+        # Call Hugging Face Inference API
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            response = await client.post(HF_API_URL, json=payload, headers=headers)
+            
+            response_time_ms = int((time.time() - start_time) * 1000)
+            
+            if response.status_code != 200:
+                log_interaction(
+                    intent="translation",
+                    tenant_id=tenant_id,
+                    success=False,
+                    error=f"API returned status {response.status_code}",
+                    response_time_ms=response_time_ms,
+                    source_lang=src_lang,
+                    target_lang=tgt_lang,
+                    fallback_used=True
+                )
+                return {
+                    "translated_text": text,  # Fallback to original
+                    "source_lang": src_lang,
+                    "target_lang": tgt_lang,
+                    "original_text": text,
+                    "available": False,
+                    "error": f"Translation API error: {response.status_code}",
+                    "response_time_ms": response_time_ms
+                }
+            
+            results = response.json()
+        
+        # Validate results
+        if not results or not isinstance(results, list) or len(results) == 0:
+            log_interaction(
+                intent="translation",
+                tenant_id=tenant_id,
+                success=False,
+                error="Empty or invalid model output",
+                response_time_ms=response_time_ms,
+                source_lang=src_lang,
+                target_lang=tgt_lang
+            )
+            return {
+                "translated_text": text,  # Fallback to original
+                "source_lang": src_lang,
+                "target_lang": tgt_lang,
+                "original_text": text,
+                "available": True,
+                "error": "Translation returned unexpected format."
+            }
+        
+        # NLLB returns format: [{'translation_text': '...'}]
+        translated = results[0].get('translation_text', '').strip()
+        
+        if not translated:
+            log_interaction(
+                intent="translation",
+                tenant_id=tenant_id,
+                success=False,
+                error="Empty translation result",
+                response_time_ms=response_time_ms,
+                source_lang=src_lang,
+                target_lang=tgt_lang
+            )
+            return {
+                "translated_text": text,  # Fallback to original
+                "source_lang": src_lang,
+                "target_lang": tgt_lang,
+                "original_text": text,
+                "available": True,
+                "error": "Translation produced empty result."
+            }
+        
+        # Log slow translations
+        if response_time_ms > 5000:  # 5 seconds
+            log_interaction(
+                intent="translation_slow",
+                tenant_id=tenant_id,
+                success=True,
+                response_time_ms=response_time_ms,
+                details="Slow translation detected",
+                source_lang=src_lang,
+                target_lang=tgt_lang,
+                text_length=len(text)
+            )
+        
+        log_interaction(
+            intent="translation",
+            tenant_id=tenant_id,
+            success=True,
+            response_time_ms=response_time_ms,
+            source_lang=src_lang,
+            target_lang=tgt_lang,
+            text_length=len(text)
+        )
+        
+        return {
+            "translated_text": translated,
+            "source_lang": src_lang,
+            "target_lang": tgt_lang,
+            "original_text": text,
+            "available": True,
+            "response_time_ms": response_time_ms
+        }
+
+    except httpx.TimeoutException:
+        response_time_ms = int((time.time() - start_time) * 1000)
+        log_interaction(
+            intent="translation",
+            tenant_id=tenant_id,
+            success=False,
+            error="Translation request timed out",
+            response_time_ms=response_time_ms,
+            source_lang=src_lang,
+            target_lang=tgt_lang,
+            fallback_used=True
+        )
+        return {
+            "translated_text": text,  # Fallback to original
+            "source_lang": src_lang,
+            "target_lang": tgt_lang,
+            "original_text": text,
+            "available": False,
+            "error": "Translation request timed out.",
+            "response_time_ms": response_time_ms
+        }
+
+    except asyncio.CancelledError:
+        log_interaction(
+            intent="translation",
+            tenant_id=tenant_id,
+            success=False,
+            error="Translation cancelled",
+            source_lang=src_lang,
+            target_lang=tgt_lang
+        )
+        raise
+        
+    except Exception as e:
+        response_time_ms = int((time.time() - start_time) * 1000)
+        
+        log_interaction(
+            intent="translation",
+            tenant_id=tenant_id,
+            success=False,
+            error=str(e),
+            response_time_ms=response_time_ms,
+            source_lang=src_lang,
+            target_lang=tgt_lang,
+            text_preview=sanitize_for_logging(text[:100]),
+            fallback_used=True
+        )
+        
+        return {
+            "translated_text": text,  # Fallback to original
+            "source_lang": src_lang,
+            "target_lang": tgt_lang,
+            "original_text": text,
+            "available": False,
+            "error": str(e),
+            "response_time_ms": response_time_ms
+        }
+
+
+async def detect_and_translate(
+    text: str,
+    target_language: str = "eng_Latn",
+    tenant_id: Optional[str] = None
+) -> Dict[str, Any]:
+    """
+    Attempts to detect the source language and translate to target.
+    
+    Note: This is a simplified heuristic-based detection. For production,
+    consider integrating a dedicated language detection model.
+    
+    Args:
+        text: The text to translate
+        target_language: Target language code
+        tenant_id: Optional tenant identifier for logging
+        
+    Returns:
+        Translation result dictionary
+    """
+    if not text or not isinstance(text, str):
+        return {
+            "translated_text": "",
+            "detected_lang": "unknown",
+            "target_lang": target_language,
+            "original_text": text if isinstance(text, str) else "",
+            "available": True,
+            "error": "Invalid text input."
+        }
+    
+    # Simple heuristic: check for common non-English characters
+    detected_lang = "eng_Latn"  # Default assumption
+    
+    # Check for Spanish characters
+    if any(char in text for char in ['¿', '¡', 'ñ', 'á', 'é', 'í', 'ó', 'ú']):
+        detected_lang = "spa_Latn"
+    # Check for Chinese characters
+    elif any('\u4e00' <= char <= '\u9fff' for char in text):
+        detected_lang = "zho_Hans"
+    # Check for Arabic script
+    elif any('\u0600' <= char <= '\u06ff' for char in text):
+        detected_lang = "arb_Arab"
+    # Check for Cyrillic (Russian)
+    elif any('\u0400' <= char <= '\u04ff' for char in text):
+        detected_lang = "rus_Cyrl"
+    # Check for Devanagari (Hindi)
+    elif any('\u0900' <= char <= '\u097f' for char in text):
+        detected_lang = "hin_Deva"
+    
+    log_interaction(
+        intent="language_detection",
+        tenant_id=tenant_id,
+        success=True,
+        detected_lang=detected_lang,
+        text_preview=sanitize_for_logging(text[:50])
+    )
+    
+    result = await translate_text(text, detected_lang, target_language, tenant_id)
+    result["detected_lang"] = detected_lang
+    
+    return result
+
+
+async def batch_translate(
+    texts: List[str],
+    source_language: str = "eng_Latn",
+    target_language: str = "spa_Latn",
+    tenant_id: Optional[str] = None
+) -> List[Dict[str, Any]]:
+    """
+    Translate multiple texts at once.
+    
+    Args:
+        texts: List of strings to translate
+        source_language: Source language code
+        target_language: Target language code
+        tenant_id: Optional tenant identifier for logging
+        
+    Returns:
+        List of translation result dictionaries
+    """
+    if not texts or not isinstance(texts, list):
+        log_interaction(
+            intent="batch_translation",
+            tenant_id=tenant_id,
+            success=False,
+            error="Invalid texts input"
+        )
+        return []
+    
+    # Filter valid texts and limit batch size
+    valid_texts = [t for t in texts if isinstance(t, str) and t.strip()]
+    if len(valid_texts) > 50:  # Batch size limit
+        valid_texts = valid_texts[:50]
+        log_interaction(
+            intent="batch_translation",
+            tenant_id=tenant_id,
+            success=None,
+            details=f"Batch size limited to 50 texts"
+        )
+    
+    if not valid_texts:
+        log_interaction(
+            intent="batch_translation",
+            tenant_id=tenant_id,
+            success=False,
+            error="No valid texts in batch"
+        )
+        return []
+    
+    start_time = time.time()
+    results = []
+    
+    for text in valid_texts:
+        result = await translate_text(text, source_language, target_language, tenant_id)
+        results.append(result)
+    
+    response_time_ms = int((time.time() - start_time) * 1000)
+    
+    log_interaction(
+        intent="batch_translation",
+        tenant_id=tenant_id,
+        success=True,
+        response_time_ms=response_time_ms,
+        batch_size=len(valid_texts),
+        source_lang=normalize_language_code(source_language),
+        target_lang=normalize_language_code(target_language)
+    )
+    
+    return results
+
+
+def get_civic_phrase(
+    phrase_key: str,
+    language: str = "eng_Latn"
+) -> str:
+    """
+    Get a pre-translated civic phrase for common queries.
+    
+    Args:
+        phrase_key: Key for the civic phrase (e.g., "voting_location")
+        language: Target language code
+        
+    Returns:
+        Translated phrase or empty string if not found
+    """
+    if not phrase_key or not isinstance(phrase_key, str):
+        return ""
+    
+    lang_code = normalize_language_code(language)
+    phrase = CIVIC_PHRASES.get(lang_code, {}).get(phrase_key, "")
+    
+    if phrase:
+        log_interaction(
+            intent="civic_phrase_lookup",
+            success=True,
+            phrase_key=phrase_key,
+            language=lang_code
+        )
+    
+    return phrase