""" DeepL Provider - Production-ready implementation. Extends TranslationProvider base class with robust error handling, retry logic, and health monitoring. Features: - Automatic Free/Pro endpoint detection based on API key format - Specific error codes for all DeepL API errors - Retry logic with exponential backoff for transient errors - Timeout configuration - Health check with caching - Structlog-compatible logging (no document content in logs) """ import os import socket import threading import time from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError from datetime import datetime, timezone from typing import Any, Dict, List, Optional from core.logging import get_logger logger = get_logger(__name__) _HAS_STRUCTLOG = True def _log_info(event: str, **kwargs): """Log info message compatible with both structlog and standard logging.""" if _HAS_STRUCTLOG: logger.info(event, **kwargs) else: logger.info(f"{event} {' '.join(f'{k}={v}' for k, v in kwargs.items())}") def _log_warning(event: str, **kwargs): """Log warning message compatible with both structlog and standard logging.""" if _HAS_STRUCTLOG: logger.warning(event, **kwargs) else: logger.warning(f"{event} {' '.join(f'{k}={v}' for k, v in kwargs.items())}") def _log_error(event: str, **kwargs): """Log error message compatible with both structlog and standard logging.""" if _HAS_STRUCTLOG: logger.error(event, **kwargs) else: logger.error(f"{event} {' '.join(f'{k}={v}' for k, v in kwargs.items())}") from .base import TranslationProvider from .schemas import ( BatchTranslationRequest, BatchTranslationResponse, ProviderHealthStatus, TranslationRequest, TranslationResponse, ) DEEPL_QUOTA_EXCEEDED = "DEEPL_QUOTA_EXCEEDED" DEEPL_INVALID_KEY = "DEEPL_INVALID_KEY" DEEPL_NETWORK_ERROR = "DEEPL_NETWORK_ERROR" DEEPL_UNSUPPORTED_LANGUAGE = "DEEPL_UNSUPPORTED_LANGUAGE" DEEPL_TEXT_TOO_LONG = "DEEPL_TEXT_TOO_LONG" _RETRYABLE_ERRORS = {DEEPL_NETWORK_ERROR, DEEPL_QUOTA_EXCEEDED} DEEPL_FREE_SUFFIX = ":fx" MAX_TEXT_LENGTH = 128 * 1024 DEEPL_SUPPORTED_LANGUAGES = { "BG", "CS", "DA", "DE", "EL", "EN-GB", "EN-US", "ES", "ET", "FI", "FR", "HU", "ID", "IT", "JA", "KO", "LT", "LV", "NB", "NL", "PL", "PT-BR", "PT-PT", "RO", "RU", "SK", "SL", "SV", "TR", "UK", "ZH", } class DeepLProviderError(Exception): """Exception raised for DeepL API errors.""" def __init__( self, code: str, message: str, details: Optional[Dict[str, Any]] = None ): self.code = code self.message = message self.details = details or {} super().__init__(message) def to_dict(self) -> Dict[str, Any]: """Convert error to dictionary format.""" result = { "error": self.code, "message": self.message, } if self.details: result["details"] = self.details return result class DeepLTranslationProvider(TranslationProvider): """ DeepL implementation using deep_translator library. Features: - Automatic Free/Pro endpoint detection based on API key format - Thread-safe translator instances per thread - Caching support (uses global cache from translation_service) - Batch translation with optimized processing - Robust error handling with specific error codes - Retry logic with exponential backoff - Configurable timeout - Health check with result caching """ def __init__( self, api_key: str, use_cache: bool = True, timeout: int = 30, max_retries: int = 3, retry_delay: float = 1.0, ): """ Initialize DeepL provider. Args: api_key: DeepL API key (Free keys end with :fx) use_cache: Whether to use translation caching (default: True) timeout: Request timeout in seconds (default: 30) max_retries: Maximum retry attempts for transient errors (default: 3) retry_delay: Initial retry delay in seconds (default: 1.0) """ if not api_key: raise ValueError("DeepL API key is required") self._api_key = api_key self._api_type = self._detect_api_type(api_key) self._local = threading.local() self._use_cache = use_cache self._provider_name = "deepl" self._cache = None self.timeout = timeout self.max_retries = max_retries self.retry_delay = retry_delay self._health_cache: Dict[str, Any] = {} self._health_cache_ttl = 60 self._health_cache_lock = threading.Lock() if use_cache: self._init_cache() def _detect_api_type(self, api_key: str) -> str: """ Detect if API key is Free or Pro based on suffix. Free tier keys end with ':fx', Pro keys do not. Args: api_key: DeepL API key Returns: "free" or "pro" """ if api_key.endswith(DEEPL_FREE_SUFFIX): return "free" return "pro" def _get_api_url(self) -> str: """ Get correct API URL based on key type. Note: deep_translator handles this internally, but we log it. Returns: API URL for Free or Pro tier """ if self._api_type == "free": return "https://api-free.deepl.com/v2/translate" return "https://api.deepl.com/v2/translate" def _init_cache(self): """Initialize or get the translation cache.""" from services.translation_service import _translation_cache self._cache = _translation_cache def _normalize_language_code(self, lang_code: str) -> str: """ Normalize language code for DeepL. DeepL uses uppercase language codes (e.g., "EN-US", "FR"). Args: lang_code: Input language code (e.g., "en", "en-US", "EN-us") Returns: Normalized language code for DeepL """ if not lang_code or lang_code.lower() == "auto": return "" lang_upper = lang_code.upper() if lang_upper in DEEPL_SUPPORTED_LANGUAGES: return lang_upper base_lang = lang_upper.split("-")[0] if base_lang == "EN": return "EN-US" elif base_lang == "PT": return "PT-BR" elif base_lang in { "BG", "CS", "DA", "DE", "EL", "ES", "ET", "FI", "FR", "HU", "ID", "IT", "JA", "KO", "LT", "LV", "NB", "NL", "PL", "RO", "RU", "SK", "SL", "SV", "TR", "UK", "ZH", }: return base_lang return lang_upper def _is_language_supported(self, lang_code: str) -> bool: """ Check if a language code is supported by DeepL. Args: lang_code: Language code to check Returns: True if supported, False otherwise """ if not lang_code: return True normalized = self._normalize_language_code(lang_code) return normalized in DEEPL_SUPPORTED_LANGUAGES def _get_translator(self, source_language: str, target_language: str): """Get or create a translator instance for the current thread.""" from deep_translator import DeepLTranslator source_lang = self._normalize_language_code(source_language) target_lang = self._normalize_language_code(target_language) key = f"{source_lang}_{target_lang}" if not hasattr(self._local, "translators"): self._local.translators = {} if key not in self._local.translators: self._local.translators[key] = DeepLTranslator( api_key=self._api_key, source=source_lang if source_lang else "auto", target=target_lang, ) return self._local.translators[key] def _make_api_request( self, text: str, source_language: str, target_language: str ) -> str: """ Make API request with error mapping. Raises: DeepLProviderError: For any API errors with specific codes """ if len(text.encode("utf-8")) > MAX_TEXT_LENGTH: raise DeepLProviderError( code=DEEPL_TEXT_TOO_LONG, message="Texte trop long (max 128KB par requête).", details={"text_length": len(text), "max_length": MAX_TEXT_LENGTH}, ) if not self._is_language_supported(target_language): raise DeepLProviderError( code=DEEPL_UNSUPPORTED_LANGUAGE, message=f"Langue '{target_language}' non supportée par DeepL.", details={"unsupported_language": target_language}, ) try: translator = self._get_translator(source_language, target_language) with ThreadPoolExecutor(max_workers=1) as executor: future = executor.submit(translator.translate, text) return future.result(timeout=self.timeout) except Exception as e: error_str = str(e).lower() if ( "quota" in error_str or "limit" in error_str or "429" in error_str or "456" in error_str ): raise DeepLProviderError( code=DEEPL_QUOTA_EXCEEDED, message="Quota DeepL dépassé. Réessayez demain.", details={"provider": "deepl", "api_type": self._api_type}, ) elif ( "auth" in error_str or "key" in error_str or "invalid" in error_str or "401" in error_str or "403" in error_str ): raise DeepLProviderError( code=DEEPL_INVALID_KEY, message="Clé API DeepL invalide. Contactez l'administrateur.", details={"provider": "deepl"}, ) elif "language" in error_str or "not supported" in error_str: raise DeepLProviderError( code=DEEPL_UNSUPPORTED_LANGUAGE, message=f"Langue '{target_language}' non supportée par DeepL.", details={"unsupported_language": target_language}, ) elif ( isinstance(e, (socket.timeout, TimeoutError, FuturesTimeoutError)) or "timeout" in error_str ): raise DeepLProviderError( code=DEEPL_NETWORK_ERROR, message="Service DeepL indisponible. Réessayez.", details={"provider": "deepl", "error_type": "timeout"}, ) else: raise DeepLProviderError( code=DEEPL_NETWORK_ERROR, message="Service DeepL indisponible. Réessayez.", details={"provider": "deepl", "original_error": str(e)[:100]}, ) def get_name(self) -> str: """Return provider name.""" return self._provider_name def is_available(self) -> bool: """ Check if DeepL is available (API key configured and API reachable). Performs a minimal translate call to verify the API is actually reachable. Uses cached result if available and not expired (TTL 60s). """ current_time = time.time() with self._health_cache_lock: if "is_available" in self._health_cache: cached = self._health_cache["is_available"] if current_time - cached["timestamp"] < self._health_cache_ttl: return cached["value"] available = False try: translator = self._get_translator("en", "fr") with ThreadPoolExecutor(max_workers=1) as executor: future = executor.submit(translator.translate, "a") future.result(timeout=5) available = True except Exception as e: _log_warning( "deepl_availability_check_failed", error=str(e)[:100], ) with self._health_cache_lock: self._health_cache["is_available"] = { "value": available, "timestamp": current_time, } return available def translate_text(self, request: TranslationRequest) -> TranslationResponse: """ Translate a single text string using DeepL. API Usage Notes: - DeepL Free tier: 500,000 characters/month - DeepL Pro: ~€25 per million characters - 128KB max per request Optimization: Skips API call if source == target language. Args: request: TranslationRequest with text and language info Returns: TranslationResponse with translated text """ text = request.text target_language = request.target_language source_language = request.source_language or "auto" if not text or not text.strip(): return TranslationResponse( translated_text=text, provider_name=self._provider_name, from_cache=False, ) norm_source = self._normalize_language_code(source_language) norm_target = self._normalize_language_code(target_language) if norm_source and norm_source == norm_target: _log_info( "deepl_translation_skip", source_target_lang=target_language, text_length=len(text), ) return TranslationResponse( translated_text=text, provider_name=self._provider_name, from_cache=False, source_language=source_language, ) if self._use_cache and self._cache: cached = self._cache.get( text, target_language, source_language, self._provider_name ) if cached is not None: return TranslationResponse( translated_text=cached, provider_name=self._provider_name, from_cache=True, ) last_error: Optional[DeepLProviderError] = None retries = 0 while retries <= self.max_retries: try: result = self._make_api_request(text, source_language, target_language) if self._use_cache and self._cache: self._cache.set( text, target_language, source_language, self._provider_name, result, ) _log_info( "deepl_translation_success", chars=len(text), source_lang=source_language, target_lang=target_language, api_type=self._api_type, retries=retries, ) return TranslationResponse( translated_text=result, provider_name=self._provider_name, from_cache=False, ) except DeepLProviderError as e: last_error = e if e.code not in _RETRYABLE_ERRORS: break retries += 1 if retries <= self.max_retries: delay = self.retry_delay * (2 ** (retries - 1)) _log_info( "deepl_translation_retry", attempt=retries, delay_s=round(delay, 2), error_code=e.code, text_length=len(text), source_lang=source_language, target_lang=target_language, ) time.sleep(delay) except Exception as e: last_error = DeepLProviderError( code=DEEPL_NETWORK_ERROR, message="Service DeepL indisponible. Réessayez.", details={"original_error": str(e)[:100]}, ) retries += 1 if retries <= self.max_retries: delay = self.retry_delay * (2 ** (retries - 1)) time.sleep(delay) if last_error: _log_error( "deepl_translation_failed", error_code=last_error.code, text_length=len(text), source_lang=source_language, target_lang=target_language, retries=retries, ) return TranslationResponse( translated_text=text, provider_name=self._provider_name, from_cache=False, error=last_error.message, error_code=last_error.code, error_details=last_error.details, ) return TranslationResponse( translated_text=text, provider_name=self._provider_name, from_cache=False, error="Unknown error", error_code=DEEPL_NETWORK_ERROR, ) def translate_batch( self, requests: List[TranslationRequest] ) -> List[TranslationResponse]: """ Translate multiple texts with optimized batch processing. Args: requests: List of TranslationRequest objects Returns: List of TranslationResponse objects """ if not requests: return [] return [self.translate_text(req) for req in requests] def health_check(self) -> ProviderHealthStatus: """ Return health status details for the provider. Performs a lightweight check to verify the provider is operational. Includes cached result for efficiency. Returns: ProviderHealthStatus with availability and latency information """ current_time = time.time() with self._health_cache_lock: if "health_check" in self._health_cache: cached = self._health_cache["health_check"] if current_time - cached["timestamp"] < self._health_cache_ttl: return cached["value"] start_time = time.time() last_check_iso = datetime.now(timezone.utc).isoformat() try: available = self.is_available() latency_ms = (time.time() - start_time) * 1000 status = ProviderHealthStatus( name=self._provider_name, available=available, latency_ms=round(latency_ms, 2), error=None if available else "Provider not available", last_check=last_check_iso, ) except Exception as e: latency_ms = (time.time() - start_time) * 1000 status = ProviderHealthStatus( name=self._provider_name, available=False, latency_ms=round(latency_ms, 2), error=str(e)[:100], last_check=last_check_iso, ) with self._health_cache_lock: self._health_cache["health_check"] = { "value": status, "timestamp": current_time, } return status def register_deepl_provider(): """ Register the DeepL provider in the global registry. This function should be called during module initialization to make the provider available through the registry. """ from .registry import registry provider = get_deepl_provider() if provider: registry.register("deepl", provider) return provider _provider_instance = None _provider_instance_lock = threading.Lock() def get_deepl_provider() -> Optional[DeepLTranslationProvider]: """Get or create the DeepL provider instance (reads config from env). Thread-safe.""" global _provider_instance if _provider_instance is None: with _provider_instance_lock: if _provider_instance is None: from .config import ProvidersConfig if not ProvidersConfig.DEEPL_API_KEY: return None _provider_instance = DeepLTranslationProvider( api_key=ProvidersConfig.DEEPL_API_KEY, use_cache=True, timeout=getattr(ProvidersConfig, "DEEPL_TIMEOUT", 30), max_retries=getattr(ProvidersConfig, "DEEPL_MAX_RETRIES", 3), retry_delay=getattr(ProvidersConfig, "DEEPL_RETRY_DELAY", 1.0), ) return _provider_instance class LegacyDeepLAdapter: """ Exposes the new DeepLTranslationProvider via the legacy interface used by translation_service: .translate(text, target_lang, source_lang) -> str and .translate_batch(texts, target_lang, source_lang) -> List[str]. Raises TranslationProviderError on failure so the API can return 4xx/502. """ def __init__(self): self._provider = get_deepl_provider() self.provider_name = "deepl" def translate( self, text: str, target_language: str, source_language: str = "auto" ) -> str: if not self._provider: from utils.exceptions import TranslationProviderError raise TranslationProviderError( "DEEPL_NOT_CONFIGURED", "DeepL provider not configured. Set DEEPL_API_KEY.", None, ) req = TranslationRequest( text=text, target_language=target_language, source_language=source_language, ) resp = self._provider.translate_text(req) if resp.error: from utils.exceptions import TranslationProviderError raise TranslationProviderError( resp.error_code or "UNKNOWN", resp.error or "Translation failed", resp.error_details, ) return resp.translated_text def translate_batch( self, texts: List[str], target_language: str, source_language: str = "auto", batch_size: int = 50, ) -> List[str]: if not self._provider: from utils.exceptions import TranslationProviderError raise TranslationProviderError( "DEEPL_NOT_CONFIGURED", "DeepL provider not configured. Set DEEPL_API_KEY.", None, ) requests = [ TranslationRequest( text=t, target_language=target_language, source_language=source_language, ) for t in texts ] responses = self._provider.translate_batch(requests) result = [] for r in responses: if r.error: from utils.exceptions import TranslationProviderError raise TranslationProviderError( r.error_code or "UNKNOWN", r.error or "Translation failed", r.error_details, ) result.append(r.translated_text) return result def get_legacy_deepl_adapter() -> LegacyDeepLAdapter: """Return an adapter so the legacy translation_service can use the new provider.""" return LegacyDeepLAdapter()