ddidderr
38894c2549
Chunked text output previously joined every word into one line and discarded speaker labels. Add a shared plain-text formatter that wraps lines, breaks paragraphs at speaker changes and pauses, and splits long turns while preserving word order. Display provider speaker IDs with chunk scope where needed, without implying that IDs identify the same person across requests. Request structured output for short diarized recordings too, and warn when speaker labels are missing or incomplete. Keep SRT formatting separate and preserve existing line breaks for short requests without diarization. Test Plan: - python3 -B -m unittest -q: all 43 tests passed. - bash -n transcribe and git diff --cached --check: passed. - Reflowed the existing transcript locally and verified all 15,888 words were preserved; source audio and generated transcripts are not committed. - A live excerpt check was blocked before upload by unavailable 1Password access; actual provider diarization remains unverified.
90 lines
3.0 KiB
Python
90 lines
3.0 KiB
Python
"""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)
|