"""Container-local Whisper jobs; the Java API remains the only public endpoint.""" from contextlib import contextmanager, asynccontextmanager import hashlib import logging import os from pathlib import Path import re import sqlite3 import threading import uuid from fastapi import FastAPI, File, Form, Header, HTTPException, UploadFile from starlette.middleware.trustedhost import TrustedHostMiddleware from summary import SummaryError, load_summarizer LOG = logging.getLogger("swyxweb.whisper") MAX_BYTES = 500 * 1024 * 1024 LANGUAGES = {"auto", "de", "en", "fr", "es", "it", "nl", "pl"} STYLES = {"standard", "short", "meeting"} PUBLIC_FIELDS = ("id", "filename", "status", "stage", "error", "transcript", "formatted_transcript", "summary") class Jobs: def __init__(self, root, transcribe, summarize=None): self.root = Path(root) self.root.mkdir(parents=True, exist_ok=True) self.transcribe = transcribe self.summarize = summarize self.stop = threading.Event() self.wake = threading.Event() with self.db() as db: db.execute("""CREATE TABLE IF NOT EXISTS jobs ( id TEXT PRIMARY KEY, recording_key TEXT UNIQUE NOT NULL, filename TEXT NOT NULL, digest TEXT NOT NULL, language TEXT NOT NULL, style TEXT NOT NULL, status TEXT NOT NULL, stage TEXT NOT NULL, error TEXT NOT NULL DEFAULT '', transcript TEXT NOT NULL DEFAULT '', formatted_transcript TEXT NOT NULL DEFAULT '', summary TEXT NOT NULL DEFAULT '', created TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP)""") if "transcribed" not in {column["name"] for column in db.execute("PRAGMA table_info(jobs)")}: db.execute("ALTER TABLE jobs ADD COLUMN transcribed INTEGER NOT NULL DEFAULT 0") db.execute("UPDATE jobs SET transcribed=1 WHERE status='done' OR transcript<>''") # A killed container must not strand an acknowledged recording. db.execute("UPDATE jobs SET status='queued', stage='Wartet auf Verarbeitung' WHERE status IN ('transcribing', 'summarizing')") self.thread = threading.Thread(target=self.run, name="whisper", daemon=True) @contextmanager def db(self): connection = sqlite3.connect(self.root / "jobs.sqlite3", timeout=30) connection.row_factory = sqlite3.Row try: with connection: yield connection finally: connection.close() def get(self, job_id): if not re.fullmatch(r"[a-f0-9]{32}", job_id): raise HTTPException(400, "Ungültige Auftragskennung.") with self.db() as db: row = db.execute("SELECT * FROM jobs WHERE id=?", (job_id,)).fetchone() if row is None: raise HTTPException(404, "Auftrag nicht gefunden.") return {key: row[key] for key in PUBLIC_FIELDS} def upload(self, source, filename, recording_key, language, style): if language not in LANGUAGES or style not in STYLES: raise HTTPException(400, "Ungültige Sprache oder Zusammenfassungsart.") if not re.fullmatch(r"swyxweb:[a-fA-F0-9-]{36}", recording_key): raise HTTPException(400, "Ungültige Aufnahme-ID.") try: recording_key = "swyxweb:" + str(uuid.UUID(recording_key[8:])) except ValueError as error: raise HTTPException(400, "Ungültige Aufnahme-ID.") from error if not filename or not filename.lower().endswith(".wav"): raise HTTPException(400, "Bitte eine WAV-Datei auswählen.") filename = filename.replace("\\", "/").split("/")[-1] job_id = uuid.uuid4().hex path = self.root / f"{job_id}.wav" digest = hashlib.sha256() size = 0 committed = False try: with path.open("xb") as output: while chunk := source.read(1024 * 1024): if size == 0 and (len(chunk) < 12 or chunk[:4] not in (b"RIFF", b"RF64") or chunk[8:12] != b"WAVE"): raise HTTPException(400, "Die Datei hat keinen WAV-Dateikopf.") size += len(chunk) if size > MAX_BYTES: raise HTTPException(413, "Maximal 500 MB pro Aufnahme.") digest.update(chunk) output.write(chunk) if size < 12: raise HTTPException(400, "Die WAV-Datei ist leer.") output.flush() os.fsync(output.fileno()) # Persist the directory entry before acknowledging the upload. directory = os.open(self.root, os.O_RDONLY) try: os.fsync(directory) finally: os.close(directory) with self.db() as db: db.execute("BEGIN IMMEDIATE") existing = db.execute("SELECT * FROM jobs WHERE recording_key=?", (recording_key,)).fetchone() if existing is not None: if (existing["digest"], existing["language"], existing["style"]) != (digest.hexdigest(), language, style): raise HTTPException(409, "Diese Aufnahme-ID wurde bereits mit anderen Daten übergeben.") return {key: existing[key] for key in PUBLIC_FIELDS} db.execute("""INSERT INTO jobs (id, recording_key, filename, digest, language, style, status, stage) VALUES (?, ?, ?, ?, ?, ?, 'queued', 'Wartet auf lokale Transkription')""", (job_id, recording_key, filename, digest.hexdigest(), language, style)) committed = True finally: if not committed: path.unlink(missing_ok=True) self.wake.set() return self.get(job_id) def retry(self, job_id): self.get(job_id) with self.db() as db: updated = db.execute("""UPDATE jobs SET status='queued', stage='Wartet auf Verarbeitung', error='' WHERE id=? AND status='failed'""", (job_id,)).rowcount if not updated: raise HTTPException(409, "Nur fehlgeschlagene Aufträge können erneut gestartet werden.") self.wake.set() return self.get(job_id) def process_next(self): with self.db() as db: db.execute("BEGIN IMMEDIATE") job = db.execute("SELECT * FROM jobs WHERE status='queued' ORDER BY created, rowid LIMIT 1").fetchone() if job is None: return False db.execute("UPDATE jobs SET status=?, stage=? WHERE id=?", ("summarizing" if job["transcribed"] else "transcribing", "Zusammenfassung wird erstellt" if job["transcribed"] else "Whisper erstellt das Transkript lokal", job["id"])) transcribed = bool(job["transcribed"]) try: transcript = job["transcript"] if not transcribed: transcript = self.transcribe(self.root / f"{job['id']}.wav", job["language"]) # Save before contacting the LLM; retries never need to rerun Whisper. with self.db() as db: db.execute("UPDATE jobs SET transcript=?, transcribed=1 WHERE id=?", (transcript, job["id"])) transcribed = True summary = "" if self.summarize and transcript.strip(): with self.db() as db: db.execute("UPDATE jobs SET status='summarizing', stage='Zusammenfassung wird erstellt' WHERE id=?", (job["id"],)) summary = self.summarize(transcript, job["style"]) with self.db() as db: stage = "Transkript und Zusammenfassung fertig" if summary else ( "Keine Sprache erkannt" if not transcript.strip() else "Transkript fertig · Zusammenfassung nicht aktiviert") db.execute("UPDATE jobs SET status='done', stage=?, summary=?, error='' WHERE id=?", (stage, summary, job["id"])) except Exception as error: if transcribed: # Provider exceptions can contain credentials or transcript excerpts; do not log them. LOG.warning("Zusammenfassung fehlgeschlagen: %s (%s)", job["id"], type(error).__name__) message = str(error) if isinstance(error, SummaryError) else "Die Zusammenfassung ist fehlgeschlagen. Bitte LLM-Konfiguration prüfen und erneut versuchen." else: LOG.exception("Lokale Transkription fehlgeschlagen: %s", job["id"]) message = "Whisper konnte die WAV-Datei nicht transkribieren. Details stehen im Container-Log." with self.db() as db: db.execute("UPDATE jobs SET status='failed', stage=?, error=? WHERE id=?", ("Zusammenfassung fehlgeschlagen" if transcribed else "Transkription fehlgeschlagen", message, job["id"])) return True def run(self): while not self.stop.is_set(): self.wake.clear() if not self.process_next(): self.wake.wait(1) def close(self): self.stop.set() self.wake.set() if self.thread.is_alive(): self.thread.join(timeout=5) def load_whisper(): from faster_whisper import WhisperModel model = WhisperModel(os.environ.get("WHISPER_MODEL_PATH", "/app/models/whisper"), device="cpu", compute_type="int8", local_files_only=True, cpu_threads=int(os.environ.get("WHISPER_THREADS", "4")), num_workers=1) def transcribe(path, language): segments, _ = model.transcribe(str(path), language=None if language == "auto" else language, task="transcribe", beam_size=5, condition_on_previous_text=False, vad_filter=True) # faster-whisper returns a lazy generator; consume it before marking the job done. return " ".join(segment.text.strip() for segment in segments).strip() return transcribe class UploadLimitMiddleware: """Bound the request before the multipart parser spools it to disk.""" def __init__(self, app): self.app = app async def __call__(self, scope, receive, send): size = 0 async def limited_receive(): nonlocal size message = await receive() size += len(message.get("body", b"")) if size > MAX_BYTES + 1024 * 1024: raise HTTPException(413, "Maximal 500 MB pro Aufnahme.") return message await self.app(scope, limited_receive if scope["type"] == "http" else receive, send) def create_app(root=None, transcribe=None, summarize=None): @asynccontextmanager async def lifespan(app): jobs = Jobs(root or os.environ.get("WHISPER_DATA_DIR", "/data/transcriptions"), transcribe or load_whisper(), summarize if summarize is not None else load_summarizer()) app.state.jobs = jobs jobs.thread.start() try: yield finally: jobs.close() app = FastAPI(title="SwyxWeb Whisper", lifespan=lifespan, docs_url=None, redoc_url=None, openapi_url=None) app.add_middleware(UploadLimitMiddleware) app.add_middleware(TrustedHostMiddleware, allowed_hosts=["127.0.0.1", "localhost", "[::1]"]) @app.get("/health") def health(): if not app.state.jobs.thread.is_alive(): raise HTTPException(503, "Whisper-Verarbeitung ist nicht verfügbar.") return {"status": "ready"} @app.post("/api/jobs", status_code=202) def upload(file: UploadFile = File(...), language: str = Form("de"), style: str = Form("meeting"), idempotency_key: str = Header(...)): return app.state.jobs.upload(file.file, file.filename, idempotency_key, language, style) @app.get("/api/jobs/{job_id}") def get(job_id: str): return app.state.jobs.get(job_id) @app.post("/api/jobs/{job_id}/retry", status_code=202) def retry(job_id: str): return app.state.jobs.retry(job_id) return app app = create_app()