229 lines
11 KiB
Python
229 lines
11 KiB
Python
"""Word-timing job: planning, sidecar client, v2 record format and graceful fallback (sidecar mocked)."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
_CONFIG = tempfile.mkdtemp(prefix="ttvc-wordalign-test-")
|
|
os.environ.setdefault("CONFIG_DIR", _CONFIG)
|
|
|
|
import requests
|
|
|
|
from core import epub_export, word_alignment as wa
|
|
|
|
|
|
def units_for(record, blocks):
|
|
return epub_export.plan_sentences(blocks, record["duration_ms"], record, None)
|
|
|
|
|
|
BLOCKS = [{"id": "b1", "text": "Erster Satz hier. Zweiter Satz."}, {"id": "b2", "text": "Dritter Satz."}]
|
|
RECORD = {"version": 1, "interview": False, "duration_ms": 6000, "segments": [
|
|
{"chars": len(BLOCKS[0]["text"]), "times": [[0, 2000], [2000, 3000]]},
|
|
{"chars": len(BLOCKS[1]["text"]), "times": [[3000, 6000]]}]}
|
|
|
|
|
|
def fake_post_factory(calls):
|
|
def post(url, wav, spans, language, **_):
|
|
calls.append((url, spans, language))
|
|
# one word per token, evenly spread over the span
|
|
out = []
|
|
for sp in spans:
|
|
toks = sp["text"].split()
|
|
step = (sp["end"] - sp["start"]) / len(toks)
|
|
out.append([{"word": t, "start": round(sp["start"] + i * step, 3),
|
|
"end": round(sp["start"] + (i + 1) * step, 3), "aligned": True} for i, t in enumerate(toks)])
|
|
return out
|
|
return post
|
|
|
|
|
|
class PlanningTests(unittest.TestCase):
|
|
def test_exact_blocks_get_one_job_per_sentence(self):
|
|
units, q = units_for(RECORD, BLOCKS)
|
|
self.assertEqual(q, "exact")
|
|
jobs = wa.alignment_jobs(units, BLOCKS, False)
|
|
self.assertEqual([(j["start"], j["end"], j["text"]) for j in jobs],
|
|
[(0, 2000, "Erster Satz hier."), (2000, 3000, "Zweiter Satz."), (3000, 6000, "Dritter Satz.")])
|
|
|
|
def test_anchored_block_is_aligned_as_one_span(self):
|
|
rec = json.loads(json.dumps(RECORD))
|
|
rec["segments"][0]["times"] = [[0, 3000]] # one utterance for two sentences
|
|
units, q = units_for(rec, BLOCKS)
|
|
self.assertEqual(q, "anchored")
|
|
jobs = wa.alignment_jobs(units, BLOCKS, False)
|
|
self.assertEqual(jobs[0]["text"], "Erster Satz hier. Zweiter Satz.")
|
|
self.assertEqual(jobs[0]["units"], [0, 1])
|
|
|
|
def test_speaker_label_is_not_sent_and_gets_a_zero_length_slot(self):
|
|
blocks = [{"id": "b1", "text": "Frage Eins: Wie geht es? Gut."}]
|
|
rec = {"version": 1, "interview": True, "duration_ms": 4000, "segments": [
|
|
{"chars": len("Wie geht es? Gut."), "times": [[0, 2000], [2000, 4000]]}]}
|
|
units, q = epub_export.plan_sentences(blocks, 4000, rec, [len("Wie geht es? Gut.")])
|
|
self.assertEqual(q, "exact")
|
|
jobs = wa.alignment_jobs(units, blocks, True)
|
|
self.assertEqual(jobs[0]["text"], "Wie geht es?")
|
|
self.assertEqual(jobs[0]["lead"], 2)
|
|
|
|
|
|
class JobTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.d = Path(tempfile.mkdtemp(prefix="ttvc-wa-"))
|
|
self.addCleanup(shutil.rmtree, self.d, True)
|
|
from pydub import AudioSegment
|
|
from pydub.generators import Sine
|
|
self.mp3 = self.d / "a1.mp3"
|
|
Sine(300).to_audio_segment(duration=6000).export(self.mp3, format="mp3")
|
|
self.rec_path = self.d / "a1.timings.json"
|
|
self.rec_path.write_text(json.dumps(RECORD))
|
|
self.units, _ = units_for(RECORD, BLOCKS)
|
|
self.jobs = wa.alignment_jobs(self.units, BLOCKS, False)
|
|
|
|
def test_words_are_written_as_version_2_and_old_fields_survive(self):
|
|
calls = []
|
|
got = wa.align_words(self.mp3, self.units, self.jobs, "de", url="http://x", post=fake_post_factory(calls))
|
|
seg_words = wa.build_segment_words(self.units, got, 2)
|
|
self.assertTrue(wa.write_words(self.mp3, seg_words))
|
|
rec = json.loads(self.rec_path.read_text())
|
|
self.assertEqual(rec["version"], 2)
|
|
self.assertEqual(rec["segments"][0]["times"], RECORD["segments"][0]["times"]) # v1 data untouched
|
|
self.assertEqual(len(rec["segments"][0]["words"]), 5) # 3 + 2 tokens
|
|
self.assertEqual(len(rec["segments"][1]["words"]), 2)
|
|
first, last = rec["segments"][0]["words"][0], rec["segments"][0]["words"][-1]
|
|
self.assertAlmostEqual(first[0], 0, delta=150)
|
|
self.assertAlmostEqual(last[1], 3000, delta=150)
|
|
# the v1 reader still accepts it
|
|
from core import magazine as mag
|
|
self.assertIsNotNone(mag.read_audio_timings(self.mp3))
|
|
self.assertEqual(calls[0][2], "de")
|
|
|
|
def test_chunks_are_bounded_and_progress_reported(self):
|
|
many = [{"units": [0], "start": i * 10000, "end": i * 10000 + 9000, "text": "a", "lead": 0} for i in range(7)]
|
|
self.assertTrue(all(len(c) <= 4 for c in wa._chunks(many))) # 45 s per request
|
|
seen = []
|
|
wa.align_words(self.mp3, self.units, self.jobs, "de", url="http://x",
|
|
post=fake_post_factory([]), progress=lambda a, b: seen.append((a, b)))
|
|
self.assertEqual(seen[-1], (3, 3))
|
|
|
|
def test_incomplete_segment_gets_no_words(self):
|
|
got = {0: [[0, 1]] * 3} # sentence 1 missing for segment 0
|
|
self.assertEqual(wa.build_segment_words(self.units, got, 2), {})
|
|
|
|
|
|
class ClientFallbackTests(unittest.TestCase):
|
|
def test_retries_with_backoff_then_raises_when_sidecar_is_down(self):
|
|
sleeps = []
|
|
with mock.patch.object(wa.requests, "post", side_effect=requests.ConnectionError("down")) as p:
|
|
with self.assertRaises(wa.AlignerError):
|
|
wa._post_chunk("http://x", b"w", [{"start": 0, "end": 1, "text": "a"}], "de", sleep=sleeps.append)
|
|
self.assertEqual(p.call_count, wa.ATTEMPTS)
|
|
self.assertEqual(sleeps, [1, 2])
|
|
|
|
def test_client_error_is_not_retried(self):
|
|
resp = mock.Mock(status_code=422)
|
|
with mock.patch.object(wa.requests, "post", return_value=resp) as p:
|
|
with self.assertRaises(wa.AlignerError):
|
|
wa._post_chunk("http://x", b"w", [], "de", sleep=lambda s: None)
|
|
self.assertEqual(p.call_count, 1)
|
|
|
|
def test_busy_then_ok(self):
|
|
ok = mock.Mock(status_code=200); ok.json.return_value = {"spans": [[]]}
|
|
with mock.patch.object(wa.requests, "post", side_effect=[mock.Mock(status_code=503), ok]):
|
|
self.assertEqual(wa._post_chunk("http://x", b"w", [], "de", sleep=lambda s: None), [[]])
|
|
|
|
def test_failed_job_is_recorded_and_not_restarted_within_cooldown(self):
|
|
def run(progress):
|
|
raise wa.AlignerError("aligner unreachable")
|
|
with mock.patch.dict(os.environ, {"ALIGNER_URL": "http://x"}):
|
|
self.assertTrue(wa.should_autostart("docF", "a1"))
|
|
self.assertTrue(wa.start("docF", "a1", run))
|
|
for _ in range(100):
|
|
if wa.status("docF", "a1")["state"] != "running":
|
|
break
|
|
time.sleep(0.02)
|
|
st = wa.status("docF", "a1")
|
|
self.assertEqual(st["state"], "failed")
|
|
self.assertFalse(wa.should_autostart("docF", "a1"))
|
|
|
|
def test_disabled_without_url(self):
|
|
with mock.patch.dict(os.environ, {"ALIGNER_URL": ""}):
|
|
self.assertFalse(wa.should_autostart("docX", "a1"))
|
|
|
|
|
|
class RouteTests(unittest.TestCase):
|
|
"""GET /timings carries words when present and never waits for the job."""
|
|
|
|
def setUp(self):
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
import routes.magazine as routes
|
|
self.routes = routes
|
|
self.lib = Path(tempfile.mkdtemp(prefix="ttvc-lib-"))
|
|
self.addCleanup(shutil.rmtree, self.lib, True)
|
|
p = mock.patch.object(routes, "_LIB", self.lib); p.start(); self.addCleanup(p.stop)
|
|
app = FastAPI(); app.include_router(routes.router)
|
|
self.client = TestClient(app)
|
|
self.doc = self.client.post("/api/magazine/docs", json={"title": "T"}).json()["id"]
|
|
d = self.d = self.lib / self.doc
|
|
blocks = [{"id": "b1", "page": 1, "bbox": [0, 0, 10, 10], "type": "body", "text": BLOCKS[0]["text"], "reading_order": 1},
|
|
{"id": "b2", "page": 1, "bbox": [0, 40, 10, 10], "type": "body", "text": BLOCKS[1]["text"], "reading_order": 2}]
|
|
(d / "layout.json").write_text(json.dumps({"pages": [{"page": 1, "width": 600, "height": 800}], "blocks": blocks, "images": []}))
|
|
(d / "articles.json").write_text(json.dumps({"articles": [
|
|
{"id": "a1", "title": "T", "pages": [1], "blocks": ["b1", "b2"], "images": [], "status": "approved"}]}))
|
|
(d / "audio").mkdir(exist_ok=True)
|
|
(d / "audio" / "a1.mp3").write_bytes(b"x")
|
|
self.rec = d / "audio" / "a1.timings.json"
|
|
self.rec.write_text(json.dumps(dict(RECORD, interview=True)))
|
|
self.url = f"/api/magazine/{self.doc}/articles/a1/timings"
|
|
|
|
def test_without_sidecar_the_answer_is_unchanged_and_instant(self):
|
|
with mock.patch.dict(os.environ, {"ALIGNER_URL": ""}):
|
|
body = self.client.get(self.url).json()
|
|
self.assertTrue(body["available"])
|
|
self.assertEqual(body["words"], "none")
|
|
self.assertNotIn("words", body["sentences"][0])
|
|
|
|
def test_words_are_returned_per_sentence(self):
|
|
rec = json.loads(self.rec.read_text()); rec["version"] = 2
|
|
rec["segments"][0]["words"] = [[0, 400], [500, 900], [1000, 1900], [2100, 2500], [2600, 2900]]
|
|
rec["segments"][1]["words"] = [[3100, 3500], [3600, 5900]]
|
|
self.rec.write_text(json.dumps(rec))
|
|
body = self.client.get(self.url).json()
|
|
self.assertEqual(body["words"], "ready")
|
|
self.assertEqual(body["sentences"][0]["words"], [[0.0, 0.4], [0.5, 0.9], [1.0, 1.9]])
|
|
self.assertEqual(body["sentences"][1]["words"], [[2.1, 2.5], [2.6, 2.9]])
|
|
self.assertEqual(body["sentences"][2]["words"][1], [3.6, 5.9])
|
|
|
|
def test_mismatching_words_are_ignored(self):
|
|
rec = json.loads(self.rec.read_text()); rec["segments"][0]["words"] = [[0, 1]]
|
|
self.rec.write_text(json.dumps(rec))
|
|
with mock.patch.dict(os.environ, {"ALIGNER_URL": ""}):
|
|
body = self.client.get(self.url).json()
|
|
self.assertNotIn("words", body["sentences"][0])
|
|
|
|
def test_first_request_starts_the_job_in_background(self):
|
|
started = []
|
|
with mock.patch.dict(os.environ, {"ALIGNER_URL": "http://x"}), \
|
|
mock.patch.object(self.routes, "_start_word_alignment", lambda *a: started.append(a) or True):
|
|
body = self.client.get(self.url).json()
|
|
self.assertEqual(body["words"], "running")
|
|
self.assertEqual(len(started), 1)
|
|
self.assertTrue(body["available"])
|
|
|
|
def test_explicit_endpoint_requires_configuration_and_timings(self):
|
|
url = f"/api/magazine/{self.doc}/articles/a1/align-words"
|
|
with mock.patch.dict(os.environ, {"ALIGNER_URL": ""}):
|
|
self.assertEqual(self.client.post(url).status_code, 503)
|
|
with mock.patch.dict(os.environ, {"ALIGNER_URL": "http://x"}):
|
|
self.rec.unlink()
|
|
self.assertEqual(self.client.post(url).status_code, 404)
|
|
self.assertIn("state", self.client.get(url).json())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|