ddidderr
a5c535cfb9
Sending entire recordings in one request can exceed provider limits. Decode audio once to a shared PCM sample clock and split uploads at a configurable maximum of 1 to 30 minutes, defaulting to 30 minutes. Prefer nearby pauses and include six seconds of overlap to protect words at audio cuts while keeping each encoded MP3 within the duration cap. Request word timestamps for chunked text and SRT output. Match overlapping words and times to retain each passage once, warn when a join falls back to timestamps, and generate SRT cues on the original recording timeline. Reuse subtitle timestamp validation without reordering regressing words. Only emit output after every request and conversion succeeds. Document the Python dependency, temporary storage, boundary limitations, and lack of automatic retry or resume. Keep single-request text formatting. Test Plan: - python3 -B -m unittest -v: all 29 tests passed, including mocked provider failures, native MP3 payload checks, boundary stitching, and SRT timing. - bash -n transcribe and shellcheck transcribe: passed. - git diff --cached --check: passed. - Locally decoded and encoded the supplied 74-minute WebM into three MP3 chunks; ffprobe confirmed each duration was below 30 minutes. - Live transcription was not verified: the configured 1Password credential was unavailable.
122 lines
4.3 KiB
Python
122 lines
4.3 KiB
Python
"""Convert OpenRouter word timestamps into readable SRT cues."""
|
|
|
|
import json
|
|
import math
|
|
import re
|
|
import sys
|
|
import textwrap
|
|
|
|
LINE_WIDTH = 42
|
|
MAX_DURATION = 6.0
|
|
PAUSE = 0.7
|
|
|
|
|
|
def lines(text):
|
|
return textwrap.wrap(
|
|
text, width=LINE_WIDTH, break_long_words=False, break_on_hyphens=False
|
|
)
|
|
|
|
|
|
def timestamp(seconds):
|
|
milliseconds = math.floor(seconds * 1000 + 0.5)
|
|
seconds, milliseconds = divmod(milliseconds, 1000)
|
|
minutes, seconds = divmod(seconds, 60)
|
|
hours, minutes = divmod(minutes, 60)
|
|
return f"{hours:02}:{minutes:02}:{seconds:02},{milliseconds:03}"
|
|
|
|
|
|
def normalized_words(response, purpose="SRT output"):
|
|
"""Validate word timing and preserve transcript order despite regressions."""
|
|
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']}")
|
|
text = response.get("text", "")
|
|
if not isinstance(text, str):
|
|
raise ValueError("Transcription response must contain text as a string")
|
|
words = response.get("words")
|
|
missing = f"{purpose} requires word timestamps; the API returned none"
|
|
if not isinstance(words, list):
|
|
raise ValueError(missing)
|
|
result = []
|
|
previous_start = -1
|
|
for item in words:
|
|
if not isinstance(item, dict) or not isinstance(item.get("word"), str):
|
|
raise ValueError("Invalid subtitle word: expected word text")
|
|
start, end = item.get("start"), item.get("end")
|
|
if (
|
|
any(
|
|
type(value) not in (int, float) or not math.isfinite(value)
|
|
for value in (start, end)
|
|
)
|
|
or start < 0
|
|
or end < start
|
|
):
|
|
raise ValueError("Invalid subtitle word: expected finite 0 <= start <= end")
|
|
word = dict(item, word=" ".join(item["word"].split()))
|
|
if not word["word"]:
|
|
continue
|
|
# Keep transcript order even when the model's word alignment moves
|
|
# backwards. Sorting by time would scramble the spoken text. Clamp
|
|
# only regressing starts, retaining the end unless it now precedes
|
|
# the start. Work on a copy so the API response remains unchanged.
|
|
start = max(start, previous_start)
|
|
end = max(end, start)
|
|
word.update(start=start, end=end)
|
|
previous_start = start
|
|
result.append(word)
|
|
if not result and text.strip():
|
|
raise ValueError(missing)
|
|
return result
|
|
|
|
|
|
def convert(response):
|
|
words = normalized_words(response)
|
|
cues = []
|
|
current = []
|
|
|
|
def flush():
|
|
if current:
|
|
text = " ".join(word["word"] for word in current)
|
|
start = current[0]["start"]
|
|
end = min(max(word["end"] for word in current), start + MAX_DURATION)
|
|
# Avoid zero-length cues after rounding very short word durations.
|
|
end = max(end, math.floor(start * 1000 + 0.5) / 1000 + 0.001)
|
|
cues.append((start, end, "\n".join(lines(text))))
|
|
current.clear()
|
|
|
|
for word in words:
|
|
start, end = word["start"], word["end"]
|
|
if len(lines(word["word"])) > 2:
|
|
raise ValueError("Invalid subtitle word: text exceeds two lines")
|
|
if current:
|
|
candidate = " ".join(w["word"] for w in [*current, word])
|
|
previous = current[-1]
|
|
if (
|
|
len(lines(candidate)) > 2
|
|
or end - current[0]["start"] > MAX_DURATION
|
|
or start - previous["end"] >= PAUSE
|
|
or word.get("speaker") != previous.get("speaker")
|
|
or (
|
|
previous["end"] - current[0]["start"] >= 1.0
|
|
and re.search(r'[.!?…]["\u201d\u2019\u0027)]*$', previous["word"])
|
|
)
|
|
):
|
|
flush()
|
|
current.append(word)
|
|
flush()
|
|
return "".join(
|
|
f"{index}\n{timestamp(start)} --> {timestamp(end)}\n{text}\n\n"
|
|
for index, (start, end, text) in enumerate(cues, 1)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
with open(sys.argv[1], encoding="utf-8") as source:
|
|
result = convert(json.load(source))
|
|
except (ValueError, OSError) as error:
|
|
print(f"SRT conversion failed: {error}", file=sys.stderr)
|
|
sys.exit(1)
|
|
sys.stdout.write(result)
|