diff --git a/routes/translate_routes.py b/routes/translate_routes.py index adcd1b6..b52fcdb 100644 --- a/routes/translate_routes.py +++ b/routes/translate_routes.py @@ -368,6 +368,35 @@ async def download_from_url(url: str, timeout: int = 30) -> tuple[Path, str]: _translation_jobs: dict[str, dict] = {} _JOB_TTL_SECONDS = 3600 _last_cleanup_ts: float = 0.0 + +# Google Cloud API key validity cache — avoids probing the API on every request. +_gc_key_cache: dict[str, tuple[bool, float]] = {} +_GC_KEY_CACHE_TTL = 600 # 10 minutes + + +def _google_cloud_key_valid(api_key: str, job_id: str) -> bool: + """Check if a Google Cloud API key is valid, with a 10-minute cache.""" + import time + now = time.time() + cached = _gc_key_cache.get(api_key) + if cached: + is_valid, ts = cached + if now - ts < _GC_KEY_CACHE_TTL: + return is_valid + # Probe the API with a tiny translation + try: + from services.providers.google_cloud_provider import LegacyGoogleCloudAdapter + _test = LegacyGoogleCloudAdapter(api_key) + _test.translate("test", "fr", "en") + _gc_key_cache[api_key] = (True, now) + return True + except Exception as _gc_err: + logger.warning( + "google_cloud_key_invalid", + extra={"job_id": job_id, "error": str(_gc_err)[:200]}, + ) + _gc_key_cache[api_key] = (False, now) + return False _CLEANUP_INTERVAL_SECONDS = 300 # run cleanup every 5 minutes at most @@ -999,23 +1028,23 @@ async def _run_translation_job( _p = provider.lower() # "google" (default classic mode): use Google Cloud API key if available. - # We no longer fall back silently to the free legacy translator if the Cloud API fails. - # If the key is invalid or quota is exceeded, we want the error to bubble up to the user. + # If the Cloud API key is invalid or the API is not enabled, fall back + # to the free legacy Google Translate (deep_translator) instead of failing. if _p == "google": # the user might have set GOOGLE_API_KEY instead of GOOGLE_CLOUD_API_KEY gc_key = _cfg( getattr(_admin_cfg.google_cloud, "api_key", None), "GOOGLE_CLOUD_API_KEY", ) or os.getenv("GOOGLE_API_KEY", "").strip() - - if gc_key: + + if gc_key and _google_cloud_key_valid(gc_key, job_id): from services.providers.google_cloud_provider import LegacyGoogleCloudAdapter translation_provider = LegacyGoogleCloudAdapter(gc_key) logger.info("google_provider_using_cloud_api", extra={"job_id": job_id}) else: from services.translation_service import GoogleTranslationProvider translation_provider = GoogleTranslationProvider() - logger.info("google_provider_no_cloud_key_using_legacy", extra={"job_id": job_id}) + logger.info("google_provider_using_legacy", extra={"job_id": job_id}) elif _p in ("openrouter", "llm") and api_key: translation_provider = OpenRouterTranslationProvider(