Coverage for src / ezqt_app / services / translation / auto_translator.py: 61.72%
234 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-26 07:07 +0000
1# ///////////////////////////////////////////////////////////////
2# SERVICES.TRANSLATION.AUTO_TRANSLATOR - Automatic translation providers
3# Project: ezqt_app
4# ///////////////////////////////////////////////////////////////
6"""Automatic translation via external providers (disabled by default)."""
8from __future__ import annotations
10# ///////////////////////////////////////////////////////////////
11# IMPORTS
12# ///////////////////////////////////////////////////////////////
13# Standard library imports
14import hashlib
15import json
16import threading
17import time
18from abc import ABC, abstractmethod
19from datetime import datetime, timedelta
20from pathlib import Path
21from typing import Any
23# Third-party imports
24import requests
25from PySide6.QtCore import QObject, Signal
27# Local imports
28from ...utils.diagnostics import warn_tech, warn_user
29from ...utils.printer import get_printer
32# ///////////////////////////////////////////////////////////////
33# CLASSES
34# ///////////////////////////////////////////////////////////////
35class TranslationProvider(ABC):
36 """Base translation provider class"""
38 def __init__(self, name: str, base_url: str):
39 self.name = name
40 self.base_url = base_url
41 self.timeout = 10
42 self.rate_limit_delay = 1.0
44 @abstractmethod
45 def translate(
46 self, text: str, source_lang: str, target_lang: str
47 ) -> str | None: ...
49 def is_available(self) -> bool:
50 try:
51 response = requests.get(self.base_url, timeout=5)
52 return response.status_code == 200
53 except Exception:
54 return False
57class LibreTranslateProvider(TranslationProvider):
58 """LibreTranslate provider"""
60 def __init__(self, api_key: str | None = None, custom_server: str | None = None):
61 server = custom_server or "https://libretranslate.com"
62 super().__init__("LibreTranslate", server)
63 self.api_key = api_key
64 self.rate_limit_delay = 1.0
66 def translate(self, text: str, source_lang: str, target_lang: str) -> str | None:
67 try:
68 url = f"{self.base_url}/translate"
69 data: dict[str, Any] = {
70 "q": text,
71 "source": source_lang,
72 "target": target_lang,
73 "format": "text",
74 }
75 if self.api_key: 75 ↛ 76line 75 didn't jump to line 76 because the condition on line 75 was never true
76 data["api_key"] = self.api_key
78 response = requests.post(
79 url,
80 json=data,
81 headers={"Content-Type": "application/json"},
82 timeout=self.timeout,
83 )
84 if response.status_code == 200: 84 ↛ 86line 84 didn't jump to line 86 because the condition on line 84 was always true
85 return response.json().get("translatedText")
86 if self.base_url == "https://libretranslate.com":
87 warn_user(
88 code="translation.provider.libretranslate.http_error",
89 user_message=(
90 f"LibreTranslate error: {response.status_code}, "
91 "trying alternative server"
92 ),
93 log_message=(
94 f"LibreTranslate error: {response.status_code}, "
95 "trying alternative server"
96 ),
97 )
98 return LibreTranslateProvider(
99 custom_server="https://translate.argosopentech.com"
100 ).translate(text, source_lang, target_lang)
101 warn_tech(
102 code="translation.provider.libretranslate.http_error",
103 message=f"LibreTranslate error: {response.status_code}",
104 )
105 return None
106 except Exception as e:
107 warn_tech(
108 code="translation.provider.libretranslate.exception",
109 message="LibreTranslate exception",
110 error=e,
111 )
112 return None
115class GoogleTranslateProvider(TranslationProvider):
116 """Google Translate Web provider (unofficial)"""
118 def __init__(self):
119 super().__init__("Google Translate", "https://translate.googleapis.com")
120 self.rate_limit_delay = 0.5
122 def translate(self, text: str, source_lang: str, target_lang: str) -> str | None:
123 try:
124 response = requests.get(
125 f"{self.base_url}/translate_a/single",
126 params={
127 "client": "gtx",
128 "sl": source_lang,
129 "tl": target_lang,
130 "dt": "t",
131 "q": text,
132 },
133 timeout=self.timeout,
134 )
135 if response.status_code == 200:
136 data = response.json()
137 if data and len(data) > 0 and len(data[0]) > 0: 137 ↛ 150line 137 didn't jump to line 150 because the condition on line 137 was always true
138 return data[0][0][0]
139 else:
140 warn_tech(
141 code="translation.provider.google.http_error",
142 message=f"Google Translate error: {response.status_code}",
143 )
144 except Exception as e:
145 warn_tech(
146 code="translation.provider.google.exception",
147 message="Google Translate exception",
148 error=e,
149 )
150 return None
153class MyMemoryProvider(TranslationProvider):
154 """MyMemory provider (free, no API key required)"""
156 def __init__(self):
157 super().__init__("MyMemory", "https://api.mymemory.translated.net")
159 def translate(self, text: str, source_lang: str, target_lang: str) -> str | None:
160 try:
161 response = requests.get(
162 f"{self.base_url}/get",
163 params={"q": text, "langpair": f"{source_lang}|{target_lang}"},
164 timeout=self.timeout,
165 )
166 if response.status_code == 200: 166 ↛ 171line 166 didn't jump to line 171 because the condition on line 166 was always true
167 data = response.json()
168 if data.get("responseStatus") == 200:
169 return data.get("responseData", {}).get("translatedText")
170 else:
171 warn_tech(
172 code="translation.provider.mymemory.http_error",
173 message=f"MyMemory error: {response.status_code}",
174 )
175 except Exception as e:
176 warn_tech(
177 code="translation.provider.mymemory.exception",
178 message="MyMemory exception",
179 error=e,
180 )
181 return None
184class TranslationCache:
185 """Translation cache manager"""
187 def __init__(self, cache_file: Path):
188 self.cache_file = cache_file
189 self.cache_data: dict[str, Any] = {}
190 self.max_age_days = 30
191 self.load_cache()
193 def _get_cache_key(self, text: str, source_lang: str, target_lang: str) -> str:
194 return hashlib.md5(
195 f"{text}|{source_lang}|{target_lang}".encode(), usedforsecurity=False
196 ).hexdigest()
198 def get(self, text: str, source_lang: str, target_lang: str) -> str | None:
199 key = self._get_cache_key(text, source_lang, target_lang)
200 entry = self.cache_data.get(key)
201 if entry:
202 created_time = datetime.fromisoformat(entry["created"])
203 if datetime.now() - created_time < timedelta(days=self.max_age_days):
204 return entry["translation"]
205 del self.cache_data[key]
206 return None
208 def set(
209 self,
210 text: str,
211 source_lang: str,
212 target_lang: str,
213 translation: str,
214 provider: str,
215 ) -> None:
216 key = self._get_cache_key(text, source_lang, target_lang)
217 self.cache_data[key] = {
218 "original": text,
219 "translation": translation,
220 "source_lang": source_lang,
221 "target_lang": target_lang,
222 "provider": provider,
223 "created": datetime.now().isoformat(),
224 }
225 self.save_cache()
227 def load_cache(self) -> None:
228 try:
229 if self.cache_file.exists():
230 with open(self.cache_file, encoding="utf-8") as f:
231 self.cache_data = json.load(f)
232 except Exception as e:
233 warn_tech(
234 code="translation.cache.load_failed",
235 message="Error loading cache",
236 error=e,
237 )
238 self.cache_data = {}
240 def save_cache(self) -> None:
241 try:
242 self.cache_file.parent.mkdir(parents=True, exist_ok=True)
243 with open(self.cache_file, "w", encoding="utf-8") as f:
244 json.dump(self.cache_data, f, indent=2, ensure_ascii=False)
245 except Exception as e:
246 warn_tech(
247 code="translation.cache.save_failed",
248 message="Error saving cache",
249 error=e,
250 )
252 def clear_expired(self) -> None:
253 current_time = datetime.now()
254 expired_keys = [
255 key
256 for key, entry in self.cache_data.items()
257 if current_time - datetime.fromisoformat(entry["created"])
258 > timedelta(days=self.max_age_days)
259 ]
260 for key in expired_keys:
261 del self.cache_data[key]
262 if expired_keys:
263 self.save_cache()
266class AutoTranslator(QObject):
267 """Automatic translation manager (disabled by default).
269 Each call to :meth:`translate` spawns a lightweight daemon thread that
270 performs the HTTP round-trip in the background. Signals are emitted from
271 that thread; Qt automatically delivers them via a queued connection to
272 slots that live in the main thread, so the UI is never blocked.
274 A ``_pending`` set (guarded by ``_lock``) deduplicates in-flight requests:
275 if the same source string is requested while a thread for it is still
276 running, no second thread is spawned.
277 """
279 translation_ready = Signal(str, str)
280 translation_error = Signal(str, str)
282 def __init__(self, cache_dir: Path | None = None):
283 super().__init__()
284 if cache_dir is None:
285 cache_dir = Path.home() / ".ezqt" / "cache"
286 cache_dir.mkdir(parents=True, exist_ok=True)
287 self.cache = TranslationCache(cache_dir / "translations.json")
288 self.providers: list[TranslationProvider] = []
289 self._pending: set[str] = set()
290 self._lock = threading.Lock()
291 self._setup_providers()
292 self.enabled = False
294 def _setup_providers(self) -> None:
295 # Order matters: fastest/unofficial first, then free fallback providers.
296 # is_available() performs a synchronous HTTP GET; calling it at setup time
297 # would block the Qt main thread. Provider availability checking requires
298 # a dedicated health-check mechanism before it can be integrated.
299 self.providers = [
300 GoogleTranslateProvider(),
301 MyMemoryProvider(),
302 LibreTranslateProvider(),
303 ]
305 def add_provider(self, provider: TranslationProvider) -> None:
306 self.providers.append(provider)
308 def remove_provider(self, provider_name: str) -> None:
309 self.providers = [p for p in self.providers if p.name != provider_name]
311 def translate(self, text: str, source_lang: str, target_lang: str) -> str | None:
312 """Schedule an async translation and return ``None`` immediately.
314 If *source_lang* equals *target_lang* the method returns *text*
315 immediately (identity translation — no HTTP request is made).
316 If *text* is already cached the cached value is returned immediately.
317 Otherwise a daemon thread is started and ``None`` is returned; the
318 caller receives the result via :attr:`translation_ready`.
319 """
320 if source_lang == target_lang:
321 return text
323 cached = self.cache.get(text, source_lang, target_lang)
324 if cached:
325 return cached
327 with self._lock:
328 if text in self._pending:
329 return None
330 self._pending.add(text)
332 t = threading.Thread(
333 target=self._do_translate,
334 args=(text, source_lang, target_lang),
335 daemon=True,
336 name=f"ez-translate:{text[:30]}",
337 )
338 t.start()
339 return None
341 def _do_translate(self, text: str, source_lang: str, target_lang: str) -> None:
342 """Blocking translation worker — runs in a background daemon thread."""
343 try:
344 for provider in self.providers:
345 try:
346 translation = provider.translate(text, source_lang, target_lang)
347 if translation:
348 self.cache.set(
349 text, source_lang, target_lang, translation, provider.name
350 )
351 get_printer().debug_msg(
352 "[TranslationService] Automatic translation "
353 f"({provider.name}): '{text}' -> '{translation}'"
354 )
355 # Signal is delivered to the main thread via queued connection.
356 self.translation_ready.emit(text, translation)
357 return
358 time.sleep(provider.rate_limit_delay)
359 except Exception as e:
360 warn_tech(
361 code="translation.worker.provider_failed",
362 message=f"Translation error with {provider.name}",
363 error=e,
364 )
366 warn_user(
367 code="translation.auto.failed",
368 user_message=f"Automatic translation failed: '{text}'",
369 log_message=f"All providers failed for '{text}'",
370 )
371 self.translation_error.emit(text, "No translation found")
372 finally:
373 with self._lock:
374 self._pending.discard(text)
376 def translate_sync(
377 self, text: str, source_lang: str, target_lang: str
378 ) -> str | None:
379 """Translate text synchronously, blocking until a result is obtained.
381 Intended for use in CLI scripts, test helpers, and offline batch-processing
382 tools that run outside the Qt event loop. Each provider call is a blocking
383 HTTP request; the total wait time can reach ``len(providers) × timeout``
384 seconds if all providers fail.
386 Warning:
387 **Never call this method from the Qt main (UI) thread.** Doing so
388 blocks the event loop for the entire duration of the HTTP round-trips,
389 freezing the application UI. For in-app translation use
390 :meth:`translate` instead, which runs the request in a daemon thread.
392 Example::
394 # Appropriate usage — called from a CLI script, not from a Qt slot:
395 translator = get_auto_translator()
396 translator.enabled = True
397 result = translator.translate_sync("Hello", "en", "fr")
398 print(result) # "Bonjour"
400 Args:
401 text: The source text to translate.
402 source_lang: BCP-47 language code of the source text (e.g. ``"en"``).
403 target_lang: BCP-47 language code of the desired output (e.g. ``"fr"``).
405 Returns:
406 The translated string, or ``None`` if the translator is disabled or
407 all providers fail.
408 """
409 if not self.enabled:
410 return None
412 cached = self.cache.get(text, source_lang, target_lang)
413 if cached:
414 return cached
416 for provider in self.providers: 416 ↛ 432line 416 didn't jump to line 432 because the loop on line 416 didn't complete
417 try:
418 translation = provider.translate(text, source_lang, target_lang)
419 if translation: 419 ↛ 424line 419 didn't jump to line 424 because the condition on line 419 was always true
420 self.cache.set(
421 text, source_lang, target_lang, translation, provider.name
422 )
423 return translation
424 time.sleep(provider.rate_limit_delay)
425 except Exception as e:
426 warn_tech(
427 code="translation.sync.provider_failed",
428 message=f"Translation error with {provider.name}",
429 error=e,
430 )
432 return None
434 def save_translation_to_ts(
435 self, original: str, translated: str, target_lang: str, ts_file_path: Path
436 ) -> None:
437 """Append a single translation entry to a Qt Linguist .ts XML file."""
438 import xml.etree.ElementTree as ET
440 try:
441 if ts_file_path.exists():
442 try:
443 tree = ET.parse(ts_file_path) # noqa: S314
444 root = tree.getroot()
445 except ET.ParseError:
446 root = ET.Element("TS", {"language": target_lang, "version": "2.1"})
447 tree = ET.ElementTree(root)
448 else:
449 root = ET.Element("TS", {"language": target_lang, "version": "2.1"})
450 tree = ET.ElementTree(root)
452 context = root.find("context")
453 if context is None:
454 context = ET.SubElement(root, "context")
455 ET.SubElement(context, "name").text = "ezqt_app"
457 # Update existing entry if source already present, otherwise append.
458 for msg in context.findall("message"):
459 src = msg.find("source")
460 if src is not None and src.text == original:
461 trans = msg.find("translation")
462 if trans is not None:
463 trans.text = translated
464 break
465 else:
466 msg = ET.SubElement(context, "message")
467 ET.SubElement(msg, "source").text = original
468 ET.SubElement(msg, "translation").text = translated
470 ts_file_path.parent.mkdir(parents=True, exist_ok=True)
471 tree.write(ts_file_path, encoding="unicode", xml_declaration=True)
472 get_printer().debug_msg(
473 f"[TranslationService] Translation saved to {ts_file_path}"
474 )
475 except Exception as e:
476 warn_tech(
477 code="translation.ts.save_failed",
478 message="Error saving translation to .ts file",
479 error=e,
480 )
482 def clear_cache(self) -> None:
483 self.cache.cache_data.clear()
484 self.cache.save_cache()
485 get_printer().debug_msg("[TranslationService] Translation cache cleared")
487 def get_cache_stats(self) -> dict[str, Any]:
488 stats: dict[str, Any] = {
489 "total_entries": len(self.cache.cache_data),
490 "cache_file": str(self.cache.cache_file),
491 "max_age_days": self.cache.max_age_days,
492 }
493 provider_stats: dict[str, int] = {}
494 for entry in self.cache.cache_data.values():
495 p = entry.get("provider", "unknown")
496 provider_stats[p] = provider_stats.get(p, 0) + 1
497 stats["by_provider"] = provider_stats
498 return stats
500 def cleanup(self) -> None:
501 # Background threads are daemon threads — they exit automatically when
502 # the process exits. We only need to flush the on-disk cache.
503 self.cache.clear_expired()
506# ///////////////////////////////////////////////////////////////
507# FUNCTIONS
508# ///////////////////////////////////////////////////////////////
509def get_auto_translator() -> AutoTranslator:
510 """Return the global AutoTranslator singleton."""
511 from .._registry import ServiceRegistry
513 return ServiceRegistry.get(AutoTranslator, AutoTranslator)