felixm/blurt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Only allow keyring for credentials retrieval

Browse Source
This commit is contained in:
felixm
2026-08-15 13:29:22 -04:00
parent 0b94bcd1c0
commit 0ccada8522
2 changed files with 186 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1 -3
View File
@@ -5,8 +5,6 @@ __pycache__/
# Virtual environment
.venv/
# Blurt runtime config / secrets (written by blurt.py at runtime)
api_key.txt
openrouter_key.txt
# Blurt runtime config (per-machine)
model.txt
language.txt
blurt.py
+185 -25
View File
@@ -2,6 +2,7 @@ import base64
import io
import logging
import os
import random
import shlex
import signal
import subprocess
@@ -28,17 +29,44 @@ OPENROUTER_APP_URL = "https://git.felixm.de/felixm/blurt"
OPENROUTER_APP_TITLE = "Blurt"
DEFAULT_TRANSCRIPTION_MODEL = "whisper-1"
DEFAULT_TRANSCRIPTION_LANGUAGE = "en"
# (model_id, menu_label, provider). The provider selects both the API endpoint
# and the request format: "openai" speaks multipart file upload; "openrouter"
# speaks JSON with base64-encoded audio. OpenRouter's whisper-large-v3-turbo is
# served by Groq (216x real-time, ~12% WER) and reuses your OpenRouter key.
TRANSCRIPTION_MODELS = [
("whisper-1", "OpenAI: whisper-1", "openai"),
("gpt-4o-mini-transcribe", "OpenAI: gpt-4o-mini-transcribe", "openai"),
(
"gpt-4o-mini-transcribe-2025-12-15",
"OpenAI: gpt-4o-mini-transcribe (2025-12-15)",
"openai",
),
("gpt-4o-transcribe", "OpenAI: gpt-4o-transcribe", "openai"),
("openai/whisper-large-v3-turbo", "OpenRouter: whisper-large-v3-turbo (Groq, fast)", "openrouter"),
("gpt-4o-transcribe-diarize", "OpenAI: gpt-4o-transcribe-diarize", "openai"),
(
"openai/whisper-large-v3-turbo",
"OpenRouter: whisper-large-v3-turbo (fast)",
"openrouter",
),
(
"openai/whisper-large-v3",
"OpenRouter: whisper-large-v3 (accurate)",
"openrouter",
),
(
"openai/gpt-4o-mini-transcribe",
"OpenRouter: gpt-4o-mini-transcribe",
"openrouter",
),
(
"openai/gpt-4o-transcribe",
"OpenRouter: gpt-4o-transcribe",
"openrouter",
),
]
PROVIDER_KEY_FILES = {"openai": "api_key.txt", "openrouter": "openrouter_key.txt"}
PROVIDER_KEY_HINTS = {
"openai": ("set OPENAI_API_KEY or configure keyring/Credential Manager entry blurt/openai-api-key"),
"openrouter": (
"set OPENROUTER_API_KEY or configure Windows Credential Manager entry "
"blurt/openrouter-api-key"
),
}
TRANSCRIPTION_LANGUAGES = [
("en", "English"),
("de", "German"),
@@ -57,6 +85,7 @@ KEYRING_CREDENTIALS = [
]
OPENROUTER_KEYRING_CREDENTIALS = [
("openrouter-api-key", "felixm"),
("blurt", "openrouter-api-key"),
]
@@ -139,7 +168,8 @@ def load_api_key() -> str:
if key:
return key
return load_config_value("OPENAI_API_KEY", "api_key.txt")
logging.warning("No OpenAI API key found in OPENAI_API_KEY or configured keyring entries")
return ""
def load_openrouter_key() -> str:
@@ -152,7 +182,10 @@ def load_openrouter_key() -> str:
if key:
return key
return load_config_value("OPENROUTER_API_KEY", "openrouter_key.txt")
logging.warning(
"No OpenRouter API key found in OPENROUTER_API_KEY or configured keyring entries"
)
return ""
def load_transcription_model() -> str:
@@ -339,6 +372,45 @@ def get_platform() -> Platform:
return _make_linux_platform()
class AwakeKeeper:
"""Periodically clicks the Windows Start button area to prevent PC lock."""
def __init__(self):
from pynput.mouse import Controller, Button
self._mouse = Controller()
self._button = Button.left
self._stop_event = threading.Event()
self._thread: threading.Thread | None = None
@property
def is_active(self) -> bool:
return self._thread is not None and self._thread.is_alive()
def start(self):
if self.is_active:
return
self._stop_event.clear()
self._thread = threading.Thread(
target=self._run, daemon=True, name="AwakeKeeper"
)
self._thread.start()
logging.info("AwakeKeeper started")
def stop(self):
self._stop_event.set()
self._thread = None
logging.info("AwakeKeeper stopped")
def _run(self):
while not self._stop_event.wait(timeout=random.uniform(60, 120)):
x = random.randint(5, 20)
y = random.randint(5, 20)
self._mouse.position = (x, y)
self._mouse.click(self._button)
logging.debug("AwakeKeeper clicked at (%d, %d)", x, y)
def send_toggle():
"""Send toggle signal to running blurt instance (Linux only)."""
if sys.platform == "win32":
@@ -363,19 +435,25 @@ class BlurtApp:
self.icon: pystray.Icon | None = None
self.device_index: int | None = None
self.input_devices = self._get_input_devices()
self.transcription_model = load_transcription_model()
self.transcription_language = load_transcription_language()
self.api_keys = {
"openai": load_api_key(),
"openrouter": load_openrouter_key(),
}
self.transcription_model = self._available_model_or_fallback(
load_transcription_model()
)
self.transcription_language = load_transcription_language()
self.session = requests.Session()
self._awake_keeper: AwakeKeeper | None = (
AwakeKeeper() if sys.platform == "win32" else None
)
self.pending_transcriptions = 0
logging.info(
"BlurtApp initialized with %d input devices, model=%s, language=%s",
"BlurtApp initialized with %d input devices, model=%s, language=%s, awake_available=%s",
len(self.input_devices),
self.transcription_model,
self.transcription_language,
self._awake_keeper is not None,
)
def _get_input_devices(self) -> list[dict]:
@@ -410,25 +488,63 @@ class BlurtApp:
self.icon.title = f"{APP_NAME} - {state}"
self.icon.update_menu()
def _audio_callback(self, indata, frames, time_info, status):
if status:
print(f"sounddevice: {status}", file=sys.stderr)
self.audio_frames.append(indata.copy())
def _provider_for_model(self, model: str) -> str:
for model_id, _, provider in TRANSCRIPTION_MODELS:
if model_id == model:
return provider
return "openai"
def _available_model_or_fallback(self, model: str) -> str:
provider = self._provider_for_model(model)
if self.api_keys.get(provider):
return model
for fallback_model, _, fallback_provider in TRANSCRIPTION_MODELS:
if self.api_keys.get(fallback_provider):
logging.warning(
"Model %s requires a %s API key, which is unavailable; "
"using available model %s for this session",
model,
provider,
fallback_model,
)
return fallback_model
logging.error(
"Model %s requires a %s API key, but no API keys are available for any provider",
model,
provider,
)
return model
def _current_api_key(self) -> str:
return self.api_keys.get(self._provider_for_model(self.transcription_model), "")
def _audio_callback(self, indata, frames, time_info, status):
if status:
print(f"sounddevice: {status}", file=sys.stderr)
self.audio_frames.append(indata.copy())
def _api_key_label(self) -> str:
provider = self._provider_for_model(self.transcription_model)
provider_label = "OpenRouter" if provider == "openrouter" else "OpenAI"
key = self._current_api_key()
if not key:
return f"API key: {provider_label} (missing)"
return f"API key: {provider_label} (…{key[-4:]})"
def _start_recording(self):
if not self._current_api_key():
provider = self._provider_for_model(self.transcription_model)
key_file = PROVIDER_KEY_FILES.get(provider, "api_key.txt")
self.platform.notify("Error", f"No {provider} API key. Put your key in {key_file}")
key_hint = PROVIDER_KEY_HINTS.get(provider, f"set the {provider} API key")
logging.error(
"Recording blocked: model=%s provider=%s has no API key; "
"%s or select a model with an available key",
self.transcription_model,
provider,
key_hint,
)
self.platform.notify("Error", f"No {provider} API key; {key_hint}.")
return
self.audio_frames = []
try:
@@ -529,14 +645,18 @@ class BlurtApp:
self._update_icon()
def _transcribe_openai(self, wav_bytes: bytes) -> dict:
data = {
"model": self.transcription_model,
"language": self.transcription_language,
}
if self.transcription_model == "gpt-4o-transcribe-diarize":
data.update(response_format="diarized_json", chunking_strategy="auto")
response = self.session.post(
OPENAI_TRANSCRIPTIONS_URL,
headers={"Authorization": f"Bearer {self.api_keys['openai']}"},
files={"file": ("blurt.wav", wav_bytes, "audio/wav")},
data={
"model": self.transcription_model,
"language": self.transcription_language,
},
data=data,
timeout=30,
)
self._log_transcription_response(response)
@@ -544,8 +664,6 @@ class BlurtApp:
return response.json()
def _transcribe_openrouter(self, wav_bytes: bytes) -> dict:
# OpenRouter's transcription endpoint takes a JSON body with base64-encoded
# audio (not OpenAI's multipart upload). The response still has a "text" field.
audio_b64 = base64.b64encode(wav_bytes).decode("ascii")
response = self.session.post(
OPENROUTER_TRANSCRIPTIONS_URL,
@@ -602,13 +720,35 @@ class BlurtApp:
logging.exception("Could not open log file")
self.platform.notify("Error", f"Could not open log file: {e}")
def _toggle_awake(self, icon, item):
if self._awake_keeper is None:
return
if self._awake_keeper.is_active:
self._awake_keeper.stop()
else:
self._awake_keeper.start()
self.icon.update_menu()
def _select_model(self, model: str):
def handler(icon, item):
self.transcription_model = model
self.api_keys = {
"openai": load_api_key(),
"openrouter": load_openrouter_key(),
}
provider = self._provider_for_model(model)
if not self.api_keys.get(provider):
logging.error(
"Model selection blocked: model=%s provider=%s has no API key",
model,
provider,
)
self.platform.notify(
"Model Unavailable",
f"No {provider} API key is configured.",
)
self.icon.update_menu()
return
self.transcription_model = model
save_transcription_model(model)
self.icon.update_menu()
self.platform.notify("Model Selected", get_model_label(model))
@@ -627,6 +767,11 @@ class BlurtApp:
return self.transcription_model == model
return check
def _is_provider_available(self, provider: str):
def check(item):
return bool(self.api_keys.get(provider))
return check
def _is_language_selected(self, language: str):
def check(item):
return self.transcription_language == language
@@ -656,12 +801,13 @@ class BlurtApp:
)
model_items = []
for model, label, _ in TRANSCRIPTION_MODELS:
for model, label, provider in TRANSCRIPTION_MODELS:
model_items.append(
pystray.MenuItem(
label,
self._select_model(model),
checked=self._is_model_selected(model),
enabled=self._is_provider_available(provider),
radio=True,
)
)
@@ -677,6 +823,16 @@ class BlurtApp:
)
)
awake_items = []
if self._awake_keeper is not None:
awake_items.append(
pystray.MenuItem(
"Awake",
self._toggle_awake,
checked=lambda item: self._awake_keeper.is_active,
)
)
return pystray.Menu(
pystray.MenuItem(
lambda item: "Stop Recording" if self.recording else "Start Recording",
@@ -686,8 +842,10 @@ class BlurtApp:
pystray.Menu.SEPARATOR,
pystray.MenuItem("Input Device", pystray.Menu(*device_items)),
pystray.MenuItem("Model", pystray.Menu(*model_items)),
pystray.MenuItem(lambda item: self._api_key_label(), None, enabled=False),
pystray.MenuItem("Language", pystray.Menu(*language_items)),
pystray.MenuItem("Show Logs", self._show_logs),
*awake_items,
pystray.Menu.SEPARATOR,
pystray.MenuItem("Quit", self._on_quit),
)
@@ -695,6 +853,8 @@ class BlurtApp:
def _on_quit(self, icon, item):
if self.recording:
self._stop_recording()
if self._awake_keeper and self._awake_keeper.is_active:
self._awake_keeper.stop()
self.platform.cleanup_toggle_trigger()
icon.stop()