fix: fallback to free Google Translate when Cloud API key is invalid/blocked
All checks were successful
Deploy to Production / Build and Deploy (push) Successful in 2m5s
All checks were successful
Deploy to Production / Build and Deploy (push) Successful in 2m5s
Google Cloud Translation API was returning "Requests to this API are blocked" which got wrapped as a misleading "Erreur lors de la lecture du fichier PowerPoint". Now probes the key once (cached 10min) and falls back to deep_translator (free) when the Cloud key is invalid. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -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(
|
||||
|
||||
Reference in New Issue
Block a user