# -*- coding: utf-8 -*- """ i18n_gtranslate — Batch Google Cloud Translation - Liste les langues supportées (--list-languages) - Traduit en masse les TranslationRecord ciblés par app / view / url / template root - API: google.cloud.translate_v3.TranslationServiceClient Exemples : # A) Lister les langues supportées python manage.py i18n_gtranslate --project YOUR_GCP_PROJECT --location global --list-languages # B) Traduire toute l’app 'accounts' vers EN et DE (sans écraser l’existant) python manage.py i18n_gtranslate --app accounts --targets en,de --project YOUR_GCP_PROJECT --location global # C) Traduire les textes rattachés à une view précise python manage.py i18n_gtranslate --view accounts.views.dashboard_home --targets en --project YOUR_GCP_PROJECT # D) Par URL (résolue en view) python manage.py i18n_gtranslate --url /accounts/dashboard/ --targets en --project YOUR_GCP_PROJECT # E) Par template root (et sa closure extends/include) python manage.py i18n_gtranslate --template accounts/dashboard_home.html --targets es --project YOUR_GCP_PROJECT # F) Forcer la retraduction des lignes déjà traduites (overwrite) python manage.py i18n_gtranslate --app accounts --targets en --overwrite --project YOUR_GCP_PROJECT """ from __future__ import annotations import itertools import os import re from dataclasses import dataclass from pathlib import Path from typing import Dict, Iterable, List, Optional, Sequence, Tuple from django.core.management.base import BaseCommand, CommandError from django.db import transaction from django.urls import resolve, Resolver404 from django.template import engines from django.conf import settings from django.db import models # 1) pour models.Q dans _pick_records #=============================================== #==== GOOGLE TRANSLATION ADDON === from google.api_core.exceptions import GoogleAPICallError from google.cloud import translate_v3 as translate # pip install google-cloud-translate from google.oauth2 import service_account # 2) pour service_account.Credentials... from translation.models import ( SubProject, DjangoViewRef, HtmlTemplateRef2025, ViewTemplateMap, TranslationKey, TranslationRecord, TranslationRecordVersion, ) def _ensure_targets_for_scope(app: Optional[str], view_dotted: Optional[str], template_root: Optional[str], targets: Sequence[str]) -> Dict[str, int]: """Crée les TranslationRecord cibles manquants pour toutes les clés du périmètre.""" diag = _diagnose_scope(app, view_dotted, template_root, targets) counts = {lang: 0 for lang in targets} # récupère les id des clés dans le scope tpls = diag["templates"] kqs = TranslationKey.objects.all() if app: kqs = kqs.filter(subproject__code=app) if tpls: kqs = kqs.filter(template__template__in=tpls) key_ids = list(kqs.values_list("id", flat=True)) for lang in targets: existing = set(TranslationRecord.objects.filter( key_ref_id__in=key_ids, language=lang ).values_list("key_ref_id", flat=True)) missing_ids = [kid for kid in key_ids if kid not in existing] # bulk-create simple (vide, PENDING) to_create = [ TranslationRecord( key_ref_id=kid, language=lang, text="", is_original=False, is_translated=False, status=TranslationRecord.Status.PENDING, engine=TranslationRecord.Engine.MANUAL ) for kid in missing_ids ] if to_create: TranslationRecord.objects.bulk_create(to_create, ignore_conflicts=True) counts[lang] += len(to_create) return counts from collections import defaultdict def _scope_templates(app: Optional[str], view_dotted: Optional[str], template_root: Optional[str]) -> List[str]: """Retourne la liste des templates couverts par le périmètre passé à la commande.""" if template_root: return _closure_from_root_template(template_root) if view_dotted: v = (DjangoViewRef.objects .filter(dotted_path=view_dotted) .filter(subproject__code=app) if app else DjangoViewRef.objects.filter(dotted_path=view_dotted) ).first() if not v: return [] return list(v.templates.select_related("tpl").values_list("tpl__template", flat=True).distinct()) if app: return list(HtmlTemplateRef2025.objects.filter(subproject__code=app).values_list("template", flat=True)) return [] def _diagnose_scope(app: Optional[str], view_dotted: Optional[str], template_root: Optional[str], targets: Sequence[str]) -> Dict[str, object]: """ Bilan par TranslationKey (pivot) : - templates couverts - clés totales (ids, keys, template) - par langue : combien de records existent / manquent """ tpls = _scope_templates(app, view_dotted, template_root) kqs = TranslationKey.objects.all() if app: kqs = kqs.filter(subproject__code=app) if tpls: kqs = kqs.filter(template__template__in=tpls) key_rows = list(kqs.values_list("id", "key", "template__template")) key_ids = [r[0] for r in key_rows] id2meta = {kid: (k, tpl or "") for kid, k, tpl in key_rows} per_lang = {} for lang in targets: have = set(TranslationRecord.objects.filter(key_ref_id__in=key_ids, language=lang).values_list("key_ref_id", flat=True)) miss = [kid for kid in key_ids if kid not in have] per_lang[lang] = { "existing": len(have), "missing": len(miss), "missing_keys": [(id2meta[m][0], id2meta[m][1]) for m in miss], } return { "templates": tpls, "keys_total": len(key_ids), "per_lang": per_lang, } # ---------- Helpers sélection ---------- def _resolve_url_to_dotted(url_path: str) -> Optional[str]: """Résout /path/ -> dotted path de la vue Django.""" try: m = resolve(url_path) except Resolver404: return None view_func = m.func # CBV ? view_class = getattr(view_func, "view_class", None) if view_class: return f"{view_class.__module__}.{view_class.__name__}" # FBV return f"{view_func.__module__}.{view_func.__name__}" def _templates_for_view(subproject_code: str, dotted_path: str) -> List[str]: v = (DjangoViewRef.objects .filter(subproject__code=subproject_code, dotted_path=dotted_path) .first()) if not v: return [] # via ViewTemplateMap (direct + extends + include) return list( v.templates.select_related("tpl").values_list("tpl__template", flat=True).distinct() ) def _template_abs_path(tpl_name: str) -> Optional[str]: try: tmpl = engines["django"].get_template(tpl_name) except Exception: return None origin = getattr(tmpl, "origin", None) if origin and getattr(origin, "name", None): return origin.name inner = getattr(tmpl, "template", None) origin = getattr(inner, "origin", None) if inner else None return origin.name if origin and getattr(origin, "name", None) else None RE_EXTENDS = re.compile(r'{%\s*extends\s+["\']([^"\']+)["\']\s*%}') RE_INCLUDE = re.compile(r'{%\s*include\s+["\']([^"\']+)["\']\s*%}') def _read(tpl_name: str) -> str: p = _template_abs_path(tpl_name) return Path(p).read_text(encoding="utf-8") if p else "" def _closure_from_root_template(root: str) -> List[str]: """Lit le fichier root et suit extends/include récursivement (sans dupliquer).""" out, seen, stack = [], set(), [root] while stack: t = stack.pop() if t in seen: continue seen.add(t) out.append(t) src = _read(t) if not src: continue for m in RE_EXTENDS.finditer(src): stack.append(m.group(1)) for m in RE_INCLUDE.finditer(src): stack.append(m.group(1)) return out # ---------- Google client ---------- @dataclass class GClient: client: translate.TranslationServiceClient project: str location: str @property def parent(self) -> str: return f"projects/{self.project}/locations/{self.location}" def _build_client(project: Optional[str], location: Optional[str]) -> GClient: # 1) project : cli -> env -> settings project = ( project or os.environ.get("GCP_PROJECT") or os.environ.get("GOOGLE_CLOUD_PROJECT") or getattr(settings, "GOOGLE_CLOUD_PROJECT", None) ) if not project: raise CommandError("Spécifiez --project ou définissez settings.GOOGLE_CLOUD_PROJECT / $GOOGLE_CLOUD_PROJECT.") # 2) location : cli -> env -> settings -> 'global' location = ( location or os.environ.get("GOOGLE_CLOUD_LOCATION") or getattr(settings, "GOOGLE_CLOUD_LOCATION", None) or "global" ) # 3) credentials : ADC via env, sinon le fichier défini en settings cred_path = os.environ.get("GOOGLE_APPLICATION_CREDENTIALS") or getattr(settings, "GOOGLE_CREDENTIALS_FILE", "") if cred_path and os.path.exists(cred_path): creds = service_account.Credentials.from_service_account_file(cred_path) client = translate.TranslationServiceClient(credentials=creds) else: # ADC (utilisera GOOGLE_APPLICATION_CREDENTIALS si présent) client = translate.TranslationServiceClient() return GClient(client, project, location) #------------------------------------------------------- # ---------- Sélection des records à traduire ---------- #------------------------------------------------------- def _pick_records( *, app : str , dotted_view : str | None = None , url_path : str | None = None , template_root : str | None = None , targets : list[str] | tuple[str, ...] = () , only_missing : bool = True , status_in : list[str] | tuple[str, ...] | None = None , overwrite : bool = False , commons : bool = False , ): """ Sélectionne les TranslationRecord à traduire pour le périmètre donné. - 'missing' = is_translated=False OU text vide/blanc - si overwrite=True => on ne filtre pas par missing (mais on exclut toujours originaux/supprimés/inactifs) """ # en tout début de _pick_records(...) commons = bool(locals().get("commons", False)) # on le passera depuis handle() #------------------------------------ # 0) URL -> dotted si nécessaire #------------------------------------ if url_path and not dotted_view: dotted_view = _resolve_url_to_dotted(url_path) #--------------------------------------------------------------- # 1) Construire le périmètre de templates (root / view / app) #--------------------------------------------------------------- templates = _scope_templates(app, dotted_view, template_root) #-------------------------------- # 2) Clés pivot dans ce scope #-------------------------------- if commons: keys_qs = TranslationKey.objects.filter(scope=TranslationKey.Scope.COMMON) else: keys_qs = TranslationKey.objects.all() if app: keys_qs = keys_qs.filter(subproject__code=app) if templates: keys_qs = keys_qs.filter(template__template__in=templates) #------------------------------------------------------- # 3) Records cibles (langues demandées) + garde-fous #------------------------------------------------------- qs = ( TranslationRecord.objects .filter(language__in=targets, key_ref__in=keys_qs) .select_related("key_ref") .filter(is_original=False, is_deleted=False, is_active=True) ) #------------------------------------------------------------------ # 4) Filtre statut (on accepte aussi les statuts vides / NULL) #------------------------------------------------------------------ if status_in: allowed = [s.strip().lower() for s in status_in if s.strip()] qs = qs.filter( models.Q(status__in=allowed) | models.Q(status__isnull=True) | models.Q(status="") # statut vide affiché comme '-' en admin ) #----------------------------------------------------------------- # 5) Missing (si pas overwrite) : inclure textes vides / blancs #----------------------------------------------------------------- if only_missing and not overwrite: empty = models.Q(text__isnull=True) | models.Q(text="") | models.Q(text__regex=r"^\s+$") qs = qs.filter(models.Q(is_translated=False) | empty) #---------------------------------- # 6) ultimate control of status #---------------------------------- picked = list(qs) if not picked and keys_qs.exists(): print("[i18n] 0 record sélectionné — pensez à --status-in \"\" ou --overwrite si les statuts sont vides.") return picked # ---------- Util : source text par key ---------- def _get_source_text(rec: TranslationRecord) -> Optional[Tuple[str, str]]: """Retourne (lang_source, texte_source) pour la même clé.""" src = (TranslationRecord.objects .filter(key_ref=rec.key_ref, is_original=True) .order_by("-updated_at") .first()) if not src or not (src.text or "").strip(): return None return (src.language, src.text) # ---------- Batch / Chunk ---------- def _chunks_by_budget(items: List[TranslationRecord], budget_chars: int, source_lang: str) -> List[List[TranslationRecord]]: """ Découpe les records en paquets dont la somme des caractères source ne dépasse pas 'budget_chars'. (Recommandé ~5k ; Advanced max ~30k code points.) """ batches = [] current, total = [], 0 for r in items: st = _get_source_text(r) if not st: continue _, txt = st n = len(txt) if n > budget_chars and not current: batches.append([r]) continue if total + n > budget_chars and current: batches.append(current) current, total = [], 0 current.append(r) total += n if current: batches.append(current) return batches # ---------- Traduction ---------- def _translate_batch(gc: GClient, records: List[TranslationRecord], cache: Dict[Tuple[str,str,str], str]) -> int: """ Traduit un lot de records (même cible par paquet) avec cache (src_text, src_lang, tgt_lang). Retourne le nombre de records mis à jour. """ if not records: return 0 # Regrouper par (source_lang, target_lang) pour des appels compacts updated = 0 keyfunc = lambda r: (_get_source_text(r)[0] if _get_source_text(r) else "", r.language) for (src_lang, tgt_lang), group in itertools.groupby(sorted(records, key=keyfunc), key=keyfunc): group = list(group) contents = [] idx_map = [] # map vers records for i, r in enumerate(group): st = _get_source_text(r) if not st: continue s_lang, s_txt = st # cache ? ckey = (s_txt, s_lang, tgt_lang) if ckey in cache: continue # on traitera en post pour remplir depuis le cache contents.append(s_txt) idx_map.append(i) # Appel API si besoin translations: List[str] = [] if contents: try: resp = gc.client.translate_text( request={ "parent": gc.parent, "contents": contents, "mime_type": "text/plain", "source_language_code": src_lang or "", "target_language_code": tgt_lang, } ) translations = [t.translated_text for t in resp.translations] except GoogleAPICallError as e: # on n’arrête pas tout le batch ; on loggue et on passe print(f"[WARN] API error {e} on src={src_lang}→{tgt_lang} ({len(contents)} items)") translations = [] # Remplir le cache for i, txt in enumerate(contents): if i < len(translations): cache[(txt, src_lang, tgt_lang)] = translations[i] # Mise à jour BD (y compris les items servis depuis cache) for r in group: st = _get_source_text(r) if not st: continue s_lang, s_txt = st ckey = (s_txt, s_lang, tgt_lang) if ckey not in cache: continue new_text = cache[ckey] # idempotent/overwrite old = r.text or "" r.previous_text = old r.text = new_text r.is_translated = True r.is_machine = True r.engine = TranslationRecord.Engine.GOOGLE # Statut : on laisse en DRAFT pour revue (ou PENDING->DRAFT) if r.status == TranslationRecord.Status.PENDING: r.status = TranslationRecord.Status.DRAFT r.needs_review = True r.save(update_fields=["text","previous_text","is_translated","is_machine","engine","status","needs_review","updated_at"]) # historiser TranslationRecordVersion.objects.create( record=r, language=r.language, old_text=old, new_text=new_text, was_original=False, engine=r.engine, status=r.status ) updated += 1 return updated # ---------- Command ---------- class Command(BaseCommand): help = "Batch Google Cloud Translation: sélection par app/view/url/template, langues cibles multiples, idempotent." def add_arguments(self, parser): # Sélecteurs parser.add_argument("--app", help="SubProject.code (ex: accounts)") parser.add_argument("--view", dest="view_dotted", help="View dotted path (ex: accounts.views.dashboard_home)") parser.add_argument("--url", dest="url_path", help="URL à résoudre en view (ex: /accounts/dashboard/)") parser.add_argument("--template", dest="template_root", help="Template root (closure extends/include)") # Langues parser.add_argument("--targets", default="en,de,es", help="Langues cibles, séparées par des virgules (ex: en,de,es)") parser.add_argument("--status-in", default="pending,draft", help="Filtrer par statut (ex: pending,draft,approved)") parser.add_argument("--only-missing", action="store_true", default=True, help="(defaut) ne traduit que les enregistrements vides/non traduits") parser.add_argument("--overwrite", action="store_true", help="Réécrire même si déjà traduit (remplit previous_text)") # GCP parser.add_argument("--project", dest="project_id", help="GCP project id (ou $GOOGLE_CLOUD_PROJECT)") parser.add_argument("--location", default="global", help="Location (ex: global, europe-west1)") parser.add_argument("--list-languages", action="store_true", help="Lister les langues supportées et quitter") parser.add_argument("--max-chars-per-call", type=int, default=5000, help="Budget max par appel (code points).") # Divers parser.add_argument("--dry-run", action="store_true", help="N’écrit rien en base ; affiche un récapitulatif") # dans add_arguments() parser.add_argument("--trace", action="store_true", help="Affiche un diagnostic détaillé du périmètre (templates, clés) et des records cibles manquants par langue.") # add_arguments(...) parser.add_argument("--create-missing-targets", action="store_true", help="Avant de traduire, crée les TranslationRecord cibles manquants pour les langues demandées (ne touche pas au FR)." ) parser.add_argument("--commons", action="store_true", help="Traduire toutes les clés marquées COMMON, quel que soit le périmètre app/view/template.") def handle(self, *args, **opts): targets = [t.strip().lower() for t in (opts["targets"] or "").split(",") if t.strip()] status_in = [s.strip().lower() for s in (opts["status_in"] or "").split(",") if s.strip()] only_missing = bool(opts.get("only_missing")) overwrite = bool(opts.get("overwrite")) commons = bool(opts.get("commons")) # ------------------------------------------------ # --- Normalisations et validations conviviaux --- # ------------------------------------------------ app = (opts.get("app") or "").strip() view_in = (opts.get("view_dotted") or "").strip() template_root = (opts.get("template_root") or "").strip() # ---------------------------------------------------------- # 1) Valider --app si fourni (pas de création implicite) # ---------------------------------------------------------- if app and not SubProject.objects.filter(code=app).exists(): raise CommandError( f"SubProject inconnu: '{app}'. " f"Exécutez d'abord: manage.py scan_view --app {app} --subproject {app}" ) # ------------------------------------------------- # 2) --view court accepté si --app est présent # ------------------------------------------------- view_dotted = None if view_in: if "." in view_in: view_dotted = view_in else: if not app: raise CommandError("--view sous forme courte (ex: 'dashboard_home') nécessite --app.") view_dotted = f"{app}.views.{view_in}" # 3) Valider l'existence de la vue v_qs = DjangoViewRef.objects.filter(dotted_path=view_dotted) if app: v_qs = v_qs.filter(subproject__code=app) if not v_qs.exists(): raise CommandError( f"Vue Django inconnue: '{view_dotted}'. " f"Assurez-vous qu'elle existe et exécutez 'scan_view --app {app} --subproject {app}' " "pour créer le mapping vue↔templates." ) # ------------------ # Client GCP # ------------------ gc = _build_client(opts.get("project_id"), opts.get("location")) # 3) Lister les langues supportées ? if opts.get("list_languages"): langs = self._list_languages(gc) self.stdout.write("Codes langue supportés :") self.stdout.write(", ".join(sorted(langs))) return # ------------------------------------------- # 4) Construire la sélection de records # ------------------------------------------- app = opts.get("app") only_missing = bool(opts.get("only_missing")) and not overwrite #----------------------------------------------------- # --- Diagnostic “pivot TranslationKey” (toujours) --- #----------------------------------------------------- diag = _diagnose_scope(app=app, view_dotted=view_dotted, template_root=template_root, targets=targets) self.stdout.write(f"[scope] templates={len(diag['templates'])} keys={diag['keys_total']}") for lang in targets: pl = diag["per_lang"][lang] self.stdout.write(f"[{lang}] records existants={pl['existing']} manquants={pl['missing']}") # si trace, lister les clés manquantes (accroche sur ton cas “62 vs 66”) if opts.get("trace") and pl["missing"]: for k, tpl in pl["missing_keys"]: self.stdout.write(f" - MISSING[{lang}] {k} ({tpl})") #---------------------------------------------------------- # --- Création auto des cibles manquantes (optionnelle) --- #---------------------------------------------------------- if opts.get("create_missing_targets"): created = _ensure_targets_for_scope(app=app, view_dotted=view_dotted, template_root=template_root, targets=targets) for lang, n in created.items(): self.stdout.write(f"[{lang}] targets créés: {n}") #----------------------------------------- # ... PICKUP RECORDS TO BE TRANSLATED .... #----------------------------------------- records = _pick_records( app = app , dotted_view = view_dotted , url_path = opts.get("url_path") , template_root = template_root , targets = targets , only_missing = only_missing , status_in = status_in , overwrite = overwrite , commons = commons , # <--- ajoute cet argument ) if not records: self.stdout.write("Aucun enregistrement à traduire.") return # Si overwrite=True, on autorise tout ; sinon, on retire les déjà traduits if overwrite: records = [r for r in records if True] # tout garder else: records = [r for r in records if (not r.is_translated) or not (r.text or "").strip()] if not records: self.stdout.write("Rien à faire (déjà traduit).") return # 3) Batch par budget budget = int(opts.get("max_chars_per_call") or 5000) batches = _chunks_by_budget(records, budget, source_lang="") # source par record if opts.get("dry_run"): self.stdout.write(f"Dry-run: {len(records)} records, {len(batches)} appels API (~{budget} chars max/chunk).") # Exemple de 5 premières unités for r in records[:5]: st = _get_source_text(r) self.stdout.write(f"- {r.key_ref.key} [{st[0] if st else '?'} → {r.language}]") return # 4) Traduction cache: Dict[Tuple[str,str,str], str] = {} total_updated = 0 for batch in batches: total_updated += _translate_batch(gc, batch, cache) self.stdout.write(self.style.SUCCESS(f"Terminé. Records mis à jour: {total_updated}")) # ----- API util ----- def _list_languages(self, gc: GClient) -> List[str]: """Interroge l’API pour retourner la liste des codes langue supportés.""" resp = gc.client.get_supported_languages(request={"parent": gc.parent}) # resp.languages: iterable de SupportedLanguage (fields: language_code, display_name, support_source, support_target) return [l.language_code.lower() for l in resp.languages]