From 55239f71a05bb69c21fdc6c315b9f17be329e18b Mon Sep 17 00:00:00 2001 From: mrwnslz Date: Mon, 9 Nov 2026 11:17:23 +0100 Subject: [PATCH] =?UTF-8?q?Transkription:=20eingebaute=20Instanz=20l=C3=A4?= =?UTF-8?q?uft=20auf=20dem=20PC=20im=20Tailnet;=20ist=20sie=20nicht=20erre?= =?UTF-8?q?ichbar,=20wartet=20der=20Auftrag=20und=20versucht=20es=20alle?= =?UTF-8?q?=2010=20Minuten=20neu=20(h=C3=B6chstens=20zwei=20Tage);=20Whisp?= =?UTF-8?q?er-Server=20kann=20CUDA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- deploy/whisper/server.py | 17 +++++++--- docs/transcription.md | 22 +++++++++++-- podcasts/test_transcribe.py | 35 +++++++++++++++++++++ podcasts/transcribe.py | 63 ++++++++++++++++++++++++++++++++++--- 4 files changed, 125 insertions(+), 12 deletions(-) diff --git a/deploy/whisper/server.py b/deploy/whisper/server.py index 571e012..1e26bbe 100644 --- a/deploy/whisper/server.py +++ b/deploy/whisper/server.py @@ -12,12 +12,19 @@ import subprocess import threading import time -import numpy as np -from fastapi import FastAPI, File, Form, HTTPException, UploadFile -from faster_whisper import WhisperModel +# Windows with a GPU: where the CUDA libraries (cuBLAS, cuDNN) are, separated by ';'. +for _d in filter(None, os.environ.get('WHISPER_DLL_DIRS', '').split(';')): + os.add_dll_directory(_d) + os.environ['PATH'] = _d + os.pathsep + os.environ['PATH'] + +import numpy as np # noqa: E402 +from fastapi import FastAPI, File, Form, HTTPException, UploadFile # noqa: E402 +from faster_whisper import WhisperModel # noqa: E402 MODEL = os.environ.get('WHISPER_MODEL', 'large-v3-turbo') THREADS = int(os.environ.get('WHISPER_THREADS', '4')) +DEVICE = os.environ.get('WHISPER_DEVICE', 'cpu') # cpu or cuda +COMPUTE = os.environ.get('WHISPER_COMPUTE', 'int8') # int8_float16 on a GPU IDLE = int(os.environ.get('WHISPER_IDLE', '600')) CACHE = os.environ.get('WHISPER_CACHE', '/models') @@ -28,7 +35,7 @@ state = {'model': None, 'last': 0.0} def model(): if state['model'] is None: - state['model'] = WhisperModel(MODEL, device='cpu', compute_type='int8', cpu_threads=THREADS, + state['model'] = WhisperModel(MODEL, device=DEVICE, compute_type=COMPUTE, cpu_threads=THREADS, download_root=CACHE) return state['model'] @@ -56,7 +63,7 @@ def decode(data: bytes) -> np.ndarray: @app.get('/health') def health(): - return {'ok': True, 'model': MODEL, 'loaded': state['model'] is not None} + return {'ok': True, 'model': MODEL, 'device': DEVICE, 'loaded': state['model'] is not None} @app.post('/v1/audio/transcriptions') diff --git a/docs/transcription.md b/docs/transcription.md index 95ca239..c9f29a7 100644 --- a/docs/transcription.md +++ b/docs/transcription.md @@ -19,8 +19,8 @@ only way to start a transcription is the `POST` below, sent when the user presse The **server** transcribes. It already downloads and decodes episodes for skipping silence (`podcasts/silence.py`), and browsers and the WebView can't fetch podcast audio (CORS). It cuts the audio at pauses and sends the pieces in parallel to a Whisper instance with an -OpenAI-compatible API (`POST /v1/audio/transcriptions`): the built-in one by default (a container -beside diora-web, `DIORA_WHISPER_URL`), or any instance the user enters. YouTube episodes are +OpenAI-compatible API (`POST /v1/audio/transcriptions`): the built-in one by default (`DIORA_WHISPER_URL`), +or any instance the user enters. YouTube episodes are not transcribed: the server never fetches their sound (docs/youtube.md). The **device** writes the article with the user's own key (docs/ai.md). The key never reaches the @@ -242,3 +242,21 @@ using the metadata, and don't quote garbled passages. ``` Prompt (user turn): the metadata block, then `Transkript:` and the text. + +## The built-in instance (since 2026-11-09) + +Whisper does not fit on the Strato server: with about 2 GB of memory free, large-v3-turbo needed +more than 2.2 GB and even `small` broke a 1.2 GB limit, and earlyoom then killed other services. +So the built-in instance runs on the user's PC (multi11, GTX 1660 Ti) and diora-prod reaches it over +the tailnet: `DIORA_WHISPER_URL=http://100.64.0.3:8766/v1` in `~/docker-compose/diora-prod/.env`. +It is `deploy/whisper/server.py` (a small OpenAI-compatible server around faster-whisper, +large-v3-turbo, int8_float16 on CUDA; one request at a time; the model is loaded on demand and freed +after 10 idle minutes), started at logon from `F:\diora-whisper` (start.cmd via start-hidden.vbs, +shortcut in the user's Startup folder; log `F:\diora-whisper\server.log`), listening only on the +tailnet address. One hour of speech takes about five minutes. + +When the PC is off, a job does not fail: the server checks the instance before downloading, and a +connection error (there or while pieces are out) puts the job back into the queue with +`error = "Whisper-Instanz nicht erreichbar, neuer Versuch um HH:MM"`, tried again every ten +minutes (`WAIT_RETRY`) for up to two days after the button was pressed (`WAIT_MAX`). Only the +built-in instance waits like this; a custom instance's connection errors fail the job as before. diff --git a/podcasts/test_transcribe.py b/podcasts/test_transcribe.py index f751854..589469c 100644 --- a/podcasts/test_transcribe.py +++ b/podcasts/test_transcribe.py @@ -299,6 +299,41 @@ class RunTests(PipelineBase): self.assertIn('Whisper', job.error) self.assertEqual(self.fake.calls, []) + def test_unreachable_builtin_waits_and_tries_again(self): + import socket + sock = socket.socket() + sock.bind(('127.0.0.1', 0)) + port = sock.getsockname()[1] + sock.close() # nothing listens there now + with override_settings(WHISPER_URL=f'http://127.0.0.1:{port}/v1'): + self.configure() + job = self.job() + transcribe.run(job.pk) + job = self.reload(job) + self.assertEqual(job.state, 'queued') + self.assertIn('nicht erreichbar', job.error) + self.assertGreater(transcribe._not_before[job.pk], time.time()) + self.assertEqual(self.fake.calls, []) # nothing downloaded while the instance is away + self.assertEqual(transcribe._next_job(), transcribe.WAIT) + transcribe._not_before[job.pk] = 0 # due again: the instance is back + with override_settings(WHISPER_URL=self.fake.whisper): + transcribe.run(job.pk) + job = self.reload(job) + self.assertEqual((job.state, job.error), ('done', '')) + transcribe._not_before.clear() + + def test_waiting_gives_up_after_two_days(self): + from datetime import timedelta + from django.utils import timezone + with override_settings(WHISPER_URL='http://127.0.0.1:9/v1'): + self.configure() + job = self.job() + TranscriptionJob.objects.filter(pk=job.pk).update(created_at=timezone.now() - timedelta(days=3)) + transcribe.run(job.pk) + job = self.reload(job) + self.assertEqual(job.state, 'error') + self.assertIn('zwei Tage', job.error) + @override_settings(WHISPER_URL='') def test_retries_then_succeeds(self): self.fake.fail_first = 2 diff --git a/podcasts/transcribe.py b/podcasts/transcribe.py index fac15be..4bc2f30 100644 --- a/podcasts/transcribe.py +++ b/podcasts/transcribe.py @@ -11,6 +11,7 @@ taken oldest first (by id). A job found `downloading` or `transcribing` when the just started (a restart killed it) is queued again. Cancelling deletes the job's row: the worker notices when it next writes progress and stops. """ +import datetime import logging import os import subprocess @@ -42,10 +43,15 @@ DECODE_TIMEOUT = 3600 OPUS_BITRATE = '32k' MAX_PROMPT = 500 # characters ALIVE_CHECK = 5 # seconds between looks whether the job was cancelled +WAIT_RETRY = 600 # seconds between tries while the built-in instance can't be reached +WAIT_MAX = 48 * 3600 # after this long since the button was pressed a waiting job gives up +WAIT_IDLE = 30 # seconds the runner sleeps while every queued job is waiting _lock = threading.Lock() _runner = None # the thread working through the queue, or None _resumed = False # whether jobs left over from before a restart were queued again +_not_before = {} # job id -> time.time() before which a waiting job isn't tried again +WAIT = 'wait' # _next_job: jobs are queued, but none is due yet class Cancelled(Exception): @@ -56,6 +62,11 @@ class Failed(Exception): """The job can't succeed; the message is shown to the user.""" +class Unreachable(Exception): + """The built-in instance can't be reached (e.g. the machine it runs on is off): the job waits + and tries again later instead of failing.""" + + class Permanent(Exception): """A request the instance refused for good (a client error): not worth a retry.""" @@ -85,11 +96,14 @@ def kick(): def _next_job(): global _runner with _lock: - job_id = (TranscriptionJob.objects.filter(state=TranscriptionJob.QUEUED) - .order_by('id').values_list('id', flat=True).first()) - if job_id is None: + queued = list(TranscriptionJob.objects.filter(state=TranscriptionJob.QUEUED) + .order_by('id').values_list('id', flat=True)) + if not queued: _runner = None # decided under the lock, so a job queued right now starts a new runner - return job_id + return None + now = time.time() + due = next((i for i in queued if _not_before.get(i, 0) <= now), None) + return due if due is not None else WAIT def _loop(): @@ -100,6 +114,9 @@ def _loop(): job_id = _next_job() if job_id is None: return + if job_id == WAIT: + time.sleep(WAIT_IDLE) + continue try: run(job_id) except Exception: # run() reports its own failures; this keeps the queue going @@ -124,6 +141,8 @@ def run(job_id): _write(job_id, state=TranscriptionJob.DONE, data=data, error='', finished_at=timezone.now()) except Cancelled: log.info('transcription job %s cancelled', job_id) + except Unreachable as e: + _wait(job_id, str(e)) except Failed as e: _fail(job_id, str(e)) except Exception as e: @@ -131,6 +150,31 @@ def run(job_id): _fail(job_id, f'{type(e).__name__}: {e}') +def _wait(job_id, reason): + """Back into the queue to try again in WAIT_RETRY seconds, or give up when it waited too long.""" + job = TranscriptionJob.objects.filter(pk=job_id).only('created_at').first() + if job is None: + return + if (timezone.now() - job.created_at).total_seconds() > WAIT_MAX: + _not_before.pop(job_id, None) + _fail(job_id, 'Whisper-Instanz zwei Tage lang nicht erreichbar') + return + _not_before[job_id] = time.time() + WAIT_RETRY + again = timezone.localtime(timezone.now() + datetime.timedelta(seconds=WAIT_RETRY)).strftime('%H:%M') + log.info('transcription job %s waits for the instance (%s)', job_id, reason) + TranscriptionJob.objects.filter(pk=job_id).update( + state=TranscriptionJob.QUEUED, chunks_done=0, chunks_total=0, started_at=None, + error=f'Whisper-Instanz nicht erreichbar, neuer Versuch um {again}') + + +def _probe(url): + """Unreachable unless something answers at the built-in instance's address.""" + try: + requests.get(url.rsplit('/audio/transcriptions', 1)[0], timeout=(5, 10)).close() + except (requests.ConnectionError, requests.Timeout) as e: + raise Unreachable(e.__class__.__name__) + + def _fail(job_id, message): TranscriptionJob.objects.filter(pk=job_id).update( state=TranscriptionJob.ERROR, error=(message or 'failed')[:500], finished_at=timezone.now()) @@ -173,6 +217,8 @@ def _transcribe(job_id): episode = job.episode conf = for_user(episode.feed.user) url, headers, builtin = _target(conf) + if builtin: + _probe(url) # before downloading: is the instance there at all? options = {'model': conf.model or DEFAULT_MODEL, 'response_format': 'verbose_json'} if conf.language: options['language'] = conf.language @@ -312,7 +358,10 @@ def _post(url, headers, builtin, name, mime, body, options): try: if builtin: # The operator's own instance (DIORA_WHISPER_URL) may sit on a private network. - r = requests.post(url, headers=headers, files=files, data=options, timeout=UPLOAD_TIMEOUT) + try: + r = requests.post(url, headers=headers, files=files, data=options, timeout=UPLOAD_TIMEOUT) + except requests.ConnectionError as e: + raise Unreachable(e.__class__.__name__) if r.status_code >= 400: if r.status_code < 500 and r.status_code not in (408, 429): raise Permanent(f'HTTP {r.status_code}') @@ -343,6 +392,10 @@ def _send(url, headers, builtin, name, mime, body, options, stop): return _post(url, headers, builtin, name, mime, body, options) except Permanent as e: raise Failed(f'Whisper: {e}') + except Unreachable: + if attempt >= RETRIES: + raise + time.sleep(RETRY_PAUSES[min(attempt, len(RETRY_PAUSES) - 1)]) except RuntimeError as e: if attempt >= RETRIES: raise Failed(f'Whisper: {e}')