ddidderr
f8e50e4a3e
SRT conversion previously rejected otherwise valid word timestamps when a start preceded the previous word's start. Clamp regressing starts to the preceding nonempty word's start and extend ends only when necessary. Preserve transcript order and the original response while continuing to reject malformed timestamps. Document the adjustment and cover regressions, cue boundaries, blank words, and invalid input. Test Plan: - python3 -B -m unittest -v: all eight tests passed. - CLI conversion of a synthetic regressing response: passed. - git diff --cached --check: passed. - Actual recording verification blocked by unavailable 1Password auth.
105 lines
3.7 KiB
Python
105 lines
3.7 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 convert(response):
|
|
if response.get("error"):
|
|
raise ValueError(f"Transcription failed: {response['error']}")
|
|
words = response.get("words")
|
|
if not isinstance(words, list) or (not words and response.get("text", "").strip()):
|
|
raise ValueError("SRT output requires word timestamps; the API returned none")
|
|
|
|
cues = []
|
|
current = []
|
|
previous_start = -1
|
|
|
|
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 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
|
|
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)
|