"""Transcription par l'API Gemini — l'audio part chez Google. Contrairement à `mlx` et `faster`, ce moteur N'EST PAS local : chaque dictée est envoyée à `generativelanguage.googleapis.com`. Il est donc toujours enveloppé d'un `TranscripteurAvecRepli` : réseau coupé, clé refusée, quota épuisé ou réponse vide, la dictée repasse par le moteur local au lieu d'être perdue. Aucune dépendance nouvelle : l'appel HTTP passe par `urllib` de la bibliothèque standard, et la clé se lit dans le trousseau macOS (`security`) ou, à défaut, dans la variable d'environnement GEMINI_API_KEY. """ import base64 import io import json import os import subprocess import sys import urllib.error import urllib.request import wave import numpy as np from core import clean_transcript URL_API = "https://generativelanguage.googleapis.com/v1beta/models/{modele}:generateContent" SERVICE_TROUSSEAU = "zonza-gemini" # Borné sous `duree_max_transcription_s` (20 s) : si l'API ne répond pas à # temps, il doit rester de quoi basculer sur le moteur local avant que la # garde du halo ne tombe. 15 s couvrent les 9 s mesurées de gemini-3.8-flash # sur une dictée de 94 s. DELAI_RESEAU_S = 15.0 CONSIGNE = ( "Tu es un moteur de dictée. Transcris mot pour mot ce que dit la personne, " "dans la langue parlée ({langue}), avec la ponctuation et les majuscules " "d'un texte écrit. Ne réponds pas, ne résume pas, ne commente pas, " "n'ajoute rien : rends UNIQUEMENT le texte dicté. Si l'enregistrement ne " "contient aucune parole, rends une chaîne vide. Termes techniques à " "orthographier exactement : {amorce}" ) class ErreurGemini(RuntimeError): """Tout échec côté Gemini : clé, réseau, quota, réponse inexploitable.""" def audio_en_wav(audio, sample_rate): """Encode un signal float32 mono dans [-1, 1] en WAV PCM 16 bits.""" pcm = (np.clip(audio, -1.0, 1.0) * 32767).astype(" ...`. Le trousseau, et non `~/.zshrc` : Zonza est lancée par le Finder, qui ne lit jamais la configuration du shell — une clé exportée dans `.zshrc` serait invisible pour l'application. """ if sys.platform != "darwin": return None try: sortie = subprocess.run( ["security", "find-generic-password", "-s", service, "-w"], capture_output=True, text=True, timeout=5, ) except Exception: return None if sortie.returncode != 0: return None return sortie.stdout.strip() or None def lire_cle(trousseau=None, environ=None, service=SERVICE_TROUSSEAU, variable="GEMINI_API_KEY"): """La clé du trousseau, sinon la variable d'environnement, sinon None.""" trousseau = trousseau or (lambda: _lire_trousseau(service)) environ = os.environ if environ is None else environ return trousseau() or environ.get(variable) or None def _envoyer(url, cle, corps, delai_s): """La clé part en en-tête, jamais dans l'URL (qui finit en journal).""" return poster_json(url, {"x-goog-api-key": cle}, corps, delai_s) def poster_json(url, entetes, corps, delai_s): """POST JSON ; toute erreur HTTP ou réseau devient une ErreurGemini.""" requete = urllib.request.Request( url, data=json.dumps(corps).encode("utf-8"), headers={"Content-Type": "application/json", **entetes}, method="POST", ) try: with urllib.request.urlopen(requete, timeout=delai_s) as reponse: return json.loads(reponse.read()) except urllib.error.HTTPError as e: brut = e.read().decode("utf-8", "replace") try: detail = json.loads(brut)["error"]["message"] except Exception: detail = " ".join(brut.split())[:300] raise ErreurGemini(f"HTTP {e.code} : {detail}") from e except (urllib.error.URLError, TimeoutError, OSError) as e: raise ErreurGemini(f"réseau : {e}") from e class TranscripteurGemini: """Implémentation du contrat Transcripteur via l'API Gemini.""" def __init__(self, modele, langue, amorce, sample_rate=16000, lire_cle=lire_cle, envoyer=_envoyer): self.modele = modele self.langue = langue self.amorce = amorce or "" self.sample_rate = sample_rate self._lire_cle = lire_cle self._envoyer = envoyer def _cle(self): cle = self._lire_cle() if not cle: raise ErreurGemini( "aucune clé Gemini : lancer `security add-generic-password " f"-s {SERVICE_TROUSSEAU} -a zonza -w` puis relancer Zonza" ) return cle def warmup(self): """Rien à charger : on vérifie seulement la clé, sans appel payant.""" self._cle() def transcrire(self, audio): corps = { "system_instruction": {"parts": [{"text": CONSIGNE.format( langue=self.langue, amorce=self.amorce)}]}, "contents": [{"role": "user", "parts": [ {"inline_data": { "mime_type": "audio/wav", "data": base64.b64encode(audio_en_wav(audio, self.sample_rate)).decode("ascii"), }}, {"text": "Transcris cet enregistrement."}, ]}], } reponse = self._envoyer(URL_API.format(modele=self.modele), self._cle(), corps, DELAI_RESEAU_S) candidats = reponse.get("candidates") or [] if not candidats: raise ErreurGemini(f"réponse sans transcription : {json.dumps(reponse)[:300]}") parts = (candidats[0].get("content") or {}).get("parts") or [] return clean_transcript("".join(p.get("text", "") for p in parts)) class TranscripteurAvecRepli: """Essaie le principal ; au moindre échec, bascule sur le moteur local. Le moteur local n'est fabriqué qu'au premier besoin : tant que Gemini répond, on n'immobilise pas 500 Mo de modèle en mémoire pour rien. """ def __init__(self, principal, fabriquer_secours): self.principal = principal self._fabriquer_secours = fabriquer_secours self._secours = None @property def secours_charge(self): return self._secours is not None def _secours_pret(self): if self._secours is None: self._secours = self._fabriquer_secours() self._secours.warmup() return self._secours def warmup(self): try: self.principal.warmup() except Exception as e: print(f"[api] indisponible ({e}) — moteur local chargé en secours") self._secours_pret() def transcrire(self, audio): try: return self.principal.transcrire(audio) except Exception as e: print(f"[api] échec ({e}) — transcription locale") return self._secours_pret().transcrire(audio)