ddidderr
24b1030776
Request speaker diarization through OpenRouter's Azure provider options for all transcription uploads. Add --no-diarization to explicitly disable it for both plain text and SRT output; returned speaker labels let existing SRT cue generation split at speaker changes. Default to 10-minute chunks when diarization is enabled because Microsoft reports failures around 15 minutes and longer. Preserve the 30-minute default with the opt-out and honor explicit --chunk-minutes values in both modes. Document the options, duration limits, and output behavior. Test Plan: - python3 -B -m unittest -v: all 32 tests passed, including request options, default upload limits, opt-out combinations, and SRT speaker boundaries. - bash -n transcribe: passed. - ./transcribe --help: verified the new option and chunk defaults. - git diff --cached --check: passed. - Provider responses were mocked; no live API request was made.
170 lines
7.9 KiB
Python
170 lines
7.9 KiB
Python
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
from test_chunks import make_audio
|
|
|
|
ROOT = Path(__file__).resolve().parent
|
|
TOOLS = ("bash", "ffmpeg", "ffprobe", "jq", "base64", "mktemp", "realpath")
|
|
|
|
|
|
@unittest.skipUnless(all(shutil.which(tool) for tool in TOOLS), "CLI media dependencies missing")
|
|
class TranscribeTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.directory = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.directory.cleanup)
|
|
self.path = Path(self.directory.name)
|
|
self.bin = self.path / "bin"
|
|
self.bin.mkdir()
|
|
for name in ("op", "curl"):
|
|
fake = self.bin / name
|
|
shutil.copyfile(ROOT / "tests" / "fake_provider.py", fake)
|
|
fake.chmod(0o755)
|
|
self.log = self.path / "requests.jsonl"
|
|
self.env = dict(os.environ, PATH=f"{self.bin}{os.pathsep}{os.environ['PATH']}",
|
|
TRANSCRIBE_TEST_LOG=str(self.log), TRANSCRIBE_TEST_MODE="ok")
|
|
self.audio = self.path / "input with spaces.wav"
|
|
|
|
def run_cli(self, *args, mode="ok", duration=70):
|
|
make_audio(self.audio, duration)
|
|
return subprocess.run([str(ROOT / "transcribe"), *args, str(self.audio)],
|
|
env=dict(self.env, TRANSCRIBE_TEST_MODE=mode),
|
|
capture_output=True, text=True, timeout=30)
|
|
|
|
def requests(self):
|
|
return [json.loads(line) for line in self.log.read_text().splitlines()]
|
|
|
|
def assert_cleaned(self):
|
|
self.assertTrue(all(not Path(r["staging"]).exists() for r in self.requests()))
|
|
|
|
def assert_diarization(self, enabled):
|
|
for request in self.requests():
|
|
self.assertEqual(request["provider"], {
|
|
"options": {"azure": {"diarization": {"enabled": enabled}}},
|
|
})
|
|
self.assertIs(
|
|
request["provider"]["options"]["azure"]["diarization"]["enabled"], enabled,
|
|
)
|
|
|
|
def test_long_text_uploads_bounded_mp3_and_stitches_once(self):
|
|
result = self.run_cli("--chunk-minutes", "1")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(result.stdout, "Hello. boundary continues. Last.\n")
|
|
self.assertIn("chunk 2/2", result.stderr)
|
|
requests = self.requests()
|
|
self.assertEqual(len(requests), 2)
|
|
self.assertTrue(all(r["upload_duration"] <= 60 for r in requests))
|
|
self.assertAlmostEqual(requests[0]["chunk"]["end"] - requests[1]["chunk"]["start"], 6)
|
|
self.assert_diarization(True)
|
|
self.assert_cleaned()
|
|
|
|
def test_long_srt_uses_original_timeline_and_continuous_numbers(self):
|
|
result = self.run_cli("--srt", "--chunk-minutes", "1")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(result.stdout, (
|
|
"1\n00:00:01,000 --> 00:00:01,400\nHello.\n\n"
|
|
"2\n00:00:56,700 --> 00:00:57,800\nboundary continues.\n\n"
|
|
"3\n00:01:05,000 --> 00:01:05,400\nLast.\n\n"
|
|
))
|
|
self.assert_diarization(True)
|
|
self.assert_cleaned()
|
|
|
|
def test_short_text_retains_provider_formatting(self):
|
|
result = self.run_cli(duration=5)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(result.stdout, "Provider's original formatting.\nNext line.\n")
|
|
self.assertEqual(len(self.requests()), 1)
|
|
self.assertIsNone(self.requests()[0]["response_format"])
|
|
self.assert_diarization(True)
|
|
self.assert_cleaned()
|
|
|
|
def test_short_srt_still_requests_word_timestamps(self):
|
|
result = self.run_cli("--srt", duration=5)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(result.stdout, "1\n00:00:01,000 --> 00:00:01,400\nHello.\n\n")
|
|
self.assertEqual(self.requests()[0]["response_format"], "verbose_json")
|
|
self.assertEqual(self.requests()[0]["timestamp_granularities"], ["segment", "word"])
|
|
self.assert_diarization(True)
|
|
|
|
def test_no_diarization_applies_to_text_srt_and_every_chunk(self):
|
|
for output in ((), ("--srt",)):
|
|
for chunked in (False, True):
|
|
with self.subTest(output=output, chunked=chunked):
|
|
self.log.unlink(missing_ok=True)
|
|
options = ("--chunk-minutes", "1") if chunked else ()
|
|
result = self.run_cli(
|
|
*output, *options, "--no-diarization", duration=70 if chunked else 5,
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertTrue(result.stdout)
|
|
self.assertEqual(len(self.requests()), 2 if chunked else 1)
|
|
self.assert_diarization(False)
|
|
for request in self.requests():
|
|
if output or chunked:
|
|
self.assertEqual(request["response_format"], "verbose_json")
|
|
self.assertEqual(request["timestamp_granularities"], ["segment", "word"])
|
|
else:
|
|
self.assertIsNone(request["response_format"])
|
|
self.assertIsNone(request["timestamp_granularities"])
|
|
self.assertLessEqual(request["upload_duration"], 60)
|
|
self.assert_cleaned()
|
|
# The opt-out must also preserve a limit specified after it.
|
|
self.log.unlink()
|
|
result = self.run_cli("--no-diarization", "--chunk-minutes", "1")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(len(self.requests()), 2)
|
|
self.assert_diarization(False)
|
|
self.assert_cleaned()
|
|
|
|
def test_default_upload_limits_depend_on_diarization(self):
|
|
for options, limit in (((), 600), (("--no-diarization",), 1800)):
|
|
with self.subTest(options=options):
|
|
self.log.unlink(missing_ok=True)
|
|
result = self.run_cli(*options, duration=limit + 10)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
requests = self.requests()
|
|
self.assertEqual(len(requests), 2)
|
|
self.assertGreaterEqual(requests[0]["upload_duration"], limit - 0.1)
|
|
self.assertTrue(all(r["upload_duration"] <= limit for r in requests))
|
|
self.assert_diarization(not options)
|
|
self.assert_cleaned()
|
|
|
|
def test_diarization_breaks_srt_at_speaker_changes(self):
|
|
result = self.run_cli("--srt", mode="speakers", duration=5)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(result.stdout, (
|
|
"1\n00:00:01,000 --> 00:00:01,400\nHello\n\n"
|
|
"2\n00:00:01,500 --> 00:00:01,900\nthere\n\n"
|
|
))
|
|
self.assert_diarization(True)
|
|
self.log.unlink()
|
|
result = self.run_cli("--no-diarization", "--srt", mode="speakers", duration=5)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(result.stdout, "1\n00:00:01,000 --> 00:00:01,900\nHello there\n\n")
|
|
self.assert_diarization(False)
|
|
|
|
def test_failed_later_chunk_has_no_partial_stdout_and_cleans_up(self):
|
|
for mode in ("http-error", "bad-json", "api-error", "missing-words"):
|
|
with self.subTest(mode=mode):
|
|
result = self.run_cli("--chunk-minutes", "1", mode=mode)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual(result.stdout, "")
|
|
self.assertTrue(result.stderr)
|
|
self.assert_cleaned()
|
|
|
|
def test_invalid_limits_fail_before_provider_access(self):
|
|
for value in ("0", "31", "abc", "1.5", "-1"):
|
|
with self.subTest(value=value):
|
|
result = self.run_cli("--chunk-minutes", value, duration=1)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("integer between 1 and 30", result.stderr)
|
|
self.assertFalse(self.log.exists())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|