| 1 | #!/usr/bin/env python3 |
| 2 | """ |
| 3 | PPT Master - Final Video Subtitles |
| 4 | |
| 5 | Align the exact narration text frozen in page-local Edge SRT files against the |
| 6 | audio track of a finished PowerPoint-exported video. This produces a delivery |
| 7 | SRT from the actual video timeline without rewriting speaker notes or relying |
| 8 | on theoretical slide offsets. |
| 9 | |
| 10 | Usage: |
| 11 | python3 scripts/video_subtitles.py <project_path> --video <video> --language <language> |
| 12 | |
| 13 | Examples: |
| 14 | python3 scripts/video_subtitles.py projects/demo --video exports/demo.mp4 --language zh --force |
| 15 | |
| 16 | Dependencies: |
| 17 | ffmpeg |
| 18 | python3 -m pip install stable-ts |
| 19 | """ |
| 20 | |
| 21 | from __future__ import annotations |
| 22 | |
| 23 | import argparse |
| 24 | import os |
| 25 | import re |
| 26 | import shutil |
| 27 | import sys |
| 28 | import tempfile |
| 29 | from dataclasses import dataclass |
| 30 | from pathlib import Path |
| 31 | from typing import Any |
| 32 | |
| 33 | from console_encoding import configure_utf8_stdio |
| 34 | |
| 35 | configure_utf8_stdio() |
| 36 | |
| 37 | |
| 38 | DEFAULT_MAX_CHARS = 20 |
| 39 | _TIMING_RE = re.compile( |
| 40 | r"^(?P<start>\d+:\d{2}:\d{2},\d{3})\s+-->\s+" |
| 41 | r"(?P<end>\d+:\d{2}:\d{2},\d{3})(?:\s+.*)?$" |
| 42 | ) |
| 43 | _CLAUSE_END = frozenset("，,；;：:") |
| 44 | |
| 45 | |
| 46 | @dataclass(frozen=True) |
| 47 | class SubtitleCue: |
| 48 | """One validated SRT cue.""" |
| 49 | |
| 50 | start_ms: int |
| 51 | end_ms: int |
| 52 | text: str |
| 53 | |
| 54 | |
| 55 | def _timestamp_to_ms(value: str) -> int: |
| 56 | hours_text, minutes_text, remainder = value.split(":") |
| 57 | seconds_text, milliseconds_text = remainder.split(",") |
| 58 | hours = int(hours_text) |
| 59 | minutes = int(minutes_text) |
| 60 | seconds = int(seconds_text) |
| 61 | milliseconds = int(milliseconds_text) |
| 62 | if minutes >= 60 or seconds >= 60: |
| 63 | raise ValueError(f"Invalid SRT timestamp: {value}") |
| 64 | return (((hours * 60) + minutes) * 60 + seconds) * 1000 + milliseconds |
| 65 | |
| 66 | |
| 67 | def _parse_srt(path: Path) -> list[SubtitleCue]: |
| 68 | """Read one strict, non-overlapping SRT file.""" |
| 69 | text = path.read_text(encoding="utf-8-sig") |
| 70 | blocks = re.split(r"\r?\n\s*\r?\n", text.strip()) |
| 71 | cues: list[SubtitleCue] = [] |
| 72 | previous_end = -1 |
| 73 | |
| 74 | for block_number, block in enumerate(blocks, 1): |
| 75 | lines = block.splitlines() |
| 76 | if len(lines) < 3: |
| 77 | raise ValueError(f"{path}: malformed SRT block {block_number}") |
| 78 | try: |
| 79 | cue_number = int(lines[0].strip()) |
| 80 | except ValueError as exc: |
| 81 | raise ValueError( |
| 82 | f"{path}: invalid cue number in block {block_number}" |
| 83 | ) from exc |
| 84 | if cue_number != block_number: |
| 85 | raise ValueError( |
| 86 | f"{path}: cue numbers must be consecutive from 1; " |
| 87 | f"block {block_number} is numbered {cue_number}" |
| 88 | ) |
| 89 | timing_match = _TIMING_RE.match(lines[1].strip()) |
| 90 | if timing_match is None: |
| 91 | raise ValueError( |
| 92 | f"{path}: invalid cue timing in block {block_number}" |
| 93 | ) |
| 94 | start_ms = _timestamp_to_ms(timing_match.group("start")) |
| 95 | end_ms = _timestamp_to_ms(timing_match.group("end")) |
| 96 | cue_text = re.sub(r"\s+", " ", " ".join(lines[2:])).strip() |
| 97 | if not cue_text: |
| 98 | raise ValueError(f"{path}: empty cue text in block {block_number}") |
| 99 | if end_ms <= start_ms: |
| 100 | raise ValueError( |
| 101 | f"{path}: cue {block_number} must end after it starts" |
| 102 | ) |
| 103 | if start_ms < previous_end: |
| 104 | raise ValueError( |
| 105 | f"{path}: cue {block_number} overlaps the preceding cue" |
| 106 | ) |
| 107 | cues.append(SubtitleCue(start_ms, end_ms, cue_text)) |
| 108 | previous_end = end_ms |
| 109 | |
| 110 | if not cues: |
| 111 | raise ValueError(f"No subtitle cues found: {path}") |
| 112 | return cues |
| 113 | |
| 114 | |
| 115 | def _display_length(text: str) -> int: |
| 116 | return sum(not character.isspace() for character in text) |
| 117 | |
| 118 | |
| 119 | def _hard_split(text: str, max_chars: int) -> list[str]: |
| 120 | """Split an overlong clause without dropping characters.""" |
| 121 | output: list[str] = [] |
| 122 | remaining = text.strip() |
| 123 | while _display_length(remaining) > max_chars: |
| 124 | visible = 0 |
| 125 | split_at = 0 |
| 126 | whitespace_split = 0 |
| 127 | for index, character in enumerate(remaining, 1): |
| 128 | if character.isspace(): |
| 129 | whitespace_split = index |
| 130 | continue |
| 131 | visible += 1 |
| 132 | if visible > max_chars: |
| 133 | break |
| 134 | split_at = index |
| 135 | if whitespace_split and whitespace_split <= split_at: |
| 136 | split_at = whitespace_split |
| 137 | if split_at <= 0: |
| 138 | raise ValueError("Unable to split an overlong subtitle clause") |
| 139 | output.append(remaining[:split_at].strip()) |
| 140 | remaining = remaining[split_at:].strip() |
| 141 | if remaining: |
| 142 | output.append(remaining) |
| 143 | return output |
| 144 | |
| 145 | |
| 146 | def _split_sentence(text: str, max_chars: int) -> list[str]: |
| 147 | """Keep one sentence unless its display length requires clause splitting.""" |
| 148 | sentence = re.sub(r"\s+", " ", text).strip() |
| 149 | if _display_length(sentence) <= max_chars: |
| 150 | return [sentence] |
| 151 | |
| 152 | clauses: list[str] = [] |
| 153 | start = 0 |
| 154 | for index, character in enumerate(sentence): |
| 155 | if character not in _CLAUSE_END: |
| 156 | continue |
| 157 | clause = sentence[start:index + 1].strip() |
| 158 | if clause: |
| 159 | clauses.append(clause) |
| 160 | start = index + 1 |
| 161 | tail = sentence[start:].strip() |
| 162 | if tail: |
| 163 | clauses.append(tail) |
| 164 | |
| 165 | atoms = [ |
| 166 | part |
| 167 | for clause in clauses |
| 168 | for part in _hard_split(clause, max_chars) |
| 169 | ] |
| 170 | output: list[str] = [] |
| 171 | for atom in atoms: |
| 172 | if not output: |
| 173 | output.append(atom) |
| 174 | continue |
| 175 | candidate = f"{output[-1]}{atom}" |
| 176 | if _display_length(candidate) <= max_chars: |
| 177 | output[-1] = candidate |
| 178 | else: |
| 179 | output.append(atom) |
| 180 | return output |
| 181 | |
| 182 | |
| 183 | def _page_subtitle_paths(subtitle_dir: Path) -> list[Path]: |
| 184 | """Resolve the ordered page-local SRT set without reading notes.""" |
| 185 | paths = [ |
| 186 | path |
| 187 | for path in sorted(subtitle_dir.glob("*.srt")) |
| 188 | if path.stem != "total" |
| 189 | ] |
| 190 | if not paths: |
| 191 | raise FileNotFoundError( |
| 192 | f"No page-local Edge SRT files found under {subtitle_dir}" |
| 193 | ) |
| 194 | return paths |
| 195 | |
| 196 | |
| 197 | def _frozen_transcript_lines( |
| 198 | subtitle_dir: Path, |
| 199 | max_chars: int, |
| 200 | ) -> list[str]: |
| 201 | """Return display lines derived from the exact text used for TTS.""" |
| 202 | lines: list[str] = [] |
| 203 | for path in _page_subtitle_paths(subtitle_dir): |
| 204 | for cue in _parse_srt(path): |
| 205 | lines.extend(_split_sentence(cue.text, max_chars)) |
| 206 | if not lines: |
| 207 | raise ValueError("The page-local SRT set contains no narration text") |
| 208 | return lines |
| 209 | |
| 210 | |
| 211 | def _require_stable_whisper() -> Any: |
| 212 | try: |
| 213 | import stable_whisper |
| 214 | except ImportError as exc: |
| 215 | raise RuntimeError( |
| 216 | "Final-video subtitle alignment requires stable-ts. " |
| 217 | "Install it with: python3 -m pip install stable-ts" |
| 218 | ) from exc |
| 219 | return stable_whisper |
| 220 | |
| 221 | |
| 222 | def _text_key(text: str) -> str: |
| 223 | return "".join(character for character in text if not character.isspace()) |
| 224 | |
| 225 | |
| 226 | def align_video_subtitles( |
| 227 | *, |
| 228 | video_path: Path, |
| 229 | subtitle_dir: Path, |
| 230 | output_path: Path, |
| 231 | language: str, |
| 232 | model_name: str, |
| 233 | device: str | None, |
| 234 | max_chars: int, |
| 235 | force: bool, |
| 236 | ) -> tuple[int, int]: |
| 237 | """Align frozen narration text to the final video's actual audio track.""" |
| 238 | if not video_path.is_file(): |
| 239 | raise FileNotFoundError(f"Finished video does not exist: {video_path}") |
| 240 | if shutil.which("ffmpeg") is None: |
| 241 | raise RuntimeError( |
| 242 | "Final-video subtitle alignment requires ffmpeg on PATH" |
| 243 | ) |
| 244 | if max_chars < 1: |
| 245 | raise ValueError("max_chars must be at least 1") |
| 246 | if output_path.resolve() == video_path.resolve(): |
| 247 | raise ValueError("Subtitle output must not overwrite the finished video") |
| 248 | if output_path.exists() and not force: |
| 249 | raise FileExistsError( |
| 250 | f"Output already exists: {output_path}; pass --force to replace it" |
| 251 | ) |
| 252 | |
| 253 | transcript_lines = _frozen_transcript_lines( |
| 254 | subtitle_dir, |
| 255 | max_chars, |
| 256 | ) |
| 257 | transcript = "\n".join(transcript_lines) |
| 258 | stable_whisper = _require_stable_whisper() |
| 259 | load_options: dict[str, Any] = {} |
| 260 | if device: |
| 261 | load_options["device"] = device |
| 262 | model = stable_whisper.load_model(model_name, **load_options) |
| 263 | result = model.align( |
| 264 | str(video_path), |
| 265 | transcript, |
| 266 | language=language, |
| 267 | original_split=True, |
| 268 | ) |
| 269 | if result is None: |
| 270 | raise RuntimeError("stable-ts could not align the narration transcript") |
| 271 | |
| 272 | output_path.parent.mkdir(parents=True, exist_ok=True) |
| 273 | descriptor, temporary_name = tempfile.mkstemp( |
| 274 | prefix=f".{output_path.stem}.", |
| 275 | suffix=".srt", |
| 276 | dir=str(output_path.parent), |
| 277 | ) |
| 278 | os.close(descriptor) |
| 279 | temporary_path = Path(temporary_name) |
| 280 | try: |
| 281 | result.to_srt_vtt( |
| 282 | str(temporary_path), |
| 283 | segment_level=True, |
| 284 | word_level=False, |
| 285 | ) |
| 286 | output_cues = _parse_srt(temporary_path) |
| 287 | output_text = _text_key("".join(cue.text for cue in output_cues)) |
| 288 | expected_text = _text_key("".join(transcript_lines)) |
| 289 | if output_text != expected_text: |
| 290 | raise RuntimeError( |
| 291 | "Aligned subtitle text differs from the frozen TTS transcript; " |
| 292 | "the final SRT was not published" |
| 293 | ) |
| 294 | if len(output_cues) != len(transcript_lines): |
| 295 | raise RuntimeError( |
| 296 | "stable-ts did not preserve the requested sentence/line " |
| 297 | "boundaries; the final SRT was not published" |
| 298 | ) |
| 299 | os.replace(temporary_path, output_path) |
| 300 | finally: |
| 301 | temporary_path.unlink(missing_ok=True) |
| 302 | return len(transcript_lines), output_cues[-1].end_ms |
| 303 | |
| 304 | |
| 305 | def build_parser() -> argparse.ArgumentParser: |
| 306 | parser = argparse.ArgumentParser( |
| 307 | description=__doc__, |
| 308 | formatter_class=argparse.RawDescriptionHelpFormatter, |
| 309 | ) |
| 310 | parser.add_argument("project_path", type=Path, help="Project directory") |
| 311 | parser.add_argument( |
| 312 | "--video", |
| 313 | required=True, |
| 314 | help="Finished PowerPoint-exported video; relative paths are project-relative", |
| 315 | ) |
| 316 | parser.add_argument( |
| 317 | "--language", |
| 318 | required=True, |
| 319 | help="Narration language passed to stable-ts, e.g. zh, en, ja, or ko", |
| 320 | ) |
| 321 | parser.add_argument( |
| 322 | "--subtitle-dir", |
| 323 | default=None, |
| 324 | help="Page-local Edge SRT directory; default: <project>/notes/subtitles", |
| 325 | ) |
| 326 | parser.add_argument( |
| 327 | "-o", |
| 328 | "--output", |
| 329 | default=None, |
| 330 | help="Final SRT path; default: beside the video with the same stem", |
| 331 | ) |
| 332 | parser.add_argument( |
| 333 | "--model", |
| 334 | default="base", |
| 335 | help="Whisper model used only for forced alignment (default: base)", |
| 336 | ) |
| 337 | parser.add_argument( |
| 338 | "--device", |
| 339 | default=None, |
| 340 | help="Optional stable-ts device, e.g. cpu or cuda", |
| 341 | ) |
| 342 | parser.add_argument( |
| 343 | "--max-chars", |
| 344 | type=int, |
| 345 | default=DEFAULT_MAX_CHARS, |
| 346 | help="Maximum non-space characters per final subtitle cue (default: 20)", |
| 347 | ) |
| 348 | parser.add_argument( |
| 349 | "--force", |
| 350 | action="store_true", |
| 351 | help="Replace an existing final SRT", |
| 352 | ) |
| 353 | return parser |
| 354 | |
| 355 | |
| 356 | def main(argv: list[str] | None = None) -> int: |
| 357 | parser = build_parser() |
| 358 | args = parser.parse_args(argv) |
| 359 | project_path = args.project_path.resolve() |
| 360 | if not project_path.is_dir(): |
| 361 | parser.error(f"Project path does not exist: {project_path}") |
| 362 | |
| 363 | video_path = Path(args.video) |
| 364 | if not video_path.is_absolute(): |
| 365 | video_path = project_path / video_path |
| 366 | video_path = video_path.resolve() |
| 367 | subtitle_dir = ( |
| 368 | Path(args.subtitle_dir) |
| 369 | if args.subtitle_dir |
| 370 | else Path("notes/subtitles") |
| 371 | ) |
| 372 | if not subtitle_dir.is_absolute(): |
| 373 | subtitle_dir = project_path / subtitle_dir |
| 374 | output_path = Path(args.output) if args.output else video_path.with_suffix(".srt") |
| 375 | if not output_path.is_absolute(): |
| 376 | output_path = project_path / output_path |
| 377 | |
| 378 | try: |
| 379 | cue_count, final_end_ms = align_video_subtitles( |
| 380 | video_path=video_path, |
| 381 | subtitle_dir=subtitle_dir.resolve(), |
| 382 | output_path=output_path.resolve(), |
| 383 | language=args.language, |
| 384 | model_name=args.model, |
| 385 | device=args.device, |
| 386 | max_chars=args.max_chars, |
| 387 | force=args.force, |
| 388 | ) |
| 389 | except (OSError, RuntimeError, ValueError) as exc: |
| 390 | print(f"Error: {exc}", file=sys.stderr) |
| 391 | return 1 |
| 392 | |
| 393 | print(output_path.resolve()) |
| 394 | print( |
| 395 | f"Aligned {cue_count} final-video subtitle cue(s); " |
| 396 | f"last cue ends at {final_end_ms / 1000:.3f}s", |
| 397 | file=sys.stderr, |
| 398 | ) |
| 399 | return 0 |
| 400 | |
| 401 | |
| 402 | if __name__ == "__main__": |
| 403 | raise SystemExit(main()) |
| 404 |