From ae5341d4419ee53b3766f83af90d91d6553b0ccc Mon Sep 17 00:00:00 2001 From: mrwnslz Date: Mon, 9 Nov 2026 10:01:35 +0100 Subject: [PATCH] =?UTF-8?q?Smarte=20Transkription=20C:=20Transkriptions-Pi?= =?UTF-8?q?peline=20(podcasts/transcribe.py),=20Endpunkte=20/transcribe/,?= =?UTF-8?q?=20/transcription/pending/,=20/article/claim/,=20Transkript-R?= =?UTF-8?q?=C3=BCckfall=20auf=20Whisper,=20transcripts=20in=20/feeds/notif?= =?UTF-8?q?ications/?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Die Folge wird geladen, zu 16 kHz mono dekodiert und an Pausen in Stücke von etwa acht Minuten geschnitten (FLAC für die eingebaute, Opus für fremde Instanzen), parallel an Whisper geschickt (drei Wiederholungen mit wachsender Pause) und per transcripts.merge geglättet. Ein Daemon-Thread arbeitet die Warteschlange ab, ein Auftrag zugleich; hängengebliebene Aufträge starten nach einem Neustart beim ersten Aufruf neu. Abbrechen löscht die Zeile des Auftrags. Gerät der Reservierung: device im Body oder Header X-Diora-Device (docs/transcription.md). Co-Authored-By: Claude Sonnet 5.5 --- docs/transcription.md | 23 +- feeds/views.py | 27 +- podcasts/test_transcribe.py | 786 ++++++++++++++++++++++++++++++++++++ podcasts/transcribe.py | 378 +++++++++++++++++ podcasts/transcription.py | 147 ++++++- podcasts/urls.py | 3 + podcasts/views.py | 40 +- 7 files changed, 1385 insertions(+), 19 deletions(-) create mode 100644 podcasts/test_transcribe.py create mode 100644 podcasts/transcribe.py diff --git a/docs/transcription.md b/docs/transcription.md index 4ac2821..b6668c5 100644 --- a/docs/transcription.md +++ b/docs/transcription.md @@ -70,8 +70,14 @@ updates `want_article` on a running one. With a transcript already there it retu `{state: "done"}` straight away (and keeps `want_article`). `GET` on the same address returns `{state, chunks_done, chunks_total, error, want_article, started_at, finished_at}`, with `state` one of `none`, `queued`, `downloading`, `transcribing`, `done`, `error`. `DELETE` cancels a -queued or running job. The server runs one job at a time; the others wait as `queued`. A YouTube -episode returns 400 `{error: "youtube"}`, an episode over 800 MB 400 `{error: "too large"}`. +queued or running job (its row is deleted, so `GET` then says `none`; a finished one stays). The server runs one job at a time; the others wait as `queued`. A YouTube +episode returns 400 `{error: "youtube"}`, an episode over 800 MB 400 `{error: "too large"}`. The size is asked with a `HEAD` +request (`Content-Length`; if that fails the job starts anyway) and checked again while +downloading, where it ends the job with the error `too large`. `article: true` is only ever +raised, never lowered, by a repeated `POST`; a `POST` after an `error` starts a new job (at the +back of the queue). Jobs run in id order, which is why the notice below can use ids as its cursor. +A feed transcript already there gives a job that is `done` with no `data`: it only holds the wish +for an article and causes no `transcripts` notice. **Transcript.** The result is kept on the job (`TranscriptionJob.data`) in the format of `EpisodeTranscript.data`: `{segments: [{start, end, speaker, text}], source: {type: "whisper", @@ -102,6 +108,11 @@ that want an article, have none and aren't reserved. `POST /podcasts/episodes//article/claim/` reserves one for 15 minutes for the calling device (200), or answers 409 when another device holds it, so phone and desktop don't both write it. `DELETE` on the claim releases it (the device gave up, e.g. no key or an error the user saw). +The device names itself with `device` in the JSON body (or, if that is missing, the +`X-Diora-Device` header): a short stable name such as `android`, `desktop` or `browser`, at most 60 +characters; a claim without one is 400. The same device claiming again renews its 15 minutes. A +claim on an episode with no open wish (no job done, no article wanted, or an article already +there) is 404. `DELETE` by a device that does not hold the claim is 409. **Notice.** `GET /feeds/notifications/` additionally takes `t_after` and returns `transcripts: [{job_id, episode_id, title, podcast, want_article}]` (jobs done after that id) and @@ -121,7 +132,13 @@ run in a thread pool of `parallel` workers; each is retried up to three times wi pause; progress is written after each. Segment times are shifted by the piece's start; the segments are joined and smoothed with `transcripts.merge`. A job runs in a daemon thread, like `silence.ensure()`; one per server. A job found `downloading` or `transcribing` after a restart -is started again. +is started again. This happens when the first +request (`GET`/`POST` on `/transcribe/` or `/transcription/pending/`) after a start reaches the server. +Cancelling deletes the job's row; the worker notices at the latest after five seconds, stops +sending pieces and removes its temporary files (they live in a temporary directory that is always +removed). Only the worker's main thread writes to the database. The settings used are those of the +episode's owner, read when the job starts. A refusal by the instance with a 4xx status other than +408 and 429 is not retried and ends the job with `Whisper: HTTP `. ## The article (`static/js/podcast-article.js`) diff --git a/feeds/views.py b/feeds/views.py index b51e476..116cf64 100644 --- a/feeds/views.py +++ b/feeds/views.py @@ -746,17 +746,32 @@ def star(request, pk): def notifications(request): """What a device should announce: posts marked when they came in, still unread, of blogs that still have the bell on, after the id the device saw - last. Without `after` (a device's first time) nothing, only where to start.""" + last. Without `after` (a device's first time) nothing, only where to start. + The same for finished transcripts (`t_after`, `transcripts`, `t_last`; jobs the + user's episodes have, docs/transcription.md).""" if (err := _auth(request)): return err qs = Item.objects.filter(user=request.user, notify=True) last = qs.order_by('-id').values_list('id', flat=True).first() or 0 + # Jobs run in id order, one at a time, so a job never finishes below an id already seen. + # (A job that is done with no data only reflects a feed transcript: no notice for it.) + from podcasts.models import TranscriptionJob + jobs = TranscriptionJob.objects.filter( + episode__feed__user=request.user, state=TranscriptionJob.DONE, data__isnull=False) + t_last = jobs.order_by('-id').values_list('id', flat=True).first() or 0 + out = {'items': [], 'last': last, 'transcripts': [], 't_last': t_last} + t_after = request.GET.get('t_after', '') + if t_after.isdigit(): + out['transcripts'] = [{ + 'job_id': j.id, 'episode_id': j.episode_id, 'title': j.episode.title, + 'podcast': j.episode.feed.title, 'want_article': j.want_article, + } for j in jobs.filter(id__gt=int(t_after)).select_related('episode__feed').order_by('id')[:50]] after = request.GET.get('after', '') - if not after.isdigit(): - return JsonResponse({'items': [], 'last': last}) - items = (qs.filter(id__gt=int(after), read_at=None, gone=False, source__notify=True) - .select_related('source').order_by('-published')[:50]) - return JsonResponse({'items': [_item_data(i) for i in items], 'last': last}) + if after.isdigit(): + items = (qs.filter(id__gt=int(after), read_at=None, gone=False, source__notify=True) + .select_related('source').order_by('-published')[:50]) + out['items'] = [_item_data(i) for i in items] + return JsonResponse(out) @csrf_exempt diff --git a/podcasts/test_transcribe.py b/podcasts/test_transcribe.py new file mode 100644 index 0000000..f751854 --- /dev/null +++ b/podcasts/test_transcribe.py @@ -0,0 +1,786 @@ +"""The transcription pipeline (podcasts/transcribe.py) against a fake Whisper instance, the +endpoints around it and the notice in /feeds/notifications/.""" +import email +import io +import json +import math +import os +import struct +import subprocess +import tempfile +import threading +import time +import wave +from datetime import timedelta +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from unittest.mock import patch + +from django.contrib.auth.models import User +from django.db import OperationalError +from django.test import TestCase, TransactionTestCase, override_settings +from django.utils import timezone + +from feeds.models import Item, Source + +from . import transcribe, transcription, youtube +from .models import ( + EpisodeArticle, EpisodeTranscript, PodcastEpisode, PodcastFeed, TranscriptionJob, TranscriptionSettings, +) + +RATE = 16000 +TOTAL = 120 +# Where the long pauses are (start, length): the pieces should be cut in these. The short +# ones (0.3 s every 4 s) are only there to tempt. +LONG_PAUSES = [(36.5, 1.6), (83.0, 1.4)] +SEGMENTS = [{'start': 0.0, 'end': 2.0, 'text': ' Hallo '}, {'start': 2.0, 'end': 5.5, 'text': 'Welt.'}] + + +def make_audio(): + """~2 minutes of tones with pauses, as mp3 bytes (and the wav as a check of the length).""" + samples = [] + t = 0.0 + pauses = [(s, s + n) for s, n in LONG_PAUSES] + for i in range(RATE * TOTAL): + t = i / RATE + silent = any(a <= t < b for a, b in pauses) or (t % 4.0) > 3.7 + value = 0 if silent else int(9000 * math.sin(2 * math.pi * (180 + 40 * ((i // 8000) % 5)) * t)) + samples.append(value) + buf = io.BytesIO() + with wave.open(buf, 'wb') as w: + w.setnchannels(1) + w.setsampwidth(2) + w.setframerate(RATE) + w.writeframes(struct.pack('<%dh' % len(samples), *samples)) + import imageio_ffmpeg + with tempfile.TemporaryDirectory() as tmp: + src, dst = os.path.join(tmp, 'a.wav'), os.path.join(tmp, 'a.mp3') + open(src, 'wb').write(buf.getvalue()) + subprocess.run([imageio_ffmpeg.get_ffmpeg_exe(), '-loglevel', 'error', '-i', src, '-b:a', '64k', dst], check=True) + return open(dst, 'rb').read() + + +_AUDIO = [] + + +def audio(): + if not _AUDIO: + _AUDIO.append(make_audio()) + return _AUDIO[0] + + +def decoded_seconds(data, suffix): + """How long the uploaded piece is, by decoding it.""" + import imageio_ffmpeg + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, 'p' + suffix) + open(path, 'wb').write(data) + out = subprocess.run([imageio_ffmpeg.get_ffmpeg_exe(), '-loglevel', 'error', '-i', path, '-f', 's16le', + '-ac', '1', '-ar', str(RATE), '-'], stdout=subprocess.PIPE, check=True).stdout + return len(out) / 2 / RATE + + +class Fake: + """A local HTTP server: GET /ep.mp3 is the test audio, POST /v1/audio/transcriptions a Whisper.""" + + def __init__(self, fail_first=0, fail_status=500, delay=0, on_call=None): + self.fail_first, self.fail_status, self.delay, self.on_call = fail_first, fail_status, delay, on_call + self.calls = [] # {'fields', 'filename', 'body', 'auth'} + self.lock = threading.Lock() + fake = self + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *a): + pass + + def do_GET(self): + if self.path == '/missing.mp3': + self.send_response(404) + self.end_headers() + return + body = audio_holder.data or audio() + self.send_response(200) + self.send_header('Content-Type', 'audio/mpeg') + self.send_header('Content-Length', str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_POST(self): + raw = self.rfile.read(int(self.headers['Content-Length'])) + msg = email.message_from_bytes( + b'Content-Type: ' + self.headers['Content-Type'].encode() + b'\r\n\r\n' + raw) + fields, filename, body = {}, '', b'' + for part in msg.get_payload(): + name = part.get_param('name', header='content-disposition') + if part.get_filename(): + filename, body = part.get_filename(), part.get_payload(decode=True) + else: + fields[name] = part.get_payload(decode=True).decode() + with fake.lock: + fake.calls.append({'fields': fields, 'filename': filename, 'body': body, + 'auth': self.headers.get('Authorization')}) + n = len(fake.calls) + if fake.on_call: + fake.on_call(n) + if fake.delay: + time.sleep(fake.delay) + if n <= fake.fail_first or fake.fail_first < 0: + self.send_response(fake.fail_status) + self.send_header('Content-Length', '0') + self.end_headers() + return + out = json.dumps({'language': 'german', 'segments': SEGMENTS}).encode() + self.send_response(200) + self.send_header('Content-Type', 'application/json') + self.send_header('Content-Length', str(len(out))) + self.end_headers() + self.wfile.write(out) + + self.server = ThreadingHTTPServer(('127.0.0.1', 0), Handler) + self.server.daemon_threads = True + self.port = self.server.server_address[1] + threading.Thread(target=self.server.serve_forever, daemon=True).start() + + @property + def whisper(self): + return f'http://127.0.0.1:{self.port}/v1' + + def url(self, path='/ep.mp3'): + return f'http://127.0.0.1:{self.port}{path}' + + def stop(self): + self.server.shutdown() + self.server.server_close() + + +def leftovers(): + base = tempfile.gettempdir() + return {d for d in os.listdir(base) if d.startswith('diora-transcribe-')} + + +FAST = dict(CHUNK_SECONDS=40, CUT_WINDOW=8, MIN_TAIL=10, RETRY_PAUSES=(0, 0, 0)) + + +def wait_for(cond, timeout=30): + end = time.time() + timeout + while time.time() < end: + if cond(): + return True + time.sleep(0.05) + return False + + +class PipelineBase(TransactionTestCase): + def setUp(self): + for k, v in FAST.items(): + patcher = patch.object(transcribe, k, v) + patcher.start() + self.addCleanup(patcher.stop) + self.user = User.objects.create_user(username='alice', password='pw12345678') + self.feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/f', title='Der Podcast', + author='Anna Autorin') + self.fake = Fake() + self.addCleanup(self.fake.stop) + self.before = leftovers() + + def job(self, audio_path='/ep.mp3', **kw): + n = PodcastEpisode.objects.count() + ep = PodcastEpisode.objects.create(feed=self.feed, guid=f'g{n}', title=f'Folge {n}', + audio_url=self.fake.url(audio_path), duration_seconds=TOTAL) + return TranscriptionJob.objects.create(episode=ep, state=TranscriptionJob.QUEUED, **kw) + + def reload(self, job): + for _ in range(50): # the in-memory test database may be locked for a moment by the worker + try: + return TranscriptionJob.objects.get(pk=job.pk) + except OperationalError: + time.sleep(0.05) + return TranscriptionJob.objects.get(pk=job.pk) + + def assertClean(self): + self.assertEqual(leftovers(), self.before) + + +class PlanTests(PipelineBase): + def test_cuts_land_in_the_long_pauses(self): + with tempfile.TemporaryDirectory() as tmp: + src, pcm = os.path.join(tmp, 'a.mp3'), os.path.join(tmp, 'a.pcm') + open(src, 'wb').write(audio()) + transcribe._decode(src, pcm) + total = os.path.getsize(pcm) / 2 / RATE + pieces = transcribe.plan(total, transcribe.pauses_of(pcm)) + self.assertAlmostEqual(total, TOTAL, delta=0.5) + self.assertEqual(len(pieces), 3) + cuts = [p[1] for p in pieces[:-1]] + for cut, (start, length) in zip(cuts, LONG_PAUSES): + self.assertTrue(start <= cut <= start + length, f'{cut} not in the pause {start}+{length}') + self.assertEqual(pieces[0][0], 0) + self.assertEqual(pieces[-1][1], total) + self.assertEqual([a[1] for a in pieces[:-1]], [b[0] for b in pieces[1:]]) + + def test_plan_without_pauses_cuts_at_the_target(self): + self.assertEqual(transcribe.plan(100, []), [(0.0, 40), (40, 80), (80, 100)]) + self.assertEqual(transcribe.plan(30, []), [(0.0, 30)]) + # a short tail is joined to the last piece + self.assertEqual(transcribe.plan(85, []), [(0.0, 40), (40, 85)]) + + +@override_settings(WHISPER_URL='') +class RunTests(PipelineBase): + def configure(self, **kw): + conf = TranscriptionSettings(user=self.user, **kw) + conf.save() + return conf + + def test_builtin_run_end_to_end(self): + with override_settings(WHISPER_URL=self.fake.whisper): + self.configure(language='de', parallel=2) + job = self.job() + started = time.time() + transcribe.run(job.pk) + took = time.time() - started + print(f'\n[pipeline run of the {TOTAL} s test audio took {took:.1f} s]') + job = self.reload(job) + self.assertEqual((job.state, job.error), ('done', '')) + self.assertEqual((job.chunks_done, job.chunks_total), (3, 3)) + self.assertIsNotNone(job.finished_at) + self.assertEqual(len(self.fake.calls), 3) + for c in self.fake.calls: + self.assertEqual(c['fields']['model'], 'whisper-1') + self.assertEqual(c['fields']['language'], 'de') + self.assertEqual(c['fields']['response_format'], 'verbose_json') + self.assertEqual(c['fields']['prompt'], 'Der Podcast. Folge 0. Anna Autorin') + self.assertTrue(c['filename'].endswith('.flac')) + self.assertIsNone(c['auth']) # the token only goes to custom instances + # the uploaded pieces are the planned ones: the same total, cut at the pauses + lengths = sorted(decoded_seconds(c['body'], '.flac') for c in self.fake.calls) + self.assertAlmostEqual(sum(lengths), TOTAL, delta=0.5) + data = job.data + self.assertEqual(data['source'], {'type': 'whisper', 'model': 'whisper-1', 'language': 'de'}) + segs = data['segments'] + starts = [s['start'] for s in segs] + self.assertEqual(starts, sorted(starts)) + self.assertTrue(all(s['speaker'] == '' for s in segs)) + # every piece answered "Hallo Welt." at 0 to 5.5 s: its start is where the piece starts, + # which lies on a cut in a long pause, and the first at zero + self.assertEqual(starts[0], 0) + cuts = starts[1:] + self.assertEqual(len(cuts), 2) + for cut, (p, n) in zip(cuts, LONG_PAUSES): + self.assertTrue(p <= cut <= p + n, f'{cut} not in {p}+{n}') + self.assertEqual(segs[0]['text'], 'Hallo Welt.') + self.assertAlmostEqual(segs[0]['end'], 5.5) + self.assertClean() + + def test_language_empty_takes_what_whisper_detects_and_custom_gets_opus_and_token(self): + conf = self.configure(engine='custom', url=self.fake.whisper, model='large-v3', language='') + conf.token_encrypted = transcription.logins.encrypt('tok') + conf.save() + job = self.job() + with patch.object(transcribe.chapters, 'assert_safe_url'), \ + patch.object(transcribe.chapters, '_assert_peer_is_safe'): + transcribe.run(job.pk) + job = self.reload(job) + self.assertEqual((job.state, job.error), ('done', '')) + self.assertEqual(job.data['source'], {'type': 'whisper', 'model': 'large-v3', 'language': 'german'}) + for c in self.fake.calls: + self.assertTrue(c['filename'].endswith('.ogg')) + self.assertNotIn('language', c['fields']) + self.assertEqual(c['auth'], 'Bearer tok') + self.assertLess(len(c['body']), decoded_seconds(c['body'], '.ogg') * 6500) # 32 kbit/s: about 4 kB a second, ogg and VBR on top + self.assertAlmostEqual(sum(decoded_seconds(c['body'], '.ogg') for c in self.fake.calls), TOTAL, delta=0.5) + self.assertClean() + + def test_custom_instance_on_a_private_address_is_refused(self): + self.configure(engine='custom', url=self.fake.whisper) + job = self.job() + transcribe.run(job.pk) + job = self.reload(job) + self.assertEqual(job.state, 'error') + self.assertIn('Whisper', job.error) + self.assertEqual(self.fake.calls, []) + + @override_settings(WHISPER_URL='') + def test_retries_then_succeeds(self): + self.fake.fail_first = 2 + with override_settings(WHISPER_URL=self.fake.whisper): + self.configure(parallel=1) + job = self.job() + transcribe.run(job.pk) + self.assertEqual(self.reload(job).state, 'done') + self.assertEqual(len(self.fake.calls), 3 + 2) + + def test_gives_up_after_three_retries(self): + self.fake.fail_first = -1 + with override_settings(WHISPER_URL=self.fake.whisper): + self.configure(parallel=1) + job = self.job() + transcribe.run(job.pk) + job = self.reload(job) + self.assertEqual(job.state, 'error') + self.assertEqual(job.error, 'Whisper: HTTP 500') + self.assertEqual(len(self.fake.calls), 4) # the first try and three more + self.assertIsNone(job.data) + self.assertClean() + + def test_a_refusal_is_not_retried(self): + self.fake.fail_first, self.fake.fail_status = -1, 401 + with override_settings(WHISPER_URL=self.fake.whisper): + self.configure(parallel=1) + job = self.job() + transcribe.run(job.pk) + job = self.reload(job) + self.assertEqual((job.state, job.error), ('error', 'Whisper: HTTP 401')) + self.assertEqual(len(self.fake.calls), 1) + + def test_download_failure(self): + with override_settings(WHISPER_URL=self.fake.whisper): + job = self.job('/missing.mp3') + transcribe.run(job.pk) + job = self.reload(job) + self.assertEqual((job.state, job.error), ('error', 'Download: HTTP 404')) + self.assertClean() + + def test_not_audio(self): + with override_settings(WHISPER_URL=self.fake.whisper), patch.object(audio_holder, 'data', b'no'): + job = self.job() + transcribe.run(job.pk) + job = self.reload(job) + self.assertEqual(job.state, 'error') + self.assertTrue(job.error.startswith('ffmpeg:'), job.error) + self.assertClean() + + def test_no_builtin_instance(self): + job = self.job() + transcribe.run(job.pk) + self.assertEqual(self.reload(job).error, 'Keine eingebaute Instanz eingerichtet') + + def test_cancel_while_running(self): + job = self.job() + + def cancel(n): + if n == 1: + TranscriptionJob.objects.filter(pk=job.pk).delete() + from django.db import connection + connection.close() + self.fake.on_call, self.fake.delay = cancel, 0.2 + with override_settings(WHISPER_URL=self.fake.whisper): + self.configure(parallel=1) + transcribe.run(job.pk) + self.assertFalse(TranscriptionJob.objects.filter(pk=job.pk).exists()) + self.assertLessEqual(len(self.fake.calls), 1) # no further piece was sent + self.assertClean() + + def test_cancelled_while_waiting_is_not_started(self): + job = self.job() + job.delete() + transcribe.run(job.pk) + self.assertEqual(self.fake.calls, []) + + +class _Holder: + """Lets a test swap the audio the fake server sends.""" + data = None + + +audio_holder = _Holder() + + +class QueueTests(PipelineBase): + """The runner and its queue. The work itself (_transcribe) is replaced by a stand-in the test + holds back, so only the runner thread writes to the database while the test looks at it.""" + + def setUp(self): + super().setUp() + self.gate = threading.Event() + self.started = [] + patcher = patch.object(transcribe, '_resumed', True) + patcher.start() + self.addCleanup(patcher.stop) + work = patch.object(transcribe, '_transcribe', side_effect=self.work) + work.start() + self.addCleanup(work.stop) + self.addCleanup(self.gate.set) + self.addCleanup(lambda: wait_for(lambda: transcribe._runner is None)) + + def work(self, job_id): + self.started.append(job_id) + self.gate.wait(30) + return {'segments': [{'start': 0, 'end': 1, 'speaker': '', 'text': 'x'}], 'source': {'type': 'whisper'}} + + def test_one_job_at_a_time_oldest_first(self): + first, second, third = self.job(), self.job(), self.job() + transcribe.kick() + self.assertTrue(wait_for(lambda: self.started == [first.pk])) + transcribe.kick() # a second runner doesn't start + time.sleep(0.3) + self.assertEqual(self.started, [first.pk]) + self.assertEqual([self.reload(j).state for j in (first, second, third)], ['downloading', 'queued', 'queued']) + self.gate.set() + self.assertTrue(wait_for(lambda: transcribe._runner is None)) + self.assertEqual(self.started, [first.pk, second.pk, third.pk]) + jobs = [self.reload(j) for j in (first, second, third)] + self.assertEqual([j.state for j in jobs], ['done'] * 3) + self.assertEqual(sorted(j.finished_at for j in jobs), [j.finished_at for j in jobs]) + + def test_a_job_queued_while_the_runner_works_is_taken_too(self): + first = self.job() + transcribe.kick() + self.assertTrue(wait_for(lambda: self.started == [first.pk])) + second = self.job() + transcribe.kick() + self.gate.set() + self.assertTrue(wait_for(lambda: transcribe._runner is None)) + self.assertEqual(self.started, [first.pk, second.pk]) + + def test_a_cancelled_waiting_job_is_skipped(self): + first, second, third = self.job(), self.job(), self.job() + transcribe.kick() + self.assertTrue(wait_for(lambda: self.started == [first.pk])) + second.delete() + self.gate.set() + self.assertTrue(wait_for(lambda: transcribe._runner is None)) + self.assertEqual(self.started, [first.pk, third.pk]) + + def test_a_crashing_job_does_not_stop_the_queue(self): + first, second = self.job(), self.job() + self.gate.set() + calls = [] + + def work(job_id): + calls.append(job_id) + if job_id == first.pk: + raise RuntimeError('boom') + return self.work(job_id) + with patch.object(transcribe, '_transcribe', side_effect=work): + transcribe.kick() + self.assertTrue(wait_for(lambda: transcribe._runner is None)) + self.assertEqual(calls, [first.pk, second.pk]) + self.assertEqual((self.reload(first).state, self.reload(first).error), ('error', 'RuntimeError: boom')) + self.assertEqual(self.reload(second).state, 'done') + + def test_stuck_jobs_start_again_after_a_restart(self): + self.gate.set() + job = self.job() + TranscriptionJob.objects.filter(pk=job.pk).update(state='transcribing', chunks_done=2, chunks_total=3) + with patch.object(transcribe, '_resumed', False): + transcribe.kick() + self.assertTrue(wait_for(lambda: transcribe._runner is None)) + self.assertEqual(self.started, [job.pk]) + self.assertEqual(self.reload(job).state, 'done') + + def test_nothing_queued_starts_no_runner(self): + transcribe.kick() + self.assertIsNone(transcribe._runner) + + +# --------------------------------------------------------------------------- endpoints + +class EndpointBase(TestCase): + def setUp(self): + self.user = User.objects.create_user(username='alice', password='pw12345678') + self.client.force_login(self.user) + self.feed = PodcastFeed.objects.create(user=self.user, rss_url='https://example.org/f', title='Der Podcast') + self.ep = self.episode('e1') + kick = patch.object(transcribe, 'kick') + self.kick = kick.start() + self.addCleanup(kick.stop) + + def episode(self, guid, **kw): + kw.setdefault('audio_url', f'https://example.org/{guid}.mp3') + return PodcastEpisode.objects.create(feed=self.feed, guid=guid, title=f'Folge {guid}', duration_seconds=600, **kw) + + def post(self, url, body=None, **extra): + return self.client.post(url, json.dumps(body or {}), content_type='application/json', **extra) + + def done_job(self, ep, **kw): + kw.setdefault('data', {'segments': [{'start': 0, 'end': 4, 'speaker': '', 'text': 'Hallo.'}], + 'source': {'type': 'whisper', 'model': 'whisper-1', 'language': 'de'}}) + return TranscriptionJob.objects.create(episode=ep, state='done', finished_at=timezone.now(), **kw) + + +class TranscribeEndpointTests(EndpointBase): + def url(self, ep=None): + return f'/podcasts/episodes/{(ep or self.ep).id}/transcribe/' + + def test_requires_login_and_ownership(self): + other = User.objects.create_user(username='bob', password='pw12345678') + self.client.force_login(other) + self.assertEqual(self.client.get(self.url()).status_code, 404) + self.assertEqual(self.post(self.url()).status_code, 404) + self.client.logout() + self.assertEqual(self.client.get(self.url()).status_code, 401) + + def test_none_before_a_job(self): + self.assertEqual(self.client.get(self.url()).json(), { + 'state': 'none', 'chunks_done': 0, 'chunks_total': 0, 'error': '', 'want_article': False, + 'started_at': None, 'finished_at': None}) + + @patch('podcasts.transcription._too_large', return_value=False) + def test_post_queues_a_job_and_starts_the_runner(self, _): + r = self.post(self.url()) + self.assertEqual(r.status_code, 200) + self.assertEqual((r.json()['state'], r.json()['want_article']), ('queued', False)) + self.assertEqual(TranscriptionJob.objects.get(episode=self.ep).state, 'queued') + self.kick.assert_called() + + @patch('podcasts.transcription._too_large', return_value=False) + def test_post_again_only_raises_the_article_wish(self, _): + self.post(self.url()) + first = TranscriptionJob.objects.get(episode=self.ep).pk + r = self.post(self.url(), {'article': True}) + self.assertTrue(r.json()['want_article']) + r = self.post(self.url(), {'article': False}) # never lowers it + self.assertTrue(r.json()['want_article']) + self.assertEqual(TranscriptionJob.objects.filter(episode=self.ep).count(), 1) + self.assertEqual(TranscriptionJob.objects.get(episode=self.ep).pk, first) + + @patch('podcasts.transcription._too_large', return_value=False) + def test_post_with_article_true_sets_the_wish(self, _): + self.assertTrue(self.post(self.url(), {'article': True}).json()['want_article']) + + def test_youtube_is_refused(self): + ep = self.episode('yt', audio_url=youtube.WATCH + 'abc123') + r = self.post(self.url(ep)) + self.assertEqual((r.status_code, r.json()), (400, {'error': 'youtube'})) + self.assertFalse(TranscriptionJob.objects.exists()) + + @patch('podcasts.transcription._too_large', return_value=True) + def test_too_large_is_refused(self, _): + r = self.post(self.url()) + self.assertEqual((r.status_code, r.json()), (400, {'error': 'too large'})) + self.assertFalse(TranscriptionJob.objects.exists()) + + def test_the_size_check_reads_the_content_length(self): + from . import silence + resp = type('R', (), {'headers': {'Content-Length': str(silence.MAX_BYTES + 1)}})() + with patch('podcasts.transcription.requests.head', return_value=resp): + self.assertTrue(transcription._too_large('https://example.org/x.mp3')) + resp.headers = {'Content-Length': '1000'} + with patch('podcasts.transcription.requests.head', return_value=resp): + self.assertFalse(transcription._too_large('https://example.org/x.mp3')) + with patch('podcasts.transcription.requests.head', side_effect=transcription.requests.ConnectionError): + self.assertFalse(transcription._too_large('https://example.org/x.mp3')) + + def test_a_whisper_transcript_is_done_already(self): + self.done_job(self.ep) + r = self.post(self.url(), {'article': True}) + self.assertEqual((r.json()['state'], r.json()['want_article']), ('done', True)) + self.kick.assert_not_called() + self.assertTrue(TranscriptionJob.objects.get(episode=self.ep).want_article) + + def test_a_feed_transcript_is_done_already(self): + self.ep.transcripts = [{'url': 'https://example.org/t.vtt', 'type': 'text/vtt'}] + self.ep.save() + r = self.post(self.url(), {'article': True}) + self.assertEqual((r.json()['state'], r.json()['want_article']), ('done', True)) + job = TranscriptionJob.objects.get(episode=self.ep) + self.assertIsNone(job.data) + self.kick.assert_not_called() + + @patch('podcasts.transcription._too_large', return_value=False) + def test_after_an_error_post_starts_again_at_the_back(self, _): + old = TranscriptionJob.objects.create(episode=self.ep, state='error', error='Whisper: HTTP 500') + r = self.post(self.url(), {'article': True}) + self.assertEqual((r.json()['state'], r.json()['error']), ('queued', '')) + new = TranscriptionJob.objects.get(episode=self.ep) + self.assertGreater(new.pk, old.pk) + + @patch('podcasts.transcription._too_large', return_value=False) + def test_delete_cancels_queued_and_running_but_not_a_finished_one(self, _): + self.post(self.url()) + self.assertEqual(self.client.delete(self.url()).json()['state'], 'none') + self.assertFalse(TranscriptionJob.objects.exists()) + TranscriptionJob.objects.create(episode=self.ep, state='transcribing', chunks_total=5, chunks_done=2) + self.assertEqual(self.client.delete(self.url()).json()['state'], 'none') + self.assertFalse(TranscriptionJob.objects.exists()) + self.done_job(self.ep) + self.assertEqual(self.client.delete(self.url()).json()['state'], 'done') + self.assertTrue(TranscriptionJob.objects.exists()) + + def test_get_shows_progress(self): + TranscriptionJob.objects.create(episode=self.ep, state='transcribing', chunks_total=38, chunks_done=7, + started_at=timezone.now()) + d = self.client.get(self.url()).json() + self.assertEqual((d['state'], d['chunks_done'], d['chunks_total']), ('transcribing', 7, 38)) + self.assertIsNotNone(d['started_at']) + self.kick.assert_called() # a restart's first request starts the queue again + + +class TranscriptTests(EndpointBase): + def transcript(self, ep=None): + return self.client.get(f'/podcasts/episodes/{(ep or self.ep).id}/transcript/') + + def detail(self, ep=None): + return self.client.get(f'/podcasts/episodes/{(ep or self.ep).id}/').json() + + def test_404_as_before_without_any(self): + self.assertEqual((self.transcript().status_code, self.transcript().json()), (404, {'error': 'no transcript'})) + d = self.detail() + self.assertEqual((d['has_transcript'], d['transcript_origin'], d['transcription'], d['article_item_id']), + (False, None, None, None)) + + def test_falls_back_to_the_whisper_transcript(self): + job = self.done_job(self.ep) + r = self.transcript() + self.assertEqual(r.status_code, 200) + self.assertEqual(r.json(), job.data) + d = self.detail() + self.assertEqual((d['has_transcript'], d['transcript_origin']), (True, 'whisper')) + self.assertEqual(d['transcription']['state'], 'done') + + def test_a_running_job_has_no_transcript_yet(self): + TranscriptionJob.objects.create(episode=self.ep, state='transcribing', data=None) + self.assertEqual(self.transcript().status_code, 404) + d = self.detail() + self.assertEqual((d['has_transcript'], d['transcript_origin'], d['transcription']['state']), + (False, None, 'transcribing')) + + def feed_transcript(self, error=''): + ep = self.ep + ep.transcripts = [{'url': 'https://example.org/t.vtt', 'type': 'text/vtt'}] + ep.save() + EpisodeTranscript.objects.create( + episode=ep, sources=['https://example.org/t.vtt'], error=error, + data=None if error else {'segments': [{'start': 0, 'end': 1, 'speaker': 'A', 'text': 'Aus dem Feed'}], + 'source': {'url': 'https://example.org/t.vtt'}}) + + def test_the_feeds_transcript_wins(self): + self.done_job(self.ep) + self.feed_transcript() + self.assertEqual(self.transcript().json()['segments'][0]['text'], 'Aus dem Feed') + self.assertEqual(self.detail()['transcript_origin'], 'feed') + + def test_an_unusable_feed_transcript_gives_way(self): + self.done_job(self.ep) + self.feed_transcript(error='invalid') + self.assertEqual(self.transcript().json()['segments'][0]['text'], 'Hallo.') + self.assertEqual(self.detail()['transcript_origin'], 'whisper') + + def test_a_feed_transcript_that_cannot_be_loaded_now_gives_way_too(self): + self.done_job(self.ep) + self.ep.transcripts = [{'url': 'https://example.org/t.vtt', 'type': 'text/vtt'}] + self.ep.save() + from . import chapters + with patch('podcasts.views.transcripts_mod.load', side_effect=chapters.FetchUnavailable('x')): + self.assertEqual(self.transcript().json()['segments'][0]['text'], 'Hallo.') + + def test_article_item_id(self): + source = Source.objects.create(user=self.user, url='diora://podcast-artikel', title='Podcast-Artikel', + kind='blog') + item = Item.objects.create(source=source, user=self.user, kind='blog', guid_hash='x', title='A', + published=timezone.now()) + EpisodeArticle.objects.create(episode=self.ep, item=item) + self.assertEqual(self.detail()['article_item_id'], item.id) + + +class WishTests(EndpointBase): + PENDING = '/podcasts/transcription/pending/' + + def claim(self, ep=None, method='post', **kw): + return getattr(self.client, method)( + f'/podcasts/episodes/{(ep or self.ep).id}/article/claim/', json.dumps(kw.pop('body', {})), + content_type='application/json', **kw) + + def test_pending_lists_open_wishes_only(self): + wished = self.done_job(self.ep, want_article=True) + e2, e3, e4, e5 = (self.episode(g) for g in ('e2', 'e3', 'e4', 'e5')) + self.done_job(e2) # no wish + TranscriptionJob.objects.create(episode=e3, state='transcribing', want_article=True) # not done + self.done_job(e4, want_article=True, claimed_by='tablet', claimed_until=timezone.now() + timedelta(minutes=5)) + job5 = self.done_job(e5, want_article=True) + source = Source.objects.create(user=self.user, url='diora://podcast-artikel', title='P', kind='blog') + item = Item.objects.create(source=source, user=self.user, kind='blog', guid_hash='x', title='A', + published=timezone.now()) + EpisodeArticle.objects.create(episode=e5, item=item) # written already + r = self.client.get(self.PENDING).json() + self.assertEqual(r, [{'episode_id': self.ep.id, 'title': 'Folge e1', 'podcast': 'Der Podcast', + 'duration_seconds': 600}]) + self.assertTrue(wished.want_article and job5.want_article) + + def test_pending_is_per_user_and_needs_login(self): + self.done_job(self.ep, want_article=True) + other = User.objects.create_user(username='bob', password='pw12345678') + self.client.force_login(other) + self.assertEqual(self.client.get(self.PENDING).json(), []) + self.client.logout() + self.assertEqual(self.client.get(self.PENDING).status_code, 401) + + def test_an_expired_claim_is_open_again(self): + self.done_job(self.ep, want_article=True, claimed_by='tablet', + claimed_until=timezone.now() - timedelta(seconds=1)) + self.assertEqual(len(self.client.get(self.PENDING).json()), 1) + + def test_claim_blocks_the_other_device_until_it_expires(self): + job = self.done_job(self.ep, want_article=True) + r = self.claim(body={'device': 'phone'}) + self.assertEqual(r.status_code, 200) + self.assertEqual(r.json()['claimed_by'], 'phone') + job.refresh_from_db() + left = job.claimed_until - timezone.now() + self.assertTrue(timedelta(minutes=14) < left <= timedelta(minutes=15)) + self.assertEqual(self.client.get(self.PENDING).json(), []) + r = self.claim(body={'device': 'desktop'}) + self.assertEqual(r.status_code, 409) + self.assertEqual(r.json()['error'], 'claimed') + self.assertEqual(self.claim(body={'device': 'phone'}).status_code, 200) # the holder renews + TranscriptionJob.objects.filter(pk=job.pk).update(claimed_until=timezone.now() - timedelta(seconds=1)) + self.assertEqual(self.claim(body={'device': 'desktop'}).status_code, 200) + job.refresh_from_db() + self.assertEqual(job.claimed_by, 'desktop') + + def test_the_device_may_come_in_a_header(self): + self.done_job(self.ep, want_article=True) + self.assertEqual(self.claim(HTTP_X_DIORA_DEVICE='phone').status_code, 200) + self.assertEqual(self.claim(HTTP_X_DIORA_DEVICE='desktop').status_code, 409) + + def test_a_device_name_is_required(self): + self.done_job(self.ep, want_article=True) + self.assertEqual(self.claim().status_code, 400) + + def test_release(self): + self.done_job(self.ep, want_article=True) + self.claim(body={'device': 'phone'}) + self.assertEqual(self.claim(method='delete', body={'device': 'desktop'}).status_code, 409) + self.assertEqual(self.claim(method='delete', body={'device': 'phone'}).status_code, 200) + self.assertEqual(len(self.client.get(self.PENDING).json()), 1) + + def test_no_wish_no_claim(self): + self.done_job(self.ep) # done, but no article wanted + self.assertEqual(self.claim(body={'device': 'phone'}).status_code, 404) + self.assertEqual(self.claim(self.episode('e9'), body={'device': 'phone'}).status_code, 404) + + +class NotificationTests(EndpointBase): + def get(self, **q): + return self.client.get('/feeds/notifications/', q).json() + + def test_first_time_only_the_cursor(self): + j = self.done_job(self.ep, want_article=True) + d = self.get() + self.assertEqual((d['transcripts'], d['t_last'], d['items']), ([], j.id, [])) + self.assertEqual(d['last'], 0) + + def test_jobs_done_after_the_cursor(self): + e2, e3 = self.episode('e2'), self.episode('e3') + a = self.done_job(self.ep) + b = self.done_job(e2, want_article=True) + TranscriptionJob.objects.create(episode=e3, state='transcribing') # not done: no notice + d = self.get(t_after=a.id) + self.assertEqual(d['transcripts'], [{'job_id': b.id, 'episode_id': e2.id, 'title': 'Folge e2', + 'podcast': 'Der Podcast', 'want_article': True}]) + self.assertEqual(d['t_last'], b.id) + self.assertEqual(self.get(t_after=b.id)['transcripts'], []) + self.assertEqual([t['job_id'] for t in self.get(t_after=0)['transcripts']], [a.id, b.id]) + + def test_only_the_users_own_and_not_feed_only_jobs(self): + other = User.objects.create_user(username='bob', password='pw12345678') + feed = PodcastFeed.objects.create(user=other, rss_url='https://example.org/o', title='O') + ep = PodcastEpisode.objects.create(feed=feed, guid='o', title='O', audio_url='https://example.org/o.mp3') + self.done_job(ep) + self.done_job(self.episode('e2'), data=None) # only the wish for an article from a feed transcript + d = self.get(t_after=0) + self.assertEqual((d['transcripts'], d['t_last']), ([], 0)) + + def test_items_still_work_beside_it(self): + d = self.get(after=0) + self.assertEqual((d['items'], d['last'], d['transcripts'], d['t_last']), ([], 0, [], 0)) diff --git a/podcasts/transcribe.py b/podcasts/transcribe.py new file mode 100644 index 0000000..fac15be --- /dev/null +++ b/podcasts/transcribe.py @@ -0,0 +1,378 @@ +"""Transcribe an episode with a Whisper instance (docs/transcription.md, "Transcribing"). + +The server downloads the episode (as silence.py does, with the same size limit), decodes it +once to 16 kHz mono, cuts it into pieces of about eight minutes where there is a pause (so no +word is cut), sends the pieces in parallel to the instance's OpenAI-compatible +`/audio/transcriptions` and joins the answers into one transcript, which is kept on the job +(TranscriptionJob.data). + +A job runs in a daemon thread, one at a time per server; the others wait as `queued` and are +taken oldest first (by id). A job found `downloading` or `transcribing` when the process has +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 logging +import os +import subprocess +import tempfile +import threading +import time +from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait + +import requests +from django.conf import settings +from django.db import close_old_connections +from django.utils import timezone + +from . import chapters, silence, transcripts +from .models import TranscriptionJob, TranscriptionSettings +from .transcription import DEFAULT_MODEL, for_user, token_of + +log = logging.getLogger(__name__) + +CHUNK_SECONDS = 8 * 60 # about this long a piece +CUT_WINDOW = 30 # a cut moves to the longest pause within this many seconds +MIN_TAIL = 60 # a last piece shorter than this is joined to the one before +RATE = 16000 # Hz, mono, 16-bit +WINDOW = silence.WINDOW +RETRIES = 3 +RETRY_PAUSES = (3, 10, 30) # seconds before each retry +UPLOAD_TIMEOUT = (10, 3600) # a CPU instance may need many minutes for a piece +DECODE_TIMEOUT = 3600 +OPUS_BITRATE = '32k' +MAX_PROMPT = 500 # characters +ALIVE_CHECK = 5 # seconds between looks whether the job was cancelled + +_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 + + +class Cancelled(Exception): + """The job's row is gone: the user cancelled it.""" + + +class Failed(Exception): + """The job can't succeed; the message is shown to the user.""" + + +class Permanent(Exception): + """A request the instance refused for good (a client error): not worth a retry.""" + + +# --------------------------------------------------------------------------- +# Queue +# --------------------------------------------------------------------------- + +def kick(): + """Make sure queued jobs are being worked on. Called by the endpoints, so after a restart + the first request brings back what was running.""" + global _runner, _resumed + with _lock: + if not _resumed: + _resumed = True + TranscriptionJob.objects.filter( + state__in=[TranscriptionJob.DOWNLOADING, TranscriptionJob.TRANSCRIBING], + ).update(state=TranscriptionJob.QUEUED, chunks_done=0, chunks_total=0) + if _runner is not None: + return + if not TranscriptionJob.objects.filter(state=TranscriptionJob.QUEUED).exists(): + return + _runner = threading.Thread(target=_loop, name='transcribe', daemon=True) + _runner.start() + + +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: + _runner = None # decided under the lock, so a job queued right now starts a new runner + return job_id + + +def _loop(): + global _runner + try: + while True: + close_old_connections() + job_id = _next_job() + if job_id is None: + return + try: + run(job_id) + except Exception: # run() reports its own failures; this keeps the queue going + log.exception('transcription job %s crashed', job_id) + except BaseException: + with _lock: + _runner = None + raise + finally: + close_old_connections() + + +def run(job_id): + """Work one job through to `done` or `error`, in this thread.""" + started = TranscriptionJob.objects.filter(pk=job_id, state=TranscriptionJob.QUEUED).update( + state=TranscriptionJob.DOWNLOADING, started_at=timezone.now(), finished_at=None, + error='', chunks_done=0, chunks_total=0) + if not started: + return # cancelled while it waited + try: + data = _transcribe(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 Failed as e: + _fail(job_id, str(e)) + except Exception as e: + log.exception('transcription job %s failed', job_id) + _fail(job_id, f'{type(e).__name__}: {e}') + + +def _fail(job_id, message): + TranscriptionJob.objects.filter(pk=job_id).update( + state=TranscriptionJob.ERROR, error=(message or 'failed')[:500], finished_at=timezone.now()) + + +def _write(job_id, **fields): + """Update the job; Cancelled when its row is gone.""" + if not TranscriptionJob.objects.filter(pk=job_id).update(**fields): + raise Cancelled() + + +def _alive(job_id): + if not TranscriptionJob.objects.filter(pk=job_id).exists(): + raise Cancelled() + + +# --------------------------------------------------------------------------- +# The work +# --------------------------------------------------------------------------- + +def _target(conf): + """(url, headers, builtin) of the instance the settings point at.""" + builtin = conf.engine == TranscriptionSettings.BUILTIN + base = settings.WHISPER_URL if builtin else conf.url + if not base: + raise Failed('Keine eingebaute Instanz eingerichtet' if builtin else 'Keine Adresse eingetragen') + headers = {'User-Agent': 'diora transcription'} + if not builtin and conf.token_encrypted: + headers['Authorization'] = 'Bearer ' + token_of(conf) + return base.rstrip('/') + '/audio/transcriptions', headers, builtin + + +def _prompt(episode): + parts = [episode.feed.title, episode.title, episode.feed.author] + return '. '.join(p.strip() for p in parts if p and p.strip())[:MAX_PROMPT] + + +def _transcribe(job_id): + job = TranscriptionJob.objects.select_related('episode__feed').get(pk=job_id) + episode = job.episode + conf = for_user(episode.feed.user) + url, headers, builtin = _target(conf) + options = {'model': conf.model or DEFAULT_MODEL, 'response_format': 'verbose_json'} + if conf.language: + options['language'] = conf.language + prompt = _prompt(episode) + if prompt: + options['prompt'] = prompt + + with tempfile.TemporaryDirectory(prefix='diora-transcribe-') as tmp: + source = os.path.join(tmp, 'episode.audio') + _download(episode.audio_url, source, job_id) + pcm = os.path.join(tmp, 'episode.pcm') + _decode(source, pcm) + os.remove(source) + total = os.path.getsize(pcm) // 2 / RATE + if total <= 0: + raise Failed('Kein Ton in der Datei') + pieces = plan(total, pauses_of(pcm)) + _write(job_id, state=TranscriptionJob.TRANSCRIBING, chunks_total=len(pieces), chunks_done=0) + + results = {} + stop = threading.Event() # set when the job ends early: no piece starts after that + suffix, mime = ('.flac', 'audio/flac') if builtin else ('.ogg', 'audio/ogg') + + def work(index): + # Only the HTTP side runs here; the database is touched by this thread alone. + if stop.is_set(): + return index, None + start, end = pieces[index] + path = _encode(pcm, start, end, tmp, index, opus=not builtin) + try: + with open(path, 'rb') as f: + body = f.read() + finally: + os.remove(path) + return index, _send(url, headers, builtin, f'piece-{index}{suffix}', mime, body, options, stop) + + pool = ThreadPoolExecutor(max_workers=max(1, min(conf.parallel, 8))) + try: + pending = {pool.submit(work, i) for i in range(len(pieces))} + while pending: + finished, pending = wait(pending, timeout=ALIVE_CHECK, return_when=FIRST_COMPLETED) + _alive(job_id) # also while a long piece is out + for f in finished: + index, answer = f.result() # the first failure ends the job + results[index] = answer + if finished: + _write(job_id, chunks_done=len(results)) + finally: + stop.set() + pool.shutdown(wait=True, cancel_futures=True) + return build(pieces, [results[i] for i in range(len(pieces))], options['model'], conf.language) + + +def _download(url, path, job_id): + size = 0 + with open(path, 'wb') as out, requests.get( + url, stream=True, timeout=(10, 60), headers={'User-Agent': 'diora transcription'}) as r: + if r.status_code >= 400: + raise Failed(f'Download: HTTP {r.status_code}') + if int(r.headers.get('Content-Length') or 0) > silence.MAX_BYTES: + raise Failed('too large') + for n, chunk in enumerate(r.iter_content(1 << 20)): + size += len(chunk) + if size > silence.MAX_BYTES: + raise Failed('too large') + out.write(chunk) + if n % 8 == 0: + _alive(job_id) + if size == 0: + raise Failed('Download: leere Datei') + + +def _run_ffmpeg(args, input=None, timeout=DECODE_TIMEOUT): + import imageio_ffmpeg + proc = subprocess.run( + [imageio_ffmpeg.get_ffmpeg_exe(), '-nostdin', '-hide_banner', '-loglevel', 'error', *args], + input=input, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout) + if proc.returncode != 0: + err = proc.stderr.decode('utf-8', 'replace').strip().splitlines() + raise Failed('ffmpeg: ' + (err[-1] if err else f'exit {proc.returncode}')[:200]) + + +def _decode(source, pcm): + """The audio at `source` as raw 16 kHz mono 16-bit samples in the file `pcm`.""" + _run_ffmpeg(['-vn', '-sn', '-dn', '-i', source, '-ac', '1', '-ar', str(RATE), '-f', 's16le', '-y', pcm]) + + +def pauses_of(pcm): + """The pauses of the decoded audio as [(middle, length)] in seconds, found with + silence.py's rules (what it cuts out of a pause is centred on the pause).""" + size = int(RATE * WINDOW) * 2 + levels = [] + with open(pcm, 'rb') as f: + while True: + data = f.read(size * 100000) + if not data: + break + for i in range(0, len(data) - size + 1, size): + levels.append(silence._level_db(data[i:i + size])) + return [((a + b) / 2, b - a) for a, b in silence.cuts(levels)] + + +def plan(total, pauses): + """[(start, end)] in seconds covering 0..total: pieces of about CHUNK_SECONDS, each cut at + the longest pause within CUT_WINDOW of where it would fall (right there when there is none).""" + cuts = [] + last = 0.0 + k = 1 + while k * CHUNK_SECONDS < total - MIN_TAIL: + target = k * CHUNK_SECONDS + k += 1 + near = [p for p in pauses if abs(p[0] - target) <= CUT_WINDOW and p[0] > last + 1] + at = max(near, key=lambda p: p[1])[0] if near else target + if at <= last or at >= total: + continue + cuts.append(at) + last = at + edges = [0.0, *cuts, total] + return [(edges[i], edges[i + 1]) for i in range(len(edges) - 1)] + + +def _encode(pcm, start, end, tmp, index, opus): + """The piece [start, end) of the decoded audio as FLAC or Opus (ogg), in a file.""" + with open(pcm, 'rb') as f: + f.seek(int(round(start * RATE)) * 2) + data = f.read(int(round((end - start) * RATE)) * 2) + out = os.path.join(tmp, f'piece-{index}.' + ('ogg' if opus else 'flac')) + codec = ['-c:a', 'libopus', '-b:a', OPUS_BITRATE, '-f', 'ogg'] if opus else ['-c:a', 'flac', '-f', 'flac'] + _run_ffmpeg(['-f', 's16le', '-ar', str(RATE), '-ac', '1', '-i', 'pipe:0', *codec, '-y', out], input=data) + return out + + +def _post(url, headers, builtin, name, mime, body, options): + """One request; the parsed answer. Permanent for a refusal that won't change, RuntimeError + for trouble that may pass.""" + files = {'file': (name, body, mime)} + 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) + 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}') + raise RuntimeError(f'HTTP {r.status_code}') + else: + try: + r = chapters._open(url, headers, method='POST', timeout=UPLOAD_TIMEOUT, files=files, data=options) + except chapters.ChaptersInvalid as e: + raise Permanent(str(e)) + except chapters.ChaptersUnavailable as e: + raise RuntimeError(str(e)) + try: + return r.json() + finally: + r.close() + except requests.RequestException as e: + raise RuntimeError(e.__class__.__name__) + except ValueError: + raise RuntimeError('Antwort ist kein JSON') + + +def _send(url, headers, builtin, name, mime, body, options, stop): + """A piece to the instance, retried up to RETRIES times with a growing pause.""" + for attempt in range(RETRIES + 1): + if stop.is_set(): + raise Cancelled() + try: + return _post(url, headers, builtin, name, mime, body, options) + except Permanent as e: + raise Failed(f'Whisper: {e}') + except RuntimeError as e: + if attempt >= RETRIES: + raise Failed(f'Whisper: {e}') + log.info('whisper piece %s failed (%s), retrying', name, e) + time.sleep(RETRY_PAUSES[min(attempt, len(RETRY_PAUSES) - 1)]) + + +def build(pieces, answers, model, language): + """{segments, source} from the pieces' places and the instance's verbose_json answers.""" + fragments = [] + detected = '' + for (start, end), answer in zip(pieces, answers): + if not isinstance(answer, dict): + continue + detected = detected or (answer.get('language') or '') + segments = answer.get('segments') + if not isinstance(segments, list): # a plain {"text": …} answer: one segment for the piece + text = (answer.get('text') or '').strip() + segments = [{'start': 0, 'end': end - start, 'text': text}] if text else [] + for s in segments: + if not isinstance(s, dict): + continue + text = ' '.join(str(s.get('text') or '').split()) + try: + a, b = float(s.get('start', 0)), float(s.get('end', 0)) + except (TypeError, ValueError): + continue + if text: + fragments.append((start + a, start + b, '', text)) + merged = transcripts.merge(fragments) + if not merged: + raise Failed('Kein Text erkannt') + return {'segments': merged, 'source': {'type': 'whisper', 'model': model, 'language': language or detected}} diff --git a/podcasts/transcription.py b/podcasts/transcription.py index d39aac4..a9b6f72 100644 --- a/podcasts/transcription.py +++ b/podcasts/transcription.py @@ -10,13 +10,14 @@ from urllib.parse import urlsplit import requests from django.conf import settings from django.http import JsonResponse +from django.utils import timezone from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_http_methods from feeds import logins from . import chapters -from .models import TranscriptionSettings +from .models import EpisodeArticle, PodcastEpisode, TranscriptionJob, TranscriptionSettings DEFAULT_MODEL = 'whisper-1' TEST_TIMEOUT = 60 # a cold CPU instance may load its model first @@ -150,3 +151,147 @@ def transcription_test(request): except requests.RequestException as e: return JsonResponse({'ok': False, 'model': model, 'error': e.__class__.__name__}) return JsonResponse({'ok': True, 'model': model, 'error': ''}) + + +# --------------------------------------------------------------------------- +# Jobs +# --------------------------------------------------------------------------- + +CLAIM_MINUTES = 15 +ACTIVE = (TranscriptionJob.QUEUED, TranscriptionJob.DOWNLOADING, TranscriptionJob.TRANSCRIBING) + + +def job_state(job): + """The job as the API shows it; state `none` when there is no job.""" + if job is None: + return {'state': TranscriptionJob.NONE, 'chunks_done': 0, 'chunks_total': 0, 'error': '', + 'want_article': False, 'started_at': None, 'finished_at': None} + return { + 'state': job.state, 'chunks_done': job.chunks_done, 'chunks_total': job.chunks_total, + 'error': job.error, 'want_article': job.want_article, + 'started_at': job.started_at.isoformat() if job.started_at else None, + 'finished_at': job.finished_at.isoformat() if job.finished_at else None, + } + + +def job_of(episode): + return TranscriptionJob.objects.filter(episode=episode).first() + + +def _too_large(url): + """Whether the server says the file at `url` is over the size limit (a failed look doesn't count).""" + from . import silence + try: + r = requests.head(url, timeout=5, allow_redirects=True, headers={'User-Agent': 'diora transcription'}) + return int(r.headers.get('Content-Length') or 0) > silence.MAX_BYTES + except (requests.RequestException, ValueError): + return False + + +@csrf_exempt +@require_http_methods(['GET', 'POST', 'DELETE']) +def episode_transcribe(request, pk): + """POST starts the episode's transcription (or just raises `want_article` on one that exists), + GET tells how far it is, DELETE cancels a queued or running one.""" + from . import transcribe, views + from .youtube import is_video + if (err := _auth(request)): + return err + ep = PodcastEpisode.objects.select_related('feed').filter(pk=pk, feed__user=request.user).first() + if ep is None: + return JsonResponse({'error': 'not found'}, status=404) + job = job_of(ep) + + if request.method == 'GET': + transcribe.kick() # after a restart the first request brings the work back + return JsonResponse(job_state(job)) + + if request.method == 'DELETE': + if job is not None and job.state != TranscriptionJob.DONE: + job.delete() # a running worker notices its row is gone and stops + job = None + return JsonResponse(job_state(job)) + + body = _body(request) + if body is None: + return JsonResponse({'error': 'invalid JSON'}, status=400) + article = body.get('article') is True + if is_video(ep.audio_url): + return JsonResponse({'error': 'youtube'}, status=400) + + has_feed = views._has_transcript(ep) + if job is not None and (job.state in ACTIVE or (job.state == TranscriptionJob.DONE and (job.data or has_feed))): + if article and not job.want_article: + job.want_article = True + job.save(update_fields=['want_article']) + return JsonResponse(job_state(job)) + if has_feed: # the feed's own transcript is there: only the article is wanted + if job is None: + job = TranscriptionJob(episode=ep) + job.state, job.want_article, job.finished_at = TranscriptionJob.DONE, article or job.want_article, timezone.now() + job.save() + return JsonResponse(job_state(job)) + + if not ep.audio_url: + return JsonResponse({'error': 'no audio'}, status=400) + if _too_large(ep.audio_url): + return JsonResponse({'error': 'too large'}, status=400) + if job is not None: + job.delete() # a failed or hollow earlier run: the new one starts at the back of the queue + job = TranscriptionJob.objects.create(episode=ep, state=TranscriptionJob.QUEUED, want_article=article) + transcribe.kick() + return JsonResponse(job_state(job)) + + +def _wishes(user): + """The user's jobs whose transcript is there and whose article is wanted but not written.""" + return (TranscriptionJob.objects + .filter(episode__feed__user=user, state=TranscriptionJob.DONE, want_article=True) + .exclude(episode__in=EpisodeArticle.objects.values('episode'))) + + +@csrf_exempt +@require_http_methods(['GET']) +def transcription_pending(request): + """The open wishes for an article that no device holds right now.""" + from . import transcribe + if (err := _auth(request)): + return err + transcribe.kick() + jobs = (_wishes(request.user).exclude(claimed_until__gt=timezone.now()) + .select_related('episode__feed').order_by('id')) + return JsonResponse([{ + 'episode_id': j.episode_id, 'title': j.episode.title, 'podcast': j.episode.feed.title, + 'duration_seconds': j.episode.duration_seconds, + } for j in jobs], safe=False) + + +@csrf_exempt +@require_http_methods(['POST', 'DELETE']) +def article_claim(request, pk): + """Reserve the wish for an article for the calling device for CLAIM_MINUTES (POST), or give + it up (DELETE). The device names itself with `device` in the JSON body or the + X-Diora-Device header. 409 when another device holds it.""" + from datetime import timedelta + if (err := _auth(request)): + return err + body = _body(request) + if body is None: + return JsonResponse({'error': 'invalid JSON'}, status=400) + device = body.get('device') or request.META.get('HTTP_X_DIORA_DEVICE') or '' + if not isinstance(device, str) or not device.strip(): + return JsonResponse({'error': 'device missing'}, status=400) + device = device.strip()[:60] + job = _wishes(request.user).filter(episode_id=pk).first() + if job is None: + return JsonResponse({'error': 'no open wish'}, status=404) + now = timezone.now() + if job.claimed_until and job.claimed_until > now and job.claimed_by != device: + return JsonResponse({'error': 'claimed', 'claimed_until': job.claimed_until.isoformat()}, status=409) + if request.method == 'DELETE': + job.claimed_by, job.claimed_until = '', None + else: + job.claimed_by, job.claimed_until = device, now + timedelta(minutes=CLAIM_MINUTES) + job.save(update_fields=['claimed_by', 'claimed_until']) + return JsonResponse({'claimed_by': job.claimed_by, + 'claimed_until': job.claimed_until.isoformat() if job.claimed_until else None}) diff --git a/podcasts/urls.py b/podcasts/urls.py index 6ffc0d7..cfa024d 100644 --- a/podcasts/urls.py +++ b/podcasts/urls.py @@ -33,6 +33,9 @@ urlpatterns = [ path('episodes//favorite/', views.set_favorite, name='podcast_set_favorite'), path('transcription/settings/', transcription.transcription_settings, name='podcast_transcription_settings'), path('transcription/test/', transcription.transcription_test, name='podcast_transcription_test'), + path('transcription/pending/', transcription.transcription_pending, name='podcast_transcription_pending'), + path('episodes//transcribe/', transcription.episode_transcribe, name='podcast_episode_transcribe'), + path('episodes//article/claim/', transcription.article_claim, name='podcast_article_claim'), path('stats/', views.podcast_stats, name='podcast_stats'), path('queue/', views.queue_get, name='podcast_queue_get'), path('queue/add/', views.queue_add, name='podcast_queue_add'), diff --git a/podcasts/views.py b/podcasts/views.py index b856077..5d249b9 100644 --- a/podcasts/views.py +++ b/podcasts/views.py @@ -29,11 +29,12 @@ from . import chapters as chapters_mod from . import silence as silence_mod from . import sponsor as sponsor_mod from . import feedext +from . import transcription as transcription_mod from . import transcripts as transcripts_mod from . import youtube as youtube_mod from . import folders as folders_mod from .models import ( - EpisodeProgress, EpisodeTranscript, PlayerDevice, PlayerState, PodcastEpisode, PodcastFeed, PodcastLater, + EpisodeArticle, EpisodeProgress, EpisodeTranscript, PlayerDevice, PlayerState, PodcastEpisode, PodcastFeed, PodcastLater, PodcastQueue, PodcastRadioHeard, MAX_SPEED, MIN_SPEED, PodcastSettings, counts_as_played, ) from .shownotes import extract_timestamps, sanitize_html @@ -1424,6 +1425,9 @@ def episode_detail(request, pk): chapters = _load_chapters(ep) progress = EpisodeProgress.objects.filter(user=request.user, episode=ep).first() + job = transcription_mod.job_of(ep) + origin = _transcript_origin(ep, job) + article = EpisodeArticle.objects.filter(episode=ep).first() return JsonResponse({ 'id': ep.id, 'title': ep.title, @@ -1431,7 +1435,10 @@ def episode_detail(request, pk): 'timestamps': extract_timestamps(ep.description, ep.duration_seconds), 'chapters': chapters, 'transcripts': ep.transcripts, - 'has_transcript': _has_transcript(ep), + 'has_transcript': origin is not None, + 'transcript_origin': origin, + 'article_item_id': article.item_id if article else None, + 'transcription': transcription_mod.job_state(job) if job else None, 'feed': {'id': ep.feed.id, 'title': ep.feed.title, 'artwork_url': ep.feed.artwork_url}, 'audio_url': ep.audio_url, 'artwork_url': ep.artwork_url, @@ -1461,6 +1468,13 @@ def _has_transcript(ep): return not (cached and cached[0] == sources and cached[1]) +def _transcript_origin(ep, job=None): + """"feed" when the feed has a usable transcript, else "whisper" when we made one, else None.""" + if _has_transcript(ep): + return 'feed' + return 'whisper' if job is not None and job.state == 'done' and job.data else None + + @csrf_exempt @require_http_methods(['GET']) def episode_silences(request, pk): @@ -1503,20 +1517,28 @@ def episode_sponsor(request, pk): def episode_transcript(request, pk): """The episode's transcript as {"segments": [{start, end, speaker, text}], "source": {url, type, language}}, fetched and parsed on first use, then - kept. 404 {"error": "no transcript" | "unsupported" | "invalid"} when - there's none to show; 502 {"error": "unavailable"} when it couldn't be - loaded right now.""" + kept. When the feed has none we can use, the one Whisper made (a finished + TranscriptionJob, docs/transcription.md). 404 {"error": "no transcript" | + "unsupported" | "invalid"} when there's none to show; 502 + {"error": "unavailable"} when it couldn't be loaded right now.""" if not request.user.is_authenticated: return JsonResponse({'error': 'authentication required'}, status=401) try: ep = PodcastEpisode.objects.get(pk=pk, feed__user=request.user) except PodcastEpisode.DoesNotExist: return JsonResponse({'error': 'not found'}, status=404) + + def fallback(error, status=404): + job = transcription_mod.job_of(ep) + if job is not None and job.state == 'done' and job.data: + return JsonResponse(job.data) + return JsonResponse({'error': error}, status=status) + if not ep.transcripts: - return JsonResponse({'error': 'no transcript'}, status=404) + return fallback('no transcript') sources = _transcript_sources(ep) if not sources: - return JsonResponse({'error': 'unsupported'}, status=404) + return fallback('unsupported') cached = EpisodeTranscript.objects.filter(episode=ep).first() if cached is not None and cached.sources == sources: @@ -1527,7 +1549,7 @@ def episode_transcript(request, pk): error = '' except chapters_mod.FetchUnavailable as e: log.info('transcript of episode %s not available: %s', ep.id, e) - return JsonResponse({'error': 'unavailable'}, status=502) + return fallback('unavailable', 502) except transcripts_mod.Unsupported as e: log.info('transcript of episode %s unsupported: %s', ep.id, e) data, error, final = None, 'unsupported', True @@ -1539,7 +1561,7 @@ def episode_transcript(request, pk): episode=ep, defaults={'sources': sources, 'data': data, 'error': error}, ) if error: - return JsonResponse({'error': error}, status=404) + return fallback(error) return JsonResponse(data)