diff --git a/README.md b/README.md index 1da127c..988f0bf 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,7 @@ # Transcribe Requires Bash, FFmpeg, curl, jq, Python 3, and standard coreutils. Keep -`chunks.py` and `subtitles.py` beside the script. The API key is read using +`chunks.py`, `subtitles.py`, and `transcript.py` beside the script. The API key is read using the 1Password CLI (`op`) from `op://API Keys/openrouter-mai-transcribe-2/credential`. ```sh @@ -24,8 +24,25 @@ Speaker diarization is **enabled by default** through `provider.options.azure.diarization.enabled`, following [OpenRouter's speaker diarization API](https://openrouter.ai/docs/guides/overview/multimodal/stt#verbose-transcripts-timestamps-and-speakers). Use `--no-diarization` to explicitly disable it for every request, with either -plain text or `--srt` output. SRT uses returned speaker labels to break cues at -speaker changes; neither output format prints speaker names or IDs. +plain text or `--srt` output. Text output prints returned IDs above each paragraph: + +```text +[Speaker 0] +How did you get started? + +[Speaker 1] +It was a long time ago. +``` + +Text wraps at 80 columns (except an unusually long unbroken word), with blank +lines at speaker changes and pauses of at least two seconds. Long turns split +at sentence endings after about 400 characters, or at 800 characters even +without punctuation. The formatter preserves the words and their order. +Unlabelled passages use `[Speaker unknown]` when other words have labels; a +response without any labels is printed without speaker markers. A warning on +stderr reports missing or incomplete labels when diarization was requested. +Speaker IDs come from the provider; they are not names or inferred from text. +SRT uses labels to break cues at speaker changes but does not print the IDs. Long recordings are automatically split into uploads of at most **10 minutes** with diarization, or **30 minutes** with `--no-diarization`, including overlap @@ -47,7 +64,9 @@ no pause. Each upload includes three seconds beyond the join, so neighbouring uploads share six seconds of audio even during continuous speech. The duration cap includes this overlap; the next chunk starts before the previous one ends. -Chunked requests always ask for word timestamps, including for plain text. +Requests with diarization and all chunked requests ask for word timestamps, +including for plain text. Short requests with `--no-diarization` retain the +provider's existing text line breaks and add wrapping to long lines. The script matches words and their times in the overlap, switches transcripts at a shared word, and keeps the repeated passage once. It prefers a matching phrase to distinguish intentional repetitions. If no words match, it uses @@ -58,8 +77,10 @@ are an error. Plain text for a chunked recording is assembled from the words. SRT timestamps are shifted back to the original recording's timeline before cue generation, with continuous cue numbering. Speaker labels, when returned, -are local to each request; the script does not identify the same speaker across -chunks. Short recordings still use one request. +are local to each request; text marks them as `[Chunk 1, Speaker 0]`, +`[Chunk 2, Speaker 0]`, etc. The same ID in different chunks does **not** establish +that it is the same person. The script does not identify speakers across +chunks. Short recordings still use one request and use `[Speaker 0]` markers. Audio is first decoded to temporary mono 16 kHz PCM (about 115 MB per hour), so all cuts share the same sample clock even when container duration metadata diff --git a/chunks.py b/chunks.py index e801c6f..8e177ee 100644 --- a/chunks.py +++ b/chunks.py @@ -9,6 +9,7 @@ import sys import wave from subtitles import convert, normalized_words +from transcript import render_text CONTEXT = 3.0 # Audio on each side of a seam: six seconds shared by two uploads. ENCODING_MARGIN = 0.1 # MP3 encoder delay and frame padding at 16 kHz. @@ -156,7 +157,7 @@ def main(): (args.manifest.parent / f"response-{i}.json").read_text(encoding="utf-8") ) for i in range(len(chunks))] response = merge(chunks, responses) - sys.stdout.write(convert(response) if args.srt else response["text"] + "\n") + sys.stdout.write(convert(response) if args.srt else render_text(response)) except (ValueError, OSError, wave.Error) as error: parser.exit(1, f"Audio chunking failed: {error}\n") diff --git a/test_transcribe.py b/test_transcribe.py index a1cfdaf..c4e8f61 100644 --- a/test_transcribe.py +++ b/test_transcribe.py @@ -53,7 +53,7 @@ class TranscribeTests(unittest.TestCase): 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.assertEqual(result.stdout, "Hello.\n\nboundary continues.\n\nLast.\n") self.assertIn("chunk 2/2", result.stderr) requests = self.requests() self.assertEqual(len(requests), 2) @@ -73,15 +73,49 @@ class TranscribeTests(unittest.TestCase): self.assert_diarization(True) self.assert_cleaned() - def test_short_text_retains_provider_formatting(self): - result = self.run_cli(duration=5) + def test_short_text_without_diarization_retains_provider_line_breaks(self): + result = self.run_cli("--no-diarization", 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_diarization(False) self.assert_cleaned() + def test_short_text_requests_and_prints_speaker_labels(self): + result = self.run_cli(mode="speakers", duration=5) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(result.stdout, "[Speaker 0]\nHello\n\n[Speaker 1]\nthere\n") + self.assertEqual(self.requests()[0]["response_format"], "verbose_json") + self.assertEqual(self.requests()[0]["timestamp_granularities"], ["segment", "word"]) + self.assert_diarization(True) + self.assertEqual(result.stderr, "") + self.assert_cleaned() + + def test_chunked_text_prints_speaker_ids_with_request_scope(self): + result = self.run_cli("--chunk-minutes", "1", mode="speakers") + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(result.stdout, ( + "[Chunk 1, Speaker 0]\nHello.\n\n" + "[Chunk 1, Speaker 0]\nboundary\n\n" + "[Chunk 2, Speaker 1]\ncontinues.\n\n" + "[Chunk 2, Speaker 1]\nLast.\n" + )) + self.assertNotIn("Warning", result.stderr) + self.assert_cleaned() + + def test_missing_speaker_labels_warn_instead_of_implying_detection_worked(self): + for output in ((), ("--srt",)): + with self.subTest(output=output): + result = self.run_cli(*output, duration=5) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("returned no speaker labels", result.stderr) + self.assertNotIn("[Speaker", result.stdout) + result = self.run_cli(mode="partial-speakers", duration=5) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("some words without speaker labels", result.stderr) + self.assertEqual(result.stdout, "[Speaker 0]\nHello\n\n[Speaker unknown]\nthere\n") + def test_short_srt_still_requests_word_timestamps(self): result = self.run_cli("--srt", duration=5) self.assertEqual(result.returncode, 0, result.stderr) diff --git a/test_transcript.py b/test_transcript.py new file mode 100644 index 0000000..1eb73ec --- /dev/null +++ b/test_transcript.py @@ -0,0 +1,74 @@ +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_speaker_ids_are_scoped_to_chunks_even_when_numbers_match(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), ( + "[Chunk 1, Speaker 0]\nFirst.\n\n[Chunk 2, Speaker 0]\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() diff --git a/tests/fake_provider.py b/tests/fake_provider.py index dba0e26..5ff2cb2 100644 --- a/tests/fake_provider.py +++ b/tests/fake_provider.py @@ -67,13 +67,16 @@ else: ("continues.", 57.4, 57.8), ("Last.", 65, 65.4), ] if start >= chunk["start"] and end <= chunk["end"] ] - if mode == "speakers": - words = [ - {"word": "Hello", "start": 1, "end": 1.4}, - {"word": "there", "start": 1.5, "end": 1.9}, - ] + if mode in ("speakers", "partial-speakers"): + if len(manifest) == 1: + words = [ + {"word": "Hello", "start": 1, "end": 1.4}, + {"word": "there", "start": 1.5, "end": 1.9}, + ] if request["provider"]["options"]["azure"]["diarization"]["enabled"]: for speaker, word in enumerate(words): - word["speaker"] = speaker + if mode == "speakers" or speaker == 0: + word["speaker"] = (speaker if len(manifest) == 1 else + int(word["start"] + chunk["start"] >= 57.3)) response = {"text": " ".join(w["word"] for w in words), "words": words} output.write_text(json.dumps(response)) diff --git a/transcribe b/transcribe index 5d0a231..7be3bee 100755 --- a/transcribe +++ b/transcribe @@ -7,7 +7,7 @@ diarization=true chunk_minutes= usage() { echo "Usage: ${0##*/} [--srt] [--no-diarization] [--chunk-minutes 1..30] AUDIO_FILE" - echo 'Transcribe audio using OPENROUTER_API_KEY; print the text to stdout.' + echo 'Transcribe audio using OPENROUTER_API_KEY; print wrapped text with speaker markers to stdout.' echo ' --srt Print SRT subtitles in short cues using word timestamps instead.' echo ' --no-diarization Disable speaker diarization (enabled by default).' echo ' --chunk-minutes N Maximum upload duration in minutes (default: 10; 30 with --no-diarization).' @@ -66,6 +66,7 @@ ffmpeg -nostdin -hide_banner -loglevel error -i "$(realpath -- "$1")" \ python3 "$script_dir/chunks.py" plan "$tmp/audio.wav" "$chunk_minutes" >"$tmp/chunks.json" count=$(jq 'length' "$tmp/chunks.json") timestamps=$srt +if "$diarization"; then timestamps=true; fi if (( count > 1 )); then timestamps=true echo "Splitting audio into $count overlapping chunks (maximum $chunk_minutes minutes each)." >&2 @@ -105,6 +106,13 @@ for ((i = 0; i < count; i++)); do elif (.text | type) == "string" then true else error("Transcription failed: missing text in response") end' \ "$tmp/response-$i.json" >/dev/null + if "$diarization" && jq -e '.text | test("\\S")' "$tmp/response-$i.json" >/dev/null; then + if ! jq -e 'any(.words[]?; .speaker != null)' "$tmp/response-$i.json" >/dev/null; then + echo "Warning: chunk $((i + 1))/$count returned no speaker labels despite diarization being enabled." >&2 + elif jq -e 'any(.words[]?; .speaker == null)' "$tmp/response-$i.json" >/dev/null; then + echo "Warning: chunk $((i + 1))/$count returned some words without speaker labels." >&2 + fi + fi done # Only emit a transcript after every chunk succeeds, including SRT conversion. @@ -115,5 +123,5 @@ if (( count > 1 )); then elif "$srt"; then python3 "$script_dir/subtitles.py" "$tmp/response-0.json" else - jq -r .text "$tmp/response-0.json" + python3 "$script_dir/transcript.py" "$tmp/response-0.json" fi diff --git a/transcript.py b/transcript.py new file mode 100644 index 0000000..61964a5 --- /dev/null +++ b/transcript.py @@ -0,0 +1,89 @@ +"""Render readable plain-text paragraphs with provider-supplied speaker IDs.""" + +import json +import re +import sys +import textwrap + +from subtitles import normalized_words + +LINE_WIDTH = 80 +PARAGRAPH_TARGET = 400 +PARAGRAPH_LIMIT = 800 +PARAGRAPH_PAUSE = 2.0 +SENTENCE_END = re.compile(r'[.!?…]["\u201d\u2019\u0027)]*$') + + +def speaker_marker(speaker): + if speaker is None: + return "[Speaker unknown]" + if isinstance(speaker, tuple): + chunk, speaker = speaker + return f"[Chunk {chunk + 1}, Speaker {speaker}]" + return f"[Speaker {speaker}]" + + +def paragraphs(words, show_speakers): + result = [] + current = [] + size = 0 + + def flush(): + if current: + body = textwrap.fill( + " ".join(word["word"] for word in current), width=LINE_WIDTH, + break_long_words=False, break_on_hyphens=False, + ) + if show_speakers: + body = speaker_marker(current[0].get("speaker")) + "\n" + body + result.append(body) + current.clear() + + for word in words: + if current: + previous = current[-1] + pause = (word.get("start", 0) - previous.get("end", 0)) >= PARAGRAPH_PAUSE + if ( + word.get("speaker") != previous.get("speaker") + or pause + or (size >= PARAGRAPH_TARGET and SENTENCE_END.search(previous["word"])) + or size + 1 + len(word["word"]) > PARAGRAPH_LIMIT + ): + flush() + size = 0 + size += len(word["word"]) + bool(current) + current.append(word) + flush() + return result + + +def render_text(response): + """Preserve word order; never infer speakers from the transcript's prose.""" + if not isinstance(response, dict): + raise ValueError("Transcription response must be a JSON object") + if response.get("error"): + raise ValueError(f"Transcription failed: {response['error']}") + if "words" in response: + words = normalized_words(response, "Text output") + blocks = paragraphs(words, any(word.get("speaker") is not None for word in words)) + return "\n\n".join(blocks) + ("\n" if blocks else "") + + text = response.get("text") + if not isinstance(text, str): + raise ValueError("Transcription response must contain text as a string") + # A short --no-diarization request has only text. Keep existing line breaks + # and add wrapping/paragraphs within long lines without inventing timing. + lines = ["\n\n".join(paragraphs( + [{"word": token} for token in line.split()], False, + )) for line in text.splitlines()] + return "\n".join(lines).rstrip() + ("\n" if text.strip() else "") + + +if __name__ == "__main__": + try: + with open(sys.argv[1], encoding="utf-8") as source: + result = render_text(json.load(source)) + except (ValueError, OSError) as error: + print(f"Text conversion failed: {error}", file=sys.stderr) + sys.exit(1) + sys.stdout.write(result)