from __future__ import annotations import json from pathlib import Path import typer from rich.console import Console from .alignment import transcribe_and_align from .exporters import ( write_accessibility_srt, write_json, write_wordbeat_ass, write_words_json, ) from .media import extract_audio, probe_fps, render_ass from .phrasing import build_phrases from .timeline import build_timeline app = typer.Typer(no_args_is_help=True, add_completion=False) console = Console() def _resolve_fps(media: Path, requested: float | None) -> float: if requested and requested > 0: return requested return probe_fps(media) @app.command() def process( media: Path = typer.Argument(..., exists=True, readable=True), output: Path = typer.Option(..., "--output", "-o"), device: str = typer.Option("cuda", help="cuda or cpu"), compute_type: str = typer.Option("float16", help="float16, float32, or int8"), model: str = typer.Option("large-v3", help="Whisper model name"), language: str | None = typer.Option(None, help="ISO language code or auto-detect"), batch_size: int = typer.Option(8, min=1), fps: float | None = typer.Option(None, help="Override source frame rate"), prereveal_ms: float = typer.Option(90.0, min=0.0, max=500.0), ) -> None: """Transcribe actual audio, force-align words, and export WordBeat files.""" output.mkdir(parents=True, exist_ok=True) wav = output / "audio-16k-mono.wav" console.print("[cyan]Extracting actual audio...[/cyan]") extract_audio(media, wav) console.print("[cyan]Transcribing and force-aligning every word...[/cyan]") transcript, words, metadata = transcribe_and_align( wav, device=device, compute_type=compute_type, model_name=model, language=language, batch_size=batch_size, ) if not words: raise typer.BadParameter("No aligned words were produced.") actual_fps = _resolve_fps(media, fps) phrases = build_phrases(words) timeline = build_timeline( str(media.resolve()), metadata.get("language", language or "unknown"), words, phrases, fps=actual_fps, desired_prereveal_ms=prereveal_ms, metadata=metadata, ) (output / "transcript.txt").write_text(transcript, encoding="utf-8") write_words_json(timeline, output / "words.json") write_json(timeline, output / "timeline.json") write_accessibility_srt(timeline, output / "captions.srt") write_wordbeat_ass(timeline, output / "wordbeat.ass") review_count = sum(1 for word in words if word.review_required) console.print(f"[green]Finished.[/green] {len(words)} aligned words, {len(phrases)} phrases.") if review_count: console.print(f"[yellow]{review_count} low-confidence words are marked for review.[/yellow]") console.print(f"Output: [bold]{output.resolve()}[/bold]") @app.command() def render( media: Path = typer.Argument(..., exists=True, readable=True), ass: Path = typer.Argument(..., exists=True, readable=True), output: Path = typer.Option(..., "--output", "-o"), ) -> None: """Burn the phrase-building ASS layer into a preview video.""" render_ass(media, ass, output) console.print(f"[green]Rendered:[/green] {output.resolve()}") @app.command("inspect") def inspect_timeline( timeline_path: Path = typer.Argument(..., exists=True, readable=True), ) -> None: """Print alignment and timing diagnostics.""" data = json.loads(timeline_path.read_text(encoding="utf-8")) words = data.get("words", []) events = data.get("events", []) console.print(f"Words: {len(words)}") console.print(f"Events: {len(events)}") low = [word for word in words if word.get("review_required")] console.print(f"Review required: {len(low)}") for word in low[:25]: console.print( f" {word['index']:>5}: {word['word']} " f"{word['start']:.3f}-{word['end']:.3f} score={word.get('score')}" ) if __name__ == "__main__": app()