"""Madaflix Translator Core. Standalone translation core inspired by the mechanisms present in the user-provided googletrans and deep-translator source trees. No googletrans/deep-translator package is required. Default mode: single_request -> one Google JSON request returns BOTH translation and detected source. Optional mode: hybrid_parallel -> Google JSON detection and deep-translator-style mobile HTML translation are executed concurrently. Environment: TRANSLATOR_MODE=single_request|hybrid_parallel TRANSLATOR_TIMEOUT=8 TRANSLATION_CACHE_TTL=300 TRANSLATION_CACHE_SIZE=1024 TRANSLATOR_PARALLEL_WORKERS=8 """ from __future__ import annotations import copy import gzip import html import http.client import json import os import socket import threading import time from collections import OrderedDict from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass from html.parser import HTMLParser from typing import Any, Dict, Optional, Tuple from urllib.parse import urlencode LANGUAGES = {'aar': 'afar', 'abk': 'abkhaz', 'ace': 'acehnese', 'ach': 'acholi', 'af': 'afrikaans', 'ak': 'twi (akan)', 'alz': 'alur', 'am': 'amharic', 'ar': 'arabic', 'as': 'assamese', 'ava': 'avar', 'awa': 'awadhi', 'ay': 'aymara', 'az': 'azerbaijani', 'bak': 'bashkir', 'bal': 'baluchi', 'ban': 'balinese', 'bbc': 'batak toba', 'bci': 'baoulé', 'be': 'belarusian', 'bem': 'bemba', 'ber': 'tamazight (tifinagh)', 'ber-latn': 'tamazight', 'bew': 'betawi', 'bg': 'bulgarian', 'bho': 'bhojpuri', 'bik': 'bikol', 'bm': 'bambara', 'bn': 'bengali', 'bod': 'tibetan', 'bre': 'breton', 'bs': 'bosnian', 'bts': 'batak simalungun', 'btx': 'batak karo', 'bua': 'buryat', 'ca': 'catalan', 'ceb': 'cebuano', 'cgg': 'kiga', 'cha': 'chamorro', 'che': 'chechen', 'chk': 'chuukese', 'chv': 'chuvash', 'ckb': 'kurdish (sorani)', 'cnh': 'hakha chin', 'co': 'corsican', 'crh': 'crimean tatar', 'crs': 'seychellois creole', 'cs': 'czech', 'cy': 'welsh', 'da': 'danish', 'de': 'german', 'din': 'dinka', 'doi': 'dogri', 'dom': 'dombe', 'dv': 'dhivehi', 'dyu': 'dyula', 'dzo': 'dzongkha', 'el': 'greek', 'en': 'english', 'eo': 'esperanto', 'es': 'spanish', 'et': 'estonian', 'eu': 'basque', 'fa': 'persian', 'fa-af': 'dari', 'fao': 'faroese', 'fi': 'finnish', 'fij': 'fijian', 'fil': 'filipino (tagalog)', 'fon': 'fon', 'fr': 'french', 'ful': 'fulani', 'fur': 'friulian', 'fy': 'frisian', 'ga': 'irish', 'gaa': 'ga', 'gd': 'scots gaelic', 'gl': 'galician', 'glv': 'manx', 'gn': 'guarani', 'gom': 'konkani', 'gu': 'gujarati', 'ha': 'hausa', 'haw': 'hawaiian', 'he': 'hebrew', 'hi': 'hindi', 'hil': 'hiligaynon', 'hmn': 'hmong', 'hr': 'croatian', 'hrx': 'hunsrik', 'ht': 'haitian creole', 'hu': 'hungarian', 'hy': 'armenian', 'iba': 'iban', 'id': 'indonesian', 'ig': 'igbo', 'ilo': 'ilocano', 'is': 'icelandic', 'it': 'italian', 'iw': 'hebrew', 'ja': 'japanese', 'jam': 'jamaican patois', 'jv': 'javanese', 'jw': 'javanese', 'ka': 'georgian', 'kac': 'jingpo', 'kal': 'kalaallisut', 'kau': 'kanuri', 'kek': "q'eqchi'", 'kha': 'khasi', 'kik': 'kikongo', 'kk': 'kazakh', 'km': 'khmer', 'kn': 'kannada', 'ko': 'korean', 'kom': 'komi', 'kri': 'krio', 'ktu': 'kituba', 'ku': 'kurdish', 'ky': 'kyrgyz', 'la': 'latin', 'lb': 'luxembourgish', 'lg': 'luganda', 'lij': 'ligurian', 'lim': 'limburgish', 'lmo': 'lombard', 'ln': 'lingala', 'lo': 'lao', 'lt': 'lithuanian', 'ltg': 'latgalian', 'luo': 'luo', 'lus': 'mizo', 'lv': 'latvian', 'mad': 'madurese', 'mah': 'marshallese', 'mai': 'maithili', 'mak': 'makassar', 'mam': 'mam', 'mfe': 'mauritian creole', 'mg': 'malagasy', 'mhr': 'meadow mari', 'mi': 'maori', 'min': 'minang', 'mk': 'macedonian', 'ml': 'malayalam', 'mn': 'mongolian', 'mni-mtei': 'meiteilon (manipuri)', 'mr': 'marathi', 'ms': 'malay', 'ms-arab': 'malay (jawi)', 'mt': 'maltese', 'mwr': 'marwadi', 'my': 'myanmar (burmese)', 'ndc-zw': 'ndau', 'nde': 'ndebele (south)', 'ne': 'nepali', 'new': 'nepalbhasa (newari)', 'nhe': 'nahuatl (eastern huasteca)', 'nl': 'dutch', 'no': 'norwegian', 'nso': 'sepedi', 'nus': 'nuer', 'ny': 'nyanja (chichewa)', 'oci': 'occitan', 'om': 'oromo', 'or': 'odia (oriya)', 'oss': 'ossetian', 'pa': 'punjabi', 'pa-arab': 'punjabi (shahmukhi)', 'pag': 'pangasinan', 'pam': 'kapampangan', 'pap': 'papiamento', 'pl': 'polish', 'por': 'portuguese (portugal)', 'ps': 'pashto', 'pt': 'portuguese (portugal, brazil)', 'qu': 'quechua', 'ro': 'romanian', 'rom': 'romani', 'ru': 'russian', 'run': 'rundi', 'rw': 'kinyarwanda', 'sa': 'sanskrit', 'sag': 'sango', 'sah': 'yakut', 'sat': 'santali', 'scn': 'sicilian', 'sd': 'sindhi', 'shn': 'shan', 'si': 'sinhala (sinhalese)', 'sk': 'slovak', 'sl': 'slovenian', 'sm': 'samoan', 'sme': 'sami (north)', 'sn': 'shona', 'so': 'somali', 'sq': 'albanian', 'sr': 'serbian', 'ssw': 'swati', 'st': 'sesotho', 'su': 'sundanese', 'sus': 'susu', 'sv': 'swedish', 'sw': 'swahili', 'szl': 'silesian', 'ta': 'tamil', 'tah': 'tahitian', 'tcy': 'tulu', 'te': 'telugu', 'tet': 'tetum', 'tg': 'tajik', 'th': 'thai', 'ti': 'tigrinya', 'tiv': 'tiv', 'tk': 'turkmen', 'tl': 'tagalog (filipino)', 'ton': 'tongan', 'tpi': 'tok pisin', 'tr': 'turkish', 'trp': 'kokborok', 'ts': 'tsonga', 'tsn': 'tswana', 'tt': 'tatar', 'tuk': 'tuvan', 'tum': 'tumbuka', 'udm': 'udmurt', 'ug': 'uyghur', 'uk': 'ukrainian', 'ur': 'urdu', 'uz': 'uzbek', 'vec': 'venetian', 'ven': 'venda', 'vi': 'vietnamese', 'war': 'waray', 'wol': 'wolof', 'xh': 'xhosa', 'yi': 'yiddish', 'yo': 'yoruba', 'yua': 'yucatec maya', 'yue': 'cantonese', 'zap': 'zapotec', 'zh': 'chinese', 'zh-cn': 'chinese (simplified)', 'zh-tw': 'chinese (traditional)', 'zu': 'zulu'} LANGUAGE_COUNTRIES = {'af': 'ZA', 'am': 'ET', 'ar': 'SA', 'az': 'AZ', 'be': 'BY', 'bg': 'BG', 'bn': 'BD', 'bs': 'BA', 'ca': 'ES', 'ceb': 'PH', 'co': 'FR', 'cs': 'CZ', 'cy': 'GB', 'da': 'DK', 'de': 'DE', 'el': 'GR', 'en': 'GB', 'es': 'ES', 'et': 'EE', 'eu': 'ES', 'fa': 'IR', 'fi': 'FI', 'fil': 'PH', 'fr': 'FR', 'fy': 'NL', 'ga': 'IE', 'gd': 'GB', 'gl': 'ES', 'gu': 'IN', 'ha': 'NG', 'haw': 'US', 'he': 'IL', 'hi': 'IN', 'hmn': 'CN', 'hr': 'HR', 'ht': 'HT', 'hu': 'HU', 'hy': 'AM', 'id': 'ID', 'ig': 'NG', 'is': 'IS', 'it': 'IT', 'iw': 'IL', 'ja': 'JP', 'jv': 'ID', 'ka': 'GE', 'kk': 'KZ', 'km': 'KH', 'kn': 'IN', 'ko': 'KR', 'ku': 'IQ', 'ky': 'KG', 'lb': 'LU', 'lo': 'LA', 'lt': 'LT', 'lv': 'LV', 'mg': 'MG', 'mi': 'NZ', 'mk': 'MK', 'ml': 'IN', 'mn': 'MN', 'mr': 'IN', 'ms': 'MY', 'mt': 'MT', 'my': 'MM', 'ne': 'NP', 'nl': 'NL', 'no': 'NO', 'ny': 'MW', 'or': 'IN', 'pa': 'IN', 'pl': 'PL', 'ps': 'AF', 'pt': 'PT', 'ro': 'RO', 'ru': 'RU', 'sd': 'PK', 'si': 'LK', 'sk': 'SK', 'sl': 'SI', 'sm': 'WS', 'sn': 'ZW', 'so': 'SO', 'sq': 'AL', 'sr': 'RS', 'st': 'LS', 'su': 'ID', 'sv': 'SE', 'sw': 'TZ', 'ta': 'IN', 'te': 'IN', 'tg': 'TJ', 'th': 'TH', 'tl': 'PH', 'tr': 'TR', 'uk': 'UA', 'ur': 'PK', 'uz': 'UZ', 'vi': 'VN', 'xh': 'ZA', 'yi': 'IL', 'yo': 'NG', 'zh': 'CN', 'zh-cn': 'CN', 'zh-tw': 'TW', 'zu': 'ZA'} class TranslationServiceError(RuntimeError): pass class UpstreamError(TranslationServiceError): pass @dataclass(frozen=True) class DetectionResult: code: str confidence: Optional[float] @dataclass(frozen=True) class JsonGoogleResult: translated_text: str source_code: str confidence: Optional[float] def _flag(country_code: Optional[str]) -> Optional[str]: if not country_code: return None code = country_code.upper() if len(code) != 2 or not code.isalpha(): return None return "".join(chr(127397 + ord(char)) for char in code) def _language_meta(code: str) -> Dict[str, Any]: normalized = (code or "unknown").lower() country_code = LANGUAGE_COUNTRIES.get(normalized) return { "code": normalized, "name": LANGUAGES.get(normalized, normalized), "country_code": country_code, "flag": _flag(country_code), } class _TTLCache: def __init__(self, max_size: int, ttl_seconds: float): self.max_size = max(0, int(max_size)) self.ttl_seconds = max(0.0, float(ttl_seconds)) self._data = OrderedDict() self._lock = threading.Lock() def get(self, key): if self.max_size == 0 or self.ttl_seconds == 0: return None now = time.monotonic() with self._lock: item = self._data.get(key) if item is None: return None created_at, value = item if now - created_at > self.ttl_seconds: self._data.pop(key, None) return None self._data.move_to_end(key) return copy.deepcopy(value) def set(self, key, value): if self.max_size == 0 or self.ttl_seconds == 0: return with self._lock: self._data[key] = (time.monotonic(), copy.deepcopy(value)) self._data.move_to_end(key) while len(self._data) > self.max_size: self._data.popitem(last=False) class _Transport: """Tiny stdlib HTTP/1.1 client with per-thread persistent connections.""" USER_AGENT = ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 Chrome/124 Safari/537.36" ) def __init__(self, timeout: float): self.timeout = timeout self._local = threading.local() def _connections(self): connections = getattr(self._local, "connections", None) if connections is None: connections = {} self._local.connections = connections return connections def _drop(self, host: str): connections = self._connections() conn = connections.pop(host, None) if conn is not None: try: conn.close() except Exception: pass def _connection(self, host: str): connections = self._connections() conn = connections.get(host) if conn is None: conn = http.client.HTTPSConnection( host, timeout=self.timeout, ) connections[host] = conn return conn def get( self, host: str, path: str, params: Dict[str, Any], accept: str, ) -> str: query = urlencode(params, doseq=True) request_path = path + "?" + query headers = { "User-Agent": self.USER_AGENT, "Accept": accept, "Accept-Language": "en-US,en;q=0.8", "Connection": "keep-alive", } last_error = None # One reconnect retry handles stale keep-alive connections. for attempt in range(2): try: conn = self._connection(host) conn.request("GET", request_path, headers=headers) response = conn.getresponse() body = response.read() status = response.status if response.getheader("Content-Encoding", "").lower() == "gzip": body = gzip.decompress(body) content_type = response.getheader("Content-Type", "") charset = "utf-8" marker = "charset=" if marker in content_type.lower(): charset = content_type.lower().split(marker, 1)[1].split(";", 1)[0].strip() text = body.decode(charset or "utf-8", errors="replace") if status == 429: raise UpstreamError("Google a renvoyé HTTP 429 (Too Many Requests).") if status < 200 or status >= 300: raise UpstreamError("Google a renvoyé HTTP {}.".format(status)) return text except UpstreamError: raise except ( http.client.HTTPException, OSError, socket.timeout, ConnectionError, ) as exc: last_error = exc self._drop(host) if attempt == 0: continue raise UpstreamError( "Erreur réseau vers {}: {}".format(host, last_error) ) class _MobileTranslationParser(HTMLParser): """Minimal replacement for BeautifulSoup for Google's mobile result div.""" TARGET_CLASSES = {"t0", "result-container"} def __init__(self): super().__init__(convert_charrefs=True) self.capture = False self.depth = 0 self.parts = [] self.done = False def handle_starttag(self, tag, attrs): if self.done: return attrs_dict = dict(attrs) if self.capture: self.depth += 1 return if tag.lower() != "div": return classes = set((attrs_dict.get("class") or "").split()) if classes.intersection(self.TARGET_CLASSES): self.capture = True self.depth = 1 def handle_startendtag(self, tag, attrs): return def handle_endtag(self, tag): if not self.capture or self.done: return self.depth -= 1 if self.depth <= 0: self.capture = False self.done = True def handle_data(self, data): if self.capture and not self.done: self.parts.append(data) def result(self) -> Optional[str]: value = html.unescape("".join(self.parts)).strip() return value or None def _looks_like_error_page(value: str) -> bool: normalized = html.unescape(value or "").lower().replace("’", "'") if not normalized.strip(): return True error_markers = ( "that's an error", "there was an error", "please try again later", "server error", "service unavailable", "too many requests", "unusual traffic", ) if any(marker in normalized for marker in error_markers): return True if "error" in normalized: return True return False def _confidence_from_data(data: Any) -> Optional[float]: """Parse confidence similarly to googletrans, with defensive fallbacks.""" try: language_data = data[8] if len(language_data[0]) > 1: raw = language_data[-2] else: raw = language_data[-2][0] value = float(raw) if 0.0 <= value <= 1.0: return value except (TypeError, ValueError, IndexError, KeyError): pass # Defensive search for a likely probability in language metadata only. try: language_data = data[8] except (TypeError, IndexError): return None values = [] def walk(node): if isinstance(node, bool): return if isinstance(node, (int, float)): number = float(node) if 0.0 <= number <= 1.0: values.append(number) elif isinstance(node, list): for item in node: walk(item) walk(language_data) return values[-1] if values else None class TranslationService: GOOGLE_API_HOST = "translate.googleapis.com" GOOGLE_API_PATH = "/translate_a/single" GOOGLE_MOBILE_HOST = "translate.google.com" GOOGLE_MOBILE_PATH = "/m" def __init__(self): self.mode = os.getenv( "TRANSLATOR_MODE", "single_request", ).strip().lower() if self.mode not in {"single_request", "hybrid_parallel"}: raise RuntimeError( "TRANSLATOR_MODE doit être single_request ou hybrid_parallel." ) timeout = float(os.getenv("TRANSLATOR_TIMEOUT", "8")) cache_ttl = float(os.getenv("TRANSLATION_CACHE_TTL", "300")) cache_size = int(os.getenv("TRANSLATION_CACHE_SIZE", "1024")) workers = int(os.getenv("TRANSLATOR_PARALLEL_WORKERS", "8")) self.transport = _Transport(timeout=timeout) self.cache = _TTLCache(cache_size, cache_ttl) # ThreadPoolExecutor creates worker threads lazily. self.executor = ThreadPoolExecutor( max_workers=max(2, workers), thread_name_prefix="translator", ) @staticmethod def _normalize_target(target: str) -> str: value = (target or "").strip().lower().replace("_", "-") if not value: raise TranslationServiceError( "La langue cible 'target' est obligatoire." ) # Compatible aliases. aliases = { "zh-cn": "zh-cn", "zh-tw": "zh-tw", } value = aliases.get(value, value) if value not in LANGUAGES: raise TranslationServiceError( "Langue cible non supportée: {}.".format(target) ) return value def _google_json(self, text: str, target: str) -> JsonGoogleResult: # This is the important optimization: # the SAME request gives us translated text + detected source. params = { "client": "gtx", "sl": "auto", "tl": target, "hl": target, "dt": ["t", "ld"], "ie": "UTF-8", "oe": "UTF-8", "otf": 1, "ssel": 0, "tsel": 0, "q": text, } raw = self.transport.get( self.GOOGLE_API_HOST, self.GOOGLE_API_PATH, params, "application/json,text/plain,*/*", ) if _looks_like_error_page(raw): raise UpstreamError( "Google JSON a renvoyé une page d'erreur." ) try: data = json.loads(raw) except json.JSONDecodeError as exc: raise UpstreamError( "Réponse JSON Google invalide." ) from exc try: translated = "".join( str(part[0] or "") for part in data[0] if isinstance(part, list) and part ).strip() except (TypeError, IndexError) as exc: raise UpstreamError( "Format de traduction Google inattendu." ) from exc if not translated: raise UpstreamError( "Google n'a renvoyé aucun texte traduit." ) source_code = "unknown" try: if isinstance(data[2], str) and data[2]: source_code = data[2].lower() except (TypeError, IndexError): pass # Fallback source parser from language metadata. if source_code == "unknown": try: candidate = data[8][0] if isinstance(candidate, str): source_code = candidate.lower() elif isinstance(candidate, list) and candidate: source_code = str(candidate[0]).lower() except (TypeError, IndexError): pass return JsonGoogleResult( translated_text=translated, source_code=source_code, confidence=_confidence_from_data(data), ) def _google_mobile(self, text: str, target: str) -> str: # Minimal reimplementation of deep-translator GoogleTranslator's # translate.google.com/m strategy. raw = self.transport.get( self.GOOGLE_MOBILE_HOST, self.GOOGLE_MOBILE_PATH, { "sl": "auto", "tl": target, "q": text, }, "text/html,application/xhtml+xml", ) if _looks_like_error_page(raw): raise UpstreamError( "Google Mobile a renvoyé une page d'erreur." ) parser = _MobileTranslationParser() parser.feed(raw) translated = parser.result() if not translated: raise UpstreamError( "Google Mobile: aucune traduction trouvée." ) return translated def _single_request(self, text: str, target: str) -> Dict[str, Any]: result = self._google_json(text, target) return self._build_response( text=text, target=target, translated_text=result.translated_text, source_code=result.source_code, confidence=result.confidence, provider="GoogleTranslateJSON", mode="single_request", ) def _hybrid_parallel(self, text: str, target: str) -> Dict[str, Any]: # Strictly closer to the original architecture: # - JSON/googletrans mechanism for detection # - Mobile/deep-translator mechanism for translation # but BOTH network requests run in parallel. json_future = self.executor.submit( self._google_json, text, target, ) mobile_future = self.executor.submit( self._google_mobile, text, target, ) json_result = None mobile_translation = None json_error = None mobile_error = None try: json_result = json_future.result() except Exception as exc: json_error = exc try: mobile_translation = mobile_future.result() except Exception as exc: mobile_error = exc if json_result is not None and mobile_translation: return self._build_response( text=text, target=target, translated_text=mobile_translation, source_code=json_result.source_code, confidence=json_result.confidence, provider="GoogleMobile", mode="hybrid_parallel", ) # Free fallback: JSON request already contains a translation. if json_result is not None: return self._build_response( text=text, target=target, translated_text=json_result.translated_text, source_code=json_result.source_code, confidence=json_result.confidence, provider="GoogleTranslateJSONFallback", mode="hybrid_parallel", ) # If the JSON endpoint failed but Mobile translated, return the # translation and mark detection unavailable rather than lying. if mobile_translation: return self._build_response( text=text, target=target, translated_text=mobile_translation, source_code="unknown", confidence=None, provider="GoogleMobile", mode="hybrid_parallel", ) raise TranslationServiceError( "Les deux mécanismes ont échoué. JSON: {} | Mobile: {}".format( json_error, mobile_error, ) ) def _build_response( self, text: str, target: str, translated_text: str, source_code: str, confidence: Optional[float], provider: str, mode: str, ) -> Dict[str, Any]: source_meta = _language_meta(source_code) source_meta["confidence"] = confidence target_meta = _language_meta(target) return { "text": text, "translated_text": translated_text, "source": source_meta, "target": target_meta, "engines": { "core": "madaflix-translator-core", "detection": "google-json-core", "translation": "google-core", "translation_provider": provider, "mode": mode, }, } def translate(self, text: str, target: str) -> Dict[str, Any]: clean_text = (text or "").strip() if not clean_text: raise TranslationServiceError( "Le champ 'text' ne peut pas être vide." ) # Keep the same practical limit as deep-translator GoogleTranslator. if len(clean_text) > 5000: raise TranslationServiceError( "Le texte ne peut pas dépasser 5000 caractères." ) normalized_target = self._normalize_target(target) cache_key = (clean_text, normalized_target, self.mode) cached = self.cache.get(cache_key) if cached is not None: cached["engines"]["cache"] = "hit" return cached started = time.perf_counter() try: if self.mode == "hybrid_parallel": result = self._hybrid_parallel( clean_text, normalized_target, ) else: result = self._single_request( clean_text, normalized_target, ) except TranslationServiceError: raise except Exception as exc: raise TranslationServiceError( "Erreur de traduction: {}".format(exc) ) from exc result["engines"]["cache"] = "miss" result["engines"]["elapsed_ms"] = round( (time.perf_counter() - started) * 1000, 2, ) self.cache.set(cache_key, result) return result