# 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