Initial ai typewriter implementation

src/ai_typewriter/__init__.py

@@ -0,0 +1,3 @@
+"""AI Typewriter package."""
+
+__version__ = "0.1.0"

src/ai_typewriter/ai_client.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+import requests
+
+from .config import AppConfig
+
+
+class AIClientError(RuntimeError):
+    """Raised when the configured AI backend cannot return text."""
+
+
+def ask_ai(prompt: str, config: AppConfig) -> str:
+    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}")
+
+
+def _ask_ollama(prompt: str, config: AppConfig) -> str:
+    url = f"{config.server_url}/api/chat"
+    payload = {
+        "model": config.model,
+        "stream": False,
+        "messages": [
+            {"role": "system", "content": config.system_prompt},
+            {"role": "user", "content": prompt},
+        ],
+    }
+    try:
+        response = requests.post(url, json=payload, timeout=config.request_timeout_seconds)
+        response.raise_for_status()
+        data = response.json()
+    except requests.RequestException as exc:
+        raise AIClientError(f"Erreur Ollama: {exc}") from exc
+    except ValueError as 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():
+        raise AIClientError("Réponse Ollama vide ou inattendue")
+    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'.")
+
+    base = config.server_url or "https://generativelanguage.googleapis.com"
+    url = f"{base}/v1beta/models/{config.model}:generateContent"
+    payload = {
+        "systemInstruction": {"parts": [{"text": config.system_prompt}]},
+        "contents": [{"role": "user", "parts": [{"text": prompt}]}],
+        "generationConfig": {"temperature": 0.2},
+    }
+    try:
+        response = requests.post(
+            url,
+            params={"key": config.api_key},
+            json=payload,
+            timeout=config.request_timeout_seconds,
+        )
+        response.raise_for_status()
+        data = response.json()
+    except requests.RequestException as exc:
+        raise AIClientError(f"Erreur Gemini: {exc}") from exc
+    except ValueError as exc:
+        raise AIClientError("Réponse Gemini invalide: JSON illisible") from exc
+
+    try:
+        text = data["candidates"][0]["content"]["parts"][0]["text"]
+    except (KeyError, IndexError, TypeError) as exc:
+        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()

src/ai_typewriter/app.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+import argparse
+import logging
+import sys
+from threading import Lock
+
+import keyboard
+
+from .ai_client import AIClientError, ask_ai
+from .clipboard_capture import capture_selection
+from .config import AppConfig, load_config
+from .key_stepper import KeyStepper
+
+LOG = logging.getLogger("ai_typewriter")
+
+
+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_selection(self.config.copy_wait_seconds, self.config.restore_clipboard)
+            LOG.info("Texte capturé: %d caractères.", len(selected))
+            answer = ask_ai(selected, self.config)
+            LOG.info("Réponse reçue: %d caractères. Mode dactylographie actif.", len(answer))
+            if self._stepper is not None:
+                self._stepper.stop()
+            self._stepper = KeyStepper(answer, 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.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")
+    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",
+    )
+    try:
+        config = load_config(args.config)
+    except Exception as exc:
+        LOG.error("%s", exc)
+        return 2
+
+    if args.ask is not None:
+        try:
+            print(ask_ai(args.ask, config))
+            return 0
+        except AIClientError as exc:
+            LOG.error("%s", exc)
+            return 1
+
+    AITypewriterApp(config).run()
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv[1:]))

src/ai_typewriter/clipboard_capture.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+import time
+
+import keyboard
+import pyperclip
+
+
+def capture_selection(copy_wait_seconds: float = 1.0, restore_clipboard: bool = True) -> str:
+    previous = pyperclip.paste()
+    keyboard.send("ctrl+c")
+
+    deadline = time.monotonic() + max(copy_wait_seconds, 0.05)
+    captured = previous
+    while time.monotonic() < deadline:
+        current = pyperclip.paste()
+        if current != previous:
+            captured = current
+            break
+        time.sleep(0.03)
+    else:
+        captured = pyperclip.paste()
+
+    if restore_clipboard:
+        try:
+            pyperclip.copy(previous)
+        except pyperclip.PyperclipException:
+            pass
+    return captured or ""

src/ai_typewriter/config.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Literal
+
+Provider = Literal["ollama", "gemini"]
+
+
+@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 = 120.0
+    copy_wait_seconds: float = 1.0
+    type_delay_seconds: float = 0.0
+    restore_clipboard: bool = True
+    system_prompt: str = (
+        "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 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=float(data.get("request_timeout_seconds", 120)),
+        copy_wait_seconds=float(data.get("copy_wait_seconds", 1)),
+        type_delay_seconds=float(data.get("type_delay_seconds", 0)),
+        restore_clipboard=bool(data.get("restore_clipboard", True)),
+        system_prompt=str(data.get("system_prompt", AppConfig.system_prompt)),
+    )

src/ai_typewriter/key_stepper.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from threading import Lock
+from typing import Optional
+
+import keyboard
+
+
+@dataclass
+class KeyStepper:
+    text: str
+    delay: float = 0.0
+    _index: int = 0
+    _hook: Optional[object] = None
+    _lock: Lock = field(default_factory=Lock)
+    _injecting: bool = False
+
+    def start(self) -> None:
+        if not self.text:
+            return
+        if self._hook is not None:
+            return
+        # suppress=True blocks the physical key while the callback types the next AI character.
+        self._hook = keyboard.hook(self._on_event, suppress=True)
+
+    def stop(self) -> None:
+        hook = self._hook
+        self._hook = None
+        if hook is not None:
+            keyboard.unhook(hook)
+
+    def _on_event(self, event: keyboard.KeyboardEvent) -> None:
+        if self._injecting or event.event_type != keyboard.KEY_DOWN:
+            return
+
+        with self._lock:
+            if self._index >= len(self.text):
+                self.stop()
+                return
+            char = self.text[self._index]
+            self._index += 1
+
+        self._injecting = True
+        try:
+            self._type_char(char)
+        finally:
+            self._injecting = False
+
+        if self._index >= len(self.text):
+            self.stop()
+
+    def _type_char(self, char: str) -> None:
+        if char == "\n":
+            keyboard.send("enter")
+        elif char == "	":
+            keyboard.send("tab")
+        else:
+            keyboard.write(char, delay=self.delay, exact=True)