Render transcript-wide speaker numbers instead of exposing upload chunks in reader-facing headers. Reconcile adjacent requests using at least two unambiguous word-and-time matches in their shared audio, including when provider speaker IDs change between requests. Reject conflicting mappings and never treat matching numeric provider IDs as identity evidence. Assign fresh IDs where continuity cannot be established and report that uncertainty on stderr. Number only speakers retained after overlap trimming. Document that overlap matching cannot identify a returning speaker who was absent from the overlap, and add the reader-facing boundary to AGENTS.md. Test Plan: - python3 -B -m unittest -v: all 46 tests passed, including swapped IDs, ambiguous evidence, continuity across three requests, and plain headers. - git diff --cached --check: passed. - Tests use synthetic speaker labels; live diarization accuracy was not verified because provider credentials are unavailable in this environment.
75 lines
3.4 KiB
Python
75 lines
3.4 KiB
Python
import copy
|
|
import unittest
|
|
|
|
from chunks import merge
|
|
from transcript import render_text
|
|
|
|
|
|
def word(text, start, speaker=None):
|
|
return {"word": text, "start": start, "end": start + 0.4, "speaker": speaker}
|
|
|
|
|
|
class TranscriptTests(unittest.TestCase):
|
|
def test_turns_preserve_zero_speaker_id_and_do_not_invent_unknown_ids(self):
|
|
response = {"words": [
|
|
word("Hello", 1, 0), word("there.", 1.5, 0),
|
|
word("Hi!", 2, 1), word("Unlabelled", 2.5), word("Again.", 3, 0),
|
|
]}
|
|
original = copy.deepcopy(response)
|
|
self.assertEqual(render_text(response), (
|
|
"[Speaker 0]\nHello there.\n\n[Speaker 1]\nHi!\n\n"
|
|
"[Speaker unknown]\nUnlabelled\n\n[Speaker 0]\nAgain.\n"
|
|
))
|
|
self.assertEqual(response, original)
|
|
|
|
def test_unmatched_speakers_get_separate_ids_without_chunk_headers(self):
|
|
response = merge(
|
|
[{"start": 0, "end": 63, "seam": 0}, {"start": 57, "end": 100, "seam": 60}],
|
|
[{"words": [word("First.", 1, 0)]}, {"words": [word("Second.", 10, 0)]}],
|
|
)
|
|
self.assertEqual(render_text(response), (
|
|
"[Speaker 0]\nFirst.\n\n[Speaker 1]\nSecond.\n"
|
|
))
|
|
|
|
def test_long_unpunctuated_turn_is_wrapped_and_split_without_losing_words(self):
|
|
tokens = [f"word{i}" for i in range(400)]
|
|
result = render_text({"words": [word(token, i * 0.5) for i, token in enumerate(tokens)]})
|
|
self.assertTrue(all(len(line) <= 80 for line in result.splitlines()))
|
|
self.assertGreater(len(result.split("\n\n")), 1)
|
|
self.assertEqual(result.split(), tokens)
|
|
self.assertNotIn("Speaker", result)
|
|
|
|
def test_long_paragraph_prefers_sentence_boundary(self):
|
|
first = ["word"] * 85 + ["end."]
|
|
tokens = first + ["Next", "sentence."]
|
|
result = render_text({"words": [word(token, i * 0.5) for i, token in enumerate(tokens)]})
|
|
blocks = result.split("\n\n")
|
|
self.assertEqual([block.split() for block in blocks], [first, ["Next", "sentence."]])
|
|
|
|
def test_pause_breaks_paragraph_and_repeats_the_speaker_marker(self):
|
|
self.assertEqual(render_text({"words": [word("Before", 1, 0), word("After", 4, 0)]}),
|
|
"[Speaker 0]\nBefore\n\n[Speaker 0]\nAfter\n")
|
|
|
|
def test_text_only_preserves_existing_breaks_and_reflows_long_lines(self):
|
|
text = "Original line.\nNext line.\n\n" + "A sentence with several words. " * 50
|
|
result = render_text({"text": text})
|
|
self.assertTrue(result.startswith("Original line.\nNext line.\n\n"))
|
|
self.assertTrue(all(len(line) <= 80 for line in result.splitlines()))
|
|
self.assertGreater(result.count("\n\n"), 1)
|
|
self.assertEqual(result.split(), text.split())
|
|
|
|
def test_empty_and_long_unbroken_word(self):
|
|
self.assertEqual(render_text({"text": ""}), "")
|
|
self.assertEqual(render_text({"text": "", "words": []}), "")
|
|
self.assertEqual(render_text({"text": "x" * 100}), "x" * 100 + "\n")
|
|
|
|
def test_bad_timing_is_not_silently_replaced_with_untimed_text(self):
|
|
for response in ({"text": "Speech", "words": []}, {"text": "Speech", "words": None},
|
|
{"words": [word("Speech", -1)]}, {"text": None}, {"error": "failed"}):
|
|
with self.subTest(response=response), self.assertRaises(ValueError):
|
|
render_text(response)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|