From 50b12795d3c2f4a7f7004d89ce555bcdae1e676a Mon Sep 17 00:00:00 2001 From: Ralph Mayola Date: Fri, 25 Sep 2026 08:35:11 +0200 Subject: [PATCH] Moteur OpenRouter (google/gemini-3.5-flash-lite), par defaut Meme consigne, meme encodage, meme repli local que le moteur Gemini direct ; seuls l'adresse, l'authentification (Bearer) et le format de requete (input_audio, compatible OpenAI) changent. Cle : trousseau zonza-openrouter, puis OPENROUTER_API_KEY. Un 200 portant un objet error est traite comme un echec, pas comme une transcription vide. Co-Authored-By: Claude Opus 5.5 --- core.py | 10 ++-- moteur/__init__.py | 6 ++- moteur/gemini.py | 28 ++++++---- moteur/openrouter.py | 79 +++++++++++++++++++++++++++++ tests/test_modele_par_moteur.py | 2 +- tests/test_openrouter.py | 90 +++++++++++++++++++++++++++++++++ 6 files changed, 198 insertions(+), 17 deletions(-) create mode 100644 moteur/openrouter.py create mode 100644 tests/test_openrouter.py diff --git a/core.py b/core.py index a28ae64..6e7e9ed 100644 --- a/core.py +++ b/core.py @@ -7,9 +7,10 @@ import re CONFIG = { # Moteur de transcription. None = detection materielle (mlx, faster-cpu, - # faster-gpu, tous locaux). "gemini" = l'audio part chez Google, avec le - # moteur local en repli si l'API ne repond pas (cf moteur/gemini.py). - "moteur": "gemini", + # faster-gpu, tous locaux). "gemini" (API Google directe) ou "openrouter" + # (meme modele, via OpenRouter) = l'audio quitte le Mac, avec le moteur + # local en repli si l'API ne repond pas (cf moteur/gemini.py). + "moteur": "openrouter", # Forcage manuel du modele. None = on suit MODELES_PAR_MOTEUR, qui choisit # d'apres le moteur detecte. Renseigner cette cle prime sur la table. "model": None, @@ -194,6 +195,9 @@ MODELES_PAR_MOTEUR = { # partir de texte, elle ne transcrit pas. Il faut un modele qui ACCEPTE # l'audio en entree. "gemini": "gemini-3.1-flash-lite", + # Flash-Lite le plus recent qui accepte l'audio chez OpenRouter (25/09) ; + # moins cher en audio que 3.1 (0,30 $ contre 0,50 $ le million de jetons). + "openrouter": "google/gemini-3.5-flash-lite", } diff --git a/moteur/__init__.py b/moteur/__init__.py index 54a6576..298320d 100644 --- a/moteur/__init__.py +++ b/moteur/__init__.py @@ -76,14 +76,16 @@ def choisir_transcripteur(nom, modele, langue, amorce): peripherique = "cuda" if nom == "faster-gpu" else "cpu" type_calcul = "float16" if nom == "faster-gpu" else "int8" return TranscripteurFasterWhisper(modele, langue, amorce, peripherique, type_calcul) - if nom == "gemini": + if nom in ("gemini", "openrouter"): # Toujours enveloppe d'un repli : la dictee ne doit jamais etre perdue # parce que le reseau ou l'API tombe (cf moteur/gemini.py). from core import modele_pour_moteur from moteur.gemini import TranscripteurAvecRepli, TranscripteurGemini + from moteur.openrouter import TranscripteurOpenRouter + classe = TranscripteurGemini if nom == "gemini" else TranscripteurOpenRouter local = _moteur_local() return TranscripteurAvecRepli( - TranscripteurGemini(modele, langue, amorce), + classe(modele, langue, amorce), fabriquer_secours=lambda: choisir_transcripteur( local, modele_pour_moteur(local), langue, amorce), ) diff --git a/moteur/gemini.py b/moteur/gemini.py index 0655d70..daa8dfd 100644 --- a/moteur/gemini.py +++ b/moteur/gemini.py @@ -58,8 +58,8 @@ def audio_en_wav(audio, sample_rate): return tampon.getvalue() -def _lire_trousseau(): - """Clé rangée par `security add-generic-password -s zonza-gemini ...`. +def _lire_trousseau(service=SERVICE_TROUSSEAU): + """Clé rangée par `security add-generic-password -s ...`. 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` @@ -69,7 +69,7 @@ def _lire_trousseau(): return None try: sortie = subprocess.run( - ["security", "find-generic-password", "-s", SERVICE_TROUSSEAU, "-w"], + ["security", "find-generic-password", "-s", service, "-w"], capture_output=True, text=True, timeout=5, ) except Exception: @@ -79,19 +79,25 @@ def _lire_trousseau(): return sortie.stdout.strip() or None -def lire_cle(trousseau=None, environ=None): - """La clé du trousseau, sinon GEMINI_API_KEY, sinon None.""" - trousseau = trousseau or _lire_trousseau +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("GEMINI_API_KEY") or None + return trousseau() or environ.get(variable) or None def _envoyer(url, cle, corps, delai_s): - """POST JSON. La clé part en en-tête, jamais dans l'URL (qui finit en journal).""" + """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", "x-goog-api-key": cle}, + headers={"Content-Type": "application/json", **entetes}, method="POST", ) try: @@ -179,12 +185,12 @@ class TranscripteurAvecRepli: try: self.principal.warmup() except Exception as e: - print(f"[gemini] indisponible ({e}) — moteur local chargé en secours") + 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"[gemini] échec ({e}) — transcription locale") + print(f"[api] échec ({e}) — transcription locale") return self._secours_pret().transcrire(audio) diff --git a/moteur/openrouter.py b/moteur/openrouter.py new file mode 100644 index 0000000..bcf2fd6 --- /dev/null +++ b/moteur/openrouter.py @@ -0,0 +1,79 @@ +"""Transcription par OpenRouter — même famille Gemini, autre guichet. + +Comme `moteur/gemini.py`, ce moteur N'EST PAS local : l'audio part chez +OpenRouter, qui le relaie à Google. L'encodage, la consigne, le repli local et +la lecture de la clé sont partagés avec le moteur Gemini direct ; seuls +l'adresse, l'authentification et le format de requête (compatible OpenAI, +partie `input_audio`) diffèrent. +""" +import base64 +import json + +from core import clean_transcript +from moteur.gemini import ( + CONSIGNE, DELAI_RESEAU_S, ErreurGemini, audio_en_wav, lire_cle, poster_json, +) + +URL_API = "https://openrouter.ai/api/v1/chat/completions" +SERVICE_TROUSSEAU = "zonza-openrouter" + + +def lire_cle_openrouter(trousseau=None, environ=None): + """La clé du trousseau `zonza-openrouter`, sinon OPENROUTER_API_KEY.""" + return lire_cle(trousseau, environ, service=SERVICE_TROUSSEAU, + variable="OPENROUTER_API_KEY") + + +def _envoyer(url, entetes, corps, delai_s): + return poster_json(url, entetes, corps, delai_s) + + +class TranscripteurOpenRouter: + """Implémentation du contrat Transcripteur via OpenRouter.""" + + def __init__(self, modele, langue, amorce, sample_rate=16000, + lire_cle=lire_cle_openrouter, 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é OpenRouter : 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): + wav = base64.b64encode(audio_en_wav(audio, self.sample_rate)).decode("ascii") + corps = { + "model": self.modele, + "messages": [ + {"role": "system", "content": CONSIGNE.format( + langue=self.langue, amorce=self.amorce)}, + {"role": "user", "content": [ + {"type": "input_audio", "input_audio": {"data": wav, "format": "wav"}}, + {"type": "text", "text": "Transcris cet enregistrement."}, + ]}, + ], + } + entetes = {"Authorization": f"Bearer {self._cle()}", "X-Title": "Zonza"} + reponse = self._envoyer(URL_API, entetes, corps, DELAI_RESEAU_S) + # OpenRouter peut rendre un 200 portant un objet `error` quand le + # fournisseur en aval tombe : ce n'est PAS une transcription vide. + if "error" in reponse: + erreur = reponse["error"] + raise ErreurGemini(f"OpenRouter : {erreur.get('message', erreur)}") + choix = reponse.get("choices") or [] + if not choix: + raise ErreurGemini(f"réponse sans transcription : {json.dumps(reponse)[:300]}") + return clean_transcript(choix[0].get("message", {}).get("content") or "") diff --git a/tests/test_modele_par_moteur.py b/tests/test_modele_par_moteur.py index 3a79e84..5b8ce33 100644 --- a/tests/test_modele_par_moteur.py +++ b/tests/test_modele_par_moteur.py @@ -37,7 +37,7 @@ def test_un_moteur_inconnu_est_refuse_clairement(): def test_la_table_ne_contient_que_les_moteurs_reconnus(): - assert set(MODELES_PAR_MOTEUR) == {"mlx", "faster-cpu", "faster-gpu", "gemini"} + assert set(MODELES_PAR_MOTEUR) == {"mlx", "faster-cpu", "faster-gpu", "gemini", "openrouter"} def test_le_reglage_manuel_prime_sur_la_table(): diff --git a/tests/test_openrouter.py b/tests/test_openrouter.py new file mode 100644 index 0000000..6d260ed --- /dev/null +++ b/tests/test_openrouter.py @@ -0,0 +1,90 @@ +"""Moteur OpenRouter : même modèle Gemini, facturé et routé par OpenRouter. + +Aucun test ne touche le réseau : l'envoi HTTP et la lecture de la clé sont +injectés. Le repli local, lui, est déjà éprouvé dans test_gemini.py. +""" +import base64 + +import numpy as np +import pytest + +from moteur import choisir_transcripteur +from moteur.gemini import ErreurGemini, TranscripteurAvecRepli +from moteur.openrouter import TranscripteurOpenRouter, lire_cle_openrouter + + +def _reponse(texte): + return {"choices": [{"message": {"role": "assistant", "content": texte}}]} + + +def _transcripteur(envoyer, cle="cle-test"): + return TranscripteurOpenRouter( + "google/gemini-x", "fr", "Vocabulaire : Gitea, iCloud.", + lire_cle=lambda: cle, envoyer=envoyer, + ) + + +def test_la_requete_porte_l_audio_le_modele_et_le_vocabulaire(): + vu = {} + + def envoyer(url, entetes, corps, delai_s): + vu.update(url=url, entetes=entetes, corps=corps) + return _reponse("Bonjour.") + + _transcripteur(envoyer).transcrire(np.zeros(1600, dtype="float32")) + + assert vu["url"] == "https://openrouter.ai/api/v1/chat/completions" + assert vu["entetes"]["Authorization"] == "Bearer cle-test" + assert vu["corps"]["model"] == "google/gemini-x" + systeme, utilisateur = vu["corps"]["messages"] + assert "Gitea" in systeme["content"] and "iCloud" in systeme["content"] + audio = next(p["input_audio"] for p in utilisateur["content"] if p["type"] == "input_audio") + assert audio["format"] == "wav" + assert base64.b64decode(audio["data"])[:4] == b"RIFF" + + +def test_la_cle_ne_part_jamais_dans_l_url(): + vu = {} + _transcripteur(lambda url, e, c, d: vu.setdefault("url", url) and _reponse("x")) \ + .transcrire(np.zeros(1600, dtype="float32")) + assert "cle-test" not in vu["url"] + + +def test_le_texte_rendu_est_nettoye(): + t = _transcripteur(lambda *a: _reponse(" Je pousse sur Gitea. ")) + assert t.transcrire(np.zeros(1600, dtype="float32")) == "Je pousse sur Gitea." + + +def test_une_erreur_rendue_dans_un_200_est_une_erreur_explicite(): + # OpenRouter peut répondre 200 avec un objet `error` (fournisseur en panne). + t = _transcripteur(lambda *a: {"error": {"code": 502, "message": "fournisseur indisponible"}}) + with pytest.raises(ErreurGemini) as e: + t.transcrire(np.zeros(1600, dtype="float32")) + assert "fournisseur indisponible" in str(e.value) + + +def test_une_reponse_sans_choix_est_une_erreur_explicite(): + t = _transcripteur(lambda *a: {"choices": []}) + with pytest.raises(ErreurGemini): + t.transcrire(np.zeros(1600, dtype="float32")) + + +def test_une_cle_absente_est_signalee_des_le_warmup_sans_appel(): + appels = [] + t = _transcripteur(lambda *a: appels.append(a), cle=None) + with pytest.raises(ErreurGemini) as e: + t.warmup() + assert "zonza-openrouter" in str(e.value) + assert appels == [] + + +def test_la_cle_se_lit_sous_son_propre_nom(): + assert lire_cle_openrouter(trousseau=lambda: None, + environ={"OPENROUTER_API_KEY": "or", "GEMINI_API_KEY": "g"}) == "or" + assert lire_cle_openrouter(trousseau=lambda: None, environ={"GEMINI_API_KEY": "g"}) is None + + +def test_la_fabrique_rend_openrouter_enveloppe_d_un_repli(): + t = choisir_transcripteur(nom="openrouter", modele="google/gemini-x", langue="fr", amorce="") + assert isinstance(t, TranscripteurAvecRepli) + assert isinstance(t.principal, TranscripteurOpenRouter)