#!/usr/bin/env python3 """Convert whitespace-delimited word timestamps in JSON to readable SRT cues.""" from __future__ import annotations import argparse import json import math import sys from pathlib import Path from typing import TypedDict class WordTimestamp(TypedDict): word: str start_ms: int end_ms: int class Cue(TypedDict): text: str start_ms: int end_ms: int TERMINAL_PUNCTUATION = (".", "!", "?") CLOSING_PUNCTUATION = set(",.!?;:%)]}") def seconds_to_milliseconds(seconds: float) -> int: return int(seconds * 1000 + 0.5) def read_word_timestamps(path: Path) -> list[WordTimestamp]: raw = json.loads(path.read_text(encoding="utf-8")) if not isinstance(raw, list) or not raw: raise ValueError("input must be a non-empty JSON array") words: list[WordTimestamp] = [] previous_start_ms = -1 previous_end_ms = -1 previous_end_seconds = -1.0 for index, item in enumerate(raw): if not isinstance(item, dict): raise ValueError(f"word[{index}] must be an object; got {item!r}") word = item.get("word") start = item.get("start") end = item.get("end") if not isinstance(word, str) or not word.strip(): raise ValueError(f"word[{index}].word must be a non-empty string; got {word!r}") if "\n" in word or "\r" in word: raise ValueError(f"word[{index}].word must stay on one line; got {word!r}") for field, value in (("start", start), ("end", end)): if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value): raise ValueError(f"word[{index}].{field} must be a finite number; got {value!r}") start_seconds = float(start) end_seconds = float(end) if start_seconds < 0: raise ValueError(f"word[{index}].start must be non-negative; got {start!r}") start_ms = seconds_to_milliseconds(start_seconds) end_ms = seconds_to_milliseconds(end_seconds) if end_ms <= start_ms: raise ValueError( f"word[{index}] must have positive duration after millisecond rounding; " f"start={start!r}, end={end!r}, " f"rounded_start_ms={start_ms}, rounded_end_ms={end_ms}" ) if start_ms < previous_start_ms: raise ValueError( f"word[{index}] is unsorted after millisecond rounding; " f"start={start!r} ({start_ms}ms), previous_start_ms={previous_start_ms}" ) if start_ms < previous_end_ms: raise ValueError( f"word[{index}] overlaps the previous word; " f"start={start!r} ({start_ms}ms), " f"previous_end={previous_end_seconds:g} ({previous_end_ms}ms)" ) words.append({"word": word.strip(), "start_ms": start_ms, "end_ms": end_ms}) previous_start_ms = start_ms previous_end_ms = end_ms previous_end_seconds = end_seconds return words def join_tokens(tokens: list[str]) -> str: text = "" for token in tokens: if text and token[0] not in CLOSING_PUNCTUATION: text += " " text += token return text def ends_sentence(token: str) -> bool: return token.rstrip("\"'”’)]}").endswith(TERMINAL_PUNCTUATION) def split_into_cues( words: list[WordTimestamp], *, max_gap_ms: int = 800, max_duration_ms: int = 6000, max_chars_per_line: int = 42, max_lines: int = 2, ) -> list[list[WordTimestamp]]: if max_gap_ms <= 0 or max_duration_ms <= 0 or max_chars_per_line <= 0 or max_lines <= 0: raise ValueError("segmentation limits must all be positive") if max_lines != 2: raise ValueError("this example supports exactly two display lines per cue") for index, item in enumerate(words): if len(item["word"]) > max_chars_per_line: raise ValueError( f"word[{index}] exceeds max_chars_per_line={max_chars_per_line}; " f"word={item['word']!r}" ) duration_ms = item["end_ms"] - item["start_ms"] if duration_ms > max_duration_ms: raise ValueError( f"word[{index}] lasts {duration_ms}ms and exceeds " f"max_duration_ms={max_duration_ms}ms" ) cues: list[list[WordTimestamp]] = [] current: list[WordTimestamp] = [] for item in words: if current: previous = current[-1] projected_text = join_tokens([entry["word"] for entry in [*current, item]]) try: wrap_cue_text(projected_text, max_chars_per_line, max_lines) projected_text_fits = True except ValueError: projected_text_fits = False starts_new_cue = ( ends_sentence(previous["word"]) or item["start_ms"] - previous["end_ms"] >= max_gap_ms or item["end_ms"] - current[0]["start_ms"] > max_duration_ms or not projected_text_fits ) if starts_new_cue: cues.append(current) current = [] current.append(item) if current: cues.append(current) return cues def wrap_cue_text(text: str, max_chars_per_line: int = 42, max_lines: int = 2) -> str: if len(text) <= max_chars_per_line: return text if max_lines != 2: raise ValueError("this example supports exactly two display lines per cue") words = text.split() candidates: list[tuple[int, int, str, str]] = [] for split_index in range(1, len(words)): first = " ".join(words[:split_index]) second = " ".join(words[split_index:]) if len(first) <= max_chars_per_line and len(second) <= max_chars_per_line: candidates.append((abs(len(first) - len(second)), max(len(first), len(second)), first, second)) if not candidates: raise ValueError( f"cue cannot fit within {max_lines} lines of {max_chars_per_line} characters: {text!r}" ) _, _, first_line, second_line = min(candidates) return f"{first_line}\n{second_line}" def format_srt_timestamp(milliseconds: int) -> str: hours, remainder = divmod(milliseconds, 3_600_000) minutes, remainder = divmod(remainder, 60_000) whole_seconds, milliseconds = divmod(remainder, 1000) return f"{hours:02d}:{minutes:02d}:{whole_seconds:02d},{milliseconds:03d}" def build_srt(words: list[WordTimestamp]) -> str: cue_words = split_into_cues(words) cues: list[Cue] = [] for group in cue_words: text = join_tokens([item["word"] for item in group]) cues.append( { "text": wrap_cue_text(text), "start_ms": group[0]["start_ms"], "end_ms": group[-1]["end_ms"], } ) blocks = [ "\n".join( [ str(index), f"{format_srt_timestamp(cue['start_ms'])} --> " f"{format_srt_timestamp(cue['end_ms'])}", cue["text"], ] ) for index, cue in enumerate(cues, start=1) ] return "\n\n".join(blocks) + "\n" def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description=( "Convert normalized whitespace-delimited word/start/end objects into SRT." ) ) parser.add_argument("input", type=Path, help="path to the word-timestamp JSON file") parser.add_argument("-o", "--output", type=Path, help="optional SRT output path") return parser.parse_args() def main() -> int: args = parse_args() try: srt = build_srt(read_word_timestamps(args.input)) if args.output: args.output.write_text(srt, encoding="utf-8") else: print(srt, end="") except (OSError, json.JSONDecodeError, ValueError) as error: print(f"error: {error}", file=sys.stderr) return 2 return 0 if __name__ == "__main__": raise SystemExit(main())