2b38717c05
Queue reviewable Whisper drafts from the song list and lyrics editor. Preserve line breaks within one timed cue across editing, saving, reporting, and service views. Add storage and listening analytics with a durable metadata collector, related-search depth, video limits, thumbnail storage, and a browsable metadata library.
67 lines
3.2 KiB
Python
67 lines
3.2 KiB
Python
import importlib.util
|
|
import pathlib
|
|
import types
|
|
import unittest
|
|
|
|
spec = importlib.util.spec_from_file_location('auto_lyrics', pathlib.Path(__file__).with_name('auto_lyrics.py'))
|
|
worker = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(worker)
|
|
|
|
class FakeApi:
|
|
def __init__(self):
|
|
self.calls = []
|
|
def call(self, method, path, body=None, raw=False):
|
|
self.calls.append((method, path, body))
|
|
if raw:
|
|
return 200, b'fixture audio'
|
|
if path.startswith('/api/notes/'):
|
|
return 200, {'lyrics': {'rev': 5, 'data': {'lines': [{'t': 0, 'text': 'Keep the original'}]}}}
|
|
return 200, {'ok': True}
|
|
|
|
class FakeModel:
|
|
def transcribe(self, path, **options):
|
|
assert options['vad_filter'] is False
|
|
words = [types.SimpleNamespace(word=word, start=i, end=i + .8) for i, word in enumerate('Because You are God You can do anything'.split())]
|
|
return [types.SimpleNamespace(words=words)], types.SimpleNamespace(language='en', duration=10)
|
|
|
|
class WhisperRequests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.args = types.SimpleNamespace(overwrite=False, no_web=False, language=None, min_words=1, dry_run=False)
|
|
self.api = FakeApi()
|
|
self.job = {'id': 'fixture-job', 'videoId': '0gfX0dFLaBc', 'lease': 'fixture-lease', 'audioPath': '/api/media/0gfX0dFLaBc?a=1'}
|
|
|
|
def test_manual_request_uses_whisper_despite_existing_lyrics_and_never_publishes(self):
|
|
result = worker.transcribe_one(self.args, self.api, FakeModel(), self.job['videoId'], draft_job=self.job)
|
|
self.assertIsInstance(result, dict)
|
|
self.assertTrue(result['lines'])
|
|
self.assertFalse(any(method == 'PUT' or path.endswith('/lyrics/web') for method, path, body in self.api.calls))
|
|
|
|
def test_requested_job_reports_stages_and_completes_a_draft(self):
|
|
worker.process_requested(self.args, self.api, self.job, lambda: FakeModel())
|
|
posts = [body for method, path, body in self.api.calls if path.startswith('/api/lyrics-worker/jobs/')]
|
|
self.assertEqual(posts[0]['stage'], 'downloading-audio')
|
|
self.assertEqual(posts[1]['stage'], 'transcribing')
|
|
self.assertEqual(posts[-1]['status'], 'complete')
|
|
self.assertEqual(posts[-1]['lease'], 'fixture-lease')
|
|
|
|
def test_rejected_draft_is_reported_failed_instead_of_retrying_forever(self):
|
|
original = self.api.call
|
|
def reject(method, path, body=None, raw=False):
|
|
if body and body.get('status') == 'complete':
|
|
return 400, {'error': 'too many lines'}
|
|
return original(method, path, body, raw)
|
|
self.api.call = reject
|
|
worker.process_requested(self.args, self.api, self.job, lambda: FakeModel())
|
|
self.assertEqual(self.api.calls[-1][2]['status'], 'failed')
|
|
self.assertEqual(self.api.calls[-1][2]['error'], 'too many lines')
|
|
|
|
def test_no_vocals_fails_without_touching_saved_lyrics(self):
|
|
self.args.min_words = 25
|
|
worker.process_requested(self.args, self.api, self.job, lambda: FakeModel())
|
|
last = self.api.calls[-1][2]
|
|
self.assertEqual(last['status'], 'failed')
|
|
self.assertFalse(any(method == 'PUT' for method, path, body in self.api.calls))
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|