From 81632348ae92ec703b6144051a34e5878a00b845 Mon Sep 17 00:00:00 2001 From: Julio Rios Date: Fri, 25 Sep 2026 09:52:43 -0500 Subject: [PATCH 1/2] Windows improves --- .github/workflows/ci.yml | 30 + README.es.md | 40 +- README.md | 40 +- codex_notifier/__init__.py | 3 + codex_notifier/channels/__init__.py | 1 + codex_notifier/channels/discord.py | 17 + codex_notifier/channels/teams.py | 10 + codex_notifier/channels/voice.py | 207 +++++ codex_notifier/channels/webhook.py | 52 ++ codex_notifier/cli.py | 192 +++++ codex_notifier/config.py | 109 +++ codex_notifier/files.py | 88 ++ codex_notifier/payloads.py | 133 +++ codex_notifier/state.py | 77 ++ codex_notifier/trace_logging.py | 78 ++ config.example.json | 2 + install.py | 77 +- notifier.py | 1248 ++++----------------------- tests/test_hardening.py | 112 +++ tests/test_install.py | 22 + 20 files changed, 1407 insertions(+), 1131 deletions(-) create mode 100644 .github/workflows/ci.yml create mode 100644 codex_notifier/__init__.py create mode 100644 codex_notifier/channels/__init__.py create mode 100644 codex_notifier/channels/discord.py create mode 100644 codex_notifier/channels/teams.py create mode 100644 codex_notifier/channels/voice.py create mode 100644 codex_notifier/channels/webhook.py create mode 100644 codex_notifier/cli.py create mode 100644 codex_notifier/config.py create mode 100644 codex_notifier/files.py create mode 100644 codex_notifier/payloads.py create mode 100644 codex_notifier/state.py create mode 100644 codex_notifier/trace_logging.py create mode 100644 tests/test_hardening.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..3f649fd --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,30 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + branches: [main] + +jobs: + tests: + strategy: + fail-fast: false + matrix: + include: + - os: ubuntu-latest + python: "3.10" + - os: ubuntu-latest + python: "3.13" + - os: windows-latest + python: "3.10" + - os: windows-latest + python: "3.13" + runs-on: ${{ matrix.os }} + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python }} + - name: Run tests + run: python -m unittest discover -s tests -v diff --git a/README.es.md b/README.es.md index 2399a07..ab3a293 100644 --- a/README.es.md +++ b/README.es.md @@ -23,7 +23,8 @@ configuración local. Las alertas de voz se silencian entre las 23:00 y las 07:00. Mencionan el proyecto, la duración y el título del chat cuando Codex lo proporciona. Teams y -Discord incluyen el mensaje final del asistente como un resumen limitado. +Discord incluyen por defecto el mensaje final como resumen limitado; configura +`include_summary` como `false` para omitir por completo el texto de la respuesta. Cada canal se ejecuta de forma independiente: el fallo de un webhook no impide que funcionen la voz o el otro webhook. El notificador mantiene un registro @@ -34,7 +35,7 @@ final del asistente. - Python 3.10 o posterior. - Codex con acceso a `~/.codex/config.toml`. -- Voz en Windows: SAPI, normalmente incluido con Windows. +- Voz en Windows: SAPI, normalmente incluido con Windows; Piper es opcional. - Voz en Linux: Piper (recomendado), `spd-say`, `espeak-ng` o `espeak`. Teams y Discord no necesitan un motor de voz local. @@ -55,11 +56,17 @@ El instalador: 3. Copia `config.example.json` a la ruta de configuración local si todavía no existe. -Los archivos existentes de Codex se respaldan antes de modificarlos. Una -configuración existente del notificador se conserva salvo que se use +Los archivos existentes de Codex se respaldan antes de modificarlos. Las +escrituras usan archivos temporales y reemplazo atómico, por lo que un fallo no +destruye el destino. Una configuración existente del notificador se conserva salvo que se use explícitamente `--force-config`. Después de instalar o cambiar el hook, abre un chat nuevo y usa `/hooks` para revisarlo y confiar en él. +El instalador modifica únicamente `notify` en la raíz de `config.toml`; no +reemplaza claves `notify` dentro de secciones TOML ni hooks ajenos. En POSIX, +las carpetas de estado son privadas y las marcas, logs y configuraciones nuevas +usan permisos solo para el propietario cuando corresponde. + Opciones del instalador: ```text @@ -89,6 +96,7 @@ canales: "enabled": false, "minimum_seconds": 300, "notify_when_duration_unknown": false, + "include_summary": true, "summary_max_chars": 1800, "summary_tail_chars": 600, "webhook_url": "" @@ -97,6 +105,7 @@ canales: "enabled": false, "minimum_seconds": 300, "notify_when_duration_unknown": false, + "include_summary": true, "summary_max_chars": 1800, "webhook_url": "" }, @@ -122,6 +131,9 @@ canales: `minimum_seconds` es inclusivo. `notify_when_duration_unknown` omite únicamente el umbral de duración: el canal debe seguir habilitado, los canales de webhook necesitan una URL HTTPS válida y la voz sigue respetando el horario silencioso. +`include_summary` es `true` por compatibilidad. Con `false`, Teams omite su +sección de resumen y Discord omite la descripción del embed, pero conserva los +metadatos. Ambos canales siguen aceptando cualquier URL HTTPS. El instalador no combina claves nuevas dentro de una configuración existente. Al actualizar, compara tu archivo local con `config.example.json` y añade @@ -162,9 +174,12 @@ PowerShell para usar directamente el dispositivo de audio de Windows. completo). `null` conserva el volumen predeterminado. La opción se aplica a Piper en Linux/WSL y a SAPI en Windows. -En Windows, `spanish_voice` y `english_voice` pueden contener parte del nombre -de una voz SAPI instalada. Los valores vacíos seleccionan automáticamente una -voz del idioma correspondiente. +En Windows, si `piper_executable` está vacío se conserva SAPI: `spanish_voice` y +`english_voice` son fragmentos opcionales del nombre de una voz instalada. Si +no está vacío, se resuelve por PATH o como ruta expandida a `piper.exe`, y esos +campos son rutas a modelos `.onnx` de español/inglés. Un Piper configurado pero +inválido produce un error visible y no cambia silenciosamente a SAPI; SAPI +continúa siendo el valor predeterminado sin configuración. Configura `quiet_start` y `quiet_end` con la misma hora para deshabilitar el horario silencioso. @@ -206,18 +221,25 @@ python notifier.py voice-test es python notifier.py voice-test en ``` -Cada comando selecciona el modelo Piper correspondiente al idioma. En Linux -nativo reproduce el WAV temporal con `paplay`, `pw-play`, `aplay` o `ffplay`. +Cada comando selecciona el modelo Piper correspondiente cuando Piper está +configurado. En Windows sin Piper prueba SAPI; en Linux reproduce el WAV +temporal con `paplay`, `pw-play`, `aplay` o `ffplay`. Estos comandos reproducen audio; las pruebas automatizadas simulan las llamadas de voz y webhook. +`notify` devuelve éxito a Codex aunque falle un canal; los fallos se registran +de forma independiente. La sintaxis manual inválida devuelve un código +distinto de cero, al igual que `voice-test` si la voz no puede ejecutarse. + ## Ejecutar las pruebas ```powershell python -m unittest discover -s tests -v ``` +GitHub Actions ejecuta esta suite en Ubuntu y Windows con Python 3.10 y 3.13. + ## Cómo funciona El hook `UserPromptSubmit` registra una marca de inicio sin guardar el prompt. diff --git a/README.md b/README.md index 6022a71..6a651aa 100644 --- a/README.md +++ b/README.md @@ -21,7 +21,8 @@ The installer copies `config.example.json` when no local configuration exists. Voice alerts are muted from 23:00 to 07:00. They mention the project, duration, and chat title when Codex provides one. Teams and Discord include the final -assistant message as a bounded summary. +assistant message as a bounded summary by default; set `include_summary` to +`false` in either channel to omit assistant-response text completely. Each channel runs independently: a failed webhook does not prevent voice or the other webhook from running. The notifier keeps a privacy-conscious daily trace @@ -31,7 +32,7 @@ log and never stores the prompt or final assistant message. - Python 3.10 or later. - Codex with an accessible `~/.codex/config.toml`. -- Windows voice: SAPI, normally included with Windows. +- Windows voice: SAPI, normally included with Windows; Piper is optional. - Linux voice: Piper (recommended), `spd-say`, `espeak-ng`, or `espeak`. Teams and Discord do not require a local speech engine. @@ -52,11 +53,17 @@ The installer: 3. Copies `config.example.json` to the local configuration path when that file does not already exist. -Existing Codex files are backed up before modification. An existing notifier -configuration is preserved unless `--force-config` is explicitly used. After +Existing Codex files are backed up before modification. Writes use temporary +files and atomic replacement, so a failed write leaves the destination intact. +An existing notifier configuration is preserved unless `--force-config` is explicitly used. After installing or changing the hook, open a new chat and use `/hooks` to review and trust it. +The installer changes only a root-level `notify` setting in `config.toml`; +`notify` keys inside TOML sections and unrelated hooks are preserved. State +directories are private on POSIX and marker/log/config files are written with +owner-only permissions where that platform supports them. + Installer options: ```text @@ -85,6 +92,7 @@ Default locations: "enabled": false, "minimum_seconds": 300, "notify_when_duration_unknown": false, + "include_summary": true, "summary_max_chars": 1800, "summary_tail_chars": 600, "webhook_url": "" @@ -93,6 +101,7 @@ Default locations: "enabled": false, "minimum_seconds": 300, "notify_when_duration_unknown": false, + "include_summary": true, "summary_max_chars": 1800, "webhook_url": "" }, @@ -118,6 +127,9 @@ Default locations: `minimum_seconds` is inclusive. `notify_when_duration_unknown` bypasses only the duration threshold: the channel must still be enabled, webhook channels still need a valid HTTPS URL, and voice still respects quiet hours. +`include_summary` defaults to `true` for compatibility. With `false`, Teams +omits its summary section and Discord omits the embed description; metadata is +still sent. Teams and Discord continue accepting any HTTPS webhook URL. The installer does not merge new keys into an existing configuration. When upgrading, compare your local file with `config.example.json` and add any new @@ -158,8 +170,12 @@ it uses the Windows audio device directly. volume). `null` preserves the default volume. The option applies to Piper on Linux/WSL and SAPI on Windows. -On Windows, `spanish_voice` and `english_voice` can contain part of an installed -SAPI voice name. Empty values select a matching language voice automatically. +On Windows, an empty `piper_executable` keeps the existing SAPI behavior: +`spanish_voice` and `english_voice` are optional fragments of installed SAPI +voice names. If `piper_executable` is non-empty, it is resolved through PATH or +as an expanded path to `piper.exe`, and those fields instead name existing +Spanish/English `.onnx` models. Piper failures are reported as errors and do +not silently fall back to SAPI. SAPI remains the zero-configuration default. Set `quiet_start` and `quiet_end` to the same time to disable quiet hours. @@ -198,17 +214,25 @@ python notifier.py voice-test es python notifier.py voice-test en ``` -Each command selects the matching Piper model. Native Linux plays the temporary -WAV with `paplay`, `pw-play`, `aplay`, or `ffplay`. +Each command selects the matching Piper model when Piper is configured. On +Windows without Piper it tests SAPI; native Linux plays a Piper WAV with +`paplay`, `pw-play`, `aplay`, or `ffplay`. These commands play audio; automated tests mock speech and webhook calls. +`notify` returns success to Codex even when an individual channel fails; those +failures are logged independently. Invalid manual syntax returns non-zero, and +`voice-test` returns non-zero when the selected voice cannot execute. + ## Run tests ```powershell python -m unittest discover -s tests -v ``` +GitHub Actions runs this suite on Ubuntu and Windows with Python 3.10 and +Python 3.13. + ## How it works The `UserPromptSubmit` hook records a start marker without storing the prompt. diff --git a/codex_notifier/__init__.py b/codex_notifier/__init__.py new file mode 100644 index 0000000..828f4c3 --- /dev/null +++ b/codex_notifier/__init__.py @@ -0,0 +1,3 @@ +"""Maintainable implementation package for Codex Notifier.""" + +__version__ = "1.0" diff --git a/codex_notifier/channels/__init__.py b/codex_notifier/channels/__init__.py new file mode 100644 index 0000000..bd27b58 --- /dev/null +++ b/codex_notifier/channels/__init__.py @@ -0,0 +1 @@ +"""External notification and local voice channels.""" diff --git a/codex_notifier/channels/discord.py b/codex_notifier/channels/discord.py new file mode 100644 index 0000000..732d90a --- /dev/null +++ b/codex_notifier/channels/discord.py @@ -0,0 +1,17 @@ +"""Discord webhook transport.""" + +from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit + +from .webhook import send_webhook_notification + + +def send(webhook_url: str, payload: dict) -> None: + parts = urlsplit(webhook_url) + query = dict(parse_qsl(parts.query, keep_blank_values=True)) + query["wait"] = "true" + confirmed_url = urlunsplit((parts.scheme, parts.netloc, parts.path, + urlencode(query), parts.fragment)) + send_webhook_notification("Discord", confirmed_url, payload) + + +send_discord_notification = send diff --git a/codex_notifier/channels/teams.py b/codex_notifier/channels/teams.py new file mode 100644 index 0000000..b1693a4 --- /dev/null +++ b/codex_notifier/channels/teams.py @@ -0,0 +1,10 @@ +"""Microsoft Teams webhook transport.""" + +from .webhook import send_webhook_notification + + +def send(webhook_url: str, payload: dict) -> None: + send_webhook_notification("Teams", webhook_url, payload) + + +send_teams_notification = send diff --git a/codex_notifier/channels/voice.py b/codex_notifier/channels/voice.py new file mode 100644 index 0000000..78a6823 --- /dev/null +++ b/codex_notifier/channels/voice.py @@ -0,0 +1,207 @@ +"""SAPI, Piper, and platform audio playback.""" + +from __future__ import annotations + +import base64 +import os +import re +import shutil +import subprocess +import tempfile +from pathlib import Path +from typing import Any + +from ..state import text + +SPEECH_TIMEOUT_SECONDS = 30 +SPANISH_WORDS = {"al", "cambios", "completado", "con", "corregido", "de", "el", "en", "esta", "este", "fue", "la", "las", "listo", "los", "para", "por", "pruebas", "que", "se", "sin", "una", "y", "ya"} +ENGLISH_WORDS = {"a", "and", "changes", "completed", "done", "for", "from", "has", "fixed", "in", "is", "of", "on", "that", "the", "tests", "this", "to", "was", "ready", "with", "without"} + + +def detect_language(value: str, configured: str = "auto") -> str: + requested = configured.strip().lower() + if requested in {"es", "spanish", "español"}: + return "es" + if requested in {"en", "english", "inglés", "ingles"}: + return "en" + lowered = value.lower() + words = re.findall(r"[a-záéíóúüñ]+", lowered) + spanish_score = sum(word in SPANISH_WORDS for word in words) + english_score = sum(word in ENGLISH_WORDS for word in words) + if re.search(r"[áéíóúüñ¿¡]", lowered): + spanish_score += 2 + return "en" if english_score > spanish_score else "es" + + +def configured_volume(voice_config: dict[str, Any]) -> float | None: + value = voice_config.get("volume") + if not isinstance(value, (int, float)) or isinstance(value, bool): + return None + return min(1.0, max(0.0, float(value))) + + +def configured_voice(voice_config: dict[str, Any], language: str) -> str: + return text(voice_config.get("spanish_voice" if language == "es" else "english_voice")) + + +def resolve_executable(configured: str, default: str) -> str | None: + requested = configured or default + expanded = Path(requested).expanduser() + if expanded.is_absolute() or expanded.parent != Path("."): + return str(expanded) if expanded.is_file() else None + return shutil.which(requested) + + +def _creation_flags() -> int: + return getattr(subprocess, "CREATE_NO_WINDOW", 0) + + +def speak_sapi_windows(message: str, *, language: str, preferred_voice: str = "", + volume: float | None = None) -> None: + if os.name != "nt": + raise RuntimeError("La voz local solo está disponible en Windows.") + environment = os.environ.copy() + environment["CODEX_NOTIFIER_SPEECH_B64"] = base64.b64encode(message.encode("utf-8")).decode("ascii") + environment["CODEX_NOTIFIER_SPEECH_LANGUAGE"] = language + environment["CODEX_NOTIFIER_SPEECH_VOICE"] = preferred_voice + environment["CODEX_NOTIFIER_SPEECH_VOLUME"] = "" if volume is None else str(round(min(1.0, max(0.0, volume)) * 100)) + script = ( + "$bytes=[Convert]::FromBase64String($env:CODEX_NOTIFIER_SPEECH_B64);" + "$text=[Text.Encoding]::UTF8.GetString($bytes);$voice=New-Object -ComObject SAPI.SpVoice;" + "$wantedName=$env:CODEX_NOTIFIER_SPEECH_VOICE;$wantedLanguage=$env:CODEX_NOTIFIER_SPEECH_LANGUAGE;" + "$wantedVolume=$env:CODEX_NOTIFIER_SPEECH_VOLUME;$selected=$null;" + "foreach($candidate in $voice.GetVoices()){$description=$candidate.GetDescription();" + "if($wantedName -and $description -like ('*'+$wantedName+'*')){$selected=$candidate;break}" + "if(-not $selected){try{$lcid=[Convert]::ToInt32($candidate.GetAttribute('Language'),16);" + "$culture=[Globalization.CultureInfo]::GetCultureInfo($lcid);" + "if($culture.TwoLetterISOLanguageName -eq $wantedLanguage){$selected=$candidate}}catch{}}};" + "if($selected){$voice.Voice=$selected};if($wantedVolume -ne ''){$voice.Volume=[int]$wantedVolume};" + "[void]$voice.Speak($text)" + ) + subprocess.run(["powershell.exe", "-NoLogo", "-NoProfile", "-NonInteractive", "-Command", script], + env=environment, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, creationflags=_creation_flags(), check=True, + timeout=SPEECH_TIMEOUT_SECONDS) + + +def play_wav_windows(wav_path: Path, *, powershell: str | None = None) -> None: + executable = powershell or shutil.which("powershell.exe") or "powershell.exe" + encoded_path = base64.b64encode(str(wav_path).encode("utf-8")).decode("ascii") + script = ( + "$ErrorActionPreference='Stop';$bytes=[Convert]::FromBase64String($env:CODEX_NOTIFIER_WAV_B64);" + "$path=[Text.Encoding]::UTF8.GetString($bytes);$player=New-Object System.Media.SoundPlayer $path;" + "$player.Load();$player.PlaySync()" + ) + environment = os.environ.copy() + environment["CODEX_NOTIFIER_WAV_B64"] = encoded_path + subprocess.run([executable, "-NoLogo", "-NoProfile", "-NonInteractive", "-Command", script], + env=environment, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, creationflags=_creation_flags(), check=True, + timeout=SPEECH_TIMEOUT_SECONDS) + + +def _play_wav_wsl(wav_path: Path, powershell: str, wslpath: str) -> None: + result = subprocess.run([wslpath, "-w", str(wav_path)], stdin=subprocess.DEVNULL, + capture_output=True, text=True, check=True, timeout=5) + windows_path = result.stdout.strip() + encoded_path = base64.b64encode(windows_path.encode("utf-8")).decode("ascii") + script = ("$ErrorActionPreference='Stop';$bytes=[Convert]::FromBase64String('" + encoded_path + "');" + "$path=[Text.Encoding]::UTF8.GetString($bytes);$player=New-Object System.Media.SoundPlayer $path;" + "$player.Load();$player.PlaySync()") + subprocess.run([powershell, "-NoLogo", "-NoProfile", "-NonInteractive", "-Command", script], + stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + creationflags=_creation_flags(), check=True, timeout=SPEECH_TIMEOUT_SECONDS) + + +def generate_piper_wav(message: str, *, language: str = "es", voice_config: dict[str, Any], + temporary_dir: str | Path) -> Path: + model_value = configured_voice(voice_config, language) + key = "spanish_voice" if language == "es" else "english_voice" + if not model_value: + raise RuntimeError(f"Configura voice.{key} para usar Piper.") + model = Path(model_value).expanduser() + if not model.is_file(): + raise RuntimeError(f"No se encontró el modelo de Piper: {model}") + configured = text(voice_config.get("piper_executable")) + piper = resolve_executable(configured, "piper") + if not piper: + raise RuntimeError(f"No se encontró el ejecutable de Piper configurado: {configured or 'piper'}") + wav_path = Path(temporary_dir) / "speech.wav" + command = [piper, "-m", str(model), "-f", str(wav_path)] + volume = configured_volume(voice_config) + if volume is not None: + command.extend(["--volume", str(volume)]) + command.extend(["--", message]) + try: + subprocess.run(command, cwd=str(temporary_dir), stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + check=True, timeout=SPEECH_TIMEOUT_SECONDS, + creationflags=_creation_flags()) + except (OSError, subprocess.SubprocessError) as exc: + raise RuntimeError(f"Piper no pudo generar el audio: {exc}") from exc + return wav_path + + +def speak_piper(message: str, *, language: str = "es", voice_config: dict[str, Any], + windows: bool | None = None) -> None: + with tempfile.TemporaryDirectory(prefix="codex-notifier-") as temporary_dir: + wav_path = generate_piper_wav(message, language=language, voice_config=voice_config, + temporary_dir=temporary_dir) + use_windows = os.name == "nt" if windows is None else windows + if use_windows: + play_wav_windows(wav_path) + return + play_wav_linux(wav_path) + + +def play_wav_linux(wav_path: Path) -> None: + powershell = shutil.which("powershell.exe") + wslpath = shutil.which("wslpath") + if powershell and wslpath: + _play_wav_wsl(wav_path, powershell, wslpath) + return + player = next(((name, shutil.which(name)) for name in ("paplay", "pw-play", "aplay", "ffplay") if shutil.which(name)), None) + if not player: + raise RuntimeError("No se encontró un reproductor compatible: instala paplay, pw-play, aplay o ffplay. En WSL también se admite powershell.exe.") + name, executable = player + command = [executable, "-nodisp", "-autoexit", str(wav_path)] if name == "ffplay" else [executable, str(wav_path)] + subprocess.run(command, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, check=True, timeout=SPEECH_TIMEOUT_SECONDS, + creationflags=_creation_flags()) + + +def speak_linux(message: str, *, language: str = "es", voice_config: dict[str, Any] | None = None) -> None: + settings = {} if voice_config is None else voice_config + if text(settings.get("piper_executable")): + speak_piper(message, language=language, voice_config=settings, windows=False) + return + dispatcher = shutil.which("spd-say") + if dispatcher: + command = [dispatcher, "-w", "-l", language, message] + else: + espeak = shutil.which("espeak-ng") or shutil.which("espeak") + if not espeak: + raise RuntimeError("Instala speech-dispatcher (spd-say) o espeak-ng para usar voz en Linux.") + command = [espeak, "-v", language, message] + subprocess.run(command, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, check=True, timeout=SPEECH_TIMEOUT_SECONDS, + creationflags=_creation_flags()) + + +def speak(message: str, *, language: str = "es", preferred_voice: str = "", + voice_config: dict[str, Any] | None = None) -> None: + settings = {} if voice_config is None else voice_config + if os.name == "nt": + if text(settings.get("piper_executable")): + speak_piper(message, language=language, voice_config=settings, windows=True) + else: + speak_sapi_windows(message, language=language, preferred_voice=preferred_voice, + volume=configured_volume(settings)) + return + if os.name == "posix": + speak_linux(message, language=language, voice_config=settings) + return + raise RuntimeError(f"La voz local no es compatible con la plataforma {os.name}.") + + +speak_windows = speak_sapi_windows diff --git a/codex_notifier/channels/webhook.py b/codex_notifier/channels/webhook.py new file mode 100644 index 0000000..7eef0f4 --- /dev/null +++ b/codex_notifier/channels/webhook.py @@ -0,0 +1,52 @@ +"""Bounded webhook retry transport. URL policy is intentionally caller-owned.""" + +from __future__ import annotations + +import json +import time +import email.utils +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen + +RETRYABLE_STATUSES = {429, 500, 502, 503, 504} +MAX_RETRY_DELAY_SECONDS = 5.0 +RETRY_DELAY_SECONDS = 1.0 + + +def _retry_after(value: str | None) -> float: + if not value: + return RETRY_DELAY_SECONDS + try: + return min(MAX_RETRY_DELAY_SECONDS, max(0.0, float(value))) + except ValueError: + try: + when = email.utils.parsedate_to_datetime(value).timestamp() + return min(MAX_RETRY_DELAY_SECONDS, max(0.0, when - time.time())) + except (TypeError, ValueError, OverflowError): + return RETRY_DELAY_SECONDS + + +def send_webhook_notification(service: str, webhook_url: str, payload: dict) -> None: + request = Request(webhook_url, data=json.dumps(payload, ensure_ascii=False).encode("utf-8"), + headers={"Content-Type": "application/json; charset=utf-8", + "User-Agent": "Codex-Notifier/1.0"}, method="POST") + for attempt in range(2): + try: + with urlopen(request, timeout=5) as response: + if 200 <= response.status < 300: + return + if response.status not in RETRYABLE_STATUSES or attempt == 1: + raise RuntimeError(f"{service} respondió con HTTP {response.status}") + headers = response.headers or {} + delay = _retry_after(headers.get("Retry-After")) + except HTTPError as exc: + if exc.code not in RETRYABLE_STATUSES or attempt == 1: + raise RuntimeError(f"{service} respondió con HTTP {exc.code}") from exc + headers = exc.headers or {} + delay = _retry_after(headers.get("Retry-After")) + except URLError as exc: + if attempt == 1: + raise RuntimeError(f"No se pudo conectar con {service}: {exc.reason}") from exc + delay = RETRY_DELAY_SECONDS + if delay: + time.sleep(delay) diff --git a/codex_notifier/cli.py b/codex_notifier/cli.py new file mode 100644 index 0000000..20a76f9 --- /dev/null +++ b/codex_notifier/cli.py @@ -0,0 +1,192 @@ +"""Command dispatch and completion orchestration.""" + +from __future__ import annotations + +import json +import sys +from datetime import datetime +from pathlib import Path +from typing import Any, Callable +from urllib.parse import urlsplit + +from .channels import discord, teams +from .channels.voice import detect_language, speak +from .config import (DEFAULT_DISCORD_MIN_SECONDS, DEFAULT_QUIET_END, + DEFAULT_QUIET_START, DEFAULT_TEAMS_MIN_SECONDS, + DEFAULT_VOICE_MIN_SECONDS, config_bool, config_section, + config_seconds, config_text, load_config) +from .payloads import build_discord_payload, build_payload +from .state import consume_start, record_start, text +from .trace_logging import safe_error, write_trace_log + + +def parse_clock(value: str, default: str) -> int: + try: + hours, minutes = value.strip().split(":", 1) + total = int(hours) * 60 + int(minutes) + if not 0 <= total < 24 * 60: + raise ValueError + return total + except (AttributeError, TypeError, ValueError): + fallback_hours, fallback_minutes = default.split(":", 1) + return int(fallback_hours) * 60 + int(fallback_minutes) + + +def is_quiet_time(moment: datetime, config: dict[str, Any] | None = None) -> bool: + settings = {} if config is None else config + start = parse_clock(config_text(settings, "voice", "quiet_start", DEFAULT_QUIET_START), DEFAULT_QUIET_START) + end = parse_clock(config_text(settings, "voice", "quiet_end", DEFAULT_QUIET_END), DEFAULT_QUIET_END) + if start == end: + return False + current = moment.hour * 60 + moment.minute + return start <= current < end if start < end else current >= start or current < end + + +def notification_channels(duration_seconds: float | None, moment: datetime, + config: dict[str, Any] | None = None) -> tuple[bool, bool, bool]: + settings = load_config() if config is None else config + def due(section: str, default: float, unknown: bool) -> bool: + return (config_bool(settings, section, "notify_when_duration_unknown", unknown) + if duration_seconds is None else duration_seconds >= config_seconds(settings, section, "minimum_seconds", default)) + teams_due = due("teams", DEFAULT_TEAMS_MIN_SECONDS, False) + discord_due = due("discord", DEFAULT_DISCORD_MIN_SECONDS, False) + voice_due = due("voice", DEFAULT_VOICE_MIN_SECONDS, True) + teams_enabled = config_bool(settings, "teams", "enabled", bool(config_text(settings, "teams", "webhook_url"))) + discord_enabled = config_bool(settings, "discord", "enabled", bool(config_text(settings, "discord", "webhook_url"))) + return (teams_enabled and teams_due, discord_enabled and discord_due, + config_bool(settings, "voice", "enabled", True) and voice_due and not is_quiet_time(moment, settings)) + + +def spoken_duration(duration_seconds: float, language: str) -> str: + total_seconds = max(0, round(duration_seconds)) + minutes, seconds = divmod(total_seconds, 60) + hours, minutes = divmod(minutes, 60) + if language == "en": + parts = ([f"{hours} hour" + ("s" if hours != 1 else "")] if hours else []) + if minutes: + parts.append(f"{minutes} minute" + ("s" if minutes != 1 else "")) + if seconds or not parts: + parts.append(f"{seconds} second" + ("s" if seconds != 1 else "")) + return " ".join(parts) + parts = ([f"{hours} hora" + ("s" if hours != 1 else "")] if hours else []) + if minutes: + parts.append(f"{minutes} minuto" + ("s" if minutes != 1 else "")) + if seconds or not parts: + parts.append(f"{seconds} segundo" + ("s" if seconds != 1 else "")) + return " ".join(parts) + + +def voice_message(notification: dict[str, Any], duration_seconds: float | None, + marker: dict[str, Any], language: str) -> str: + cwd = text(notification.get("cwd")) or text(marker.get("cwd")) + project = (Path(cwd).name[:60] if cwd else "actual") + title = text(notification.get("thread-title") or notification.get("thread_title") or notification.get("title")) + extra = (f" Task: {title}." if language == "en" else f" Tarea: {title}.") if title else "" + if language == "en": + base = (f"Codex finished the task for project {project}. The duration could not be determined." + if duration_seconds is None else f"Codex finished the task for project {project} after {spoken_duration(duration_seconds, language)}.") + else: + base = (f"Codex terminó la tarea del proyecto {project}. No se pudo determinar la duración." + if duration_seconds is None else f"Codex terminó la tarea del proyecto {project} después de {spoken_duration(duration_seconds, language)}.") + return base + extra + + +def handle_completion(notification: dict[str, Any], *, now: float | None = None, + load=load_config, consume=consume_start, + speak_fn=speak, send_teams_fn=teams.send, + send_discord_fn=discord.send, log_fn=write_trace_log) -> None: + if notification.get("type") != "agent-turn-complete": + return + duration, marker = consume(notification, now=now) + try: + config = load() + except (OSError, TypeError, json.JSONDecodeError) as exc: + print(f"Configuración inválida: {exc}", file=sys.stderr) + config = {} + moment = datetime.now().astimezone() + chat_id = text(notification.get("thread-id") or notification.get("thread_id")) or text(marker.get("session_id")) + teams_due, discord_due, voice_due = notification_channels(duration, moment, config) + statuses = {"teams": "not_due", "discord": "not_due", "voice": "not_due"} + errors = {"teams": "", "discord": "", "voice": ""} + summary_secret = text(notification.get("last-assistant-message")) + if voice_due: + try: + voice_config = config_section(config, "voice") + language = detect_language(text(notification.get("last-assistant-message")), text(voice_config.get("language")) or "auto") + preferred = text(voice_config.get("spanish_voice" if language == "es" else "english_voice")) + speak_fn(voice_message(notification, duration, marker, language), language=language, + preferred_voice=preferred, voice_config=voice_config) + statuses["voice"] = "sent" + except Exception as exc: + statuses["voice"] = "failed" + errors["voice"] = safe_error(exc, summary_secret) + print(f"No se pudo reproducir el aviso de voz: {errors['voice']}", file=sys.stderr) + for name, due_flag, sender, builder in (("teams", teams_due, send_teams_fn, build_payload), + ("discord", discord_due, send_discord_fn, build_discord_payload)): + if not due_flag: + continue + webhook_url = config_text(config, name, "webhook_url") + if webhook_url and urlsplit(webhook_url).scheme == "https": + try: + sender(webhook_url, builder(notification, duration, marker, config)) + statuses[name] = "sent" + except Exception as exc: + statuses[name] = "failed" + errors[name] = safe_error(exc, webhook_url, summary_secret) + print(f"No se pudo enviar la notificación a {name.title()}: {errors[name]}", file=sys.stderr) + else: + statuses[name] = "failed" + errors[name] = f"El webhook de {name.title()} debe usar HTTPS." if webhook_url else f"No hay un webhook de {name.title()} configurado." + print(errors[name], file=sys.stderr) + try: + log_fn(config, moment=moment, chat_id=chat_id, duration_seconds=duration, + teams_status=statuses["teams"], discord_status=statuses["discord"], + voice_status=statuses["voice"], teams_error=errors["teams"], + discord_error=errors["discord"], voice_error=errors["voice"]) + except OSError as exc: + print(f"No se pudo escribir el log del notificador: {exc}", file=sys.stderr) + + +def _json_stdin() -> dict[str, Any]: + payload = json.load(sys.stdin) + if not isinstance(payload, dict): + raise TypeError("el payload debe ser un objeto JSON") + return payload + + +def _json_argument(value: str) -> dict[str, Any]: + payload = json.loads(value) + if not isinstance(payload, dict): + raise TypeError("el payload debe ser un objeto JSON") + return payload + + +def main(argv: list[str] | None = None, *, speak_fn=speak) -> int: + arguments = sys.argv[1:] if argv is None else argv + command = arguments[0] if arguments else "" + try: + if command == "record-start" and len(arguments) == 1: + record_start(_json_stdin()) + return 0 + if command == "notify" and len(arguments) == 2: + handle_completion(_json_argument(arguments[1])) + return 0 + if command == "voice-test" and len(arguments) in {1, 2}: + requested = arguments[1] if len(arguments) == 2 else "es" + if requested.lower() not in {"es", "en"}: + raise ValueError("voice-test requiere es o en") + config = load_config() + language = detect_language("", requested) + voice_config = config_section(config, "voice") + preferred = text(voice_config.get("spanish_voice" if language == "es" else "english_voice")) + message = "The Codex voice notification is working." if language == "en" else "La notificación por voz de Codex está funcionando." + speak_fn(message, language=language, preferred_voice=preferred, voice_config=voice_config) + return 0 + except (json.JSONDecodeError, TypeError, ValueError) as exc: + print(f"Payload o sintaxis inválida: {exc}", file=sys.stderr) + return 2 + except Exception as exc: + print(f"Error del notificador: {exc}", file=sys.stderr) + return 1 if command == "voice-test" else 0 + print("Uso: notifier.py record-start | notify '' | voice-test [es|en]", file=sys.stderr) + return 2 diff --git a/codex_notifier/config.py b/codex_notifier/config.py new file mode 100644 index 0000000..e63ee33 --- /dev/null +++ b/codex_notifier/config.py @@ -0,0 +1,109 @@ +"""Configuration paths and conservative typed accessors.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +from typing import Any + +CONFIG_PATH_ENV = "CODEX_NOTIFIER_CONFIG" +STATE_DIR_ENV = "CODEX_NOTIFIER_STATE_DIR" + +DEFAULT_TEAMS_MIN_SECONDS = 300.0 +DEFAULT_DISCORD_MIN_SECONDS = 300.0 +DEFAULT_VOICE_MIN_SECONDS = 30.0 +DEFAULT_QUIET_START = "23:00" +DEFAULT_QUIET_END = "07:00" +DEFAULT_SUMMARY_MAX_CHARS = 1800 +DEFAULT_SUMMARY_TAIL_CHARS = 600 +MIN_SUMMARY_MAX_CHARS = 100 +MAX_SUMMARY_MAX_CHARS = 6000 +MAX_DISCORD_SUMMARY_MAX_CHARS = 4096 +DEFAULT_LOG_RETENTION_DAYS = 7 +MAX_LOG_RETENTION_DAYS = 365 + + +def default_config_path(*, platform_name: str | None = None, + environment: dict[str, str] | None = None, + home: Path | None = None) -> Path: + platform = os.name if platform_name is None else platform_name + env = os.environ if environment is None else environment + user_home = Path.home() if home is None else home + if platform == "nt": + local_app_data = env.get("LOCALAPPDATA", "").strip() + if local_app_data: + return Path(local_app_data) / "CodexNotifier" / "config.json" + return user_home / "AppData" / "Local" / "CodexNotifier" / "config.json" + xdg = env.get("XDG_CONFIG_HOME", "").strip() + return (Path(xdg) if xdg else user_home / ".config") / "codex-notifier" / "config.json" + + +def config_path() -> Path: + configured = os.environ.get(CONFIG_PATH_ENV, "").strip() + return Path(configured).expanduser() if configured else default_config_path() + + +def load_config() -> dict[str, Any]: + path = config_path() + if not path.exists(): + return {} + content = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(content, dict): + raise TypeError(f"{path} debe contener un objeto JSON") + return content + + +def config_section(config: dict[str, Any], name: str) -> dict[str, Any]: + section = config.get(name, {}) + return section if isinstance(section, dict) else {} + + +def config_bool(config: dict[str, Any], section_name: str, key: str, default: bool) -> bool: + value = config_section(config, section_name).get(key) + return value if isinstance(value, bool) else default + + +def config_seconds(config: dict[str, Any], section_name: str, key: str, default: float) -> float: + value = config_section(config, section_name).get(key) + if isinstance(value, (int, float)) and not isinstance(value, bool): + return max(0.0, float(value)) + return default + + +def config_int(config: dict[str, Any], section_name: str, key: str, default: int, + *, minimum: int = 0, maximum: int | None = None) -> int: + value = config_section(config, section_name).get(key) + if not isinstance(value, (int, float)) or isinstance(value, bool): + value = default + result = max(minimum, int(value)) + return min(result, maximum) if maximum is not None else result + + +def config_text(config: dict[str, Any], section_name: str, key: str, default: str = "") -> str: + value = config_section(config, section_name).get(key) + return value.strip() if isinstance(value, str) else default + + +def default_state_dir(*, platform_name: str | None = None, + environment: dict[str, str] | None = None, + home: Path | None = None) -> Path: + platform = os.name if platform_name is None else platform_name + env = os.environ if environment is None else environment + user_home = Path.home() if home is None else home + if platform == "nt": + local_app_data = env.get("LOCALAPPDATA", "").strip() + if local_app_data: + return Path(local_app_data) / "CodexNotifier" + return user_home / "AppData" / "Local" / "CodexNotifier" + xdg = env.get("XDG_STATE_HOME", "").strip() + return (Path(xdg) if xdg else user_home / ".local" / "state") / "codex-notifier" + + +def state_dir() -> Path: + configured = os.environ.get(STATE_DIR_ENV, "").strip() + return Path(configured).expanduser() if configured else default_state_dir() + + +def log_dir() -> Path: + return state_dir() / "logs" diff --git a/codex_notifier/files.py b/codex_notifier/files.py new file mode 100644 index 0000000..808ecf6 --- /dev/null +++ b/codex_notifier/files.py @@ -0,0 +1,88 @@ +"""Small filesystem helpers used by installation and notifier state.""" + +from __future__ import annotations + +import os +import tempfile +from pathlib import Path + + +def secure_directory(path: Path, *, platform_name: str | None = None) -> Path: + """Create a private POSIX directory; Windows ACL behavior is unchanged.""" + path.mkdir(parents=True, exist_ok=True) + platform = os.name if platform_name is None else platform_name + if platform != "nt": + path.chmod(0o700) + return path + + +def secure_file(path: Path, mode: int = 0o600, *, platform_name: str | None = None) -> None: + platform = os.name if platform_name is None else platform_name + if platform != "nt": + path.chmod(mode) + + +def atomic_write_bytes( + path: Path, + data: bytes, + *, + mode: int | None = None, + platform_name: str | None = None, +) -> None: + """Write in the destination directory and replace it only after close.""" + path.parent.mkdir(parents=True, exist_ok=True) + temporary_name: str | None = None + try: + descriptor, temporary_name = tempfile.mkstemp( + prefix=f".{path.name}.", suffix=".tmp", dir=path.parent + ) + temporary = Path(temporary_name) + if mode is not None: + secure_file(temporary, mode, platform_name=platform_name) + with os.fdopen(descriptor, "wb") as stream: + stream.write(data) + stream.flush() + os.fsync(stream.fileno()) + if mode is not None: + secure_file(temporary, mode, platform_name=platform_name) + os.replace(temporary, path) + if mode is not None: + secure_file(path, mode, platform_name=platform_name) + temporary_name = None + finally: + if temporary_name: + try: + Path(temporary_name).unlink(missing_ok=True) + except OSError: + pass + + +def atomic_write_text( + path: Path, + text: str, + *, + mode: int | None = None, + platform_name: str | None = None, +) -> None: + atomic_write_bytes( + path, + text.encode("utf-8"), + mode=mode, + platform_name=platform_name, + ) + + +def atomic_copy( + source: Path, + destination: Path, + *, + mode: int | None = None, + platform_name: str | None = None, +) -> None: + with source.open("rb") as stream: + atomic_write_bytes( + destination, + stream.read(), + mode=mode, + platform_name=platform_name, + ) diff --git a/codex_notifier/payloads.py b/codex_notifier/payloads.py new file mode 100644 index 0000000..11a2f53 --- /dev/null +++ b/codex_notifier/payloads.py @@ -0,0 +1,133 @@ +"""Webhook payload construction without retaining assistant content.""" + +from __future__ import annotations + +from datetime import datetime +from pathlib import Path +from typing import Any + +from .config import (DEFAULT_SUMMARY_MAX_CHARS, DEFAULT_SUMMARY_TAIL_CHARS, + MAX_DISCORD_SUMMARY_MAX_CHARS, MAX_SUMMARY_MAX_CHARS, + MIN_SUMMARY_MAX_CHARS, config_bool, config_int) +from .state import text + +MAX_TITLE_CHARS = 120 + + +def truncate(value: str, limit: int) -> str: + if len(value) <= limit: + return value + return value[: limit - 1].rstrip() + "…" + + +def summary_excerpt(value: str, limit: int, tail_chars: int) -> str: + if len(value) <= limit: + return value + omitted = len(value) + for _ in range(10): + omitted_text = f"{omitted:,}".replace(",", ".") + separator = f"\n\n[… {omitted_text} caracteres omitidos …]\n\n" + available = limit - len(separator) + if available < 2: + return truncate(value, limit) + tail_length = min(max(0, tail_chars), available - 1) + head_length = available - tail_length + new_omitted = len(value) - head_length - tail_length + if new_omitted == omitted: + tail = value[-tail_length:] if tail_length else "" + return value[:head_length] + separator + tail + omitted = new_omitted + return truncate(value, limit) + + +def duration_text(duration_seconds: float | None) -> str: + if duration_seconds is None: + return "No disponible" + total_seconds = max(0, round(duration_seconds)) + minutes, seconds = divmod(total_seconds, 60) + hours, minutes = divmod(minutes, 60) + parts: list[str] = [] + if hours: + parts.append(f"{hours} h") + if minutes: + parts.append(f"{minutes} min") + if seconds or not parts: + parts.append(f"{seconds} s") + return " ".join(parts) + + +def chat_title(notification: dict[str, Any]) -> str: + for key in ("thread-title", "thread_title", "title"): + title = text(notification.get(key)) + if title: + return truncate(title, MAX_TITLE_CHARS) + return "" + + +def build_payload(notification: dict[str, Any], duration_seconds: float | None, + marker: dict[str, Any], config: dict[str, Any] | None = None) -> dict[str, Any]: + settings = {} if config is None else config + completed_at = datetime.now().astimezone().strftime("%d/%m/%Y %H:%M %Z") + chat_id = text(notification.get("thread-id")) or text(marker.get("session_id")) + title = chat_title(notification) + cwd = text(notification.get("cwd")) or text(marker.get("cwd")) + facts = [ + {"title": "Estado", "value": "Completado"}, + {"title": "Duración", "value": duration_text(duration_seconds)}, + ] + if chat_id: + facts.append({"title": "ID del chat", "value": chat_id}) + if title: + facts.append({"title": "Título", "value": title}) + if cwd: + facts.append({"title": "Proyecto", "value": Path(cwd).name or cwd}) + facts.append({"title": "Fecha y hora", "value": completed_at}) + body: list[dict[str, Any]] = [ + {"type": "TextBlock", "text": "✅ Turno de Codex completado", "size": "Large", + "weight": "Bolder", "color": "Good", "wrap": True}, + {"type": "FactSet", "facts": facts}, + ] + if config_bool(settings, "teams", "include_summary", True): + maximum = config_int(settings, "teams", "summary_max_chars", + DEFAULT_SUMMARY_MAX_CHARS, minimum=MIN_SUMMARY_MAX_CHARS, + maximum=MAX_SUMMARY_MAX_CHARS) + tail = config_int(settings, "teams", "summary_tail_chars", DEFAULT_SUMMARY_TAIL_CHARS) + summary = summary_excerpt(text(notification.get("last-assistant-message")) or + "El turno terminó sin un mensaje final para resumir.", maximum, tail) + body.extend([ + {"type": "TextBlock", "text": "Resumen de lo realizado", "weight": "Bolder", + "wrap": True, "separator": True}, + {"type": "TextBlock", "text": summary, "wrap": True}, + ]) + return {"type": "message", "attachments": [{ + "contentType": "application/vnd.microsoft.card.adaptive", "contentUrl": None, + "content": {"$schema": "http://adaptivecards.io/schemas/adaptive-card.json", + "type": "AdaptiveCard", "version": "1.2", "msteams": {"width": "Full"}, + "body": body}, + }]} + + +def build_discord_payload(notification: dict[str, Any], duration_seconds: float | None, + marker: dict[str, Any], config: dict[str, Any] | None = None) -> dict[str, Any]: + settings = {} if config is None else config + chat_id = text(notification.get("thread-id")) or text(marker.get("session_id")) + title = chat_title(notification) + cwd = text(notification.get("cwd")) or text(marker.get("cwd")) + fields = [{"name": "Estado", "value": "Completado", "inline": True}, + {"name": "Duración", "value": duration_text(duration_seconds), "inline": True}] + if cwd: + fields.append({"name": "Proyecto", "value": truncate(Path(cwd).name or cwd, 100), "inline": True}) + if title: + fields.append({"name": "Título", "value": title, "inline": False}) + if chat_id: + fields.append({"name": "ID del chat", "value": truncate(chat_id, 200), "inline": False}) + embed: dict[str, Any] = {"title": "✅ Turno de Codex completado", "color": 3061878, + "fields": fields, + "timestamp": datetime.now().astimezone().isoformat(timespec="seconds")} + if config_bool(settings, "discord", "include_summary", True): + maximum = config_int(settings, "discord", "summary_max_chars", + DEFAULT_SUMMARY_MAX_CHARS, minimum=MIN_SUMMARY_MAX_CHARS, + maximum=MAX_DISCORD_SUMMARY_MAX_CHARS) + embed["description"] = truncate(text(notification.get("last-assistant-message")) or + "El turno terminó sin un mensaje final para resumir.", maximum) + return {"username": "Codex Notifier", "allowed_mentions": {"parse": []}, "embeds": [embed]} diff --git a/codex_notifier/state.py b/codex_notifier/state.py new file mode 100644 index 0000000..30d0e43 --- /dev/null +++ b/codex_notifier/state.py @@ -0,0 +1,77 @@ +"""Private turn-marker lifecycle.""" + +from __future__ import annotations + +import hashlib +import json +import os +import time +from pathlib import Path +from typing import Any + +from .config import state_dir +from .files import atomic_write_text, secure_directory + +STALE_MARKER_SECONDS = 48 * 60 * 60 + + +def text(value: Any) -> str: + if value is None: + return "" + if isinstance(value, str): + return value.strip() + return json.dumps(value, ensure_ascii=False).strip() + + +def marker_path(turn_id: str) -> Path: + digest = hashlib.sha256(turn_id.encode("utf-8")).hexdigest() + return state_dir() / "turns" / f"{digest}.json" + + +def cleanup_stale_markers(now: float) -> None: + marker_dir = state_dir() / "turns" + if not marker_dir.exists(): + return + for marker in marker_dir.glob("*.json"): + try: + if now - marker.stat().st_mtime > STALE_MARKER_SECONDS: + marker.unlink(missing_ok=True) + except OSError: + continue + + +def record_start(payload: dict[str, Any], *, now: float | None = None) -> bool: + if payload.get("hook_event_name") != "UserPromptSubmit": + return False + turn_id = text(payload.get("turn_id")) + if not turn_id: + return False + timestamp = time.time() if now is None else now + marker = marker_path(turn_id) + secure_directory(state_dir()) + secure_directory(marker.parent) + content = { + "started_at": timestamp, + "session_id": text(payload.get("session_id")), + "turn_id": turn_id, + "cwd": text(payload.get("cwd")), + } + atomic_write_text(marker, json.dumps(content, ensure_ascii=False), mode=0o600) + cleanup_stale_markers(timestamp) + return True + + +def consume_start(notification: dict[str, Any], *, now: float | None = None) -> tuple[float | None, dict[str, Any]]: + turn_id = text(notification.get("turn-id") or notification.get("turn_id")) + if not turn_id: + return None, {} + marker = marker_path(turn_id) + try: + content = json.loads(marker.read_text(encoding="utf-8")) + started_at = float(content["started_at"]) + except (OSError, ValueError, TypeError, KeyError, json.JSONDecodeError): + return None, {} + finally: + marker.unlink(missing_ok=True) + finished_at = time.time() if now is None else now + return max(0.0, finished_at - started_at), content diff --git a/codex_notifier/trace_logging.py b/codex_notifier/trace_logging.py new file mode 100644 index 0000000..c5050af --- /dev/null +++ b/codex_notifier/trace_logging.py @@ -0,0 +1,78 @@ +"""Privacy-safe daily JSONL trace logging.""" + +from __future__ import annotations + +import json +import os +import re +from datetime import datetime, timedelta +from typing import Any + +from .config import (DEFAULT_LOG_RETENTION_DAYS, MAX_LOG_RETENTION_DAYS, + config_bool, config_int, log_dir, state_dir) +from .files import secure_directory, secure_file + +MAX_LOG_ERROR_CHARS = 300 + + +def truncate(value: str, limit: int) -> str: + if len(value) <= limit: + return value + return value[: limit - 1].rstrip() + "…" + + +def safe_error(exc: Exception, *secrets: str) -> str: + message = f"{type(exc).__name__}: {exc}" + for secret in secrets: + if secret: + message = message.replace(secret, "[redacted]") + # Error strings from subprocesses can contain configured model paths. + message = re.sub(r"[A-Za-z]:\\[^\s,'\]]+", "[path redacted]", message) + message = re.sub(r"(? None: + oldest = moment.date() - timedelta(days=retention_days - 1) + for path in directory.glob("notifier-????-??-??.jsonl"): + try: + file_date = datetime.strptime(path.stem[9:], "%Y-%m-%d").date() + if file_date < oldest: + path.unlink(missing_ok=True) + except (OSError, ValueError): + continue + + +def write_trace_log(config: dict[str, Any], *, moment: datetime, chat_id: str, + duration_seconds: float | None, teams_status: str, + discord_status: str, voice_status: str, + teams_error: str = "", discord_error: str = "", + voice_error: str = "") -> None: + if not config_bool(config, "logging", "enabled", True): + return + retention_days = config_int(config, "logging", "retention_days", + DEFAULT_LOG_RETENTION_DAYS, minimum=1, + maximum=MAX_LOG_RETENTION_DAYS) + secure_directory(state_dir()) + directory = secure_directory(log_dir()) + cleanup_old_logs(directory, moment, retention_days) + entry: dict[str, Any] = { + "timestamp": moment.isoformat(timespec="seconds"), + "chat_id": chat_id, + "duration_seconds": round(duration_seconds, 3) if duration_seconds is not None else None, + "teams_status": teams_status, + "discord_status": discord_status, + "voice_status": voice_status, + } + if teams_error: + entry["teams_error"] = teams_error + if discord_error: + entry["discord_error"] = discord_error + if voice_error: + entry["voice_error"] = voice_error + path = directory / f"notifier-{moment.date().isoformat()}.jsonl" + with path.open("a", encoding="utf-8", newline="\n") as stream: + stream.write(json.dumps(entry, ensure_ascii=False) + "\n") + stream.flush() + os.fsync(stream.fileno()) + secure_file(path) diff --git a/config.example.json b/config.example.json index cd39049..65da417 100644 --- a/config.example.json +++ b/config.example.json @@ -3,6 +3,7 @@ "enabled": false, "minimum_seconds": 300, "notify_when_duration_unknown": false, + "include_summary": true, "summary_max_chars": 1800, "summary_tail_chars": 600, "webhook_url": "" @@ -11,6 +12,7 @@ "enabled": false, "minimum_seconds": 300, "notify_when_duration_unknown": false, + "include_summary": true, "summary_max_chars": 1800, "webhook_url": "" }, diff --git a/install.py b/install.py index c3f6c43..7a0f715 100644 --- a/install.py +++ b/install.py @@ -14,6 +14,8 @@ from pathlib import Path from typing import Callable, Sequence +from codex_notifier.files import atomic_copy, atomic_write_text + PROJECT_ROOT = Path(__file__).resolve().parent NOTIFIER_PATH = PROJECT_ROOT / "notifier.py" @@ -136,26 +138,82 @@ def _backup(path: Path, timestamp: str) -> Path: return backup +def _strip_toml_comment(line: str) -> str: + """Return the part before an unquoted TOML comment marker.""" + quote = "" + escaped = False + for index, character in enumerate(line): + if quote == '"' and escaped: + escaped = False + continue + if quote == '"' and character == "\\": + escaped = True + continue + if character in {'"', "'"}: + if not quote: + quote = character + elif quote == character: + quote = "" + elif character == "#" and not quote: + return line[:index] + return line + + +def _is_toml_table_header(line: str) -> bool: + return _strip_toml_comment(line).strip().startswith("[") + + +def _inline_comment(line: str) -> str: + content = line.rstrip("\r\n") + stripped = _strip_toml_comment(content) + return content[len(stripped.rstrip()):] + + +def update_root_notify(existing: str, notify_line: str) -> str: + """Update only a bare root-level ``notify`` before the first TOML table.""" + lines = existing.splitlines(keepends=True) + first_table = next((index for index, line in enumerate(lines) if _is_toml_table_header(line)), len(lines)) + root_index = next( + (index for index, line in enumerate(lines[:first_table]) + if re.match(r"^[ \t]*notify[ \t]*=", _strip_toml_comment(line))), + None, + ) + if root_index is not None: + original = lines[root_index] + ending = "\r\n" if original.endswith("\r\n") else "\n" if original.endswith("\n") else "" + lines[root_index] = notify_line + _inline_comment(original) + ending + return "".join(lines) + + newline = "\r\n" if "\r\n" in existing else "\n" + insertion = notify_line + newline + if first_table < len(lines): + lines.insert(first_table, insertion) + return "".join(lines) + return insertion + existing + + def update_codex_config( config_path: Path, python_command: Sequence[str], notifier_path: Path, *, timestamp: str, + platform_name: str | None = None, ) -> tuple[bool, Path | None]: config_path.parent.mkdir(parents=True, exist_ok=True) existing = config_path.read_text(encoding="utf-8") if config_path.exists() else "" notify_arguments = [*python_command, str(notifier_path), "notify"] notify_line = "notify = " + json.dumps(notify_arguments, ensure_ascii=False) - pattern = re.compile(r"^[ \t]*notify[ \t]*=.*$", re.MULTILINE) - if pattern.search(existing): - updated = pattern.sub(lambda _: notify_line, existing, count=1) - else: - updated = notify_line + os.linesep + existing + updated = update_root_notify(existing, notify_line) if updated == existing: return False, None backup = _backup(config_path, timestamp) if config_path.exists() else None - config_path.write_text(updated, encoding="utf-8", newline="") + atomic_write_text( + config_path, + updated, + mode=0o600, + platform_name=os.name if platform_name is None else platform_name, + ) return True, backup @@ -226,7 +284,7 @@ def update_hooks( if updated == existing: return False, None backup = _backup(hooks_path, timestamp) if hooks_path.exists() else None - hooks_path.write_text(updated, encoding="utf-8", newline="") + atomic_write_text(hooks_path, updated, mode=0o600, platform_name=platform_name) return True, backup @@ -242,9 +300,7 @@ def install_local_config( return False, None destination.parent.mkdir(parents=True, exist_ok=True) backup = _backup(destination, timestamp) if destination.exists() else None - shutil.copyfile(example_path, destination) - if platform_name != "nt": - destination.chmod(0o600) + atomic_copy(example_path, destination, mode=0o600, platform_name=platform_name) return True, backup @@ -266,6 +322,7 @@ def install( python_command, notifier_path, timestamp=stamp, + platform_name=platform, ) hooks_changed, hooks_backup = update_hooks( codex_home / "hooks.json", diff --git a/notifier.py b/notifier.py index bd7e6a3..dbb4310 100644 --- a/notifier.py +++ b/notifier.py @@ -1,734 +1,163 @@ -"""Filtered Codex completion notifications for webhooks and local speech. +"""Compatibility entry point for Codex Notifier. -The script has two entry points: - -* ``record-start`` reads a Codex UserPromptSubmit hook payload from stdin. -* ``notify`` reads the agent-turn-complete payload appended by Codex to the - configured ``notify`` command. - -Only Python's standard library is required. Windows speech uses the built-in -SAPI voice through Windows PowerShell. +The implementation lives in :mod:`codex_notifier`; these aliases keep the +historic script, helper names, and integration points working. """ from __future__ import annotations -import base64 -import hashlib -import json import os -import re -import shutil -import subprocess import sys -import tempfile -import time -from datetime import datetime, timedelta +from datetime import datetime from pathlib import Path from typing import Any -from urllib.error import HTTPError, URLError -from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit -from urllib.request import Request, urlopen - - -STATE_DIR_ENV = "CODEX_NOTIFIER_STATE_DIR" -CONFIG_PATH_ENV = "CODEX_NOTIFIER_CONFIG" - -DEFAULT_TEAMS_MIN_SECONDS = 300.0 -DEFAULT_DISCORD_MIN_SECONDS = 300.0 -DEFAULT_VOICE_MIN_SECONDS = 30.0 -DEFAULT_QUIET_START = "23:00" -DEFAULT_QUIET_END = "07:00" -STALE_MARKER_SECONDS = 48 * 60 * 60 -MAX_TITLE_CHARS = 120 -DEFAULT_SUMMARY_MAX_CHARS = 1800 -DEFAULT_SUMMARY_TAIL_CHARS = 600 -MIN_SUMMARY_MAX_CHARS = 100 -MAX_SUMMARY_MAX_CHARS = 6000 -MAX_DISCORD_SUMMARY_MAX_CHARS = 4096 -DEFAULT_LOG_RETENTION_DAYS = 7 -MAX_LOG_RETENTION_DAYS = 365 -MAX_LOG_ERROR_CHARS = 300 -SPEECH_TIMEOUT_SECONDS = 30 - -SPANISH_WORDS = { - "al", - "cambios", - "completado", - "con", - "corregido", - "de", - "el", - "en", - "esta", - "este", - "fue", - "la", - "las", - "listo", - "los", - "para", - "por", - "pruebas", - "que", - "se", - "sin", - "una", - "y", - "ya", -} -ENGLISH_WORDS = { - "a", - "and", - "changes", - "completed", - "done", - "for", - "from", - "has", - "fixed", - "in", - "is", - "of", - "on", - "that", - "the", - "tests", - "this", - "to", - "was", - "ready", - "with", - "without", -} + +from codex_notifier import cli as _cli +from codex_notifier import config as _config +from codex_notifier import payloads as _payloads +from codex_notifier import state as _state +from codex_notifier import trace_logging as _trace +from codex_notifier.channels import voice as _voice +from codex_notifier.channels.webhook import send_webhook_notification as _transport + +# These imports remain module attributes for compatibility with callers that +# patched notifier.subprocess/shutil/tempfile in the original monolith. +import base64 # noqa: F401 +import re # noqa: F401 +import shutil # noqa: F401 +import subprocess # noqa: F401 +import tempfile # noqa: F401 + +STATE_DIR_ENV = _config.STATE_DIR_ENV +CONFIG_PATH_ENV = _config.CONFIG_PATH_ENV +DEFAULT_TEAMS_MIN_SECONDS = _config.DEFAULT_TEAMS_MIN_SECONDS +DEFAULT_DISCORD_MIN_SECONDS = _config.DEFAULT_DISCORD_MIN_SECONDS +DEFAULT_VOICE_MIN_SECONDS = _config.DEFAULT_VOICE_MIN_SECONDS +DEFAULT_QUIET_START = _config.DEFAULT_QUIET_START +DEFAULT_QUIET_END = _config.DEFAULT_QUIET_END +STALE_MARKER_SECONDS = _state.STALE_MARKER_SECONDS +MAX_TITLE_CHARS = _payloads.MAX_TITLE_CHARS +DEFAULT_SUMMARY_MAX_CHARS = _config.DEFAULT_SUMMARY_MAX_CHARS +DEFAULT_SUMMARY_TAIL_CHARS = _config.DEFAULT_SUMMARY_TAIL_CHARS +MIN_SUMMARY_MAX_CHARS = _config.MIN_SUMMARY_MAX_CHARS +MAX_SUMMARY_MAX_CHARS = _config.MAX_SUMMARY_MAX_CHARS +MAX_DISCORD_SUMMARY_MAX_CHARS = _config.MAX_DISCORD_SUMMARY_MAX_CHARS +DEFAULT_LOG_RETENTION_DAYS = _config.DEFAULT_LOG_RETENTION_DAYS +MAX_LOG_RETENTION_DAYS = _config.MAX_LOG_RETENTION_DAYS +MAX_LOG_ERROR_CHARS = _trace.MAX_LOG_ERROR_CHARS +SPEECH_TIMEOUT_SECONDS = _voice.SPEECH_TIMEOUT_SECONDS def _text(value: Any) -> str: - if value is None: - return "" - if isinstance(value, str): - return value.strip() - return json.dumps(value, ensure_ascii=False).strip() + return _state.text(value) def _truncate(value: str, limit: int) -> str: - if len(value) <= limit: - return value - return value[: limit - 1].rstrip() + "…" + return _payloads.truncate(value, limit) def _summary_excerpt(value: str, limit: int, tail_chars: int) -> str: - """Keep the beginning and end of a summary within a character limit.""" - if len(value) <= limit: - return value - - omitted = len(value) - for _ in range(10): - omitted_text = f"{omitted:,}".replace(",", ".") - separator = f"\n\n[… {omitted_text} caracteres omitidos …]\n\n" - available = limit - len(separator) - if available < 2: - return _truncate(value, limit) - tail_length = min(max(0, tail_chars), available - 1) - head_length = available - tail_length - new_omitted = len(value) - head_length - tail_length - if new_omitted == omitted: - tail = value[-tail_length:] if tail_length else "" - return value[:head_length] + separator + tail - omitted = new_omitted - - return _truncate(value, limit) - - -def default_config_path( - *, - platform_name: str | None = None, - environment: dict[str, str] | None = None, - home: Path | None = None, -) -> Path: - platform = os.name if platform_name is None else platform_name - env = os.environ if environment is None else environment - user_home = Path.home() if home is None else home - if platform == "nt": - local_app_data = env.get("LOCALAPPDATA", "").strip() - if local_app_data: - return Path(local_app_data) / "CodexNotifier" / "config.json" - return user_home / "AppData" / "Local" / "CodexNotifier" / "config.json" - xdg_config_home = env.get("XDG_CONFIG_HOME", "").strip() - config_home = Path(xdg_config_home) if xdg_config_home else user_home / ".config" - return config_home / "codex-notifier" / "config.json" + return _payloads.summary_excerpt(value, limit, tail_chars) + + +def default_config_path(**kwargs: Any) -> Path: + return _config.default_config_path(**kwargs) def config_path() -> Path: - configured = os.environ.get(CONFIG_PATH_ENV, "").strip() - if configured: - return Path(configured).expanduser() - return default_config_path() + return _config.config_path() def load_config() -> dict[str, Any]: - path = config_path() - if not path.exists(): - return {} - content = json.loads(path.read_text(encoding="utf-8")) - if not isinstance(content, dict): - raise TypeError(f"{path} debe contener un objeto JSON") - return content + return _config.load_config() def _config_section(config: dict[str, Any], name: str) -> dict[str, Any]: - section = config.get(name, {}) - return section if isinstance(section, dict) else {} - - -def _config_bool( - config: dict[str, Any], - section_name: str, - key: str, - default: bool, -) -> bool: - value = _config_section(config, section_name).get(key) - return value if isinstance(value, bool) else default - - -def _config_seconds( - config: dict[str, Any], section_name: str, key: str, default: float -) -> float: - value = _config_section(config, section_name).get(key) - if isinstance(value, (int, float)) and not isinstance(value, bool): - return max(0.0, float(value)) - return default - - -def _config_int( - config: dict[str, Any], - section_name: str, - key: str, - default: int, - *, - minimum: int = 0, - maximum: int | None = None, -) -> int: - value = _config_section(config, section_name).get(key) - if not isinstance(value, (int, float)) or isinstance(value, bool): - value = default - result = max(minimum, int(value)) - return min(result, maximum) if maximum is not None else result - - -def _config_text( - config: dict[str, Any], - section_name: str, - key: str, - default: str = "", -) -> str: - value = _config_section(config, section_name).get(key) - return value.strip() if isinstance(value, str) else default - - -def default_state_dir( - *, - platform_name: str | None = None, - environment: dict[str, str] | None = None, - home: Path | None = None, -) -> Path: - platform = os.name if platform_name is None else platform_name - env = os.environ if environment is None else environment - user_home = Path.home() if home is None else home - if platform == "nt": - local_app_data = env.get("LOCALAPPDATA", "").strip() - if local_app_data: - return Path(local_app_data) / "CodexNotifier" - return user_home / "AppData" / "Local" / "CodexNotifier" - xdg_state_home = env.get("XDG_STATE_HOME", "").strip() - state_home = Path(xdg_state_home) if xdg_state_home else user_home / ".local" / "state" - return state_home / "codex-notifier" + return _config.config_section(config, name) + + +def _config_bool(config: dict[str, Any], section_name: str, key: str, default: bool) -> bool: + return _config.config_bool(config, section_name, key, default) + + +def _config_seconds(config: dict[str, Any], section_name: str, key: str, default: float) -> float: + return _config.config_seconds(config, section_name, key, default) + + +def _config_int(config: dict[str, Any], section_name: str, key: str, default: int, *, minimum: int = 0, maximum: int | None = None) -> int: + return _config.config_int(config, section_name, key, default, minimum=minimum, maximum=maximum) + + +def _config_text(config: dict[str, Any], section_name: str, key: str, default: str = "") -> str: + return _config.config_text(config, section_name, key, default) + + +def default_state_dir(**kwargs: Any) -> Path: + return _config.default_state_dir(**kwargs) def state_dir() -> Path: - configured = os.environ.get(STATE_DIR_ENV, "").strip() - if configured: - return Path(configured).expanduser() - return default_state_dir() + return _config.state_dir() def log_dir() -> Path: - return state_dir() / "logs" + return _config.log_dir() def _safe_error(exc: Exception, *secrets: str) -> str: - message = f"{type(exc).__name__}: {exc}" - for secret in secrets: - if secret: - message = message.replace(secret, "[redacted]") - return _truncate(message, MAX_LOG_ERROR_CHARS) - - -def _cleanup_old_logs(directory: Path, moment: datetime, retention_days: int) -> None: - oldest = moment.date() - timedelta(days=retention_days - 1) - for path in directory.glob("notifier-????-??-??.jsonl"): - try: - file_date = datetime.strptime(path.stem[9:], "%Y-%m-%d").date() - if file_date < oldest: - path.unlink(missing_ok=True) - except (OSError, ValueError): - continue - - -def write_trace_log( - config: dict[str, Any], - *, - moment: datetime, - chat_id: str, - duration_seconds: float | None, - teams_status: str, - discord_status: str, - voice_status: str, - teams_error: str = "", - discord_error: str = "", - voice_error: str = "", -) -> None: - if not _config_bool(config, "logging", "enabled", True): - return - retention_days = _config_int( - config, - "logging", - "retention_days", - DEFAULT_LOG_RETENTION_DAYS, - minimum=1, - maximum=MAX_LOG_RETENTION_DAYS, - ) - directory = log_dir() - directory.mkdir(parents=True, exist_ok=True) - _cleanup_old_logs(directory, moment, retention_days) - entry: dict[str, Any] = { - "timestamp": moment.isoformat(timespec="seconds"), - "chat_id": chat_id, - "duration_seconds": ( - round(duration_seconds, 3) if duration_seconds is not None else None - ), - "teams_status": teams_status, - "discord_status": discord_status, - "voice_status": voice_status, - } - if teams_error: - entry["teams_error"] = teams_error - if discord_error: - entry["discord_error"] = discord_error - if voice_error: - entry["voice_error"] = voice_error - path = directory / f"notifier-{moment.date().isoformat()}.jsonl" - with path.open("a", encoding="utf-8", newline="\n") as stream: - stream.write(json.dumps(entry, ensure_ascii=False) + "\n") - if os.name != "nt": - path.chmod(0o600) + return _trace.safe_error(exc, *secrets) -def _marker_path(turn_id: str) -> Path: - digest = hashlib.sha256(turn_id.encode("utf-8")).hexdigest() - return state_dir() / "turns" / f"{digest}.json" +def write_trace_log(config: dict[str, Any], **kwargs: Any) -> None: + return _trace.write_trace_log(config, **kwargs) -def _cleanup_stale_markers(now: float) -> None: - marker_dir = state_dir() / "turns" - if not marker_dir.exists(): - return - for marker in marker_dir.glob("*.json"): - try: - if now - marker.stat().st_mtime > STALE_MARKER_SECONDS: - marker.unlink(missing_ok=True) - except OSError: - continue +def _marker_path(turn_id: str) -> Path: + return _state.marker_path(turn_id) def record_start(payload: dict[str, Any], *, now: float | None = None) -> bool: - """Persist the start of a Codex turn without storing the user prompt.""" - if payload.get("hook_event_name") != "UserPromptSubmit": - return False - turn_id = _text(payload.get("turn_id")) - if not turn_id: - return False - - timestamp = time.time() if now is None else now - marker = _marker_path(turn_id) - marker.parent.mkdir(parents=True, exist_ok=True) - content = { - "started_at": timestamp, - "session_id": _text(payload.get("session_id")), - "turn_id": turn_id, - "cwd": _text(payload.get("cwd")), - } - temporary = marker.with_suffix(f".{os.getpid()}.tmp") - temporary.write_text(json.dumps(content, ensure_ascii=False), encoding="utf-8") - os.replace(temporary, marker) - _cleanup_stale_markers(timestamp) - return True - - -def consume_start( - notification: dict[str, Any], *, now: float | None = None -) -> tuple[float | None, dict[str, Any]]: - """Read and remove the matching marker, returning elapsed seconds.""" - turn_id = _text(notification.get("turn-id") or notification.get("turn_id")) - if not turn_id: - return None, {} - marker = _marker_path(turn_id) - try: - content = json.loads(marker.read_text(encoding="utf-8")) - started_at = float(content["started_at"]) - except (OSError, ValueError, TypeError, KeyError, json.JSONDecodeError): - return None, {} - finally: - marker.unlink(missing_ok=True) - - finished_at = time.time() if now is None else now - return max(0.0, finished_at - started_at), content - - -def _parse_clock(value: str, default: str) -> int: - try: - hours, minutes = value.strip().split(":", 1) - total = int(hours) * 60 + int(minutes) - if not 0 <= total < 24 * 60: - raise ValueError - return total - except (AttributeError, TypeError, ValueError): - fallback_hours, fallback_minutes = default.split(":", 1) - return int(fallback_hours) * 60 + int(fallback_minutes) + return _state.record_start(payload, now=now) + + +def consume_start(notification: dict[str, Any], *, now: float | None = None) -> tuple[float | None, dict[str, Any]]: + return _state.consume_start(notification, now=now) def is_quiet_time(moment: datetime, config: dict[str, Any] | None = None) -> bool: - settings = {} if config is None else config - start = _parse_clock( - _config_text( - settings, - "voice", - "quiet_start", - DEFAULT_QUIET_START, - ), - DEFAULT_QUIET_START, - ) - end = _parse_clock( - _config_text( - settings, - "voice", - "quiet_end", - DEFAULT_QUIET_END, - ), - DEFAULT_QUIET_END, - ) - if start == end: - return False - current = moment.hour * 60 + moment.minute - if start < end: - return start <= current < end - return current >= start or current < end - - -def notification_channels( - duration_seconds: float | None, - moment: datetime, - config: dict[str, Any] | None = None, -) -> tuple[bool, bool, bool]: - settings = load_config() if config is None else config - teams_webhook_configured = bool( - _config_text(settings, "teams", "webhook_url") - ) - discord_webhook_configured = bool( - _config_text(settings, "discord", "webhook_url") - ) - teams_duration_due = ( - _config_bool( - settings, - "teams", - "notify_when_duration_unknown", - False, - ) - if duration_seconds is None - else duration_seconds - >= _config_seconds( - settings, - "teams", - "minimum_seconds", - DEFAULT_TEAMS_MIN_SECONDS, - ) - ) - discord_duration_due = ( - _config_bool( - settings, - "discord", - "notify_when_duration_unknown", - False, - ) - if duration_seconds is None - else duration_seconds - >= _config_seconds( - settings, - "discord", - "minimum_seconds", - DEFAULT_DISCORD_MIN_SECONDS, - ) - ) - voice_duration_due = ( - _config_bool( - settings, - "voice", - "notify_when_duration_unknown", - True, - ) - if duration_seconds is None - else duration_seconds - >= _config_seconds( - settings, - "voice", - "minimum_seconds", - DEFAULT_VOICE_MIN_SECONDS, - ) - ) - teams = _config_bool( - settings, "teams", "enabled", teams_webhook_configured - ) and teams_duration_due - discord = _config_bool( - settings, - "discord", - "enabled", - discord_webhook_configured, - ) and discord_duration_due - voice = ( - _config_bool(settings, "voice", "enabled", True) - and voice_duration_due - and not is_quiet_time(moment, settings) - ) - return teams, discord, voice + return _cli.is_quiet_time(moment, config) + + +def notification_channels(duration_seconds: float | None, moment: datetime, config: dict[str, Any] | None = None) -> tuple[bool, bool, bool]: + return _cli.notification_channels(duration_seconds, moment, config) def _duration_text(duration_seconds: float | None) -> str: - if duration_seconds is None: - return "No disponible" - total_seconds = max(0, round(duration_seconds)) - minutes, seconds = divmod(total_seconds, 60) - hours, minutes = divmod(minutes, 60) - parts: list[str] = [] - if hours: - parts.append(f"{hours} h") - if minutes: - parts.append(f"{minutes} min") - if seconds or not parts: - parts.append(f"{seconds} s") - return " ".join(parts) + return _payloads.duration_text(duration_seconds) def _spoken_duration(duration_seconds: float, language: str) -> str: - total_seconds = max(0, round(duration_seconds)) - minutes, seconds = divmod(total_seconds, 60) - hours, minutes = divmod(minutes, 60) - parts: list[str] = [] - if language == "en": - if hours: - parts.append(f"{hours} hour" + ("s" if hours != 1 else "")) - if minutes: - parts.append(f"{minutes} minute" + ("s" if minutes != 1 else "")) - if seconds or not parts: - parts.append(f"{seconds} second" + ("s" if seconds != 1 else "")) - else: - if hours: - parts.append(f"{hours} hora" + ("s" if hours != 1 else "")) - if minutes: - parts.append(f"{minutes} minuto" + ("s" if minutes != 1 else "")) - if seconds or not parts: - parts.append(f"{seconds} segundo" + ("s" if seconds != 1 else "")) - return " ".join(parts) - - -def detect_language(text: str, configured: str = "auto") -> str: - requested = configured.strip().lower() - if requested in {"es", "spanish", "español"}: - return "es" - if requested in {"en", "english", "inglés", "ingles"}: - return "en" - - lowered = text.lower() - words = re.findall(r"[a-záéíóúüñ]+", lowered) - spanish_score = sum(word in SPANISH_WORDS for word in words) - english_score = sum(word in ENGLISH_WORDS for word in words) - if re.search(r"[áéíóúüñ¿¡]", lowered): - spanish_score += 2 - return "en" if english_score > spanish_score else "es" + return _cli.spoken_duration(duration_seconds, language) + + +def detect_language(value: str, configured: str = "auto") -> str: + return _voice.detect_language(value, configured) def _chat_title(notification: dict[str, Any]) -> str: - for key in ("thread-title", "thread_title", "title"): - title = _text(notification.get(key)) - if title: - return _truncate(title, MAX_TITLE_CHARS) - return "" - - -def build_payload( - notification: dict[str, Any], - duration_seconds: float | None, - marker: dict[str, Any], - config: dict[str, Any] | None = None, -) -> dict[str, Any]: - settings = {} if config is None else config - completed_at = datetime.now().astimezone().strftime("%d/%m/%Y %H:%M %Z") - chat_id = _text(notification.get("thread-id")) or _text(marker.get("session_id")) - chat_title = _chat_title(notification) - cwd = _text(notification.get("cwd")) or _text(marker.get("cwd")) - summary_max_chars = _config_int( - settings, - "teams", - "summary_max_chars", - DEFAULT_SUMMARY_MAX_CHARS, - minimum=MIN_SUMMARY_MAX_CHARS, - maximum=MAX_SUMMARY_MAX_CHARS, - ) - summary_tail_chars = _config_int( - settings, - "teams", - "summary_tail_chars", - DEFAULT_SUMMARY_TAIL_CHARS, - ) - summary = _summary_excerpt( - _text(notification.get("last-assistant-message")) - or "El turno terminó sin un mensaje final para resumir.", - summary_max_chars, - summary_tail_chars, - ) - facts = [ - {"title": "Estado", "value": "Completado"}, - {"title": "Duración", "value": _duration_text(duration_seconds)}, - ] - if chat_id: - facts.append({"title": "ID del chat", "value": chat_id}) - if chat_title: - facts.append({"title": "Título", "value": chat_title}) - if cwd: - facts.append({"title": "Proyecto", "value": Path(cwd).name or cwd}) - facts.append({"title": "Fecha y hora", "value": completed_at}) - - return { - "type": "message", - "attachments": [ - { - "contentType": "application/vnd.microsoft.card.adaptive", - "contentUrl": None, - "content": { - "$schema": "http://adaptivecards.io/schemas/adaptive-card.json", - "type": "AdaptiveCard", - "version": "1.2", - "msteams": {"width": "Full"}, - "body": [ - { - "type": "TextBlock", - "text": "✅ Turno de Codex completado", - "size": "Large", - "weight": "Bolder", - "color": "Good", - "wrap": True, - }, - {"type": "FactSet", "facts": facts}, - { - "type": "TextBlock", - "text": "Resumen de lo realizado", - "weight": "Bolder", - "wrap": True, - "separator": True, - }, - {"type": "TextBlock", "text": summary, "wrap": True}, - ], - }, - } - ], - } - - -def build_discord_payload( - notification: dict[str, Any], - duration_seconds: float | None, - marker: dict[str, Any], - config: dict[str, Any] | None = None, -) -> dict[str, Any]: - settings = {} if config is None else config - chat_id = _text(notification.get("thread-id")) or _text(marker.get("session_id")) - chat_title = _chat_title(notification) - cwd = _text(notification.get("cwd")) or _text(marker.get("cwd")) - summary_max_chars = _config_int( - settings, - "discord", - "summary_max_chars", - DEFAULT_SUMMARY_MAX_CHARS, - minimum=MIN_SUMMARY_MAX_CHARS, - maximum=MAX_DISCORD_SUMMARY_MAX_CHARS, - ) - summary = _truncate( - _text(notification.get("last-assistant-message")) - or "El turno terminó sin un mensaje final para resumir.", - summary_max_chars, - ) - fields = [ - {"name": "Estado", "value": "Completado", "inline": True}, - { - "name": "Duración", - "value": _duration_text(duration_seconds), - "inline": True, - }, - ] - if cwd: - fields.append( - { - "name": "Proyecto", - "value": _truncate(Path(cwd).name or cwd, 100), - "inline": True, - } - ) - if chat_title: - fields.append({"name": "Título", "value": chat_title, "inline": False}) - if chat_id: - fields.append( - {"name": "ID del chat", "value": _truncate(chat_id, 200), "inline": False} - ) - return { - "username": "Codex Notifier", - "allowed_mentions": {"parse": []}, - "embeds": [ - { - "title": "✅ Turno de Codex completado", - "description": summary, - "color": 3061878, - "fields": fields, - "timestamp": datetime.now().astimezone().isoformat(timespec="seconds"), - } - ], - } - - -def _send_webhook_notification( - service: str, webhook_url: str, payload: dict[str, Any] -) -> None: - request = Request( - webhook_url, - data=json.dumps(payload, ensure_ascii=False).encode("utf-8"), - headers={ - "Content-Type": "application/json; charset=utf-8", - "User-Agent": "Codex-Notifier/1.0", - }, - method="POST", - ) - for attempt in range(2): - try: - with urlopen(request, timeout=5) as response: - if 200 <= response.status < 300: - return - raise RuntimeError(f"{service} respondió con HTTP {response.status}") - except HTTPError as exc: - if exc.code not in {429, 500, 502, 503, 504} or attempt == 1: - raise RuntimeError(f"{service} respondió con HTTP {exc.code}") from exc - except URLError as exc: - if attempt == 1: - raise RuntimeError( - f"No se pudo conectar con {service}: {exc.reason}" - ) from exc - time.sleep(1) + return _payloads.chat_title(notification) + + +def build_payload(notification: dict[str, Any], duration_seconds: float | None, marker: dict[str, Any], config: dict[str, Any] | None = None) -> dict[str, Any]: + return _payloads.build_payload(notification, duration_seconds, marker, config) + + +def build_discord_payload(notification: dict[str, Any], duration_seconds: float | None, marker: dict[str, Any], config: dict[str, Any] | None = None) -> dict[str, Any]: + return _payloads.build_discord_payload(notification, duration_seconds, marker, config) + + +def _send_webhook_notification(service: str, webhook_url: str, payload: dict[str, Any]) -> None: + _transport(service, webhook_url, payload) def send_teams_notification(webhook_url: str, payload: dict[str, Any]) -> None: @@ -736,455 +165,66 @@ def send_teams_notification(webhook_url: str, payload: dict[str, Any]) -> None: def send_discord_notification(webhook_url: str, payload: dict[str, Any]) -> None: + from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit parts = urlsplit(webhook_url) query = dict(parse_qsl(parts.query, keep_blank_values=True)) query["wait"] = "true" - confirmed_url = urlunsplit( - (parts.scheme, parts.netloc, parts.path, urlencode(query), parts.fragment) - ) + confirmed_url = urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(query), parts.fragment)) _send_webhook_notification("Discord", confirmed_url, payload) -def _voice_message( - notification: dict[str, Any], - duration_seconds: float | None, - marker: dict[str, Any], - language: str, -) -> str: - cwd = _text(notification.get("cwd")) or _text(marker.get("cwd")) - project = _truncate(Path(cwd).name, 60) if cwd else "actual" - title = _chat_title(notification) - extra = "" - if title: - extra = f" Task: {title}." if language == "en" else f" Tarea: {title}." - if language == "en": - if duration_seconds is None: - base = ( - f"Codex finished the task for project {project}. " - "The duration could not be determined." - ) - else: - duration = _spoken_duration(duration_seconds, language) - base = f"Codex finished the task for project {project} after {duration}." - return base + extra - if duration_seconds is None: - base = ( - f"Codex terminó la tarea del proyecto {project}. " - "No se pudo determinar la duración." - ) - else: - duration = _spoken_duration(duration_seconds, language) - base = f"Codex terminó la tarea del proyecto {project} después de {duration}." - return base + extra +def _voice_message(notification: dict[str, Any], duration_seconds: float | None, marker: dict[str, Any], language: str) -> str: + return _cli.voice_message(notification, duration_seconds, marker, language) def _configured_volume(voice_config: dict[str, Any]) -> float | None: - value = voice_config.get("volume") - if not isinstance(value, (int, float)) or isinstance(value, bool): - return None - return min(1.0, max(0.0, float(value))) - - -def speak_windows( - message: str, - *, - language: str = "es", - preferred_voice: str = "", - volume: float | None = None, -) -> None: - if os.name != "nt": - raise RuntimeError("La voz local solo está disponible en Windows.") - encoded = base64.b64encode(message.encode("utf-8")).decode("ascii") - environment = os.environ.copy() - environment["CODEX_NOTIFIER_SPEECH_B64"] = encoded - environment["CODEX_NOTIFIER_SPEECH_LANGUAGE"] = language - environment["CODEX_NOTIFIER_SPEECH_VOICE"] = preferred_voice - environment["CODEX_NOTIFIER_SPEECH_VOLUME"] = ( - "" if volume is None else str(round(min(1.0, max(0.0, volume)) * 100)) - ) - script = ( - "$bytes=[Convert]::FromBase64String($env:CODEX_NOTIFIER_SPEECH_B64);" - "$text=[Text.Encoding]::UTF8.GetString($bytes);" - "$voice=New-Object -ComObject SAPI.SpVoice;" - "$wantedName=$env:CODEX_NOTIFIER_SPEECH_VOICE;" - "$wantedLanguage=$env:CODEX_NOTIFIER_SPEECH_LANGUAGE;" - "$wantedVolume=$env:CODEX_NOTIFIER_SPEECH_VOLUME;" - "$selected=$null;" - "foreach($candidate in $voice.GetVoices()){" - "$description=$candidate.GetDescription();" - "if($wantedName -and $description -like ('*'+$wantedName+'*')){" - "$selected=$candidate;break}" - "if(-not $selected){try{" - "$lcid=[Convert]::ToInt32($candidate.GetAttribute('Language'),16);" - "$culture=[Globalization.CultureInfo]::GetCultureInfo($lcid);" - "if($culture.TwoLetterISOLanguageName -eq $wantedLanguage){" - "$selected=$candidate}}catch{}}};" - "if($selected){$voice.Voice=$selected};" - "if($wantedVolume -ne ''){$voice.Volume=[int]$wantedVolume};" - "[void]$voice.Speak($text)" - ) - creation_flags = getattr(subprocess, "CREATE_NO_WINDOW", 0) - subprocess.run( - [ - "powershell.exe", - "-NoLogo", - "-NoProfile", - "-NonInteractive", - "-Command", - script, - ], - env=environment, - stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - creationflags=creation_flags, - check=True, - timeout=SPEECH_TIMEOUT_SECONDS, - ) + return _voice.configured_volume(voice_config) + + +def speak_windows(message: str, *, language: str = "es", preferred_voice: str = "", volume: float | None = None) -> None: + return _voice.speak_sapi_windows(message, language=language, preferred_voice=preferred_voice, volume=volume) def _configured_voice(voice_config: dict[str, Any], language: str) -> str: - key = "spanish_voice" if language == "es" else "english_voice" - return _text(voice_config.get(key)) + return _voice.configured_voice(voice_config, language) def _resolve_executable(configured: str, default: str) -> str | None: - requested = configured or default - if Path(requested).expanduser().parent != Path("."): - path = Path(requested).expanduser() - return str(path) if path.is_file() else None - return shutil.which(requested) - - -def speak_piper( - message: str, *, language: str = "es", voice_config: dict[str, Any] -) -> None: - model_name = _configured_voice(voice_config, language) - if not model_name: - key = "spanish_voice" if language == "es" else "english_voice" - raise RuntimeError(f"Configura voice.{key} para usar Piper.") - model = Path(model_name).expanduser() - if not model.is_file(): - raise RuntimeError(f"No se encontró el modelo de Piper: {model}") - - configured_executable = _text(voice_config.get("piper_executable")) - piper = _resolve_executable(configured_executable, "piper") - if not piper: - raise RuntimeError( - "No se encontró Piper. Instala piper-tts o configura " - "voice.piper_executable." - ) - - powershell = shutil.which("powershell.exe") - wslpath = shutil.which("wslpath") - player = None - if not (powershell and wslpath): - player = next( - ( - (name, shutil.which(name)) - for name in ("paplay", "pw-play", "aplay", "ffplay") - if shutil.which(name) - ), - None, - ) - if not (powershell and wslpath) and not player: - raise RuntimeError( - "No se encontró un reproductor compatible: instala paplay, pw-play, " - "aplay o ffplay. En WSL también se admite powershell.exe." - ) - - with tempfile.TemporaryDirectory(prefix="codex-notifier-") as temporary_dir: - wav_path = Path(temporary_dir) / "speech.wav" - piper_command = [piper, "-m", str(model), "-f", str(wav_path)] - volume = _configured_volume(voice_config) - if volume is not None: - piper_command.extend(["--volume", str(volume)]) - piper_command.extend(["--", message]) - subprocess.run( - piper_command, - cwd=temporary_dir, - stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - check=True, - timeout=SPEECH_TIMEOUT_SECONDS, - ) - if powershell and wslpath: - windows_path = subprocess.run( - [wslpath, "-w", str(wav_path)], - stdin=subprocess.DEVNULL, - capture_output=True, - text=True, - check=True, - timeout=5, - ).stdout.strip() - encoded_path = base64.b64encode(windows_path.encode("utf-8")).decode( - "ascii" - ) - script = ( - "$ErrorActionPreference='Stop';" - f"$bytes=[Convert]::FromBase64String('{encoded_path}');" - "$path=[Text.Encoding]::UTF8.GetString($bytes);" - "$player=New-Object System.Media.SoundPlayer $path;" - "$player.Load();$player.PlaySync()" - ) - subprocess.run( - [ - powershell, - "-NoLogo", - "-NoProfile", - "-NonInteractive", - "-Command", - script, - ], - stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - check=True, - timeout=SPEECH_TIMEOUT_SECONDS, - ) - else: - player_name, player_path = player - player_command = ( - [player_path, "-nodisp", "-autoexit", str(wav_path)] - if player_name == "ffplay" - else [player_path, str(wav_path)] - ) - subprocess.run( - player_command, - stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - check=True, - timeout=SPEECH_TIMEOUT_SECONDS, - ) - - -def speak_linux( - message: str, - *, - language: str = "es", - voice_config: dict[str, Any] | None = None, -) -> None: - settings = {} if voice_config is None else voice_config - if _text(settings.get("piper_executable")): - speak_piper(message, language=language, voice_config=settings) - return - - speech_dispatcher = shutil.which("spd-say") - if speech_dispatcher: - command = [speech_dispatcher, "-w", "-l", language, message] - else: - espeak = shutil.which("espeak-ng") or shutil.which("espeak") - if not espeak: - raise RuntimeError( - "Instala speech-dispatcher (spd-say) o espeak-ng para usar voz en Linux." - ) - command = [espeak, "-v", language, message] - subprocess.run( - command, - stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - check=True, - timeout=SPEECH_TIMEOUT_SECONDS, - ) - - -def speak( - message: str, - *, - language: str = "es", - preferred_voice: str = "", - voice_config: dict[str, Any] | None = None, -) -> None: - settings = {} if voice_config is None else voice_config - if os.name == "nt": - speak_windows( - message, - language=language, - preferred_voice=preferred_voice, - volume=_configured_volume(settings), - ) - return - if os.name == "posix": - speak_linux(message, language=language, voice_config=settings) - return - raise RuntimeError(f"La voz local no es compatible con la plataforma {os.name}.") + return _voice.resolve_executable(configured, default) + + +def generate_piper_wav(message: str, *, language: str = "es", voice_config: dict[str, Any], temporary_dir: str | Path) -> Path: + return _voice.generate_piper_wav(message, language=language, voice_config=voice_config, temporary_dir=temporary_dir) + + +def play_wav_windows(wav_path: Path, *, powershell: str | None = None) -> None: + return _voice.play_wav_windows(wav_path, powershell=powershell) + + +def play_wav_linux(wav_path: Path) -> None: + return _voice.play_wav_linux(wav_path) + + +def speak_piper(message: str, *, language: str = "es", voice_config: dict[str, Any]) -> None: + return _voice.speak_piper(message, language=language, voice_config=voice_config) + + +def speak_linux(message: str, *, language: str = "es", voice_config: dict[str, Any] | None = None) -> None: + return _voice.speak_linux(message, language=language, voice_config=voice_config) + + +def speak(message: str, *, language: str = "es", preferred_voice: str = "", voice_config: dict[str, Any] | None = None) -> None: + return _voice.speak(message, language=language, preferred_voice=preferred_voice, voice_config=voice_config) def handle_completion(notification: dict[str, Any], *, now: float | None = None) -> None: - if notification.get("type") != "agent-turn-complete": - return - duration, marker = consume_start(notification, now=now) - try: - config = load_config() - except (OSError, TypeError, json.JSONDecodeError) as exc: - print(f"Configuración inválida: {exc}", file=sys.stderr) - config = {} - - moment = datetime.now().astimezone() - chat_id = ( - _text(notification.get("thread-id") or notification.get("thread_id")) - or _text(marker.get("session_id")) - ) - teams_due, discord_due, voice_due = notification_channels( - duration, moment, config - ) - teams_status = "not_due" - discord_status = "not_due" - voice_status = "not_due" - teams_error = "" - discord_error = "" - voice_error = "" - - # Voice runs first and synchronously so a slow or failed webhook cannot - # delay or cancel the local notification. - if voice_due: - try: - voice_config = _config_section(config, "voice") - language = detect_language( - _text(notification.get("last-assistant-message")), - _text(voice_config.get("language")) or "auto", - ) - preferred_voice = _text( - voice_config.get( - "spanish_voice" if language == "es" else "english_voice" - ) - ) - speak( - _voice_message(notification, duration, marker, language), - language=language, - preferred_voice=preferred_voice, - voice_config=voice_config, - ) - voice_status = "sent" - except Exception as exc: - voice_status = "failed" - voice_error = _safe_error(exc) - print(f"No se pudo reproducir el aviso de voz: {exc}", file=sys.stderr) - - if teams_due: - webhook_url = _config_text(config, "teams", "webhook_url") - if webhook_url and urlsplit(webhook_url).scheme == "https": - try: - send_teams_notification( - webhook_url, - build_payload(notification, duration, marker, config), - ) - teams_status = "sent" - except Exception as exc: - teams_status = "failed" - teams_error = _safe_error(exc, webhook_url) - print(f"No se pudo enviar la notificación a Teams: {exc}", file=sys.stderr) - else: - teams_status = "failed" - teams_error = ( - "El webhook de Teams debe usar HTTPS." - if webhook_url - else "No hay un webhook de Teams configurado." - ) - print(teams_error, file=sys.stderr) - - if discord_due: - webhook_url = _config_text(config, "discord", "webhook_url") - if webhook_url and urlsplit(webhook_url).scheme == "https": - try: - send_discord_notification( - webhook_url, - build_discord_payload(notification, duration, marker, config), - ) - discord_status = "sent" - except Exception as exc: - discord_status = "failed" - discord_error = _safe_error(exc, webhook_url) - print( - f"No se pudo enviar la notificación a Discord: {exc}", - file=sys.stderr, - ) - else: - discord_status = "failed" - discord_error = ( - "El webhook de Discord debe usar HTTPS." - if webhook_url - else "No hay un webhook de Discord configurado." - ) - print(discord_error, file=sys.stderr) - - try: - write_trace_log( - config, - moment=moment, - chat_id=chat_id, - duration_seconds=duration, - teams_status=teams_status, - discord_status=discord_status, - voice_status=voice_status, - teams_error=teams_error, - discord_error=discord_error, - voice_error=voice_error, - ) - except OSError as exc: - print(f"No se pudo escribir el log del notificador: {exc}", file=sys.stderr) - - -def _read_json_stdin() -> dict[str, Any]: - payload = json.load(sys.stdin) - if not isinstance(payload, dict): - raise TypeError("el payload debe ser un objeto JSON") - return payload - - -def _read_json_argument(index: int) -> dict[str, Any]: - payload = json.loads(sys.argv[index]) - if not isinstance(payload, dict): - raise TypeError("el payload debe ser un objeto JSON") - return payload + _cli.handle_completion(notification, now=now, load=load_config, consume=consume_start, + speak_fn=speak, send_teams_fn=send_teams_notification, + send_discord_fn=send_discord_notification, log_fn=write_trace_log) def main() -> int: - command = sys.argv[1] if len(sys.argv) > 1 else "" - try: - if command == "record-start": - record_start(_read_json_stdin()) - return 0 - if command == "notify" and len(sys.argv) == 3: - handle_completion(_read_json_argument(2)) - return 0 - if command == "voice-test": - config = load_config() - requested = sys.argv[2] if len(sys.argv) > 2 else "es" - language = detect_language("", requested) - voice_config = _config_section(config, "voice") - preferred_voice = _text( - voice_config.get( - "spanish_voice" if language == "es" else "english_voice" - ) - ) - message = ( - "The Codex voice notification is working." - if language == "en" - else "La notificación por voz de Codex está funcionando." - ) - speak( - message, - language=language, - preferred_voice=preferred_voice, - voice_config=voice_config, - ) - return 0 - except (json.JSONDecodeError, TypeError, ValueError) as exc: - print(f"Payload de Codex inválido: {exc}", file=sys.stderr) - return 0 - except Exception as exc: - print(f"Error del notificador: {exc}", file=sys.stderr) - return 0 - - print( - "Uso: notifier.py record-start | notify '' | voice-test [es|en]", - file=sys.stderr, - ) - return 0 + return _cli.main(speak_fn=speak) if __name__ == "__main__": diff --git a/tests/test_hardening.py b/tests/test_hardening.py new file mode 100644 index 0000000..88e66be --- /dev/null +++ b/tests/test_hardening.py @@ -0,0 +1,112 @@ +from __future__ import annotations + +import json +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +from codex_notifier.channels import webhook +from codex_notifier.channels import voice +from codex_notifier import cli +from codex_notifier.files import atomic_write_text +import notifier + + +class HardeningTests(unittest.TestCase): + def test_summary_disabled_is_absent_from_both_serialized_payloads(self) -> None: + secret = "SENSITIVE-ASSISTANT-SUMMARY-42" + notification = {"last-assistant-message": secret, "thread-id": "chat"} + teams = json.dumps(notifier.build_payload(notification, 10, {}, {"teams": {"include_summary": False}})) + discord = json.dumps(notifier.build_discord_payload(notification, 10, {}, {"discord": {"include_summary": False}})) + self.assertNotIn(secret, teams) + self.assertNotIn(secret, discord) + self.assertNotIn("Resumen de lo realizado", teams) + self.assertNotIn("description", discord) + + @patch("notifier.send_teams_notification") + def test_disabled_summary_is_not_written_to_channel_error_logs(self, send) -> None: + secret = "SENSITIVE-ASSISTANT-SUMMARY-43" + send.side_effect = RuntimeError(secret) + with tempfile.TemporaryDirectory() as temporary: + with patch.dict(os.environ, { + notifier.STATE_DIR_ENV: temporary, + notifier.CONFIG_PATH_ENV: str(Path(temporary) / "config.json"), + }): + Path(temporary, "config.json").write_text(json.dumps({ + "teams": {"enabled": True, "minimum_seconds": 0, + "notify_when_duration_unknown": True, + "include_summary": False, + "webhook_url": "https://example.invalid/hook"}, + "voice": {"enabled": False}, + })) + notifier.handle_completion({ + "type": "agent-turn-complete", "thread-id": "chat", + "last-assistant-message": secret, + }) + log = next(Path(temporary, "logs").glob("*.jsonl")) + self.assertNotIn(secret, log.read_text()) + + @unittest.skipIf(os.name == "nt", "POSIX mode bits are not Windows semantics") + def test_state_directory_and_marker_are_private(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + with patch.dict(os.environ, {notifier.STATE_DIR_ENV: temporary}): + notifier.record_start({ + "hook_event_name": "UserPromptSubmit", + "session_id": "session", + "turn_id": "turn", + "cwd": "/work/project", + }, now=1) + turns = Path(temporary) / "turns" + marker = next(turns.glob("*.json")) + self.assertEqual(turns.stat().st_mode & 0o777, 0o700) + self.assertEqual(marker.stat().st_mode & 0o777, 0o600) + + @unittest.skipIf(os.name == "nt", "POSIX mode bits are not Windows semantics") + def test_atomic_write_sets_private_file_mode(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + path = Path(temporary) / "config.json" + atomic_write_text(path, "{}\n", mode=0o600, platform_name="posix") + self.assertEqual(path.read_text(), "{}\n") + self.assertEqual(path.stat().st_mode & 0o777, 0o600) + + @patch("codex_notifier.channels.webhook.time.sleep") + @patch("codex_notifier.channels.webhook.urlopen") + def test_retry_after_is_used_for_transient_http_error(self, urlopen, sleep) -> None: + from urllib.error import HTTPError + error = HTTPError("https://example.invalid", 503, "busy", {"Retry-After": "2"}, None) + response = MagicMock() + response.status = 200 + response.__enter__.return_value = response + urlopen.side_effect = [error, response] + webhook.send_webhook_notification("Teams", "https://example.invalid", {}) + sleep.assert_called_once_with(2.0) + + @patch("codex_notifier.channels.voice.speak_sapi_windows") + @patch("codex_notifier.channels.voice.speak_piper") + @patch("codex_notifier.channels.voice.os.name", "nt") + def test_windows_voice_defaults_to_sapi_and_uses_piper_when_configured(self, piper, sapi) -> None: + voice.speak("hello", language="en", voice_config={}) + sapi.assert_called_once() + piper.assert_not_called() + sapi.reset_mock() + voice.speak("hello", language="en", voice_config={"piper_executable": "piper"}) + piper.assert_called_once() + sapi.assert_not_called() + + @patch("codex_notifier.channels.voice.speak_sapi_windows") + @patch("codex_notifier.channels.voice.speak_piper", side_effect=RuntimeError("bad Piper")) + @patch("codex_notifier.channels.voice.os.name", "nt") + def test_explicit_piper_failure_does_not_fallback_to_sapi(self, piper, sapi) -> None: + with self.assertRaisesRegex(RuntimeError, "bad Piper"): + voice.speak("hello", voice_config={"piper_executable": "piper"}) + sapi.assert_not_called() + + def test_manual_cli_errors_have_nonzero_exit_codes(self) -> None: + self.assertNotEqual(cli.main(["unknown-command"]), 0) + self.assertNotEqual(cli.main(["voice-test", "fr"]), 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_install.py b/tests/test_install.py index b76f28d..7f9aad4 100644 --- a/tests/test_install.py +++ b/tests/test_install.py @@ -178,6 +178,28 @@ def test_existing_local_config_is_preserved(self) -> None: self.assertIsNone(backup) self.assertEqual(destination.read_text(encoding="utf-8"), '{"keep":true}\n') + def test_root_notify_scanner_does_not_touch_section_values(self) -> None: + notify = 'notify = ["python", "notifier.py", "notify"]' + cases = { + "": notify + "\n", + 'notify = ["old"]\nmodel = "x"\n': notify + "\nmodel = \"x\"\n", + 'model = "x"\n[section]\nnotify = ["section"]\n': 'model = "x"\n' + notify + '\n[section]\nnotify = ["section"]\n', + '# notify = ["comment"]\n[section]\nkey = true\n': '# notify = ["comment"]\n' + notify + '\n[section]\nkey = true\n', + 'notify = ["root"]\n[section]\nnotify = ["section"]\n': notify + '\n[section]\nnotify = ["section"]\n', + } + for original, expected in cases.items(): + with self.subTest(original=original): + self.assertEqual(install.update_root_notify(original, notify), expected) + self.assertEqual(install.update_root_notify(expected, notify), expected) + + def test_root_notify_preserves_inline_comment(self) -> None: + notify = 'notify = ["python", "notifier.py", "notify"]' + original = 'notify = ["old"] # keep this comment\n[section]\nvalue = 1\n' + self.assertEqual( + install.update_root_notify(original, notify), + notify + ' # keep this comment\n[section]\nvalue = 1\n', + ) + if __name__ == "__main__": unittest.main() From 9003877447bfd084558ad4dcbd6971d44bb489bf Mon Sep 17 00:00:00 2001 From: Julio Rios Date: Fri, 25 Sep 2026 10:21:35 -0500 Subject: [PATCH 2/2] More improves --- .github/workflows/ci.yml | 8 ++++---- README.es.md | 2 +- README.md | 4 ++-- codex_notifier/cli.py | 5 ++--- codex_notifier/payloads.py | 12 +++++++++--- 5 files changed, 18 insertions(+), 13 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3f649fd..3e536e6 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -12,14 +12,14 @@ jobs: fail-fast: false matrix: include: - - os: ubuntu-latest - python: "3.10" - os: ubuntu-latest python: "3.13" - - os: windows-latest - python: "3.10" + - os: ubuntu-latest + python: "3.14" - os: windows-latest python: "3.13" + - os: windows-latest + python: "3.14" runs-on: ${{ matrix.os }} steps: - uses: actions/checkout@v4 diff --git a/README.es.md b/README.es.md index ab3a293..c76dcd7 100644 --- a/README.es.md +++ b/README.es.md @@ -238,7 +238,7 @@ distinto de cero, al igual que `voice-test` si la voz no puede ejecutarse. python -m unittest discover -s tests -v ``` -GitHub Actions ejecuta esta suite en Ubuntu y Windows con Python 3.10 y 3.13. +GitHub Actions ejecuta esta suite en Ubuntu y Windows con Python 3.13 y 3.14. ## Cómo funciona diff --git a/README.md b/README.md index 6a651aa..c5c0e07 100644 --- a/README.md +++ b/README.md @@ -230,8 +230,8 @@ failures are logged independently. Invalid manual syntax returns non-zero, and python -m unittest discover -s tests -v ``` -GitHub Actions runs this suite on Ubuntu and Windows with Python 3.10 and -Python 3.13. +GitHub Actions runs this suite on Ubuntu and Windows with Python 3.13 and +Python 3.14. ## How it works diff --git a/codex_notifier/cli.py b/codex_notifier/cli.py index 20a76f9..35c081e 100644 --- a/codex_notifier/cli.py +++ b/codex_notifier/cli.py @@ -5,7 +5,6 @@ import json import sys from datetime import datetime -from pathlib import Path from typing import Any, Callable from urllib.parse import urlsplit @@ -15,7 +14,7 @@ DEFAULT_QUIET_START, DEFAULT_TEAMS_MIN_SECONDS, DEFAULT_VOICE_MIN_SECONDS, config_bool, config_section, config_seconds, config_text, load_config) -from .payloads import build_discord_payload, build_payload +from .payloads import build_discord_payload, build_payload, project_name from .state import consume_start, record_start, text from .trace_logging import safe_error, write_trace_log @@ -79,7 +78,7 @@ def spoken_duration(duration_seconds: float, language: str) -> str: def voice_message(notification: dict[str, Any], duration_seconds: float | None, marker: dict[str, Any], language: str) -> str: cwd = text(notification.get("cwd")) or text(marker.get("cwd")) - project = (Path(cwd).name[:60] if cwd else "actual") + project = (project_name(cwd)[:60] if cwd else "actual") title = text(notification.get("thread-title") or notification.get("thread_title") or notification.get("title")) extra = (f" Task: {title}." if language == "en" else f" Tarea: {title}.") if title else "" if language == "en": diff --git a/codex_notifier/payloads.py b/codex_notifier/payloads.py index 11a2f53..f53b486 100644 --- a/codex_notifier/payloads.py +++ b/codex_notifier/payloads.py @@ -3,7 +3,7 @@ from __future__ import annotations from datetime import datetime -from pathlib import Path +from pathlib import PurePosixPath, PureWindowsPath from typing import Any from .config import (DEFAULT_SUMMARY_MAX_CHARS, DEFAULT_SUMMARY_TAIL_CHARS, @@ -14,6 +14,12 @@ MAX_TITLE_CHARS = 120 +def project_name(cwd: str) -> str: + """Return the final component of a Windows or POSIX working directory.""" + path_type = PureWindowsPath if "\\" in cwd else PurePosixPath + return path_type(cwd).name or cwd + + def truncate(value: str, limit: int) -> str: if len(value) <= limit: return value @@ -80,7 +86,7 @@ def build_payload(notification: dict[str, Any], duration_seconds: float | None, if title: facts.append({"title": "Título", "value": title}) if cwd: - facts.append({"title": "Proyecto", "value": Path(cwd).name or cwd}) + facts.append({"title": "Proyecto", "value": project_name(cwd)}) facts.append({"title": "Fecha y hora", "value": completed_at}) body: list[dict[str, Any]] = [ {"type": "TextBlock", "text": "✅ Turno de Codex completado", "size": "Large", @@ -116,7 +122,7 @@ def build_discord_payload(notification: dict[str, Any], duration_seconds: float fields = [{"name": "Estado", "value": "Completado", "inline": True}, {"name": "Duración", "value": duration_text(duration_seconds), "inline": True}] if cwd: - fields.append({"name": "Proyecto", "value": truncate(Path(cwd).name or cwd, 100), "inline": True}) + fields.append({"name": "Proyecto", "value": truncate(project_name(cwd), 100), "inline": True}) if title: fields.append({"name": "Título", "value": title, "inline": False}) if chat_id: