255 lines
12 KiB
Python
255 lines
12 KiB
Python
"""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()
|