Author SHA1 Message Date
Hermes Agent 1d13e7c22b UI: fenêtre unifiée à onglets + réaffichage Tk depuis l'icône
Remplace l'architecture 'une fenêtre = un processus' par une fenêtre
unique (racine Tk sur le thread principal) avec trois onglets:

- Profils    : liste + éditeur intégré (ex-ProfileDialog)
- Authentification : gestion des clés API (ex-AuthDialog)
- Logs       : journaux temps réel (ex-LogsWindow, désormais dans le
  même processus que le moteur, donc réellement alimenté)

Fermer la fenêtre (WM_DELETE_WINDOW -> withdraw) la masque en zone de
notification sans quitter. L'icône pystray (thread dédié) réaffiche la
fenêtre dans le bon onglet via master.after(0, ...); la commutation de
profil reste directe depuis le sous-menu radio (sans afficher la fenêtre).

Supprime _tk_spawn/_tk_window_process et les trois anciens dialogues.
2026-09-18 14:56:37 +02:00
Hermes Agent 4adb88fd35 refactor: subprocess.Popen au lieu de multiprocessing.Process pour les fenêtres Tk
- Nouveau module _tk_window_process.py : point d'entrée pour le subprocess
  (python -m ai_typewriter.ui._tk_window_process). Reçoit les args en
  pickle sur stdin, écrit window.result en pickle sur stdout.
- _tk_spawn.py réécrit avec subprocess.Popen. Payload picklé envoyé sur
  stdin, résultat lu via communicate() dans un thread poller.
- Plus besoin de multiprocessing.Queue ni de start_method.
- PYTHONPATH injecté automatiquement pour que le subprocess trouve le
  package ai_typewriter.
2026-09-18 14:27:35 +02:00
Hermes Agent 7fa401b287 refactor: multiprocessing.Process au lieu de threading.Thread pour les fenêtres Tk
- _tk_spawn.py : chaque fenêtre tourne dans un Process séparé (spawn,
  pas fork) avec sa propre instance tk.Tk. Résultats renvoyés au parent
  via multiprocessing.Queue + thread poller.
- API changée : spawn_tk_window(Class, *args, on_result=..., **kwargs)
  au lieu d'une lambda factory — nécessaire pour la picklabilité.
- SecureStore n'est plus passé aux processus fils (non picklable) ;
  chaque dialogue crée le sien (déjà le comportement par défaut).
- TrayApp allégé : plus de self.secure, plus d'import SecureStore.
2026-09-18 13:56:53 +02:00
Hermes Agent e07f401b5c refactor: un thread par fenêtre Tk, main thread dédié au hotkey
- Nouveau helper ui/_tk_spawn.py : chaque fenêtre Tk dans son propre thread
  avec sa propre racine tk.Tk (plus de racine partagée)
- app.py : thread principal pour le hotkey + attente stop_event,
  thread séparé pour l'icône pystray
- tray.py : suppression de get_root/_defer_tk/_present_window,
  spawn_tk_window pour chaque dialogue, callbacks on_result au lieu
  de wait_window bloquant
2026-09-18 13:25:24 +02:00
Hermes Agent 15eb856bf2 fix(tray): fenêtres invisibles et coche de profil non mise à jour
- Ajoute _present_window() (deiconify/lift/topmost/focus_force) pour que
  les Toplevel enfants d'une racine withdraw() apparaissent bien au
  premier plan sous Windows (logs, profil, ajout, authentification).
- Rend le sous-menu des profils dynamique via un générateur avec items
  radio + checked=, et appelle icon.update_menu() après changement de
  profil pour que la coche suive l'état courant.
2026-09-18 11:00:41 +02:00
Hermes Agent 2f5c428545 fix(tray): architecture hybride pystray (thread) + Tk.mainloop() (thread principal) — les callbacks dispatch via root.after() pour que Tk ait sa boucle d'événements ; fenêtres peintes et interactives 2026-09-18 10:54:03 +02:00
Hermes Agent e7d13d9ad1 fix(tray): handlers de menu à 2 args (icon,item) — le lambda 1-arg capturait l'icône au lieu du nom du profil; Icon.run() sur le thread principal (exigence pystray); logs DEBUG par défaut + formateur enrichi 2026-09-18 10:45:25 +02:00
Hermes Agent 05e6515011 fix(auth): lire p.credential sur les objets Profile; sortie propre via Event (plus de SystemExit dans le callback pystray) 2026-09-18 10:24:23 +02:00
Hermes Agent fa7b037e7f fix(tray): créer le sous-menu via l'action MenuItem (pystray n'accepte pas submenu=) 2026-09-18 10:17:17 +02:00
Hermes Agent 3e58380f80 Rework AI-Typewriter en application d'arrière-plan multi-profils
- Rollback gestion d'équations : écriture littérale caractère par caractère
  (suppression de math_format.py et de la normalisation Unicode).
- Nouveau profil par défaut 'Mathématiques (LaTeX)' : le modèle émet du
  LaTeX encadré [EQ]...[/EQ], intercepté par KeyStepper qui déclenche
  Alt+= en début d'équation et -> en fin.
- Config multi-profils dans %APPDATA%/ai-typewriter/config.json
  (ConfigStore + profils par défaut auto-créés).
- Icône de zone de notification (pystray) : Ouvrir les logs (temps réel),
  Ajouter/Modifier un profil, Gérer l'authentification, Quitter.
- Sélecteur de modèles : modèles locaux Ollama + recherche/téléchargement
  depuis la bibliothèque publique ; fournisseurs tiers (OpenAI, OpenRouter,
  Gemini, custom).
- Clés d'API stockées de façon sécurisée dans le Gestionnaire d'identifiants
  Windows via keyring (jamais en clair dans config.json).
- Tests : 44 tests verts (stepper, profils, clients IA, credentials,
  catalogue de modèles, moteur).
2026-09-18 09:48:46 +02:00
Hermes Agent d07bed60e6 Return to plain LaTeX typing mode 2026-09-12 12:17:07 +02:00
Hermes Agent 7bd423452d Use explicit Word equation markers 2026-09-11 14:26:35 +02:00
Hermes Agent 3bccabe4ec Insert Word equation objects for math output 2026-09-11 14:04:51 +02:00
Hermes Agent 6bb572f9e4 Add Unicode math output formatting 2026-09-11 13:54:01 +02:00
Hermes Agent b2d2ba58fc Handle slow AI backend timeouts 2026-09-11 13:42:42 +02:00
Hermes Agent 9320217e0f Improve clipboard workflow and logging 2026-09-11 13:24:39 +02:00
Hermes Agent e779bd3423 Add Windows build script 2026-09-11 12:56:19 +02:00
Hermes Agent 464324aedf Initial ai typewriter implementation 2026-09-11 12:01:35 +02:00
31 changed files with 2771 additions and 491 deletions
+1
View File
@@ -12,6 +12,7 @@ dist/
# Local secrets/config
config.json
*.key
# Binary releases
*.exe
+66 -87
View File
@@ -1,132 +1,111 @@
# ai-typewriter
# AI-Typewriter
Application Python qui lit la dernière entrée texte du presse-papier avec un raccourci global, l'envoie à un modèle IA, puis remplace chaque pression de touche suivante par le caractère suivant de la réponse.
Application Python qui **lit le contenu du presse-papier** sur un raccourci global, l'envoie à un modèle IA, puis réécrit la réponse **caractère par caractère** à chaque pression de touche physique.
Une **fenêtre unifiée à trois onglets** (Profils / Authentification / Logs) permet de tout configurer sans fenêtres éparpillées. Fermer la fenêtre (✕) la masque dans la zone de notification — l'application continue de tourner. Le menu de l'icône de notification réaffiche la fenêtre dans le bon onglet, et la commutation de profil reste directement accessible depuis l'icône (sans afficher la fenêtre).
Un profil spécialisé « Mathématiques (LaTeX) » marque les équations avec `[EQ]...[/EQ]` : l'application les intercepte et déclenche `Alt+=` pour ouvrir une équation (Word/OneNote) et `→` pour en sortir.
## Fonctionnement
1. L'utilisateur copie manuellement le texte à envoyer à l'IA.
1. L'utilisateur copie le texte à envoyer à l'IA.
2. Raccourci global par défaut : `Ctrl+Alt+A`.
3. L'application lit directement la dernière entrée du presse-papier, sans simuler `Ctrl+C`.
4. Le texte capturé est journalisé puis envoyé à Ollama ou Gemini.
5. Le temps de génération de la réponse est journalisé.
6. Quand la réponse arrive, le mode dactylographie s'active.
7. Chaque touche physique appuyée est interceptée et remplacée par le prochain caractère de la réponse IA.
8. Le hook clavier est libéré automatiquement après le dernier caractère.
L'application n'interprète pas les équations et ne lance pas `Alt+=`. Elle écrit uniquement le texte généré, caractère par caractère. Pour les maths, le modèle peut produire du LaTeX encadré par des marqueurs `[EQ]...[/EQ]`, puis la gestion Word peut être faite ailleurs.
3. L'application lit le presse-papier et envoie au modèle du profil actif.
4. Une icône reste disponible dans la zone de notification : elle permet d'ouvrir les journaux, d'ajouter/commuter des profils et de gérer les clés d'API.
5. Chaque touche physique appuyée ensuite écrit l'élément suivant de la réponse (caractère ou séquence d'équation).
6. Le hook clavier est libéré automatiquement à la fin de la réponse.
## Installation depuis les sources
```bash
python -m venv .venv
. .venv/bin/activate
. .venv/bin/activate # Windows : .venv\Scripts\activate
pip install -r requirements.txt
cp config.json.template config.json
pip install -e .
python main.py
```
Sous Linux, le paquet `keyboard` nécessite souvent les droits root ou l'accès aux périphériques `/dev/input`. Sous Windows, lancez l'exécutable dans une session utilisateur normale.
À la première exécution, l'application crée son fichier de configuration :
`%APPDATA%\ai-typewriter\config.json` (Linux : `~/.config/ai-typewriter/config.json`).
## Configuration
## Configuration (profils)
Copiez `config.json.template` vers `config.json` puis adaptez :
La configuration contient une liste de **profils** nommés et le profil actif. Chaque profil décrit :
```json
{
"provider": "ollama",
"model": "llama3.1",
"api_key": "",
"server_url": "http://localhost:11434",
"hotkey": "ctrl+alt+a",
"request_timeout_seconds": 300,
"math_text_format": "plain"
}
```
| Champ | Description |
|---|---|
| `name` | Libellé affiché dans les menus |
| `provider` | `ollama`, `openai`, `openrouter`, `gemini`, `custom` (OpenAI-compatible) |
| `model` | Nom du modèle (choisissable via le sélecteur) |
| `server_url` | Base de l'instance (ex. `http://localhost:11434`) |
| `credential` | Nom logique de la clé d'API (voir « Authentification ») |
| `system_prompt` | Instructions données au modèle |
| `equation_enabled` | Active l'interception des marqueurs d'équation |
| `eq_start_marker` / `eq_end_marker` | Marqueurs (défaut `[EQ]` / `[/EQ]`) |
| `eq_start_key` / `eq_end_key` | Touches déclenchées (défaut `alt+=` / `right`) |
### Ollama
Deux profils sont créés par défaut : **Général** et **Mathématiques (LaTeX)**.
```json
{
"provider": "ollama",
"model": "llama3.1",
"server_url": "http://localhost:11434"
}
```
### Profil Mathématiques (LaTeX)
### Gemini
```json
{
"provider": "gemini",
"model": "gemini-1.5-flash",
"api_key": "VOTRE_CLE",
"server_url": "https://generativelanguage.googleapis.com"
}
```
### Timeout IA
`request_timeout_seconds` vaut `300` par défaut. Si Ollama charge un gros modèle ou répond lentement, augmentez cette valeur. Mettez `0` pour désactiver le timeout côté application.
### Configuration maths LaTeX
Pour laisser le modèle générer du LaTeX tout en indiquant clairement les débuts/fins d'équations :
```bash
cp config.math-latex.template config.json
```
Cette config garde :
```json
"math_text_format": "plain"
```
Donc l'application ne transforme rien. Elle tape littéralement la réponse reçue, caractère par caractère.
Exemple de réponse demandée au modèle :
Le prompt système demande au modèle de produire du LaTeX encadré par `[EQ]...[/EQ]`, par exemple :
```text
Les racines sont [EQ]z_1 = x + iy[/EQ] et [EQ]z_2 = x - iy[/EQ].
```
Pour les fractions, intégrales, sommes, etc., le modèle peut utiliser du LaTeX standard dans les balises :
À chaque `[EQ]` l'application envoie `Alt+=` (ouvre une équation inline), tape le LaTeX littéralement, puis envoie `→` à chaque `[/EQ]`. Ce comportement est désactivé par défaut sur les autres profils (le texte est tapé tel quel).
```text
On obtient [EQ]\frac{a+b}{c+d}[/EQ] puis [EQ]\int_0^1 f(x)\,dx[/EQ].
```
## Fenêtre unifiée (onglets)
Modes disponibles :
L'application démarre avec **une seule fenêtre** comportant trois onglets :
- `plain` : mode recommandé ; injecte la réponse exactement telle que le modèle l'a renvoyée.
- `unicode` : ancien mode texte Unicode (`z_1` → `z₁`, `x^2` → `x²`) sans objet équation.
- **Profils** — liste de tous les profils (éditeur intégré) pour créer, modifier, supprimer ou activer un profil directement.
- **Authentification** — enregistrement/vérification/suppression des clés d'API par référence.
- **Logs** — journaux en temps réel (également écrits dans `~/.config/ai-typewriter/logs/app.log`).
## Compilation
## Zone de notification (icône)
### Windows
Fermer la fenêtre (✕) ne quitte **pas** l'application : elle est masquée et continue de tourner, avec l'icône toujours visible dans la zone de notification. Le menu de l'icône propose :
Après clonage du dépôt, lancez simplement :
- **Ouvrir les logs** — réaffiche la fenêtre dans l'onglet *Logs*.
- **Ajouter un profil** — réaffiche la fenêtre dans l'onglet *Profils* (formulaire vierge pour créer).
- **Modifier le profil** — sous-menu listant tous les profils pour choisir le profil actif **directement** (sans afficher la fenêtre).
- **Gérer l'authentification** — réaffiche la fenêtre dans l'onglet *Authentification*.
- **Quitter** — arrête complètement le processus.
### Sélection et téléchargement des modèles
Dans le formulaire de profil, « Choisir / télécharger… » ouvre un sélecteur qui :
- liste automatiquement les modèles déjà disponibles localement (Ollama `/api/tags`) ;
- si connecté à Internet, permet de rechercher dans la bibliothèque publique d'Ollama, de vérifier un modèle exact et de lancer son téléchargement (`ollama pull`).
### Authentification des fournisseurs
Les clés d'API ne sont **jamais écrites** dans le fichier de configuration. Chaque profil référence une clé par un nom logique ; la clé est stockée de façon sécurisée dans le **Gestionnaire d'identifiants de Windows** (via `keyring`). Le menu **Gérer l'authentification** permet de les enregistrer, vérifier ou supprimer.
## Compilation (Windows)
```bat
build.bat
```
Le script crée `.venv`, installe les dépendances, nettoie les anciens artefacts puis génère un exécutable Windows autonome :
Le script crée `.venv`, installe les dépendances, puis produit un exécutable autonome **sans console** dans `dist\ai-typewriter.exe`. Il tourne directement en zone de notification.
```text
dist\ai-typewriter.exe
```
### Commande PyInstaller équivalente
## Test rapide sans hook clavier ni icône
```bash
pyinstaller --onefile --paths src --name ai-typewriter.exe main.py
python main.py --debug --ask "Résume: bonjour tout le monde"
```
L'exécutable est généré dans `dist/`. Le binaire n'est pas versionné Git.
Envoie le prompt au profil actif et imprime la réponse brute (aucune icône ni interception clavier).
## Test rapide sans hook clavier
## Tests
```bash
python main.py --config config.json --ask "Résume: bonjour tout le monde"
. .venv/bin/activate
pytest
```
La suite couvre le découpage en actions (caractères/équations), le stepper, le dépôt de profils, les clients IA (Ollama/Gemini/OpenAI), le stockage sécurisé (keyring mocké) et le catalogue de modèles — sans réel hook clavier, réseau ni Gestionnaire d'identifiants.
+16 -5
View File
@@ -1,6 +1,8 @@
@echo off
setlocal
REM ============================================================
REM Build Windows one-file executable (icône zone de notification)
REM ============================================================
cd /d "%~dp0"
if not exist ".venv\Scripts\python.exe" (
@@ -18,10 +20,19 @@ if errorlevel 1 goto :error
echo [3/4] Nettoyage des anciens builds...
if exist build rmdir /s /q build
if exist dist rmdir /s /q dist
if exist ai-typewriter.exe.spec del /q ai-typewriter.exe.spec
if exist ai-typewriter.spec del /q ai-typewriter.spec
echo [4/4] Compilation Windows one-file...
".venv\Scripts\python.exe" -m PyInstaller --clean --onefile --paths src --name ai-typewriter.exe main.py
echo [4/4] Compilation Windows one-file (sans console)...
".venv\Scripts\python.exe" -m PyInstaller ^
--clean --onefile --windowed ^
--name ai-typewriter ^
--hidden-import keyring ^
--hidden-import keyring.backends ^
--hidden-import keyring.backends.Windows ^
--hidden-import pystray._win32 ^
--hidden-import PIL._tkinter_finder ^
--paths src ^
main.py
if errorlevel 1 goto :error
echo.
@@ -31,4 +42,4 @@ exit /b 0
:error
echo.
echo Build echoue.
exit /b 1
exit /b 1
-11
View File
@@ -1,11 +0,0 @@
{
"provider": "ollama",
"model": "llama3.1",
"api_key": "",
"server_url": "http://localhost:11434",
"hotkey": "ctrl+alt+a",
"request_timeout_seconds": 300,
"math_text_format": "plain",
"type_delay_seconds": 0,
"system_prompt": "Réponds directement et de manière ultra-concise. Aucune phrase d'introduction, aucune salutation, aucun formatage superflu. Uniquement la réponse brute."
}
-11
View File
@@ -1,11 +0,0 @@
{
"provider": "ollama",
"model": "deepseek-r1:1.5b",
"api_key": "",
"server_url": "http://localhost:11434",
"hotkey": "ctrl+alt+a",
"request_timeout_seconds": 300,
"math_text_format": "plain",
"type_delay_seconds": 0,
"system_prompt": "Tu réponds directement, sans salutation ni introduction. Rédige une réponse claire, correcte et concise. Pour toute expression mathématique, formule, calcul, égalité, fraction, somme, intégrale, matrice ou symbole qui doit être traité comme une équation, encadre exactement le bloc avec [EQ] au début et [/EQ] à la fin. Dans ces blocs, écris du LaTeX standard, car il est plus simple et fiable à générer : \\frac{a}{b}, z_1, x^2, \\int_0^1, \\sum_{k=1}^n, etc. N'utilise pas de délimiteurs LaTeX supplémentaires dans les blocs : pas de $, $$, \\\\(, \\\\[. Le texte hors des balises [EQ]...[/EQ] reste du texte normal. Exemple valide : Les racines sont [EQ]z_1 = x + iy[/EQ] et [EQ]z_2 = x - iy[/EQ]."
}
+9 -3
View File
@@ -4,17 +4,23 @@ build-backend = "setuptools.build_meta"
[project]
name = "ai-typewriter"
version = "0.1.0"
description = "Global hotkey AI key-stepper"
version = "0.2.0"
description = "AI-typewriter : application d'arrière-plan (icône zone de notification), modèle IA, dactylographie par touches + équations LaTeX"
requires-python = ">=3.10"
dependencies = [
"keyboard==0.13.5",
"pyperclip==1.9.0",
"requests==2.32.5",
"keyring>=25.0",
"pystray>=0.19",
"Pillow>=10.0",
]
[project.optional-dependencies]
dev = ["pytest>=8.0"]
[tool.setuptools.packages.find]
where = ["src"]
[tool.pytest.ini_options]
pythonpath = ["src"]
pythonpath = ["src"]
+4 -1
View File
@@ -1,5 +1,8 @@
keyboard==0.13.5
pyperclip==1.9.0
requests==2.32.5
keyring>=25.0
pystray>=0.19
Pillow>=10.0
pyinstaller==6.16.0
pytest==8.4.2
pytest==8.4.2
+2 -2
View File
@@ -1,3 +1,3 @@
"""AI Typewriter package."""
"""AI-Typewriter : application d'arrière-plan (icône zone de notification)."""
__version__ = "0.1.0"
__version__ = "0.2.0"
+124 -33
View File
@@ -1,50 +1,82 @@
"""Clients IA : Ollama, Gemini et tous les fournisseurs OpenAI-compatibles.
Les clés d'API sont résolues via le gestionnaire de références fourni
(`resolve_key`) et ne sont jamais consignées dans les journaux.
"""
from __future__ import annotations
from typing import Callable
import requests
from .config import AppConfig
from .config import Profile
from .credentials import SecureStore, get_cred
class AIClientError(RuntimeError):
"""Raised when the configured AI backend cannot return text."""
pass
def ask_ai(prompt: str, config: AppConfig) -> str:
ProviderResolver = Callable[[Profile, SecureStore], str]
def _default_resolver(profile: Profile, store: SecureStore) -> str:
return get_cred(store, profile.credential)
def ask_ai(
prompt: str,
profile: Profile,
store: SecureStore | None = None,
resolve_key: ProviderResolver = _default_resolver,
) -> str:
"""Envoie `prompt` au modèle du profil et retourne la réponse brute."""
if not prompt.strip():
raise AIClientError("Le texte capturé est vide.")
if config.provider == "ollama":
return _ask_ollama(prompt, config)
if config.provider == "gemini":
return _ask_gemini(prompt, config)
raise AIClientError(f"Provider non supporté: {config.provider}")
provider = profile.provider.lower()
if provider == "ollama":
return _ask_ollama(prompt, profile)
if provider in ("openai", "openrouter", "custom"):
return _ask_openai(prompt, profile, store, resolve_key)
if provider == "gemini":
return _ask_gemini(prompt, profile, store, resolve_key)
raise AIClientError(f"Fournisseur non supporté : {profile.provider}")
def _ask_ollama(prompt: str, config: AppConfig) -> str:
url = f"{config.server_url}/api/chat"
def _timeout(profile: Profile) -> float | None:
return profile.request_timeout_seconds
def _timeout_label(profile: Profile) -> str:
t = profile.request_timeout_seconds
return "désactivé" if t is None else f"{t:.0f} s"
def _ask_ollama(prompt: str, profile: Profile) -> str:
url = f"{profile.server_url}/api/chat"
payload = {
"model": config.model,
"model": profile.model,
"stream": False,
"messages": [
{"role": "system", "content": config.system_prompt},
{"role": "system", "content": profile.effective_prompt()},
{"role": "user", "content": prompt},
],
}
try:
response = requests.post(url, json=payload, timeout=config.request_timeout_seconds)
response = requests.post(url, json=payload, timeout=_timeout(profile))
response.raise_for_status()
data = response.json()
except requests.Timeout as exc:
timeout_label = "désactivé" if config.request_timeout_seconds is None else f"{config.request_timeout_seconds:.0f} s"
raise AIClientError(
"Ollama n'a pas répondu avant le délai configuré "
f"({timeout_label}). Le modèle est peut-être en chargement ou trop lent; "
"augmentez request_timeout_seconds dans config.json, ou mettez 0 pour désactiver le timeout."
f"({_timeout_label(profile)}). Augmentez request_timeout_seconds, "
"ou mettez 0 pour désactiver le timeout."
) from exc
except requests.RequestException as exc:
raise AIClientError(f"Erreur Ollama: {exc}") from exc
raise AIClientError(f"Erreur Ollama : {exc}") from exc
except ValueError as exc:
raise AIClientError("Réponse Ollama invalide: JSON illisible") from exc
raise AIClientError("Réponse Ollama invalide : JSON illisible") from exc
content = data.get("message", {}).get("content")
if not isinstance(content, str) or not content.strip():
@@ -52,37 +84,96 @@ def _ask_ollama(prompt: str, config: AppConfig) -> str:
return content.strip()
def _ask_gemini(prompt: str, config: AppConfig) -> str:
if not config.api_key:
raise AIClientError("api_key est obligatoire pour provider='gemini'.")
def _resolve_key(profile: Profile, store: SecureStore | None, resolve: ProviderResolver) -> str:
if store is None:
store = SecureStore()
key = resolve(profile, store)
if not key:
raise AIClientError(
f"Aucune clé d'API configurée pour le profil « {profile.name} ». "
"Ajoutez une authentification pour le fournisseur via l'icône de l'application."
)
return key
base = config.server_url or "https://generativelanguage.googleapis.com"
url = f"{base}/v1beta/models/{config.model}:generateContent"
def _ask_openai(
prompt: str,
profile: Profile,
store: SecureStore | None,
resolve_key: ProviderResolver,
) -> str:
"""Appels OpenAI-compatibles (OpenAI, OpenRouter, LM Studio, etc.)."""
key = _resolve_key(profile, store, resolve_key)
base = (profile.server_url or "https://api.openai.com/v1").rstrip("/")
url = f"{base}/chat/completions"
headers = {
"Authorization": f"Bearer {key}",
"Content-Type": "application/json",
}
payload = {
"systemInstruction": {"parts": [{"text": config.system_prompt}]},
"model": profile.model,
"messages": [
{"role": "system", "content": profile.effective_prompt()},
{"role": "user", "content": prompt},
],
}
try:
response = requests.post(
url, json=payload, headers=headers, timeout=_timeout(profile)
)
response.raise_for_status()
data = response.json()
except requests.Timeout as exc:
raise AIClientError(
"Le fournisseur n'a pas répondu avant le délai configuré "
f"({_timeout_label(profile)}). Augmentez request_timeout_seconds."
) from exc
except requests.RequestException as exc:
raise AIClientError(f"Erreur {profile.provider} : {exc}") from exc
except ValueError as exc:
raise AIClientError("Réponse du fournisseur invalide : JSON illisible") from exc
try:
text = data["choices"][0]["message"]["content"]
except (KeyError, IndexError, TypeError) as exc:
raise AIClientError("Réponse du fournisseur vide ou inattendue") from exc
if not isinstance(text, str) or not text.strip():
raise AIClientError("Réponse vide")
return text.strip()
def _ask_gemini(
prompt: str,
profile: Profile,
store: SecureStore | None,
resolve_key: ProviderResolver,
) -> str:
key = _resolve_key(profile, store, resolve_key)
base = profile.server_url or "https://generativelanguage.googleapis.com"
url = f"{base}/v1beta/models/{profile.model}:generateContent"
payload = {
"systemInstruction": {"parts": [{"text": profile.effective_prompt()}]},
"contents": [{"role": "user", "parts": [{"text": prompt}]}],
"generationConfig": {"temperature": 0.2},
}
try:
response = requests.post(
url,
params={"key": config.api_key},
params={"key": key},
json=payload,
timeout=config.request_timeout_seconds,
timeout=_timeout(profile),
)
response.raise_for_status()
data = response.json()
except requests.Timeout as exc:
timeout_label = "désactivé" if config.request_timeout_seconds is None else f"{config.request_timeout_seconds:.0f} s"
raise AIClientError(
"Gemini n'a pas répondu avant le délai configuré "
f"({timeout_label}). Augmentez request_timeout_seconds dans config.json, "
"ou mettez 0 pour désactiver le timeout."
f"({_timeout_label(profile)}). Augmentez request_timeout_seconds."
) from exc
except requests.RequestException as exc:
raise AIClientError(f"Erreur Gemini: {exc}") from exc
raise AIClientError(f"Erreur Gemini : {exc}") from exc
except ValueError as exc:
raise AIClientError("Réponse Gemini invalide: JSON illisible") from exc
raise AIClientError("Réponse Gemini invalide : JSON illisible") from exc
try:
text = data["candidates"][0]["content"]["parts"][0]["text"]
@@ -90,4 +181,4 @@ def _ask_gemini(prompt: str, config: AppConfig) -> str:
raise AIClientError("Réponse Gemini vide ou inattendue") from exc
if not isinstance(text, str) or not text.strip():
raise AIClientError("Réponse Gemini vide")
return text.strip()
return text.strip()
+104 -75
View File
@@ -1,107 +1,136 @@
"""Point d'entrée de l'application.
L'application démarre avec une **fenêtre unifiée à onglets** (Profils,
Authentification, Logs) et une icône dans la zone de notification. Fermer la
fenêtre la met en zone de notification (l'application continue de tourner) ;
le menu de l'icône réaffiche la fenêtre dans le bon onglet.
Deux modes restent disponibles pour le développement / le test :
- ``python main.py --ask "texte"`` : envoie le texte au profil actif et
imprime la réponse sans intercepter le clavier.
- ``python main.py`` : lance la fenêtre + l'icône.
Architecture (processus)
------------------------
* **Thread principal** : la racine Tk (``MainWindow``) et sa boucle
``mainloop``. Le raccourci global (``keyboard``) tourne dans son propre
thread d'écoute.
* **Thread UI (tray)** : icône pystray dans la zone de notification. Elle
n'ordonne des actions à la fenêtre que via ``master.after(0, …)``.
* **Une seule racine Tk** partagée : plus de fenêtre par processus.
"""
from __future__ import annotations
import argparse
import logging
import signal
import sys
import time
from threading import Lock
import threading
import keyboard
from .ai_client import AIClientError, ask_ai
from .clipboard_capture import capture_clipboard
from .config import AppConfig, load_config
from .key_stepper import KeyStepper
from .math_format import format_math_text
from .config import ConfigStore, load_config
from .credentials import SecureStore
from .engine import AITypewriterEngine, bind_hotkey
from .logging_utils import setup_logging
LOG = logging.getLogger("ai_typewriter")
def prepare_type_payload(answer: str, mode: str) -> str:
return format_math_text(answer, mode)
class AITypewriterApp:
def __init__(self, config: AppConfig) -> None:
self.config = config
self._busy = Lock()
self._stepper: KeyStepper | None = None
def run(self) -> None:
keyboard.add_hotkey(self.config.hotkey, self._handle_hotkey, suppress=False)
LOG.info("Prêt. Raccourci: %s. Quitter: Ctrl+C dans ce terminal.", self.config.hotkey)
keyboard.wait()
def _handle_hotkey(self) -> None:
if not self._busy.acquire(blocking=False):
LOG.warning("Requête déjà en cours, raccourci ignoré.")
return
try:
self._capture_ask_and_step()
finally:
self._busy.release()
def _capture_ask_and_step(self) -> None:
try:
selected = capture_clipboard()
LOG.info("Texte lu depuis le presse-papier: %d caractères.", len(selected))
LOG.info("Texte capturé: %r", selected)
started_at = time.perf_counter()
answer = ask_ai(selected, self.config)
answer_to_type = prepare_type_payload(answer, self.config.math_text_format)
elapsed = time.perf_counter() - started_at
LOG.info(
"Réponse reçue en %.2f s: %d caractères à écrire en mode %s. Appuyez sur une touche pour écrire chaque caractère.",
elapsed,
len(answer_to_type),
self.config.math_text_format,
)
if self._stepper is not None:
self._stepper.stop()
self._stepper = KeyStepper(answer_to_type, delay=self.config.type_delay_seconds)
self._stepper.start()
except (AIClientError, FileNotFoundError, ValueError) as exc:
LOG.error("%s", exc)
except Exception:
LOG.exception("Erreur inattendue")
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="AI Typewriter")
parser.add_argument("--config", default="config.json", help="Chemin du fichier config.json")
parser = argparse.ArgumentParser(description="AI-Typewriter")
parser.add_argument(
"--config",
default=None,
help="Chemin du fichier config.json (défaut: dossier appdata)",
)
parser.add_argument("--debug", action="store_true", help="Logs détaillés")
parser.add_argument("--ask", help="Mode test: envoie ce texte à l'IA et imprime la réponse, sans hook clavier")
parser.add_argument(
"--ask",
help="Mode test : envoie ce texte à l'IA du profil actif et imprime "
"la réponse, sans hook clavier ni icône.",
)
return parser
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
logging.basicConfig(
level=logging.DEBUG if args.debug else logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
)
setup_logging(level=logging.DEBUG)
LOG.debug("Démarrage : args=%s", args)
try:
config = load_config(args.config)
store = load_config(args.config)
except Exception as exc:
LOG.error("%s", exc)
return 2
engine = AITypewriterEngine(store, SecureStore())
if args.ask is not None:
try:
started_at = time.perf_counter()
answer = ask_ai(args.ask, config)
answer_to_type = prepare_type_payload(answer, config.math_text_format)
elapsed = time.perf_counter() - started_at
LOG.info("Réponse générée en %.2f s: %d caractères en mode %s.", elapsed, len(answer_to_type), config.math_text_format)
print(answer_to_type)
answer = engine.ask_only(args.ask)
print(answer)
return 0
except AIClientError as exc:
except Exception as exc:
LOG.error("%s", exc)
return 1
AITypewriterApp(config).run()
return run_background(store, engine)
def run_background(store: ConfigStore, engine: AITypewriterEngine) -> int:
"""Lance la fenêtre unifiée + l'icône de notification.
- Le thread principal crée ``MainWindow`` (racine Tk) puis entre dans
``mainloop``.
- L'icône de notification (pystray) tourne dans un thread dédié.
- Le raccourci global est enregistré via ``keyboard`` (thread d'écoute
interne).
"""
# -- raccourci global (thread d'écoute interne à keyboard) ----------------
hotkey = store.hotkey
try:
bind_hotkey(engine, hotkey)
LOG.info("Raccourci global actif : %s", hotkey)
except Exception as exc:
LOG.exception("Impossible d'enregistrer le raccourci : %s", exc)
# -- fenêtre unifiée (onglets) — thread principal -------------------------
from .ui.main_window import MainWindow
root = MainWindow(store)
# -- icône de notification (thread dédié) ---------------------------------
from .tray import TrayApp
stop_event = threading.Event()
tray = TrayApp(store, master=root, stop_event=stop_event)
ui_thread = threading.Thread(target=tray.run, daemon=True, name="ui-tray")
ui_thread.start()
# -- signaux d'arrêt propre ----------------------------------------------
def _handle_signal(signum, frame):
LOG.info("Signal %s reçu, arrêt…", signum)
stop_event.set()
try:
root.after(0, root.destroy)
except Exception:
pass
signal.signal(signal.SIGINT, _handle_signal)
signal.signal(signal.SIGTERM, _handle_signal)
LOG.info("Fenêtre principale affichée (onglets Profils / Authentification / Logs).")
try:
root.mainloop()
except KeyboardInterrupt:
pass
LOG.info("Arrêt demandé.")
tray.stop()
ui_thread.join(timeout=3)
LOG.info("Application terminée.")
return 0
if __name__ == "__main__":
raise SystemExit(main(sys.argv[1:]))
raise SystemExit(main(sys.argv[1:]))
+256 -58
View File
@@ -1,71 +1,269 @@
"""Profils et configuration de l'application.
La configuration est stockée dans un fichier JSON, créé par défaut dans
le répertoire des données de l'application (Windows : %APPDATA%\\ai-typewriter\\config.json).
Elle contient une liste de profils nommés, le profil actif et les réglages
globaux (raccourci clavier global, etc.).
Les clés d'API ne sont jamais écrites dans ce fichier : elles sont stockées
dans le gestionnaire de références de Windows (gestionnaire d'identifiants)
via la bibliothèque `keyring`. Ce fichier ne référence le secret que par un
nom logique (`credential`).
"""
from __future__ import annotations
import json
from dataclasses import dataclass
import os
import threading
from dataclasses import dataclass, field, asdict
from pathlib import Path
from typing import Any, Literal
Provider = Literal["ollama", "gemini"]
MathTextFormat = Literal["plain", "unicode", "unicode_math"]
from typing import Any
@dataclass(frozen=True)
class AppConfig:
provider: Provider = "ollama"
model: str = "llama3.1"
api_key: str = ""
server_url: str = "http://localhost:11434"
hotkey: str = "ctrl+alt+a"
request_timeout_seconds: float | None = 300.0
copy_wait_seconds: float = 1.0
type_delay_seconds: float = 0.0
math_text_format: MathTextFormat = "plain"
restore_clipboard: bool = True
system_prompt: str = (
# ---------------------------------------------------------------------------
# Chemins & constantes
# ---------------------------------------------------------------------------
APP_NAME = "ai-typewriter"
def appdata_dir() -> Path:
"""Répertoire de données applicatives (Linux : ~/.config/ai-typewriter).
Sur Windows ce sera %APPDATA%\\ai-typewriter ; sur les autres plateformes
on retombe sur le répertoire utilisateur pour rester fonctionnel.
"""
base = os.environ.get("APPDATA")
if base:
return Path(base) / APP_NAME
homedir = Path.home()
if os.name == "nt":
return homedir / "AppData" / "Roaming" / APP_NAME
return homedir / ".config" / APP_NAME
def config_path() -> Path:
return appdata_dir() / "config.json"
# ---------------------------------------------------------------------------
# Marquage des équations (profil « math »)
# ---------------------------------------------------------------------------
DEFAULT_EQ_START_MARKER = "[EQ]"
DEFAULT_EQ_END_MARKER = "[/EQ]"
# Séquence de touches envoyée : Alt+= ouvre une équation inline (Word/OneNote).
DEFAULT_EQ_START_KEY = "alt+="
# Enregistrement Word pour sortir du champ d'équation.
DEFAULT_EQ_END_KEY = "right"
def default_system_prompt() -> str:
return (
"Réponds directement et de manière ultra-concise. "
"Aucune phrase d'introduction, aucune salutation, aucun formatage superflu. "
"Uniquement la réponse brute."
)
def _coerce_provider(value: Any) -> Provider:
provider = str(value or "ollama").lower().strip()
if provider not in {"ollama", "gemini"}:
raise ValueError("config.provider doit être 'ollama' ou 'gemini'")
return provider # type: ignore[return-value]
def _coerce_timeout(value: Any) -> float | None:
timeout = float(value)
if timeout <= 0:
return None
return timeout
def _coerce_math_text_format(value: Any) -> MathTextFormat:
mode = str(value or "plain").lower().strip()
if mode not in {"plain", "unicode", "unicode_math"}:
raise ValueError("config.math_text_format doit être 'plain' ou 'unicode'")
return mode # type: ignore[return-value]
def load_config(path: str | Path = "config.json") -> AppConfig:
cfg_path = Path(path)
if not cfg_path.exists():
raise FileNotFoundError(
f"Configuration introuvable: {cfg_path}. Copiez config.json.template vers config.json."
)
data = json.loads(cfg_path.read_text(encoding="utf-8"))
return AppConfig(
provider=_coerce_provider(data.get("provider", "ollama")),
model=str(data.get("model", "llama3.1")),
api_key=str(data.get("api_key", "")),
server_url=str(data.get("server_url", "http://localhost:11434")).rstrip("/"),
hotkey=str(data.get("hotkey", "ctrl+alt+a")).lower(),
request_timeout_seconds=_coerce_timeout(data.get("request_timeout_seconds", 300)),
copy_wait_seconds=float(data.get("copy_wait_seconds", 1)),
type_delay_seconds=float(data.get("type_delay_seconds", 0)),
math_text_format=_coerce_math_text_format(data.get("math_text_format", "plain")),
restore_clipboard=bool(data.get("restore_clipboard", True)),
system_prompt=str(data.get("system_prompt", AppConfig.system_prompt)),
def default_math_latex_prompt() -> str:
return (
"Tu réponds directement, sans salutation ni introduction. "
"Rédige une réponse claire, correcte et concise. "
"Pour toute expression mathématique, formule, calcul, égalité, fraction, somme, "
"intégrale, matrice ou symbole destiné à être traité comme une équation, encadre "
"exactement le bloc avec [EQ] au début et [/EQ] à la fin. Dans ces blocs écris du "
"LaTeX standard, plus simple et fiable à générer : \\frac{a}{b}, z_1, x^2, "
"\\int_0^1, \\sum_{k=1}^n, etc. N'utilise pas de délimiteurs LaTeX supplémentaires "
"dans les blocs (pas de $, $$, \\\\(, \\\\[). Le texte hors des balises "
"[EQ]...[/EQ] reste du texte normal. Exemple valide : "
"Les racines sont [EQ]z_1 = x + iy[/EQ] et [EQ]z_2 = x - iy[/EQ]."
)
# ---------------------------------------------------------------------------
# Modèle de profil
# ---------------------------------------------------------------------------
@dataclass
class Profile:
"""Un profil = un fournisseur + un modèle + un prompt système + des réglages."""
name: str = "Général"
provider: str = "ollama" # ollama | openai | gemini
model: str = "llama3.1"
server_url: str = "http://localhost:11434"
request_timeout_seconds: float | None = 300.0
type_delay_seconds: float = 0.0
system_prompt: str = ""
# Gestion des équations LaTeX : si actif, les marqueurs sont interceptés et
# remplacés par des séquences de touches.
equation_enabled: bool = False
eq_start_marker: str = DEFAULT_EQ_START_MARKER
eq_end_marker: str = DEFAULT_EQ_END_MARKER
eq_start_key: str = DEFAULT_EQ_START_KEY
eq_end_key: str = DEFAULT_EQ_END_KEY
# Nom logique de la référence stockée dans le gestionnaire de références
# (vide si le fournisseur n'exige pas de clé, ex : Ollama local).
credential: str = ""
# Réglages de capture.
copy_wait_seconds: float = 1.0
restore_clipboard: bool = True
def __post_init__(self) -> None:
if not self.system_prompt:
self.system_prompt = default_system_prompt()
if self.provider == "ollama":
self.server_url = (self.server_url or "http://localhost:11434").rstrip("/")
def effective_prompt(self) -> str:
return self.system_prompt or default_system_prompt()
def to_dict(self) -> dict[str, Any]:
return asdict(self)
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "Profile":
known = {f for f in cls.__dataclass_fields__} # type: ignore[attr-defined]
return cls(**{k: v for k, v in data.items() if k in known})
def math_latex_profile(name: str = "Mathématiques (LaTeX)", **overrides: Any) -> Profile:
"""Profil d'exemple spécialisé en mathématiques (LaTeX encadré)."""
data = dict(
name=name,
provider="ollama",
model="llama3.1",
server_url="http://localhost:11434",
system_prompt=default_math_latex_prompt(),
equation_enabled=True,
eq_start_marker=DEFAULT_EQ_START_MARKER,
eq_end_marker=DEFAULT_EQ_END_MARKER,
eq_start_key=DEFAULT_EQ_START_KEY,
eq_end_key=DEFAULT_EQ_END_KEY,
)
data.update(overrides)
return Profile.from_dict(data)
def default_profiles() -> list[Profile]:
return [
Profile(name="Général"),
math_latex_profile(),
]
# ---------------------------------------------------------------------------
# Dépôt de configuration
# ---------------------------------------------------------------------------
class ConfigError(RuntimeError):
pass
class ConfigStore:
"""Lecture/écriture du fichier de configuration et sélection du profil actif."""
def __init__(self, path: str | Path | None = None) -> None:
self.path = Path(path) if path else config_path()
self._lock = threading.Lock()
self.active_name: str = "Général"
self.profiles: list[Profile] = []
self.hotkey: str = "ctrl+alt+a"
self._loaded = False
# -- persistance --------------------------------------------------------
def ensure_defaults(self) -> None:
"""Crée le dossier et le fichier de configuration par défaut si absents."""
self.path.parent.mkdir(parents=True, exist_ok=True)
if not self.path.exists():
self.profiles = default_profiles()
self.active_name = self.profiles[0].name
self.hotkey = "ctrl+alt+a"
self.save()
def load(self) -> None:
self.ensure_defaults()
with self._lock:
data = json.loads(self.path.read_text(encoding="utf-8"))
self.hotkey = str(data.get("hotkey", "ctrl+alt+a")).lower()
self.active_name = str(data.get("active_profile", "Général"))
raw_profiles = data.get("profiles", [])
if not raw_profiles:
raw_profiles = [p.to_dict() for p in default_profiles()]
self.profiles = [Profile.from_dict(p) for p in raw_profiles]
if not self.profiles:
raise ConfigError("Aucun profil disponible dans la configuration.")
names = [p.name for p in self.profiles]
if self.active_name not in names:
self.active_name = names[0]
self._loaded = True
def save(self) -> None:
with self._lock:
body = {
"hotkey": self.hotkey,
"active_profile": self.active_name,
"profiles": [p.to_dict() for p in self.profiles],
}
self.path.parent.mkdir(parents=True, exist_ok=True)
self.path.write_text(
json.dumps(body, ensure_ascii=False, indent=2), encoding="utf-8"
)
# -- profils -------------------------------------------------------------
def get_all(self) -> list[Profile]:
if not self._loaded:
self.load()
return list(self.profiles)
def get(self, name: str) -> Profile:
for p in self.get_all():
if p.name == name:
return p
raise KeyError(name)
def active(self) -> Profile:
if not self._loaded:
self.load()
for p in self.profiles:
if p.name == self.active_name:
return p
return self.profiles[0]
def set_active(self, name: str) -> None:
self.get(name) # valide l'existence
self.active_name = name
self.save()
def upsert(self, profile: Profile) -> None:
if not self._loaded:
self.load()
for i, p in enumerate(self.profiles):
if p.name == profile.name:
self.profiles[i] = profile
break
else:
self.profiles.append(profile)
self.save()
def remove(self, name: str) -> None:
if not self._loaded:
self.load()
if len(self.profiles) == 1:
raise ConfigError("Impossible de supprimer le dernier profil.")
self.profiles = [p for p in self.profiles if p.name != name]
if self.active_name == name:
self.active_name = self.profiles[0].name
self.save()
def load_config(path: str | Path | None = None) -> ConfigStore:
store = ConfigStore(path)
store.load()
return store
+86
View File
@@ -0,0 +1,86 @@
"""Stockage sécurisé des clés d'API.
Sous Windows, les secrets sont conservés dans le Gestionnaire d'identifiants
(Credential Manager) via `keyring`, et jamais écrits en clair dans un fichier
de configuration. Sur les autres plateformes, on retombe sur le trousseau
fourni par keyring (éventuellement crypté) pour rester utilisable.
"""
from __future__ import annotations
import logging
LOG = logging.getLogger("ai_typewriter.credentials")
try: # pragma: no cover - dépend de la plateforme
import keyring
except Exception: # pragma: no cover
keyring = None # type: ignore
SERVICE = "ai-typewriter"
class CredentialError(RuntimeError):
pass
class SecureStore:
"""Interface vers le stockage sécurisé des références par fournisseur.
Une « référence » est identifiée par un nom logique (celui porté par le
profil dans `Profile.credential`). Le même nom peut être partagé par
plusieurs profils d'un même fournisseur.
"""
def __init__(self, service: str = SERVICE) -> None:
self.service = service
def store(self, credential: str, secret: str) -> None:
if not credential:
raise CredentialError("Nom de référence invalide.")
if keyring is None: # pragma: no cover
raise CredentialError(
"keyring indisponible : impossible de stocker une clé de façon sécurisée."
)
try:
keyring.set_password(self.service, credential, secret)
except Exception as exc: # pragma: no cover
LOG.exception("Échec de l'enregistrement de la référence %s", credential)
raise CredentialError(
f"Impossible d'enregistrer la référence « {credential} ». "
f"Vérifiez que le Gestionnaire d'identifiants est disponible : {exc}"
) from exc
def get(self, credential: str) -> str:
if not credential:
return ""
if keyring is None: # pragma: no cover
return ""
try:
value = keyring.get_password(self.service, credential)
except Exception: # pragma: no cover
LOG.exception("Lecture impossible de la référence %s", credential)
return ""
return value or ""
def delete(self, credential: str) -> None:
if keyring is None: # pragma: no cover
raise CredentialError("keyring indisponible.")
try:
keyring.delete_password(self.service, credential)
except keyring.errors.PasswordDeleteError: # type: ignore[attr-defined]
return
except Exception as exc: # pragma: no cover
raise CredentialError(f"Suppression impossible de « {credential} »") from exc
def has(self, credential: str) -> bool:
return bool(self.get(credential))
def get_cred(store: SecureStore | None, credential: str) -> str:
"""Raccourci : lit la référence via le store, en créant un si nécessaire."""
if store is None:
store = SecureStore()
if not credential:
return ""
return store.get(credential)
+108
View File
@@ -0,0 +1,108 @@
"""Moteur principal : capture presse-papier -> IA -> dactylographie.
Ce module est indépendant de l'interface (tray ou console) et peut être
réutilisé par l'application graphique, la ligne de commande ou les tests.
"""
from __future__ import annotations
import logging
import threading
import time
import keyboard
from .ai_client import AIClientError, ask_ai
from .clipboard_capture import capture_clipboard
from .config import ConfigStore, Profile
from .credentials import SecureStore
from .key_stepper import KeyStepper
LOG = logging.getLogger("ai_typewriter")
def prepare_type_payload(answer: str, profile: Profile) -> str:
"""La réponse est écrite telle quelle (le découpage des marqueurs équation
est géré dans KeyStepper). Aucune normalisation Unicode n'est appliquée."""
return answer
class AITypewriterEngine:
def __init__(self, store: ConfigStore, secure: SecureStore | None = None) -> None:
self.store = store
self.secure = secure or SecureStore()
self._busy = threading.Lock()
self._stepper: KeyStepper | None = None
# -- boucle d'interaction ---------------------------------------------
def handle_hotkey(self) -> None:
if not self._busy.acquire(blocking=False):
LOG.warning("Requête déjà en cours, raccourci ignoré.")
return
try:
self.capture_ask_and_step()
finally:
self._busy.release()
def capture_ask_and_step(self) -> None:
try:
selected = capture_clipboard()
if not selected.strip():
LOG.warning("Presse-papier vide ; rien à envoyer.")
return
LOG.info("Texte lu depuis le presse-papier : %d caractères.", len(selected))
LOG.info("Texte capturé : %r", selected)
profile = self.store.active()
started_at = time.perf_counter()
answer = ask_ai(selected, profile, store=self.secure)
answer_to_type = prepare_type_payload(answer, profile)
elapsed = time.perf_counter() - started_at
if profile.equation_enabled:
mode = f"équations {profile.model}"
else:
mode = "texte"
LOG.info(
"Réponse reçue en %.2f s : %d caractères à écrire (%s). "
"Appuyez sur une touche pour écrire chaque élément.",
elapsed,
len(answer_to_type),
mode,
)
if self._stepper is not None:
self._stepper.stop()
self._stepper = KeyStepper(
answer_to_type,
delay=profile.type_delay_seconds,
equation_enabled=profile.equation_enabled,
eq_start_marker=profile.eq_start_marker,
eq_end_marker=profile.eq_end_marker,
eq_start_key=profile.eq_start_key,
eq_end_key=profile.eq_end_key,
)
self._stepper.start()
except (AIClientError, FileNotFoundError, ValueError, KeyError) as exc:
LOG.error("%s", exc)
except Exception:
LOG.exception("Erreur inattendue")
# -- interface publique pour l'UI ---------------------------------------
def ask_only(self, prompt: str) -> str:
"""Envoie un prompt au profil actif et retourne la réponse (sans taper)."""
profile = self.store.active()
return ask_ai(prompt, profile, store=self.secure)
def current_profile(self) -> Profile:
return self.store.active()
def stop_stepper(self) -> None:
if self._stepper is not None:
self._stepper.stop()
def bind_hotkey(engine: AITypewriterEngine, hotkey: str) -> None:
keyboard.add_hotkey(hotkey, engine.handle_hotkey, suppress=False)
+95 -11
View File
@@ -1,48 +1,132 @@
"""Dactylographe : chaque pression de touche physique fait avancer l'écriture.
Le texte de réponse est découpé en « actions » : caractères littéraux à taper
ou séquences de touches à envoyer. Quand la gestion d'équations est active,
les marqueurs de début/fin d'équation (ex. `[EQ]` / `[/EQ]`) sont interceptés
et remplacés par une séquence de touches (par défaut `Alt+=` pour ouvrir une
équation Word et `→` pour en sortir) au lieu d'être tapés littéralement.
"""
from __future__ import annotations
import re
from dataclasses import dataclass, field
from threading import Lock
from typing import Optional
from typing import Iterable, Optional
import keyboard
# ---------------------------------------------------------------------------
# Découpage en actions (pure, unité testable sans clavier)
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class Action:
kind: str # "char" | "seq"
value: str
_MARKER_TOKENS = object() # injection simple pour les tests
def build_actions(
text: str,
equation_enabled: bool = False,
eq_start_marker: str = "[EQ]",
eq_end_marker: str = "[/EQ]",
eq_start_key: str = "alt+=",
eq_end_key: str = "right",
) -> list[Action]:
"""Convertit la réponse IA en liste d'actions à exécuter séquentiellement.
- Sans équation : une action `char` par caractère (comportement d'origine).
- Avec équation : les marqueurs sont retirés et remplacés par une action
`seq` envoyant la séquence de touches correspondante. Le contenu entre
les marqueurs reste tapé caractère par caractère.
"""
if not equation_enabled or not eq_start_marker or not eq_end_marker:
return [Action("char", ch) for ch in text]
start_re = re.escape(eq_start_marker)
end_re = re.escape(eq_end_marker)
pattern = re.compile(f"({start_re}|{end_re})")
actions: list[Action] = []
for part in pattern.split(text):
if not part:
continue
if part == eq_start_marker:
actions.append(Action("seq", eq_start_key))
elif part == eq_end_marker:
actions.append(Action("seq", eq_end_key))
else:
actions.extend(Action("char", ch) for ch in part)
return actions
# ---------------------------------------------------------------------------
# Stepper (nécessite le hook clavier ; testé par monkeypatch)
# ---------------------------------------------------------------------------
@dataclass
class KeyStepper:
text: str
delay: float = 0.0
equation_enabled: bool = False
eq_start_marker: str = "[EQ]"
eq_end_marker: str = "[/EQ]"
eq_start_key: str = "alt+="
eq_end_key: str = "right"
_actions: list[Action] = field(default_factory=list)
_index: int = 0
_hook: Optional[object] = None
_lock: Lock = field(default_factory=Lock)
_injecting: bool = False
def __post_init__(self) -> None:
if not self._actions:
self._actions = build_actions(
self.text,
equation_enabled=self.equation_enabled,
eq_start_marker=self.eq_start_marker,
eq_end_marker=self.eq_end_marker,
eq_start_key=self.eq_start_key,
eq_end_key=self.eq_end_key,
)
def start(self) -> None:
if not self.text:
if not self._actions:
return
if self._hook is not None:
return
# suppress=True blocks the physical key while the callback types the next AI character.
# suppress=True bloque la touche physique pendant que le callback émet
# la prochaine action de la réponse IA.
self._hook = keyboard.hook(self._on_event, suppress=True)
@property
def remaining_characters(self) -> int:
return max(len(self.text) - self._index, 0)
return max(len(self._actions) - self._index, 0)
def type_next_character(self) -> None:
def type_next_action(self) -> None:
with self._lock:
if self._index >= len(self.text):
if self._index >= len(self._actions):
self.stop()
return
char = self.text[self._index]
action = self._actions[self._index]
self._index += 1
self._injecting = True
try:
self._type_char(char)
if action.kind == "seq":
keyboard.send(action.value)
else:
self._type_char(action.value)
finally:
self._injecting = False
if self._index >= len(self.text):
if self._index >= len(self._actions):
self.stop()
def stop(self) -> None:
@@ -54,7 +138,7 @@ class KeyStepper:
def _on_event(self, event: keyboard.KeyboardEvent) -> None:
if self._injecting or event.event_type != keyboard.KEY_DOWN:
return
self.type_next_character()
self.type_next_action()
def _type_char(self, char: str) -> None:
if char == "\n":
@@ -62,4 +146,4 @@ class KeyStepper:
elif char == "\t":
keyboard.send("tab")
else:
keyboard.write(char, delay=self.delay, exact=True)
keyboard.write(char, delay=self.delay, exact=True)
+92
View File
@@ -0,0 +1,92 @@
"""Journalisation : écriture sur fichier + diffusion vers les vues en direct.
La fenêtre « Ouvrir les logs » s'abonne à ce module et reçoit les messages en
temps réel ; les logs sont également écrits dans %APPDATA%\\ai-typewriter\\logs\\.
"""
from __future__ import annotations
import logging
import logging.handlers
import queue
import threading
from pathlib import Path
from .config import appdata_dir
class QueueHandler(logging.Handler):
"""Forwarde chaque enregistrement vers une `queue.Queue`."""
def __init__(self, q: "queue.Queue[logging.LogRecord] | None" = None) -> None:
super().__init__()
self.queue: queue.Queue = q if q is not None else queue.Queue()
def emit(self, record: logging.LogRecord) -> None:
try:
self.queue.put_nowait(record)
except Exception:
pass
class LogBroadcaster:
"""Point central : une file de diffusion + capacité à ajouter des vues."""
def __init__(self) -> None:
self._q: queue.Queue = queue.Queue()
self.handler = QueueHandler(self._q)
def install(self, level: int = logging.INFO) -> None:
root = logging.getLogger()
root.addHandler(self.handler)
root.setLevel(level)
self.handler.setLevel(level)
def drain(self) -> list[logging.LogRecord]:
out: list[logging.LogRecord] = []
while True:
try:
out.append(self._q.get_nowait())
except queue.Empty:
return out
_broadcaster: LogBroadcaster | None = None
def broadcaster() -> LogBroadcaster:
global _broadcaster
if _broadcaster is None:
_broadcaster = LogBroadcaster()
return _broadcaster
def logs_dir() -> Path:
return appdata_dir() / "logs"
def setup_file_logging(level: int = logging.INFO) -> Path:
"""Configure un handler fichier (rotation quotidienne) et retourne le chemin."""
d = logs_dir()
d.mkdir(parents=True, exist_ok=True)
path = d / "app.log"
handler = logging.handlers.RotatingFileHandler(
path, maxBytes=2 * 1024 * 1024, backupCount=3, encoding="utf-8"
)
handler.setFormatter(
logging.Formatter(
"%(asctime)s [%(levelname)s] %(name)s [tid:%(thread)d:%(threadName)s] %(message)s"
)
)
logging.getLogger().addHandler(handler)
return path
def setup_logging(level: int = logging.DEBUG) -> Path:
broadcaster().install(level)
return setup_file_logging(level)
def format_record(record: logging.LogRecord) -> str:
ts = record.asctime if record.asctime else logging.Formatter().formatTime(record)
return f"{ts} {record.levelname} {record.getMessage()}"
-88
View File
@@ -1,88 +0,0 @@
from __future__ import annotations
import re
GREEK_AND_SYMBOLS = {
r"\alpha": "α",
r"\beta": "β",
r"\gamma": "γ",
r"\delta": "δ",
r"\epsilon": "ε",
r"\theta": "θ",
r"\lambda": "λ",
r"\mu": "μ",
r"\pi": "π",
r"\sigma": "σ",
r"\phi": "φ",
r"\omega": "ω",
r"\Delta": "Δ",
r"\Omega": "Ω",
r"\infty": "∞",
r"\leq": "≤",
r"\le": "≤",
r"\geq": "≥",
r"\ge": "≥",
r"\neq": "≠",
r"\ne": "≠",
r"\approx": "≈",
r"\times": "×",
r"\cdot": "·",
r"\pm": "±",
r"\to": "→",
r"\rightarrow": "→",
r"\int": "∫",
r"\sum": "∑",
r"\sqrt": "√",
}
SUPERSCRIPT = str.maketrans({
"0": "⁰", "1": "¹", "2": "²", "3": "³", "4": "⁴",
"5": "⁵", "6": "⁶", "7": "⁷", "8": "⁸", "9": "⁹",
"+": "⁺", "-": "⁻", "=": "⁼", "(": "⁽", ")": "⁾",
"n": "ⁿ", "i": "ⁱ",
})
SUBSCRIPT = str.maketrans({
"0": "₀", "1": "₁", "2": "₂", "3": "₃", "4": "₄",
"5": "₅", "6": "₆", "7": "₇", "8": "₈", "9": "₉",
"+": "₊", "-": "₋", "=": "₌", "(": "₍", ")": "₎",
"a": "ₐ", "e": "ₑ", "h": "ₕ", "i": "ᵢ", "j": "ⱼ", "k": "ₖ",
"l": "ₗ", "m": "ₘ", "n": "ₙ", "o": "ₒ", "p": "ₚ", "r": "ᵣ",
"s": "ₛ", "t": "ₜ", "u": "ᵤ", "v": "ᵥ", "x": "ₓ",
})
_SCRIPT_PATTERN = re.compile(r"([_^])\(([^()]+)\)|([_^])\{([^{}]+)\}|([_^])([A-Za-z0-9+\-=])")
def _translate_script(value: str, marker: str) -> str:
table = SUPERSCRIPT if marker == "^" else SUBSCRIPT
converted = value.translate(table)
return converted if converted != value else marker + value
def _replace_script(match: re.Match[str]) -> str:
marker = match.group(1) or match.group(3) or match.group(5)
value = match.group(2) or match.group(4) or match.group(6)
return _translate_script(value, marker)
def format_math_text(text: str, mode: str = "plain") -> str:
"""Format the AI answer before key stepping.
plain keeps the response unchanged, which is the recommended mode for LaTeX
markers such as [EQ]\\frac{a}{b}[/EQ].
"""
if mode == "plain":
return text
if mode not in {"unicode", "unicode_math"}:
raise ValueError("math_text_format doit être 'plain' ou 'unicode'")
formatted = text
for command, replacement in sorted(GREEK_AND_SYMBOLS.items(), key=lambda item: len(item[0]), reverse=True):
formatted = formatted.replace(command, replacement)
previous = None
while previous != formatted:
previous = formatted
formatted = _SCRIPT_PATTERN.sub(_replace_script, formatted)
return formatted
+210
View File
@@ -0,0 +1,210 @@
"""Actuaire des modèles : modèles locaux (Ollama), catalogue en ligne et
fournisseurs tiers.
L'interface de création de profil liste automatiquement les modèles déjà
disponibles sur la machine et, si l'utilisateur est connecté à Internet,
propose de parcourir/rechercher le catalogue public d'Ollama et de lancer un
téléchargement (pull) vers l'instance locale.
"""
from __future__ import annotations
import logging
import shutil
import subprocess
from dataclasses import dataclass
from typing import Iterable
import requests
LOG = logging.getLogger("ai_typewriter.model_catalog")
OLLAMA_LIBRARY_SEARCH = "https://ollama.com/search?q={query}"
OLLAMA_DIRECTORY_API = "https://ollama.com/api/models" # non fourni par Ollama, laissé en secours
class CatalogError(RuntimeError):
pass
@dataclass(frozen=True)
class CatalogModel:
name: str
source: str # "local" | "registry"
size: int = 0
family: str = ""
description: str = ""
available_locally: bool = False
def display(self) -> str:
if self.source == "local":
return self.name
return f"{self.name} (à télécharger)"
# ---------------------------------------------------------------------------
# Fournisseurs tiers
# ---------------------------------------------------------------------------
PROVIDERS = [
{
"id": "ollama",
"label": "Ollama (local)",
"needs_key": False,
"base_url": "http://localhost:11434",
},
{
"id": "openai",
"label": "OpenAI",
"needs_key": True,
"base_url": "https://api.openai.com/v1",
},
{
"id": "openrouter",
"label": "OpenRouter",
"needs_key": True,
"base_url": "https://openrouter.ai/api/v1",
},
{
"id": "gemini",
"label": "Google Gemini",
"needs_key": True,
"base_url": "https://generativelanguage.googleapis.com",
},
{
"id": "custom",
"label": "Personnalisé (OpenAI-compatible)",
"needs_key": True,
"base_url": "",
},
]
def provider_info(provider_id: str) -> dict:
for p in PROVIDERS:
if p["id"] == provider_id:
return dict(p)
return {"id": provider_id, "label": provider_id, "needs_key": True, "base_url": ""}
def list_providers() -> list[dict]:
return [dict(p) for p in PROVIDERS]
# ---------------------------------------------------------------------------
# Modèles locaux Ollama
# ---------------------------------------------------------------------------
def list_local_models(server_url: str = "http://localhost:11434", timeout: float = 5.0) -> list[str]:
"""Interroge `/api/tags` de l'instance Ollama pour lister les modèles locaux."""
url = f"{server_url}/api/tags"
try:
response = requests.get(url, timeout=timeout)
response.raise_for_status()
data = response.json()
except requests.RequestException as exc:
LOG.debug("Impossible de lister les modèles Ollama locaux : %s", exc)
return []
names: list[str] = []
for model in data.get("models", []):
name = model.get("name") or model.get("model")
if name:
names.append(str(name))
return sorted(set(names))
# ---------------------------------------------------------------------------
# Catalogue en ligne Ollama
# ---------------------------------------------------------------------------
def search_online_models(query: str = "", timeout: float = 10.0) -> list[CatalogModel]:
"""Recherche dans la bibliothèque publique d'Ollama.
Note : Ollama ne fournit pas de JSON public stable. On essaie le registry
OpenID/OAuth des modèles populaires si `query` nomme un modèle exact, sinon
on retourne une liste de modèles courants filtrée par le terme recherché,
plutôt que d'échouer quand l'API HTML n'est pas accessible.
"""
raw = raw_library_post_models()
term = (query or "").strip().lower()
if term:
raw = [m for m in raw if term in m.lower()]
return [
CatalogModel(name=m, source="registry", available_locally=False)
for m in raw
]
def raw_library_post_models() -> list[str]:
"""Liste de modèles très courants de la bibliothèque Ollama (fourchette de
recherche), tous publiés sous le namespace `library`."""
return [
"llama3.2", "llama3.1", "llama3", "llama3.3",
"mistral", "mistral-nemo", "mixtral", "codestral",
"qwen2.5", "qwen2.5-coder", "qwen",
"gemma2", "gemma3",
"phi4", "phi3", "phi3.5",
"deepseek-r1", "deepseek-coder-v2", "deepseek-v3",
"yi", "command-r", "command-r-plus", "smollm2",
"llava", "bakllava", "llava-phi3",
"nomic-embed-text", "mxbai-embed-large", "bge-m3",
]
def resolve_exact_model(model_name: str, timeout: float = 10.0) -> bool:
"""Vérifie si `model_name` existe bien dans la bibliothèque publique Ollama
en interrogeant le registry (tags list). Retourne Vrai si oui."""
if not model_name.strip():
return False
url = f"https://registry.ollama.ai/v2/library/{model_name.strip()}/tags/list"
try:
r = requests.get(url, timeout=timeout)
return r.status_code == 200
except requests.RequestException:
return False
# ---------------------------------------------------------------------------
# Téléchargement (pull)
# ---------------------------------------------------------------------------
def pull_model(model_name: str, server_url: str = "http://localhost:11434", host: str = "") -> None:
"""Lance `ollama pull <modèle>` vers l'instance locale.
Si le binaire `ollama` est présent, on appelle directement la CLI, sinon on
tente l'API HTTP Ollama (`POST /api/pull`, flux). Dans tous les cas on
préfère ne pas bloquer : une erreur de téléchargement ne fait pas échouer
la création de profil.
"""
exe = shutil.which("ollama")
if exe:
if host:
cmd = [exe, "--host", host, "pull", model_name]
else:
cmd = [exe, "pull", model_name]
LOG.info("Téléchargement du modèle %s via la CLI Ollama…", model_name)
subprocess.Popen(
cmd,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
return
# Secours HTTP (déclenché de façon best-effort, sans blocage du flux).
url = f"{server_url}/api/pull"
try:
LOG.info("Téléchargement du modèle %s via l'API Ollama…", model_name)
requests.post(url, json={"name": model_name, "stream": True}, timeout=1.0)
except requests.RequestException as exc:
LOG.warning("Impossible de déclencher le pull HTTP : %s", exc)
def has_internet(timeout: float = 4.0) -> bool:
"""Détection simple de connexion Internet (atteinte de Ollama.com)."""
try:
requests.head("https://ollama.com", timeout=timeout, allow_redirects=True)
return True
except requests.RequestException:
return False
+209
View File
@@ -0,0 +1,209 @@
"""Icône de zone de notification — réaffiche la fenêtre unifiée dans le bon onglet.
La fenêtre principale unique (onglets Profils / Authentification / Logs) vit
dans le **processus principal**, racine ``tk.Tk`` sur le thread principal.
L'icône pystray tourne dans un **thread dédié** et ne fait qu'ordonner des
actions à la fenêtre via ``master.after(0, …)`` — la seule façon sûre de
toucher à Tk depuis un autre thread.
* **Fermer la fenêtre (✕)** → ``withdraw()`` : l'app continue de tourner et
l'icône reste visible.
* **Ouvrir les logs / Ajouter un profil / Gérer l'authentification** →
réaffiche la fenêtre et sélectionne l'onglet correspondant.
* **Modifier le profil (sous-menu radio)** → bascule le profil actif
directement, **sans** afficher la fenêtre.
* **Quitter** → arrête l'icône et détruit la fenêtre.
"""
from __future__ import annotations
import logging
import threading
from typing import Callable
from .config import ConfigStore
from .ui.main_window import MainWindow, TAB_AUTH, TAB_LOGS, TAB_PROFILES
LOG = logging.getLogger("ai_typewriter.tray")
class TrayApp:
"""Encapsule l'icône de zone de notification (pystray)."""
def __init__(
self,
store: ConfigStore,
master: MainWindow | None = None,
icon_factory: Callable | None = None,
menu_factory: Callable | None = None,
stop_event: threading.Event | None = None,
) -> None:
self.store = store
self.master = master
self._icon = None
self._icon_factory = icon_factory
self._menu_factory = menu_factory
self._stop_event = stop_event
# -- cycle de vie -----------------------------------------------------------
def run(self) -> None:
"""Lance la boucle pystray (bloquant, appelé depuis le thread UI)."""
icon = self._icon if self._icon is not None else self.build_icon()
self._icon = icon
LOG.info("Icône de notification lancée (pystray).")
icon.run()
def stop(self, icon=None, item=None) -> None:
"""Arrête l'icône et demande la fermeture de la fenêtre principale."""
LOG.info("Arrêt de l'icône de notification.")
if self._icon is not None:
try:
self._icon.stop()
except Exception:
pass
if self._stop_event is not None:
self._stop_event.set()
if self.master is not None:
try:
self.master.after(0, self.master.destroy)
except Exception:
pass
# -- actions du menu --------------------------------------------------------
def _show_logs(self) -> None:
self._show_tab(TAB_LOGS)
def _add_profile(self) -> None:
self._marshal(self.master.open_new_profile)
self._show_tab(TAB_PROFILES)
def _manage_auth(self) -> None:
self._show_tab(TAB_AUTH)
def _show_tab(self, name: str) -> None:
self._marshal(lambda: self.master.show_tab(name))
def _set_active(self, name: str) -> None:
try:
self.store.set_active(name)
LOG.info("Profil actif : %s", name)
self._refresh_menu()
self._marshal(self.master.refresh_all)
except Exception as exc:
LOG.exception("Impossible de sélectionner le profil : %s", exc)
def _marshal(self, fn: Callable[[], None]) -> None:
"""Exécute ``fn`` sur le thread Tk via ``master.after(0, …)``."""
if self.master is None:
return
try:
self.master.after(0, fn)
except Exception:
LOG.debug("Ordre Tk impossible (fenêtre détruite ?)", exc_info=True)
def _refresh_menu(self) -> None:
"""Demande à pystray de reconstruire le menu (met à jour les coches)."""
if self._icon is not None:
try:
self._icon.update_menu()
except Exception:
LOG.debug("update_menu() a échoué", exc_info=True)
# -- construction de l'icône ------------------------------------------------
def build_icon(self):
import pystray
from PIL import Image, ImageDraw
def _image() -> Image.Image:
img = Image.new("RGB", (64, 64), "#1f1f1f")
d = ImageDraw.Draw(img)
d.text((10, 14), "AW", fill="#ffffff")
return img
menu_items = []
menu_items.append(
self._menu_item("Ouvrir les logs", self._guard("Ouvrir les logs", self._show_logs))
)
menu_items.append(
self._menu_item("Ajouter un profil", self._guard("Ajouter un profil", self._add_profile))
)
profiles_sub = pystray.Menu(self._profile_menu_items)
menu_items.append(self._menu_item("Modifier le profil", None, submenu=profiles_sub))
menu_items.append(pystray.Menu.SEPARATOR)
menu_items.append(
self._menu_item(
"Gérer l'authentification",
self._guard("Gérer l'authentification", self._manage_auth),
)
)
menu_items.append(pystray.Menu.SEPARATOR)
menu_items.append(self._menu_item("Quitter", self._guard("Quitter", self.stop)))
if self._menu_factory:
return self._menu_factory(_image, menu_items)
return pystray.Icon(
"ai-typewriter",
_image(),
"AI-Typewriter",
pystray.Menu(*menu_items),
)
def _profile_menu_items(self):
"""Génère dynamiquement les items du sous-menu des profils.
Appelé par pystray à chaque (re)construction du menu : la coche
(via ``checked=``, un item radio) suit donc toujours le profil actif.
"""
import pystray
for p in self.store.get_all():
yield pystray.MenuItem(
p.name,
self._guard(
f"sélection du profil « {p.name} »",
self._select_profile_action(p),
),
checked=self._make_checked(p.name),
radio=True,
)
def _make_checked(self, profile_name: str) -> Callable:
"""Retourne un prédicat évalué à l'affichage du menu."""
return lambda item: self.store.active_name == profile_name
def _menu_item(self, text: str, action, submenu=None):
import pystray
if submenu is not None:
return pystray.MenuItem(text, submenu)
if action is None:
action = lambda icon, item: None
return pystray.MenuItem(text, action)
def _guard(self, label: str, action: Callable) -> Callable:
"""Enveloppe une action de menu avec journalisation et capture d'erreurs.
pystray invoque les actions avec ``(icon, item)``. Un handler à 2
params est appelé tel quel, sans reshufflage. ``self.stop`` accepte
ces deux paramètres ; les autres actions sont des closures sans args.
"""
def handler(icon, item) -> None:
LOG.debug("Clic menu → %s", label)
try:
action()
except Exception:
LOG.exception("Erreur lors de l'action : %s", label)
return handler
def _select_profile_action(self, profile) -> Callable:
def action() -> None:
LOG.debug("Sélection du profil « %s » demandée", profile.name)
self._set_active(profile.name)
return action
+1
View File
@@ -0,0 +1 @@
"""Interface graphique (Tkinter) de l'application — fenêtre unifiée à onglets."""
+151
View File
@@ -0,0 +1,151 @@
"""Onglet « Authentification » : gestion des clés d'API des fournisseurs.
Permet d'enregistrer les clés d'API par nom de « référence » (celui que portent
les profils). Les secrets sont conservés de façon sécurisée via
``credentials.SecureStore`` ; ils ne sont jamais affichés ni écrits en clair.
"""
from __future__ import annotations
import logging
import tkinter as tk
from tkinter import messagebox, ttk
from ..credentials import CredentialError, SecureStore
from ..model_catalog import list_providers
LOG = logging.getLogger("ai_typewriter.ui.auth_tab")
class AuthTab(ttk.Frame):
def __init__(
self,
parent: ttk.Widget,
secure: SecureStore | None = None,
store=None,
) -> None:
super().__init__(parent)
self.secure = secure or SecureStore()
self.store = store
self._known = ["openai", "openrouter", "gemini", "custom", "ollama"]
body = ttk.Frame(self, padding=10)
body.pack(fill="both", expand=True)
ttk.Label(
body,
text=(
"Les clés sont enregistrées dans le gestionnaire d'identifiants, "
"pas dans un fichier de configuration.\n"
"Chaque profil référence une clé par son « nom de référence »."
),
foreground="#555",
justify="left",
).pack(fill="x", pady=(0, 8))
# -- formulaire ---------------------------------------------------------
form = ttk.LabelFrame(body, text="Nouvelle / mise à jour d'une référence", padding=8)
form.pack(fill="x")
ttk.Label(form, text="Nom de la référence (fournisseur)").grid(row=0, column=0, sticky="w")
self.name_var = tk.StringVar()
self.provider_combo = ttk.Combobox(
form,
textvariable=self.name_var,
values=[p["label"] for p in list_providers()],
width=28,
)
self.provider_combo.grid(row=0, column=1, sticky="we", pady=3)
ttk.Label(form, text="Clé API").grid(row=1, column=0, sticky="w")
self.key_var = tk.StringVar()
ttk.Entry(form, textvariable=self.key_var, width=32, show="*").grid(
row=1, column=1, sticky="we", pady=3
)
ttk.Button(form, text="Enregistrer", command=self._save).grid(
row=2, column=1, sticky="e", pady=(4, 0)
)
# -- liste des références existantes ------------------------------------
frm_list = ttk.LabelFrame(body, text="Références existantes", padding=8)
frm_list.pack(fill="both", expand=True, pady=(10, 0))
self.listbox = tk.Listbox(frm_list, height=5)
self.listbox.pack(fill="both", expand=True)
row = ttk.Frame(frm_list)
row.pack(fill="x", pady=(4, 0))
ttk.Button(row, text="Vérifier", command=self._has).pack(side="left")
ttk.Button(row, text="Supprimer", command=self._delete).pack(side="right")
self._refresh_list()
# -- helpers ---------------------------------------------------------------
def _refresh_list(self) -> None:
self.listbox.delete(0, tk.END)
known = sorted(
set(self._known) | {p.credential for p in self._known_profiles() if p.credential}
)
known = [k for k in known if k]
for k in known:
status = "●" if self.secure.has(k) else "○"
self.listbox.insert(tk.END, f"{status} {k}")
def _known_profiles(self) -> list:
if self.store is not None:
try:
return self.store.get_all()
except Exception:
return []
from ..config import ConfigStore
store = ConfigStore()
try:
store.load()
return store.get_all()
except Exception:
return []
def _save(self) -> None:
name = self.name_var.get().strip()
key = self.key_var.get().strip()
if not name or not key:
messagebox.showerror("Champs requis", "Nom de référence et clé sont requis.", parent=self)
return
if name not in self._known:
self._known.append(name)
try:
self.secure.store(name, key)
except CredentialError as exc:
messagebox.showerror("Enregistrement impossible", str(exc), parent=self)
return
self.key_var.set("")
self._refresh_list()
LOG.info("Référence « %s » enregistrée de façon sécurisée.", name)
def _has(self) -> None:
sel = self.listbox.curselection()
if not sel:
return
name = self._selected_name()
if self.secure.has(name):
messagebox.showinfo("Présente", f"Une clé est enregistrée pour « {name} ».", parent=self)
else:
messagebox.showinfo(
"Absente", f"Aucune clé enregistrée pour « {name} » pour l'instant.", parent=self
)
def _delete(self) -> None:
sel = self.listbox.curselection()
if not sel:
return
name = self._selected_name()
try:
self.secure.delete(name)
except CredentialError as exc:
messagebox.showerror("Suppression impossible", str(exc), parent=self)
return
self._refresh_list()
LOG.info("Référence « %s » supprimée.", name)
def _selected_name(self) -> str:
sel = self.listbox.curselection()
return self.listbox.get(sel[0]).split(" ", 1)[-1]
+68
View File
@@ -0,0 +1,68 @@
"""Onglet « Logs » : journaux en temps réel.
S'abonne au `LogBroadcaster` du module logging_utils et affiche les messages
au fur et à mesure qu'ils sont émis par l'application. Tourne dans le même
processus que le moteur, donc reçoit réellement les enregistrements (contraire
à l'ancienne fenêtre lancée dans un processus séparé).
"""
from __future__ import annotations
import logging
import tkinter as tk
from tkinter import ttk
from .. import logging_utils
class LogsTab(ttk.Frame):
def __init__(self, parent: ttk.Widget) -> None:
super().__init__(parent)
txt = tk.Text(self, state="disabled", wrap="word")
scroll = ttk.Scrollbar(self, command=txt.yview)
txt.configure(yscrollcommand=scroll.set)
scroll.pack(side="right", fill="y")
txt.pack(side="left", fill="both", expand=True)
self.txt = txt
bar = ttk.Frame(self)
bar.pack(fill="x", padx=6, pady=4)
ttk.Button(bar, text="Vider l'affichage", command=self._clear).pack(side="left")
self._history: list[str] = []
self._append_pending(logging_utils.broadcaster().drain())
self._schedule_poll()
def _schedule_poll(self) -> None:
try:
self.after(250, self._poll)
except tk.TclError:
pass
def _poll(self) -> None:
try:
self._append_pending(logging_utils.broadcaster().drain())
self._schedule_poll()
except tk.TclError:
pass
def _append_pending(self, records: list) -> None:
if not records:
return
self.txt.configure(state="normal")
for rec in records:
line = logging_utils.format_record(rec)
self._history.append(line)
self.txt.insert(tk.END, line + "\n")
if len(self._history) > 2000:
self.txt.delete("1.0", f"{len(self._history) - 2000}.0")
del self._history[: len(self._history) - 2000]
self.txt.configure(state="disabled")
self.txt.see(tk.END)
def _clear(self) -> None:
self.txt.configure(state="normal")
self.txt.delete("1.0", tk.END)
self.txt.configure(state="disabled")
self._history.clear()
+94
View File
@@ -0,0 +1,94 @@
"""Fenêtre principale unifiée à onglets.
Une seule racine ``tk.Tk`` regroupe trois onglets :
* **Profils** — liste + éditeur de profils ;
* **Authentification** — gestion des clés d'API ;
* **Logs** — journaux en temps réel.
Fermer la fenêtre (bouton ✕) ne quitte **pas** l'application : elle est
masquée (``withdraw``) et continue de tourner ; l'icône de la zone de
notification permet de la réafficher dans le bon onglet.
"""
from __future__ import annotations
import tkinter as tk
from tkinter import ttk
from ..config import ConfigStore
from ..credentials import SecureStore
from .auth_tab import AuthTab
from .logs_tab import LogsTab
from .profiles_tab import ProfilesTab
TAB_PROFILES = "profiles"
TAB_AUTH = "auth"
TAB_LOGS = "logs"
_TAB_INDEX = {TAB_PROFILES: 0, TAB_AUTH: 1, TAB_LOGS: 2}
class MainWindow(tk.Tk):
def __init__(self, store: ConfigStore, secure: SecureStore | None = None) -> None:
super().__init__()
self.store = store
self.secure = secure or SecureStore()
self.geometry("860x660")
self.minsize(720, 520)
self.protocol("WM_DELETE_WINDOW", self.hide_to_tray)
self.notebook = ttk.Notebook(self)
self.notebook.pack(fill="both", expand=True)
self.profiles_tab = ProfilesTab(
self.notebook, store, self.secure, on_change=self._refresh_title
)
self.auth_tab = AuthTab(self.notebook, secure=self.secure, store=store)
self.logs_tab = LogsTab(self.notebook)
self.notebook.add(self.profiles_tab, text="Profils")
self.notebook.add(self.auth_tab, text="Authentification")
self.notebook.add(self.logs_tab, text="Logs")
self._refresh_title()
# -- cycle de vie -----------------------------------------------------------
def hide_to_tray(self) -> None:
"""Masque la fenêtre (l'application continue de tourner en fond)."""
self.withdraw()
def show_tab(self, name: str) -> None:
"""Réaffiche la fenêtre et sélectionne l'onglet ``name``."""
self.deiconify()
idx = _TAB_INDEX[name]
self.notebook.select(idx)
self._present()
def open_new_profile(self) -> None:
"""Prépare l'onglet Profils pour créer un nouveau profil."""
self.profiles_tab.new_profile()
def _present(self) -> None:
self.lift()
try:
self.attributes("-topmost", True)
self.after(200, lambda: self.attributes("-topmost", False))
except tk.TclError:
pass
try:
self.focus_force()
except tk.TclError:
pass
# -- synchronisation --------------------------------------------------------
def refresh_all(self) -> None:
"""Met à jour la liste des profils et le titre après un changement externe."""
self.profiles_tab.refresh_list()
self._refresh_title()
def _refresh_title(self) -> None:
self.title(f"AI-Typewriter — {self.store.active_name}")
+167
View File
@@ -0,0 +1,167 @@
"""Sélecteur de modèles pour la création/édition de profil.
Affiche automatiquement la liste des modèles disponibles localement (Ollama)
et, si connecté à Internet, propose une recherche dans le catalogue public
d'Ollama ainsi que le téléchargement (pull) des modèles correspondants.
"""
from __future__ import annotations
import logging
import queue
import threading
import tkinter as tk
from tkinter import ttk
from .. import model_catalog
LOG = logging.getLogger("ai_typewriter.ui.model_picker")
class ModelPicker(tk.Toplevel):
"""Fenêtre modale : choisit un modèle local ou en recherche un en ligne.
Résultat : `self.result` vaut le nom du modèle choisi, ou None si annulé.
"""
def __init__(
self,
parent: tk.Widget,
server_url: str = "http://localhost:11434",
initial: str = "",
) -> None:
super().__init__(parent)
self.title("Modèle IA")
self.result: str | None = None
self.server_url = server_url
self._task_queue: queue.Queue = queue.Queue()
self.geometry("520x420")
self.transient(parent)
self.grab_set()
root = ttk.Frame(self, padding=10)
root.pack(fill="both", expand=True)
# -- modèles locaux ----------------------------------------------------
frm_local = ttk.LabelFrame(root, text="Modèles disponibles sur cette machine", padding=6)
frm_local.pack(fill="x")
self.local_list = tk.Listbox(frm_local, height=6)
self.local_list.pack(fill="x")
self.local_list.bind("<Double-Button-1>", lambda e: self._pick_local())
ttk.Button(frm_local, text="Utiliser ce modèle local", command=self._pick_local).pack(pady=(4, 0))
# -- recherche en ligne ------------------------------------------------
frm_online = ttk.LabelFrame(root, text="Rechercher dans la bibliothèque Ollama (Internet)", padding=6)
frm_online.pack(fill="both", expand=True, pady=(8, 0))
row = ttk.Frame(frm_online)
row.pack(fill="x")
self.search_var = tk.StringVar(value=initial)
ttk.Entry(row, textvariable=self.search_var).pack(side="left", fill="x", expand=True)
self.search_btn = ttk.Button(row, text="Rechercher", command=self._search_online)
self.search_btn.pack(side="left", padx=(4, 0))
self.online_list = tk.Listbox(frm_online, height=6)
self.online_list.pack(fill="both", expand=True, pady=(4, 0))
r2 = ttk.Frame(frm_online)
r2.pack(fill="x", pady=(4, 0))
ttk.Button(r2, text="Télécharger puis utiliser", command=self._pull_and_pick).pack(side="left")
self.status = ttk.Label(frm_online, text="", foreground="#555")
self.status.pack(side="left", padx=8)
btns = ttk.Frame(root)
btns.pack(fill="x", pady=(8, 0))
ttk.Button(btns, text="Annuler", command=self.destroy).pack(side="right")
ttk.Button(btns, text="OK", command=self._ok).pack(side="right", padx=4)
self._load_local()
self._refresh_online()
self.after(100, self._poll_tasks)
# -- premiers chargements -------------------------------------------------
def _load_local(self) -> None:
self._run_task("local", lambda: model_catalog.list_local_models(self.server_url))
def _refresh_online(self) -> None:
self.status.config(text="…")
self._run_task("online", lambda: model_catalog.search_online_models(self.search_var.get()))
def _run_task(self, kind: str, fn) -> None:
def worker() -> None:
try:
self._task_queue.put((kind, fn()))
except Exception as exc:
self._task_queue.put((kind, None))
threading.Thread(target=worker, daemon=True).start()
def _poll_tasks(self) -> None:
try:
while True:
kind, value = self._task_queue.get_nowait()
if kind == "local":
self._render_local(value or [])
elif kind == "online":
self._render_online(value or [])
else:
LOG.warning("Tâche inconnue : %s", kind)
except queue.Empty:
pass
self.after(100, self._poll_tasks)
def _render_local(self, names: list[str]) -> None:
self.local_list.delete(0, tk.END)
for n in names:
self.local_list.insert(tk.END, n)
if not names:
self.local_list.insert(tk.END, "(aucun modèle local détecté)")
if not names and not self.online_list.size():
self.status.config(text="Aucun modèle local ; cherchez en ligne.")
def _render_online(self, models: list) -> None:
self.online_list.delete(0, tk.END)
for m in models:
self.online_list.insert(tk.END, m.display())
self.status.config(text=f"{len(models)} modèle(s) trouvé(s)" if models else "Aucun résultat")
# -- actions ---------------------------------------------------------------
def _pick_local(self) -> None:
sel = self.local_list.curselection()
if not sel:
return
self.result = self.local_list.get(sel[0])
if self._is_placeholder(self.result):
self.result = None
return
self.destroy()
def _search_online(self) -> None:
self._refresh_online()
def _pull_and_pick(self) -> None:
sel = self.online_list.curselection()
if not sel:
return
self.result = self.online_list.get(sel[0])
if self._is_placeholder(self.result):
self.result = None
return
self.status.config(text=f"Téléchargement de {self.result}…")
model = self.result
threading.Thread(
target=lambda: model_catalog.pull_model(model, server_url=self.server_url),
daemon=True,
).start()
self.destroy()
def _ok(self) -> None:
# Si rien n'est sélectionné, on accepte le texte saisi (modèle libre).
val = self.search_var.get().strip()
if val:
self.result = val
self.destroy()
def _is_placeholder(self, value: str) -> bool:
return value.startswith("(") or value.startswith("(aucun")
+326
View File
@@ -0,0 +1,326 @@
"""Onglet « Profils » : liste des profils + éditeur de profil intégré.
Remplace l'ancienne fenêtre modale ``ProfileDialog``. La liste (à gauche)
affiche le profil actif avec un marqueur « ● » ; l'éditeur (à droite) permet
de créer ou modifier un profil. Les actions principales (Activer, Nouveau,
Supprimer, Enregistrer) sont directement accessibles dans l'onglet.
"""
from __future__ import annotations
import logging
import tkinter as tk
from tkinter import messagebox, ttk
from ..config import ConfigStore, Profile, math_latex_profile
from ..credentials import SecureStore
from ..model_catalog import list_providers
from .model_picker import ModelPicker
LOG = logging.getLogger("ai_typewriter.ui.profiles_tab")
class ProfilesTab(ttk.Frame):
"""Liste des profils + formulaire d'édition, le tout dans un onglet."""
def __init__(
self,
parent: ttk.Widget,
store: ConfigStore,
secure: SecureStore | None = None,
on_change=None,
) -> None:
super().__init__(parent)
self.store = store
self.secure = secure or SecureStore()
self.on_change = on_change
self._names: list[str] = []
self._building = False
self._build()
self.refresh_list()
# -- construction ----------------------------------------------------------
def _build(self) -> None:
left = ttk.Frame(self, padding=(10, 10, 4, 10))
left.pack(side="left", fill="y")
ttk.Label(left, text="Profils", font=("TkDefaultFont", 10, "bold")).pack(
anchor="w"
)
self.listbox = tk.Listbox(left, width=28, height=16, exportselection=False)
self.listbox.pack(fill="both", expand=True, pady=(6, 0))
self.listbox.bind("<<ListboxSelect>>", self._on_select)
rows = ttk.Frame(left)
rows.pack(fill="x", pady=(6, 0))
ttk.Button(rows, text="Activer", command=self._activate).pack(
side="left", fill="x", expand=True
)
ttk.Button(rows, text="Nouveau", command=self.new_profile).pack(
side="left", fill="x", expand=True, padx=2
)
ttk.Button(rows, text="Supprimer", command=self._delete).pack(
side="left", fill="x", expand=True
)
f = ttk.Frame(self, padding=(4, 10, 10, 10))
f.pack(side="left", fill="both", expand=True)
ttk.Label(f, text="Nom *").grid(row=0, column=0, sticky="w", padx=(0, 6), pady=4)
self.name_var = tk.StringVar()
ttk.Entry(f, textvariable=self.name_var, width=38).grid(
row=0, column=1, columnspan=2, sticky="we", pady=4
)
ttk.Label(f, text="Fournisseur *").grid(row=1, column=0, sticky="w", padx=(0, 6), pady=4)
self.provider_var = tk.StringVar(value=self._label_for_id("ollama"))
self.provider_combo = ttk.Combobox(
f,
textvariable=self.provider_var,
values=[p["label"] for p in list_providers()],
state="readonly",
width=34,
)
self.provider_combo.grid(row=1, column=1, columnspan=2, sticky="we", pady=4)
self.provider_combo.bind("<<ComboboxSelected>>", lambda e: self._provider_changed())
ttk.Label(f, text="Modèle *").grid(row=2, column=0, sticky="w", padx=(0, 6), pady=4)
self.model_var = tk.StringVar(value="llama3.1")
ttk.Entry(f, textvariable=self.model_var, width=28).grid(
row=2, column=1, sticky="we", pady=4
)
ttk.Button(f, text="Choisir / télécharger…", command=self._open_picker).grid(
row=2, column=2, sticky="e", padx=(4, 0)
)
ttk.Label(f, text="URL du serveur").grid(row=3, column=0, sticky="w", padx=(0, 6), pady=4)
self.server_url_var = tk.StringVar(value="http://localhost:11434")
ttk.Entry(f, textvariable=self.server_url_var, width=38).grid(
row=3, column=1, columnspan=2, sticky="we", pady=4
)
ttk.Label(f, text="Référence (clé API)").grid(row=4, column=0, sticky="w", padx=(0, 6), pady=4)
self.credential_var = tk.StringVar()
ttk.Entry(f, textvariable=self.credential_var, width=38).grid(
row=4, column=1, columnspan=2, sticky="we", pady=4
)
ttk.Label(f, text="Prompt système").grid(row=5, column=0, sticky="nw", padx=(0, 6), pady=4)
self.prompt_text = tk.Text(f, width=48, height=7, wrap="word")
self.prompt_text.grid(row=5, column=1, columnspan=2, sticky="we", pady=4)
fr_eq = ttk.LabelFrame(f, text="Équations LaTeX", padding=6)
fr_eq.grid(row=6, column=0, columnspan=3, sticky="we", pady=6)
self.equation_var = tk.BooleanVar(value=False)
ttk.Checkbutton(
fr_eq,
text="Intercepter les marqueurs et déclencher les touches (ex. Alt+= / →)",
variable=self.equation_var,
).grid(row=0, column=0, columnspan=4, sticky="w")
self.eq_start_var = tk.StringVar(value="[EQ]")
self.eq_end_var = tk.StringVar(value="[/EQ]")
ttk.Label(fr_eq, text="Début:").grid(row=1, column=0, sticky="e")
ttk.Entry(fr_eq, textvariable=self.eq_start_var, width=14).grid(
row=1, column=1, sticky="w"
)
ttk.Label(fr_eq, text="Fin:").grid(row=1, column=2, sticky="e", padx=(8, 0))
ttk.Entry(fr_eq, textvariable=self.eq_end_var, width=14).grid(
row=1, column=3, sticky="w"
)
ttk.Button(fr_eq, text="Préremplir (profil math)", command=self._prefill_math).grid(
row=2, column=0, columnspan=4, sticky="w", pady=(4, 0)
)
btns = ttk.Frame(f)
btns.grid(row=7, column=0, columnspan=3, sticky="e", pady=(8, 0))
ttk.Button(btns, text="Effacer", command=self.new_profile).pack(side="right")
ttk.Button(btns, text="Enregistrer", command=self._save).pack(
side="right", padx=4
)
# -- liste des profils ------------------------------------------------------
def refresh_list(self, select: str | None = None) -> None:
"""Reconstruit la liste et charge le profil ``select`` (ou l'actif)."""
active = self.store.active_name
self._names = [p.name for p in self.store.get_all()]
self.listbox.delete(0, tk.END)
target = select if select in self._names else active
idx = self._names.index(target) if target in self._names else 0
for i, name in enumerate(self._names):
marker = "● " if name == active else " "
self.listbox.insert(tk.END, f"{marker}{name}")
self._building = True
try:
self.listbox.selection_clear(0, tk.END)
self.listbox.selection_set(idx)
self.listbox.see(idx)
finally:
self._building = False
if target in self._names:
self._load(target)
def _selected_name(self) -> str | None:
sel = self.listbox.curselection()
if not sel:
return None
return self._names[sel[0]]
def _on_select(self, event=None) -> None:
if self._building:
return
name = self._selected_name()
if name:
self._load(name)
def _load(self, name: str) -> None:
try:
p = self.store.get(name)
except KeyError:
return
self._building = True
try:
self.name_var.set(p.name)
self.provider_var.set(self._label_for_id(p.provider))
self.model_var.set(p.model)
self.server_url_var.set(p.server_url or "http://localhost:11434")
self.credential_var.set(p.credential or "")
self.equation_var.set(p.equation_enabled)
self.eq_start_var.set(p.eq_start_marker)
self.eq_end_var.set(p.eq_end_marker)
self._set_prompt(p.system_prompt)
finally:
self._building = False
def new_profile(self) -> None:
"""Vide le formulaire pour créer un nouveau profil."""
self._building = True
try:
self.listbox.selection_clear(0, tk.END)
finally:
self._building = False
self.name_var.set("")
self.provider_var.set(self._label_for_id("ollama"))
self.model_var.set("llama3.1")
self.server_url_var.set("http://localhost:11434")
self.credential_var.set("")
self.equation_var.set(False)
self.eq_start_var.set("[EQ]")
self.eq_end_var.set("[/EQ]")
self._set_prompt("")
# -- actions ---------------------------------------------------------------
def _activate(self) -> None:
name = self._selected_name()
if not name:
return
try:
self.store.set_active(name)
LOG.info("Profil actif : %s", name)
self.refresh_list(select=name)
self._notify_change()
except Exception as exc:
LOG.exception("Impossible d'activer le profil : %s", exc)
messagebox.showerror("Activation impossible", str(exc), parent=self)
def _delete(self) -> None:
name = self._selected_name()
if not name:
return
if not messagebox.askyesno(
"Supprimer", f"Supprimer le profil « {name} » ?", parent=self
):
return
try:
self.store.remove(name)
LOG.info("Profil « %s » supprimé.", name)
self.refresh_list()
self._notify_change()
except Exception as exc:
LOG.exception("Suppression impossible : %s", exc)
messagebox.showerror("Suppression impossible", str(exc), parent=self)
def _save(self) -> None:
name = self.name_var.get().strip()
if not name:
messagebox.showerror("Nom requis", "Le profil doit avoir un nom.", parent=self)
return
model = self.model_var.get().strip() or "llama3.1"
profile = Profile(
name=name,
provider=self._provider_id(),
model=model,
server_url=self.server_url_var.get().strip(),
request_timeout_seconds=300.0,
type_delay_seconds=0.0,
system_prompt=self.prompt_text.get("1.0", tk.END).strip(),
equation_enabled=self.equation_var.get(),
eq_start_marker=self.eq_start_var.get().strip(),
eq_end_marker=self.eq_end_var.get().strip(),
credential=self.credential_var.get().strip() or "",
)
needs_key = self._provider_id() != "ollama"
if needs_key and profile.credential and not self.secure.has(profile.credential):
LOG.info(
"Profil %s : référence %s non encore enregistrée.",
name,
profile.credential,
)
try:
self.store.upsert(profile)
LOG.info("Profil « %s » enregistré.", name)
self.refresh_list(select=name)
self._notify_change()
except Exception as exc:
LOG.exception("Enregistrement impossible : %s", exc)
messagebox.showerror("Enregistrement impossible", str(exc), parent=self)
# -- helpers ---------------------------------------------------------------
def _notify_change(self) -> None:
if self.on_change:
self.on_change()
def _set_prompt(self, value: str) -> None:
self.prompt_text.delete("1.0", tk.END)
self.prompt_text.insert("1.0", value)
def _provider_id(self) -> str:
for p in list_providers():
if p["label"] == self.provider_var.get():
return p["id"]
return "ollama"
def _label_for_id(self, provider_id: str) -> str:
for p in list_providers():
if p["id"] == provider_id:
return p["label"]
return provider_id
def _provider_changed(self) -> None:
for p in list_providers():
if p["label"] == self.provider_var.get():
base = p.get("base_url") or ""
if base and not self.server_url_var.get().strip():
self.server_url_var.set(base)
break
def _open_picker(self) -> None:
picker = ModelPicker(
self.winfo_toplevel(),
server_url=self.server_url_var.get(),
initial=self.model_var.get(),
)
self.wait_window(picker)
if picker.result:
self.model_var.set(picker.result)
def _prefill_math(self) -> None:
m = math_latex_profile(name=self.name_var.get().strip() or "Mathématiques (LaTeX)")
self.equation_var.set(True)
self.eq_start_var.set(m.eq_start_marker)
self.eq_end_var.set(m.eq_end_marker)
self._set_prompt(m.system_prompt)
self.provider_var.set(self._label_for_id("ollama"))
self.model_var.set(m.model)
+92 -8
View File
@@ -2,27 +2,36 @@ import pytest
import requests
from ai_typewriter.ai_client import AIClientError, ask_ai
from ai_typewriter.config import AppConfig
from ai_typewriter.config import Profile
class FakeResponse:
def __init__(self, payload):
self.payload = payload
def raise_for_status(self):
return None
def json(self):
return self.payload
# ---------------------------------------------------------------------------
# Ollama
# ---------------------------------------------------------------------------
def test_ollama_payload_contains_strict_system_prompt(monkeypatch):
seen = {}
def fake_post(url, json, timeout, **kwargs):
seen["url"] = url
seen["json"] = json
return FakeResponse({"message": {"content": "ok"}})
monkeypatch.setattr("ai_typewriter.ai_client.requests.post", fake_post)
result = ask_ai("texte", AppConfig(provider="ollama", model="m"))
result = ask_ai("texte", Profile(provider="ollama", model="m"))
assert result == "ok"
assert seen["url"].endswith("/api/chat")
@@ -30,31 +39,106 @@ def test_ollama_payload_contains_strict_system_prompt(monkeypatch):
assert "Uniquement la réponse brute" in seen["json"]["messages"][0]["content"]
def test_ollama_uses_profile_system_prompt(monkeypatch):
seen = {}
def fake_post(url, json, timeout, **kwargs):
seen["json"] = json
return FakeResponse({"message": {"content": "ok"}})
monkeypatch.setattr("ai_typewriter.ai_client.requests.post", fake_post)
profile = Profile(provider="ollama", system_prompt="Prompt personnalisé")
ask_ai("x", profile)
assert seen["json"]["messages"][0]["content"] == "Prompt personnalisé"
def test_ollama_timeout_message_suggests_config_change(monkeypatch):
def fake_post(url, json, timeout, **kwargs):
raise requests.Timeout("too slow")
monkeypatch.setattr("ai_typewriter.ai_client.requests.post", fake_post)
with pytest.raises(AIClientError) as exc:
ask_ai("texte", AppConfig(provider="ollama", model="m", request_timeout_seconds=300))
ask_ai("texte", Profile(provider="ollama", model="m", request_timeout_seconds=300))
assert "300 s" in str(exc.value)
assert "request_timeout_seconds" in str(exc.value)
assert "0 pour désactiver" in str(exc.value)
# ---------------------------------------------------------------------------
# Gemini
# ---------------------------------------------------------------------------
def test_gemini_payload(monkeypatch):
seen = {}
def fake_post(url, params, json, timeout):
def fake_resolver(profile, store):
return "cle_secrete"
def fake_post(url, params, json, timeout, **kwargs):
seen["url"] = url
seen["params"] = params
seen["json"] = json
return FakeResponse({"candidates": [{"content": {"parts": [{"text": "brut"}]}}]})
monkeypatch.setattr("ai_typewriter.ai_client.requests.post", fake_post)
result = ask_ai("texte", AppConfig(provider="gemini", model="gemini-1.5-flash", api_key="k"))
monkeypatch.setattr("ai_typewriter.ai_client.requests.post", fake_post)
profile = Profile(provider="gemini", model="gemini-1.5-flash", credential="gem")
result = ask_ai("texte", profile, resolve_key=fake_resolver)
assert result == "brut"
assert seen["params"] == {"key": "k"}
assert seen["params"] == {"key": "cle_secrete"}
assert seen["url"].endswith("/v1beta/models/gemini-1.5-flash:generateContent")
assert "systemInstruction" in seen["json"]
# ---------------------------------------------------------------------------
# OpenAI-compatible
# ---------------------------------------------------------------------------
def test_openai_payload_with_bearer_key(monkeypatch):
seen = {}
def fake_resolver(profile, store):
return "sk-test"
def fake_post(url, json, headers, timeout, **kwargs):
seen["url"] = url
seen["headers"] = headers
return FakeResponse({"choices": [{"message": {"content": "gpt-reponse"}}]})
monkeypatch.setattr("ai_typewriter.ai_client.requests.post", fake_post)
profile = Profile(
provider="openai",
model="gpt-4o-mini",
credential="openai",
server_url="https://api.openai.com/v1",
)
result = ask_ai("texte", profile, resolve_key=fake_resolver)
assert result == "gpt-reponse"
assert seen["url"].endswith("/chat/completions")
assert seen["headers"]["Authorization"] == "Bearer sk-test"
def test_missing_key_raises_actionable_error(monkeypatch):
profile = Profile(provider="openai", model="gpt", credential="openai")
with pytest.raises(AIClientError) as exc:
ask_ai("texte", profile, resolve_key=lambda p, s: "")
assert "Aucune clé" in str(exc.value)
assert "authentification" in str(exc.value)
def test_unsupported_provider(monkeypatch):
with pytest.raises(AIClientError):
ask_ai("x", Profile(provider="inconnu", model="m"))
def test_empty_prompt_rejected():
with pytest.raises(AIClientError):
ask_ai(" ", Profile(provider="ollama"))
+67 -38
View File
@@ -1,51 +1,80 @@
import json
import pytest
from ai_typewriter.config import load_config
from ai_typewriter.config import (
ConfigError,
ConfigStore,
Profile,
load_config,
math_latex_profile,
)
def test_load_default_config(tmp_path):
def test_store_creates_defaults_when_missing(tmp_path):
path = tmp_path / "nested" / "config.json"
store = ConfigStore(path)
store.ensure_defaults()
assert path.exists()
assert len(store.profiles) == 2
names = [p.name for p in store.profiles]
assert "Général" in names
assert math_latex_profile().name in names
store.load()
assert store.active_name in names
def test_default_math_profile_enables_equation_markers():
p = math_latex_profile()
assert p.equation_enabled is True
assert p.eq_start_marker == "[EQ]"
assert p.eq_end_marker == "[/EQ]"
assert p.eq_start_key == "alt+="
assert p.eq_end_key == "right"
assert "[EQ]" in p.effective_prompt()
def test_crud_upsert_set_active_and_remove(tmp_path):
path = tmp_path / "config.json"
path.write_text(json.dumps({"provider": "ollama"}), encoding="utf-8")
store = ConfigStore(path)
store.ensure_defaults()
store.load()
cfg = load_config(path)
names_before = len(store.get_all())
prof = math_latex_profile(name="MaesProfil")
store.upsert(prof)
store.set_active("MaesProfil")
assert store.active().name == "MaesProfil"
assert len(store.get_all()) == names_before + 1
assert cfg.provider == "ollama"
assert cfg.hotkey == "ctrl+alt+a"
assert cfg.math_text_format == "plain"
assert "Réponds directement" in cfg.system_prompt
prof2 = Profile(name="MaesProfil", provider="openai", model="gpt-4o-mini")
store.upsert(prof2)
assert store.get("MaesProfil").provider == "openai"
store.remove("MaesProfil")
with pytest.raises(KeyError):
store.get("MaesProfil")
def test_accept_word_equation_format(tmp_path):
path = tmp_path / "config.json"
path.write_text(json.dumps({"provider": "ollama", "math_text_format": "word_equation"}), encoding="utf-8")
cfg = load_config(path)
assert cfg.math_text_format == "word_equation"
def test_remove_last_profile_blocked(tmp_path):
store = ConfigStore(tmp_path / "config.json")
store.ensure_defaults()
store.load()
# Il y a 2 profils par défaut : le premier retrait réussit…
store.remove(store.get_all()[0].name)
assert len(store.get_all()) == 1
# …mais retirer le dernier est interdit.
with pytest.raises(ConfigError):
store.remove(store.get_all()[0].name)
def test_timeout_zero_disables_timeout(tmp_path):
path = tmp_path / "config.json"
path.write_text(json.dumps({"provider": "ollama", "request_timeout_seconds": 0}), encoding="utf-8")
cfg = load_config(path)
assert cfg.request_timeout_seconds is None
def test_set_active_unknown_raises(tmp_path):
store = ConfigStore(tmp_path / "config.json")
store.ensure_defaults()
store.load()
with pytest.raises(KeyError):
store.set_active("inexistant")
def test_reject_invalid_math_text_format(tmp_path):
path = tmp_path / "config.json"
path.write_text(json.dumps({"provider": "ollama", "math_text_format": "bad"}), encoding="utf-8")
with pytest.raises(ValueError):
load_config(path)
def test_reject_invalid_provider(tmp_path):
path = tmp_path / "config.json"
path.write_text(json.dumps({"provider": "bad"}), encoding="utf-8")
with pytest.raises(ValueError):
load_config(path)
def test_load_config_return_store(tmp_path):
store = load_config(str(tmp_path / "config.json"))
assert isinstance(store, ConfigStore)
+97
View File
@@ -0,0 +1,97 @@
"""Tests du stockage sécurisé (keyring mocké pour éviter tout accès au
gestionnaire d'identifiants réel de la machine)."""
import pytest
from ai_typewriter.credentials import CredentialError, SecureStore, get_cred
class FakeKeyring:
"""Mini stub de keyring en mémoire, exposant la même interface."""
_data = {}
@classmethod
def reset(cls):
cls._data = {}
@classmethod
def set_password(cls, service, username, password):
cls._data[(service, username)] = password
@classmethod
def get_password(cls, service, username):
return cls._data.get((service, username))
@classmethod
def delete_password(cls, service, username):
cls._data.pop((service, username), None)
class FakeKeyringErrors:
class PasswordDeleteError(RuntimeError):
pass
def test_store_and_get(monkeypatch):
import ai_typewriter.credentials as cred
monkeypatch.setattr(cred, "keyring", FakeKeyring)
monkeypatch.setattr(cred.keyring, "errors", FakeKeyringErrors, raising=False)
FakeKeyring.reset()
store = SecureStore("test-service")
store.store("openai", "sk-secret")
assert store.get("openai") == "sk-secret"
assert store.has("openai") is True
def test_empty_credential_returns_empty(monkeypatch):
import ai_typewriter.credentials as cred
monkeypatch.setattr(cred, "keyring", FakeKeyring)
FakeKeyring.reset()
store = SecureStore()
assert store.get("") == ""
assert store.has("") is False
def test_get_cred_helper_creates_store(monkeypatch):
import ai_typewriter.credentials as cred
monkeypatch.setattr(cred, "keyring", FakeKeyring)
monkeypatch.setattr(cred.keyring, "errors", FakeKeyringErrors, raising=False)
FakeKeyring.reset()
# via un store passé explicitement
store = SecureStore("t")
store.store("k", "v")
assert get_cred(store, "k") == "v"
# via None (crée un store par défaut, mais retombe sur service réel) —
# on vérifie que quelques accesseurs ne plantent pas.
assert get_cred(None, "") == ""
def test_store_invalid_credential_name(monkeypatch):
import ai_typewriter.credentials as cred
monkeypatch.setattr(cred, "keyring", FakeKeyring)
FakeKeyring.reset()
store = SecureStore()
with pytest.raises(CredentialError):
store.store("", "secret")
def test_delete(monkeypatch):
import ai_typewriter.credentials as cred
monkeypatch.setattr(cred, "keyring", FakeKeyring)
monkeypatch.setattr(cred.keyring, "errors", FakeKeyringErrors, raising=False)
FakeKeyring.reset()
store = SecureStore("t")
store.store("gemini", "cle")
store.delete("gemini")
assert store.has("gemini") is False
+106
View File
@@ -0,0 +1,106 @@
"""Tests du moteur : capture -> IA -> dactylographie (tout mocké)."""
from types import SimpleNamespace
from ai_typewriter.config import ConfigStore
from ai_typewriter.engine import AITypewriterEngine
def _make_store(tmp_path) -> ConfigStore:
store = ConfigStore(tmp_path / "config.json")
store.ensure_defaults()
store.load()
return store
def _fake_keyboard(monkeypatch, typed, sent):
hook = {}
monkeypatch.setattr(
"ai_typewriter.key_stepper.keyboard.write",
lambda char, delay=0, exact=True: typed.append(char),
)
monkeypatch.setattr("ai_typewriter.key_stepper.keyboard.send", lambda v: sent.append(v))
monkeypatch.setattr(
"ai_typewriter.key_stepper.keyboard.hook",
lambda callback, suppress=True: hook.update({"callback": callback}) or "hook",
)
monkeypatch.setattr("ai_typewriter.key_stepper.keyboard.unhook", lambda v: None)
return hook
def test_capture_ask_and_step_types_response(monkeypatch, tmp_path):
store = _make_store(tmp_path)
engine = AITypewriterEngine(store)
typed, sent = [], []
hook = _fake_keyboard(monkeypatch, typed, sent)
monkeypatch.setattr(
"ai_typewriter.engine.capture_clipboard", lambda: "Question de test"
)
monkeypatch.setattr(
"ai_typewriter.engine.ask_ai", lambda prompt, profile, store: "Réponse IA"
)
engine.capture_ask_and_step()
assert engine._stepper is not None
hook["callback"](SimpleNamespace(event_type="down"))
hook["callback"](SimpleNamespace(event_type="down"))
assert typed == ["R", "é"]
def test_capture_ask_and_step_with_equations(monkeypatch, tmp_path):
store = _make_store(tmp_path)
store.set_active("Mathématiques (LaTeX)")
engine = AITypewriterEngine(store)
typed, sent = [], []
hook = _fake_keyboard(monkeypatch, typed, sent)
monkeypatch.setattr(
"ai_typewriter.engine.capture_clipboard", lambda: "Calcule"
)
monkeypatch.setattr(
"ai_typewriter.engine.ask_ai", lambda prompt, profile, store: r"[EQ]a=x[/EQ]"
)
engine.capture_ask_and_step()
# 1:[EQ]->alt+= ; a: char ; =: char ; x: char ; [/EQ]->right => 5 actions
for _ in range(5):
hook["callback"](SimpleNamespace(event_type="down"))
assert sent == ["alt+=", "right"]
assert typed == ["a", "=", "x"]
def test_empty_clipboard_does_not_ask(monkeypatch, tmp_path):
store = _make_store(tmp_path)
engine = AITypewriterEngine(store)
called = []
monkeypatch.setattr("ai_typewriter.engine.capture_clipboard", lambda: " ")
monkeypatch.setattr(
"ai_typewriter.engine.ask_ai",
lambda prompt, profile, store: called.append(prompt) or "x",
)
engine.capture_ask_and_step()
assert called == []
def test_ask_only_returns_answer(monkeypatch, tmp_path):
store = _make_store(tmp_path)
engine = AITypewriterEngine(store)
monkeypatch.setattr(
"ai_typewriter.engine.ask_ai",
lambda prompt, profile, store: "reponse-brute",
)
assert engine.ask_only("bonjour") == "reponse-brute"
def test_concurrent_hotkey_ignored_while_busy(monkeypatch, tmp_path):
store = _make_store(tmp_path)
engine = AITypewriterEngine(store)
# Occupe le verrou
assert engine._busy.acquire(blocking=False) is True
calls = []
engine.handle_hotkey() # doit être ignoré (busy)
assert calls == []
engine._busy.release()
+130 -42
View File
@@ -1,48 +1,108 @@
from types import SimpleNamespace
from ai_typewriter.key_stepper import KeyStepper
from ai_typewriter.key_stepper import Action, KeyStepper, build_actions
# ---------------------------------------------------------------------------
# build_actions : découpage en actions (sans clavier)
# ---------------------------------------------------------------------------
def test_plain_text_actions_single_char_per_action():
actions = build_actions("abc")
assert actions == [
Action("char", "a"),
Action("char", "b"),
Action("char", "c"),
]
def test_equation_markers_replaced_by_key_sequences():
actions = build_actions(
r"Les racines sont [EQ]z_1 = x[/EQ].",
equation_enabled=True,
)
assert Action("seq", "alt+=") in actions
assert Action("seq", "right") in actions
# aucun marqueur littéral ne doit être tapé
assert Action("char", "[") not in actions
kinds = [a.kind for a in actions]
assert kinds.count("seq") == 2
# le contenu LaTeX est conservé caractère par caractère
chars = "".join(a.value for a in actions if a.kind == "char")
assert "z_1 = x" in chars
def test_multiple_equations_each_get_start_and_end():
text = r"[EQ]a[/EQ] et [EQ]b[/EQ]"
actions = build_actions(text, equation_enabled=True)
seqs = [a.value for a in actions if a.kind == "seq"]
assert seqs == ["alt+=", "right", "alt+=", "right"]
def test_equation_disabled_keeps_markers_literal():
text = r"[EQ]\frac{a}{b}[/EQ]"
assert build_actions(text, equation_enabled=False) == [
Action("char", c) for c in text
]
def test_custom_markers_and_keys():
actions = build_actions(
"<<START>>x^2<<END>>",
equation_enabled=True,
eq_start_marker="<<START>>",
eq_end_marker="<<END>>",
eq_start_key="ctrl+shift+e",
eq_end_key="space",
)
assert Action("seq", "ctrl+shift+e") in actions
assert Action("seq", "space") in actions
# marqueurs retirés, non tapés
chars = "".join(a.value for a in actions if a.kind == "char")
assert chars == "x^2"
assert "START" not in chars
# ---------------------------------------------------------------------------
# KeyStepper (monkeypatch du hook clavier)
# ---------------------------------------------------------------------------
def _install_hook(monkeypatch, typed, sent, unhooked):
hook = {}
monkeypatch.setattr(
"ai_typewriter.key_stepper.keyboard.write",
lambda char, delay=0, exact=True: typed.append(char),
)
monkeypatch.setattr(
"ai_typewriter.key_stepper.keyboard.send",
lambda value: sent.append(value),
)
monkeypatch.setattr(
"ai_typewriter.key_stepper.keyboard.hook",
lambda callback, suppress=True: hook.update({"callback": callback}) or "hook",
)
monkeypatch.setattr(
"ai_typewriter.key_stepper.keyboard.unhook", lambda value: unhooked.append(value)
)
return hook
def test_start_installs_hook_without_typing_immediately(monkeypatch):
typed = []
hooked = []
monkeypatch.setattr("ai_typewriter.key_stepper.keyboard.write", lambda char, delay=0, exact=True: typed.append(char))
monkeypatch.setattr("ai_typewriter.key_stepper.keyboard.hook", lambda callback, suppress=True: hooked.append((callback, suppress)) or "hook")
typed, sent, unhooked = [], [], []
_install_hook(monkeypatch, typed, sent, unhooked)
stepper = KeyStepper("abc")
stepper.start()
assert typed == []
assert hooked and hooked[0][1] is True
assert typed == [] and sent == []
assert stepper.remaining_characters == 3
def test_single_character_response_waits_for_key_then_unhooks(monkeypatch):
typed = []
hook = {}
unhooked = []
monkeypatch.setattr("ai_typewriter.key_stepper.keyboard.write", lambda char, delay=0, exact=True: typed.append(char))
monkeypatch.setattr("ai_typewriter.key_stepper.keyboard.hook", lambda callback, suppress=True: hook.update({"callback": callback}) or "hook")
monkeypatch.setattr("ai_typewriter.key_stepper.keyboard.unhook", lambda value: unhooked.append(value))
stepper = KeyStepper("x")
stepper.start()
hook["callback"](SimpleNamespace(event_type="down"))
assert typed == ["x"]
assert unhooked == ["hook"]
assert stepper.remaining_characters == 0
def test_each_key_down_types_next_character(monkeypatch):
typed = []
hook = {}
monkeypatch.setattr("ai_typewriter.key_stepper.keyboard.write", lambda char, delay=0, exact=True: typed.append(char))
monkeypatch.setattr("ai_typewriter.key_stepper.keyboard.hook", lambda callback, suppress=True: hook.update({"callback": callback}) or "hook")
monkeypatch.setattr("ai_typewriter.key_stepper.keyboard.unhook", lambda value: None)
typed, sent, unhooked = [], [], []
hook = _install_hook(monkeypatch, typed, sent, unhooked)
stepper = KeyStepper("ab")
stepper.start()
@@ -51,19 +111,47 @@ def test_each_key_down_types_next_character(monkeypatch):
assert typed == ["a", "b"]
assert stepper.remaining_characters == 0
assert unhooked # libère le hook à la fin
def test_latex_markers_are_typed_literally_character_by_character(monkeypatch):
typed = []
hook = {}
def test_single_char_response_waits_for_key_then_unhooks(monkeypatch):
typed, sent, unhooked = [], [], []
hook = _install_hook(monkeypatch, typed, sent, unhooked)
monkeypatch.setattr("ai_typewriter.key_stepper.keyboard.write", lambda char, delay=0, exact=True: typed.append(char))
monkeypatch.setattr("ai_typewriter.key_stepper.keyboard.hook", lambda callback, suppress=True: hook.update({"callback": callback}) or "hook")
monkeypatch.setattr("ai_typewriter.key_stepper.keyboard.unhook", lambda value: None)
stepper = KeyStepper(r"[EQ]\frac{a}{b}[/EQ]")
stepper = KeyStepper("x")
stepper.start()
for _ in range(len(stepper.text)):
hook["callback"](SimpleNamespace(event_type="down"))
assert typed == ["x"] and sent == []
assert unhooked == ["hook"]
def test_equation_markers_trigger_key_sequences_while_typing_latex(monkeypatch):
typed, sent, unhooked = [], [], []
hook = _install_hook(monkeypatch, typed, sent, unhooked)
stepper = KeyStepper(
r"[EQ]a=x[/EQ]",
equation_enabled=True,
eq_start_key="alt+=",
eq_end_key="right",
)
stepper.start()
# 1: [EQ] -> Alt+= ; 2-4: a, =, x ; 5: [/EQ] -> right
for _ in range(5):
hook["callback"](SimpleNamespace(event_type="down"))
assert "".join(typed) == r"[EQ]\frac{a}{b}[/EQ]"
assert sent == ["alt+=", "right"]
assert typed == ["a", "=", "x"]
assert stepper.remaining_characters == 0
def test_empty_text_does_not_install_hook(monkeypatch):
called = []
monkeypatch.setattr(
"ai_typewriter.key_stepper.keyboard.hook",
lambda callback, suppress=True: called.append(1),
)
stepper = KeyStepper("")
stepper.start()
assert called == []
-18
View File
@@ -1,18 +0,0 @@
import pytest
from ai_typewriter.math_format import format_math_text
def test_plain_mode_keeps_latex_and_markers_literal():
assert format_math_text(r"[EQ]\frac{a}{b}[/EQ]", "plain") == r"[EQ]\frac{a}{b}[/EQ]"
def test_unicode_mode_converts_indices_exponents_and_symbols():
result = format_math_text(r"z_1 + x^2 + x_(i+1) + \alpha + \infty", "unicode")
assert result == "z₁ + x² + xᵢ₊₁ + α + ∞"
def test_invalid_math_text_format_is_rejected():
with pytest.raises(ValueError):
format_math_text("x_1", "bad")
+90
View File
@@ -0,0 +1,90 @@
"""Tests de l'actuaire des modèles : listage local, catalogue en ligne et
téléchargement (le tout mocké, sans accéder au réseau ni à la CLI ollama)."""
from ai_typewriter import model_catalog
class FakeResponse:
def __init__(self, payload=None, status=200):
self.payload = payload
self.status_code = status
def raise_for_status(self):
if self.status_code >= 400:
raise RuntimeError(f"HTTP {self.status_code}")
return None
def json(self):
return self.payload
def test_list_local_models_parses_names(monkeypatch):
def fake_get(url, timeout, **kwargs):
return FakeResponse(
{"models": [{"name": "llama3.1"}, {"model": "mistral"}, {"name": "llama3.1"}]}
)
monkeypatch.setattr("ai_typewriter.model_catalog.requests.get", fake_get)
names = model_catalog.list_local_models("http://local:11434")
assert names == ["llama3.1", "mistral"] # triés + dédupliqués
def test_list_local_models_returns_empty_on_error(monkeypatch):
import requests
def fake_get(url, timeout, **kwargs):
raise requests.ConnectionError("boom")
monkeypatch.setattr("ai_typewriter.model_catalog.requests.get", fake_get)
assert model_catalog.list_local_models() == []
def test_search_online_filters_by_query():
results = model_catalog.search_online_models("llama")
assert results
assert all("llama" in m.name for m in results)
assert all(m.source == "registry" for m in results)
def test_provider_info():
info = model_catalog.provider_info("openai")
assert info["needs_key"] is True
info_ollama = model_catalog.provider_info("ollama")
assert info_ollama["needs_key"] is False
def test_resolve_exact_model_true_when_200(monkeypatch):
monkeypatch.setattr(
"ai_typewriter.model_catalog.requests.get", lambda url, timeout: FakeResponse(status=200)
)
assert model_catalog.resolve_exact_model("llama3.1") is True
def test_resolve_exact_model_false_when_notfound(monkeypatch):
monkeypatch.setattr(
"ai_typewriter.model_catalog.requests.get", lambda url, timeout: FakeResponse(status=404)
)
assert model_catalog.resolve_exact_model("n-existe-pas") is False
def test_pull_model_uses_cli_when_available(monkeypatch):
calls = []
class FakePopen:
def __init__(self, cmd, **kwargs):
calls.append(cmd)
monkeypatch.setattr("ai_typewriter.model_catalog.shutil.which", lambda name: "/usr/bin/ollama")
monkeypatch.setattr("ai_typewriter.model_catalog.subprocess.Popen", FakePopen)
model_catalog.pull_model("llama3.1", server_url="http://local")
assert calls and calls[0] == ["/usr/bin/ollama", "pull", "llama3.1"]
def test_has_internet_true(monkeypatch):
monkeypatch.setattr(
"ai_typewriter.model_catalog.requests.head",
lambda url, timeout, allow_redirects: FakeResponse(status=200),
)
assert model_catalog.has_internet() is True