diff --git a/docs/transcription.md b/docs/transcription.md index 05b33ff..95ca239 100644 --- a/docs/transcription.md +++ b/docs/transcription.md @@ -71,8 +71,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", @@ -110,6 +116,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 @@ -129,7 +140,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 9c6ce57..b096c80 100644 --- a/feeds/views.py +++ b/feeds/views.py @@ -752,17 +752,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 2c32eed..464703f 100644 --- a/podcasts/urls.py +++ b/podcasts/urls.py @@ -34,6 +34,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 60a5549..cf31492 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 @@ -1432,6 +1433,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, @@ -1439,7 +1443,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, 'link': ep.link, @@ -1470,6 +1477,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): @@ -1512,20 +1526,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: @@ -1536,7 +1558,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 @@ -1548,7 +1570,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)