diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml index 5c7e6bd..687f843 100644 --- a/.github/workflows/test.yaml +++ b/.github/workflows/test.yaml @@ -29,3 +29,6 @@ jobs: - name: Run smoke test run: pycaps --help + + - name: Run unit tests + run: python -m unittest discover -s tests -p "test_*.py" diff --git a/.gitignore b/.gitignore index 48d73d6..2b3d369 100644 --- a/.gitignore +++ b/.gitignore @@ -14,3 +14,4 @@ demo TODO.md research .secrets +.DS_Store \ No newline at end of file diff --git a/README.md b/README.md index b0de153..7579f70 100644 --- a/README.md +++ b/README.md @@ -106,6 +106,14 @@ This command will: 3. Apply the template's styles and animations. 4. Save the result in a new file. +If you already have a transcript from another tool, you can skip built-in transcription: + +```bash +pycaps render --input my_video.mp4 --template minimalist --transcript transcript.json +``` + +Supported transcript formats: `whisper_json`, `pycaps_json`, `srt`, `vtt` (`--transcript-format` is optional and defaults to `auto`). + ### 2. Using the Python Library For full control, use the `CapsPipelineBuilder` in your Python code. diff --git a/docs/CLI.md b/docs/CLI.md index 3c0d6b2..cf8d658 100644 --- a/docs/CLI.md +++ b/docs/CLI.md @@ -52,8 +52,32 @@ This will create a new video named `output_... .mp4` in your current directory. pycaps render ... --preview-time 10.5,15 ``` - `--subtitle-data `: Skips transcription and uses a pre-generated `.json` data file. Great for re-rendering with different styles. +- `--transcript `: Uses an external transcription input and skips built-in STT. +- `--transcript-format `: Declares transcript format (defaults to `auto`). - `-v`, `--verbose`: Show detailed logs during processing. +### External Transcript Input + +If you already transcribed audio with another tool, you can pass that file directly: + +```bash +pycaps render --input my_video.mp4 --template minimalist --transcript transcript.json +``` + +Supported formats: +- `whisper_json` (Whisper response with `segments[].words[]`) +- `pycaps_json` (pycaps document JSON or lightweight `segments[].words[]`) +- `srt` +- `vtt` (including inline timestamp tags) + +Use `--transcript-format` when auto-detection is not enough: + +```bash +pycaps render --input my_video.mp4 --template minimalist --transcript subtitles.vtt --transcript-format vtt +``` + +`--subtitle-data` is different: it expects already processed pycaps subtitle data and skips both transcription and processing. `--transcript` still runs processing (splitters, tags, effects) after loading text/timings. + ## `pycaps preview-styles` This command launches a GUI to help you design your subtitle styles in real-time, without needing a video. @@ -88,4 +112,3 @@ Manage your Pycaps API key for AI features. - `pycaps config`: Shows your currently saved API key, if any. - `pycaps config --set-api-key `: Saves your API key locally. - `pycaps config --unset-api-key`: Removes your saved API key. - diff --git a/docs/EXAMPLES.md b/docs/EXAMPLES.md index 228707b..ad5464b 100644 --- a/docs/EXAMPLES.md +++ b/docs/EXAMPLES.md @@ -82,6 +82,33 @@ pipeline.run() print("Video has been rendered!") ``` +--- +## Example 2.5: Render Using an Existing Transcript + +### CLI + +```bash +pycaps render --input my_video.mp4 --template minimalist --transcript transcript.srt +``` + +### Python + +```python +from pycaps import CapsPipelineBuilder, TranscriptFormat + +pipeline = ( + CapsPipelineBuilder() + .with_input_video("my_video.mp4") + .with_transcription_file("transcript.vtt", TranscriptFormat.VTT) + .add_css("styles.css") + .build() +) + +pipeline.run() +``` + +You can also pass a dict or a `Document` directly with `with_transcription(...)`. + --- ## Example 3: Advanced JSON with Tagger and Effects @@ -200,4 +227,4 @@ builder.add_animation( pipeline = builder.build() pipeline.run() -print("Advanced pipeline finished successfully!") \ No newline at end of file +print("Advanced pipeline finished successfully!") diff --git a/src/pycaps/__init__.py b/src/pycaps/__init__.py index c64d2fa..a535b5c 100644 --- a/src/pycaps/__init__.py +++ b/src/pycaps/__init__.py @@ -1,6 +1,6 @@ from .pipeline import CapsPipeline, CapsPipelineBuilder, JsonConfigLoader from .renderer import CssSubtitleRenderer, PictexSubtitleRenderer -from .transcriber import WhisperAudioTranscriber, GoogleAudioTranscriber, AudioTranscriber, LimitByWordsSplitter, LimitByCharsSplitter, SplitIntoSentencesSplitter +from .transcriber import WhisperAudioTranscriber, GoogleAudioTranscriber, AudioTranscriber, LimitByWordsSplitter, LimitByCharsSplitter, SplitIntoSentencesSplitter, TranscriptFormat, load_transcription from .effect import * from .animation import * from .selector import WordClipSelector @@ -10,4 +10,4 @@ from .ai import LlmProvider from .template import TemplateLoader, TemplateFactory, DEFAULT_TEMPLATE_NAME -__version__ = "0.2.1" \ No newline at end of file +__version__ = "0.2.1" diff --git a/src/pycaps/cli/render_cli.py b/src/pycaps/cli/render_cli.py index 5f1d9de..f5357f8 100644 --- a/src/pycaps/cli/render_cli.py +++ b/src/pycaps/cli/render_cli.py @@ -6,6 +6,7 @@ from pycaps.common import VideoQuality from pycaps.layout import VerticalAlignmentType, SubtitleLayoutOptions from pycaps.template import TemplateLoader, DEFAULT_TEMPLATE_NAME, TemplateFactory +from pycaps.transcriber import TranscriptFormat render_app = typer.Typer() @@ -62,12 +63,20 @@ def render( preview: bool = typer.Option(False, "--preview", help="Generate a low quality preview of the rendered video", rich_help_panel="Utils"), preview_time: Optional[str] = typer.Option(None, "--preview-time", help="Generate a low quality preview of the rendered video at the given time, example: --preview-time=10,15", rich_help_panel="Utils", show_default=False), subtitle_data: Optional[str] = typer.Option(None, "--subtitle-data", help="Subtitle data file path. If provided, the rendering process will skip the transcription and tagging steps", rich_help_panel="Utils", show_default=False), + transcript: Optional[str] = typer.Option(None, "--transcript", help="Path to an external transcript file. If provided, pycaps skips built-in transcription", rich_help_panel="Utils", show_default=False), + transcript_format: TranscriptFormat = typer.Option(TranscriptFormat.AUTO, "--transcript-format", help="Transcript format: auto|whisper_json|pycaps_json|srt|vtt", rich_help_panel="Utils", show_default=False), verbose: bool = typer.Option(False, "--verbose", "-v", help="Verbose mode", rich_help_panel="Utils"), ): set_logging_level(logging.DEBUG if verbose else logging.INFO) if template_name and config_file: typer.echo("Only one of --template or --config can be provided", err=True) return None + if subtitle_data and transcript: + typer.echo("Only one of --subtitle-data or --transcript can be provided", err=True) + return None + if transcript_format != TranscriptFormat.AUTO and not transcript: + typer.echo("--transcript-format requires --transcript", err=True) + return None if not template_name and not config_file: template_name = DEFAULT_TEMPLATE_NAME @@ -85,6 +94,7 @@ def render( # TODO: this has a little issue (if you set lang via js + whisper model by cli, it will change the lang to None) if language or whisper_model or whisper_prompt: builder.with_whisper_config(language=language, model_size=whisper_model if whisper_model else "base", initial_prompt=whisper_prompt) if subtitle_data: builder.with_subtitle_data_path(subtitle_data) + if transcript: builder.with_transcription_file(transcript, transcript_format) if transcription_preview: builder.should_preview_transcription(True) if video_quality: builder.with_video_quality(video_quality) if layout_align or layout_align_offset: builder.with_layout_options(_build_layout_options(builder, layout_align, layout_align_offset)) diff --git a/src/pycaps/pipeline/caps_pipeline.py b/src/pycaps/pipeline/caps_pipeline.py index 13fb69a..3daa7c4 100644 --- a/src/pycaps/pipeline/caps_pipeline.py +++ b/src/pycaps/pipeline/caps_pipeline.py @@ -32,6 +32,7 @@ def __init__(self): self._sound_effects: List[SoundEffect] = [] self._should_save_subtitle_data: bool = True self._subtitle_data_path_for_loading: Optional[str] = None + self._transcription_for_loading: Optional[Document] = None self._should_preview_transcription: bool = False self._layout_options = SubtitleLayoutOptions() self._preview_time: Optional[Tuple[float, float]] = None @@ -254,6 +255,9 @@ def run(self) -> None: logger().info(f"Loading subtitle data from: {self._subtitle_data_path_for_loading}") document = SubtitleDataService(self._subtitle_data_path_for_loading).load() self._cut_document_for_preview_time(document) + elif self._transcription_for_loading: + logger().info("Using external transcription input.") + document = self.process_document(self._transcription_for_loading) else: initial_document = self.transcribe() document = self.process_document(initial_document) diff --git a/src/pycaps/pipeline/caps_pipeline_builder.py b/src/pycaps/pipeline/caps_pipeline_builder.py index 1d4ea9c..e155a7c 100644 --- a/src/pycaps/pipeline/caps_pipeline_builder.py +++ b/src/pycaps/pipeline/caps_pipeline_builder.py @@ -2,6 +2,8 @@ from .caps_pipeline import CapsPipeline from pycaps.layout import SubtitleLayoutOptions, LineSplitter, LayoutUpdater, PositionsCalculator from pycaps.transcriber import AudioTranscriber, BaseSegmentSplitter, WhisperAudioTranscriber, PreviewTranscriber +from pycaps.transcriber import TranscriptFormat, load_transcription +from pycaps.common import Document from typing import Optional from pycaps.animation import Animation, ElementAnimator from pycaps.common import ElementType, EventType, VideoQuality, CacheStrategy @@ -75,6 +77,17 @@ def with_subtitle_data_path(self, subtitle_data_path: str) -> "CapsPipelineBuild raise ValueError(f"Subtitle data file not found: {subtitle_data_path}") self._caps_pipeline._subtitle_data_path_for_loading = subtitle_data_path return self + + def with_transcription(self, transcription: Document | dict | str, format: TranscriptFormat | str = TranscriptFormat.AUTO) -> "CapsPipelineBuilder": + self._caps_pipeline._transcription_for_loading = load_transcription(transcription, format) + return self + + def with_transcription_file(self, path: str, format: TranscriptFormat | str = TranscriptFormat.AUTO) -> "CapsPipelineBuilder": + if not os.path.exists(path): + raise ValueError(f"Transcription file not found: {path}") + if not os.path.isfile(path): + raise ValueError(f"Transcription path is not a file: {path}") + return self.with_transcription(path, format) def should_save_subtitle_data(self, should_save: bool) -> "CapsPipelineBuilder": self._caps_pipeline._should_save_subtitle_data = should_save @@ -121,4 +134,4 @@ def build(self, preview_time: Optional[tuple[float, float]] = None) -> CapsPipel pipeline = self._caps_pipeline self._caps_pipeline = CapsPipeline() - return pipeline \ No newline at end of file + return pipeline diff --git a/src/pycaps/transcriber/__init__.py b/src/pycaps/transcriber/__init__.py index 071ea86..8ffd53c 100644 --- a/src/pycaps/transcriber/__init__.py +++ b/src/pycaps/transcriber/__init__.py @@ -5,6 +5,8 @@ from .editor import TranscriptionEditor from .preview_transcriber import PreviewTranscriber from .google_audio_transcriber import GoogleAudioTranscriber +from .transcript_format import TranscriptFormat +from .transcript_loader import load_transcription __all__ = [ "AudioTranscriber", @@ -15,5 +17,7 @@ "SplitIntoSentencesSplitter", "TranscriptionEditor", "PreviewTranscriber", - "GoogleAudioTranscriber" + "GoogleAudioTranscriber", + "TranscriptFormat", + "load_transcription", ] diff --git a/src/pycaps/transcriber/transcript_format.py b/src/pycaps/transcriber/transcript_format.py new file mode 100644 index 0000000..c84b4a4 --- /dev/null +++ b/src/pycaps/transcriber/transcript_format.py @@ -0,0 +1,9 @@ +from enum import Enum + + +class TranscriptFormat(str, Enum): + AUTO = "auto" + WHISPER_JSON = "whisper_json" + PYCAPS_JSON = "pycaps_json" + SRT = "srt" + VTT = "vtt" diff --git a/src/pycaps/transcriber/transcript_loader.py b/src/pycaps/transcriber/transcript_loader.py new file mode 100644 index 0000000..40fc077 --- /dev/null +++ b/src/pycaps/transcriber/transcript_loader.py @@ -0,0 +1,383 @@ +from __future__ import annotations + +import html +import json +import re +from pathlib import Path +from typing import Any + +from pycaps.common import Document, Line, Segment, TimeFragment, Word + +from .transcript_format import TranscriptFormat + +_TIME_EPSILON = 0.01 +_VTT_INLINE_TIME_RE = re.compile(r"<((?:\d{2}:)?\d{2}:\d{2}\.\d{3})>") + + +def load_transcription(source: Document | dict[str, Any] | str, format: TranscriptFormat | str = TranscriptFormat.AUTO) -> Document: + resolved_format = _resolve_format(format) + + if isinstance(source, Document): + return _normalize_document(source) + + if isinstance(source, dict): + return _load_from_dict(source, resolved_format) + + if isinstance(source, str): + source_path = Path(source) + if not source_path.exists(): + raise ValueError(f"Transcription file not found: {source}") + if not source_path.is_file(): + raise ValueError(f"Transcription path is not a file: {source}") + return _load_from_path(source_path, resolved_format) + + raise ValueError("Invalid transcription source. Expected Document, dict, or file path.") + + +def _load_from_path(path: Path, requested_format: TranscriptFormat) -> Document: + suffix = path.suffix.lower() + content = path.read_text(encoding="utf-8") + resolved_format = requested_format + + if requested_format == TranscriptFormat.AUTO: + if suffix == ".srt": + resolved_format = TranscriptFormat.SRT + elif suffix == ".vtt": + resolved_format = TranscriptFormat.VTT + else: + try: + data = json.loads(content) + except json.JSONDecodeError as e: + raise ValueError(f"Could not parse transcription file as JSON: {path}") from e + return _load_from_dict(data, TranscriptFormat.AUTO) + + if resolved_format == TranscriptFormat.SRT: + return _parse_srt(content) + if resolved_format == TranscriptFormat.VTT: + return _parse_vtt(content) + + if resolved_format in (TranscriptFormat.WHISPER_JSON, TranscriptFormat.PYCAPS_JSON): + try: + data = json.loads(content) + except json.JSONDecodeError as e: + raise ValueError(f"Could not parse transcription file as JSON: {path}") from e + return _load_from_dict(data, resolved_format) + + raise ValueError(f"Unsupported transcript format: {resolved_format.value}") + + +def _load_from_dict(data: dict[str, Any], requested_format: TranscriptFormat) -> Document: + resolved_format = requested_format if requested_format != TranscriptFormat.AUTO else _detect_json_format(data) + if resolved_format == TranscriptFormat.WHISPER_JSON: + document = _parse_whisper_json(data) + elif resolved_format == TranscriptFormat.PYCAPS_JSON: + document = _parse_pycaps_json(data) + else: + raise ValueError(f"Format '{resolved_format.value}' requires a text subtitle file path.") + return _normalize_document(document) + + +def _detect_json_format(data: dict[str, Any]) -> TranscriptFormat: + segments = data.get("segments") + if not isinstance(segments, list): + raise ValueError("Invalid transcript JSON: expected a top-level 'segments' array.") + + first_segment = next((segment for segment in segments if isinstance(segment, dict)), None) + if first_segment is None: + return TranscriptFormat.PYCAPS_JSON + + if isinstance(first_segment.get("lines"), list): + return TranscriptFormat.PYCAPS_JSON + + words = first_segment.get("words") + if isinstance(words, list): + if "language" in data or "text" in first_segment or "id" in first_segment: + return TranscriptFormat.WHISPER_JSON + first_word = next((word for word in words if isinstance(word, dict)), None) + if first_word and "word" in first_word and "time" not in first_word: + return TranscriptFormat.WHISPER_JSON + return TranscriptFormat.PYCAPS_JSON + + return TranscriptFormat.PYCAPS_JSON + + +def _parse_whisper_json(data: dict[str, Any]) -> Document: + segments_data = data.get("segments") + if not isinstance(segments_data, list): + raise ValueError("Invalid whisper_json: expected 'segments' array.") + + document = Document() + for segment_data in segments_data: + if not isinstance(segment_data, dict): + continue + words = _parse_words_from_entries(segment_data.get("words")) + if not words: + segment_text = str(segment_data.get("text", "")).strip() + segment_start = _to_float(segment_data.get("start")) + segment_end = _to_float(segment_data.get("end")) + if segment_text and segment_start is not None and segment_end is not None: + words = _build_words_with_proportional_timing(segment_text, segment_start, segment_end) + _append_segment_from_words(document, words) + return document + + +def _parse_pycaps_json(data: dict[str, Any]) -> Document: + segments_data = data.get("segments") + if not isinstance(segments_data, list): + raise ValueError("Invalid pycaps_json: expected 'segments' array.") + + document = Document() + for segment_data in segments_data: + if not isinstance(segment_data, dict): + continue + + words: list[Word] = [] + if isinstance(segment_data.get("lines"), list): + for line_data in segment_data["lines"]: + if not isinstance(line_data, dict): + continue + words.extend(_parse_words_from_entries(line_data.get("words"))) + elif isinstance(segment_data.get("words"), list): + words.extend(_parse_words_from_entries(segment_data.get("words"))) + elif "text" in segment_data: + segment_text = str(segment_data.get("text", "")).strip() + segment_start, segment_end = _extract_entry_time(segment_data) + if segment_text and segment_start is not None and segment_end is not None: + words.extend(_build_words_with_proportional_timing(segment_text, segment_start, segment_end)) + + _append_segment_from_words(document, words) + return document + + +def _parse_srt(content: str) -> Document: + document = Document() + for cue in _parse_subtitle_cues(content, format=TranscriptFormat.SRT): + words = _build_words_with_proportional_timing(cue["text"], cue["start"], cue["end"]) + _append_segment_from_words(document, words) + return document + + +def _parse_vtt(content: str) -> Document: + document = Document() + for cue in _parse_subtitle_cues(content, format=TranscriptFormat.VTT): + words = _parse_vtt_inline_words(cue["text"], cue["start"], cue["end"]) + if not words: + cleaned_text = _clean_caption_text(cue["text"]) + words = _build_words_with_proportional_timing(cleaned_text, cue["start"], cue["end"]) + _append_segment_from_words(document, words) + return document + + +def _parse_subtitle_cues(content: str, format: TranscriptFormat) -> list[dict[str, Any]]: + normalized = content.replace("\r\n", "\n").replace("\r", "\n") + blocks = re.split(r"\n\s*\n", normalized.strip()) + cues: list[dict[str, Any]] = [] + + for block in blocks: + lines = [line.rstrip() for line in block.split("\n") if line.strip() != ""] + if not lines: + continue + + first_line = lines[0].strip() + if format == TranscriptFormat.VTT: + if first_line.startswith("\ufeff"): + first_line = first_line.removeprefix("\ufeff") + if first_line.startswith("WEBVTT"): + continue + if first_line.startswith(("NOTE", "STYLE", "REGION")): + continue + + timing_index = 0 if "-->" in lines[0] else 1 if len(lines) > 1 and "-->" in lines[1] else -1 + if timing_index == -1: + continue + + start, end = _parse_timing_line(lines[timing_index]) + text_lines = lines[timing_index + 1 :] + text = "\n".join(text_lines).strip() + if not text: + continue + + cues.append({"start": start, "end": end, "text": text}) + + return cues + + +def _parse_timing_line(line: str) -> tuple[float, float]: + parts = line.split("-->") + if len(parts) != 2: + raise ValueError(f"Invalid subtitle timing line: {line}") + + start_raw = parts[0].strip().split(" ")[0] + end_raw = parts[1].strip().split(" ")[0] + start = _parse_timestamp(start_raw) + end = _parse_timestamp(end_raw) + return _sanitize_time_range(start, end) + + +def _parse_vtt_inline_words(cue_text: str, cue_start: float, cue_end: float) -> list[Word]: + content = cue_text.replace("\n", " ") + matches = list(_VTT_INLINE_TIME_RE.finditer(content)) + if not matches: + return [] + + words: list[Word] = [] + prefix_text = _clean_caption_text(content[: matches[0].start()]) + first_anchor = min(cue_end, _parse_timestamp(matches[0].group(1))) + words.extend(_build_words_with_proportional_timing(prefix_text, cue_start, first_anchor)) + + for index, match in enumerate(matches): + interval_start = max(cue_start, _parse_timestamp(match.group(1))) + interval_end = cue_end + if index + 1 < len(matches): + interval_end = min(cue_end, _parse_timestamp(matches[index + 1].group(1))) + + chunk_start = match.end() + chunk_end = matches[index + 1].start() if index + 1 < len(matches) else len(content) + chunk_text = _clean_caption_text(content[chunk_start:chunk_end]) + words.extend(_build_words_with_proportional_timing(chunk_text, interval_start, interval_end)) + + return words + + +def _build_words_with_proportional_timing(text: str, start: float, end: float) -> list[Word]: + words_text = _split_words(text) + if not words_text: + return [] + + start, end = _sanitize_time_range(start, end) + if len(words_text) == 1: + return [Word(text=words_text[0], time=TimeFragment(start=start, end=end))] + + weights = [max(len(word), 1) for word in words_text] + total_weight = sum(weights) + duration = end - start + + words: list[Word] = [] + consumed_weight = 0 + for index, (word_text, weight) in enumerate(zip(words_text, weights)): + word_start = start + duration * (consumed_weight / total_weight) + consumed_weight += weight + word_end = end if index == len(words_text) - 1 else start + duration * (consumed_weight / total_weight) + word_start, word_end = _sanitize_time_range(word_start, word_end) + words.append(Word(text=word_text, time=TimeFragment(start=word_start, end=word_end))) + + return words + + +def _parse_words_from_entries(entries: Any) -> list[Word]: + if not isinstance(entries, list): + return [] + + words: list[Word] = [] + for entry in entries: + if not isinstance(entry, dict): + continue + text = str(entry.get("text", entry.get("word", ""))).strip() + if not text: + continue + + start, end = _extract_entry_time(entry) + if start is None or end is None: + continue + start, end = _sanitize_time_range(start, end) + words.append(Word(text=text, time=TimeFragment(start=start, end=end))) + + return words + + +def _extract_entry_time(entry: dict[str, Any]) -> tuple[float | None, float | None]: + if isinstance(entry.get("time"), dict): + time_data = entry["time"] + return _to_float(time_data.get("start")), _to_float(time_data.get("end")) + return _to_float(entry.get("start")), _to_float(entry.get("end")) + + +def _append_segment_from_words(document: Document, words: list[Word]) -> None: + cleaned_words = [word for word in words if word.text.strip()] + if not cleaned_words: + return + + segment_start = cleaned_words[0].time.start + segment_end = cleaned_words[-1].time.end + segment_start, segment_end = _sanitize_time_range(segment_start, segment_end) + segment_time = TimeFragment(start=segment_start, end=segment_end) + + segment = Segment(time=segment_time) + line = Line(time=segment_time) + line.words.set_all(cleaned_words) + segment.lines.add(line) + document.segments.add(segment) + + +def _normalize_document(document: Document) -> Document: + normalized_document = Document() + for segment in document.segments: + words: list[Word] = [] + for line in segment.lines: + for word in line.words: + text = str(word.text).strip() + if not text: + continue + start, end = _sanitize_time_range(word.time.start, word.time.end) + words.append(Word(text=text, time=TimeFragment(start=start, end=end))) + _append_segment_from_words(normalized_document, words) + return normalized_document + + +def _split_words(text: str) -> list[str]: + return [token for token in re.split(r"\s+", text.strip()) if token] + + +def _clean_caption_text(text: str) -> str: + return html.unescape(re.sub(r"]+>", "", text)).strip() + + +def _parse_timestamp(value: str) -> float: + cleaned = value.strip().replace(",", ".") + if not cleaned: + raise ValueError("Invalid empty timestamp.") + + parts = cleaned.split(":") + if len(parts) == 3: + hours = int(parts[0]) + minutes = int(parts[1]) + seconds = float(parts[2]) + elif len(parts) == 2: + hours = 0 + minutes = int(parts[0]) + seconds = float(parts[1]) + else: + raise ValueError(f"Invalid timestamp format: {value}") + + return hours * 3600 + minutes * 60 + seconds + + +def _sanitize_time_range(start: float, end: float) -> tuple[float, float]: + start = max(0.0, float(start)) + end = max(0.0, float(end)) + if end <= start: + end = start + _TIME_EPSILON + return start, end + + +def _to_float(value: Any) -> float | None: + if value is None: + return None + try: + return float(value) + except (TypeError, ValueError): + return None + + +def _resolve_format(format: TranscriptFormat | str) -> TranscriptFormat: + if isinstance(format, TranscriptFormat): + return format + if isinstance(format, str): + normalized = format.strip().lower() + try: + return TranscriptFormat(normalized) + except ValueError as e: + raise ValueError( + "Invalid transcript format. Expected one of: auto, whisper_json, pycaps_json, srt, vtt." + ) from e + raise ValueError("Invalid transcript format type. Expected TranscriptFormat or str.") diff --git a/tests/test_pipeline_transcription_path.py b/tests/test_pipeline_transcription_path.py new file mode 100644 index 0000000..521e945 --- /dev/null +++ b/tests/test_pipeline_transcription_path.py @@ -0,0 +1,77 @@ +import unittest +from unittest.mock import MagicMock, patch + +from pycaps.common import Document, Line, Segment, TimeFragment, Word +from pycaps.pipeline.caps_pipeline import CapsPipeline + + +def _make_document(text: str, start: float = 0.0, end: float = 1.0) -> Document: + document = Document() + segment_time = TimeFragment(start=start, end=end) + segment = Segment(time=segment_time) + line = Line(time=segment_time) + line.words.add(Word(text=text, time=TimeFragment(start=start, end=end))) + segment.lines.add(line) + document.segments.add(segment) + return document + + +class PipelineTranscriptionPathTests(unittest.TestCase): + @patch("pycaps.pipeline.caps_pipeline.check_dependencies", return_value=None) + def test_external_transcription_skips_built_in_transcribe(self, _mock_dependencies): + pipeline = CapsPipeline() + external_document = _make_document("external") + processed_document = _make_document("processed") + + pipeline.prepare = MagicMock() + pipeline.transcribe = MagicMock() + pipeline.process_document = MagicMock(return_value=processed_document) + pipeline.render = MagicMock() + pipeline._transcription_for_loading = external_document + + pipeline.run() + + pipeline.transcribe.assert_not_called() + pipeline.process_document.assert_called_once_with(external_document) + pipeline.render.assert_called_once_with(processed_document) + + @patch("pycaps.pipeline.caps_pipeline.check_dependencies", return_value=None) + def test_subtitle_data_path_still_skips_transcribe_and_processing(self, _mock_dependencies): + pipeline = CapsPipeline() + loaded_document = _make_document("loaded") + + pipeline.prepare = MagicMock() + pipeline.transcribe = MagicMock() + pipeline.process_document = MagicMock() + pipeline.render = MagicMock() + pipeline._subtitle_data_path_for_loading = "subtitle_data.json" + + with patch("pycaps.pipeline.caps_pipeline.SubtitleDataService") as subtitle_data_service: + subtitle_data_service.return_value.load.return_value = loaded_document + pipeline.run() + + subtitle_data_service.assert_called_once_with("subtitle_data.json") + pipeline.transcribe.assert_not_called() + pipeline.process_document.assert_not_called() + pipeline.render.assert_called_once_with(loaded_document) + + @patch("pycaps.pipeline.caps_pipeline.check_dependencies", return_value=None) + def test_default_flow_still_transcribes_and_processes(self, _mock_dependencies): + pipeline = CapsPipeline() + transcribed_document = _make_document("transcribed") + processed_document = _make_document("processed") + + pipeline.prepare = MagicMock() + pipeline.transcribe = MagicMock(return_value=transcribed_document) + pipeline.process_document = MagicMock(return_value=processed_document) + pipeline.render = MagicMock() + + pipeline.run() + + pipeline.transcribe.assert_called_once() + pipeline.process_document.assert_called_once_with(transcribed_document) + pipeline.render.assert_called_once_with(processed_document) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_render_cli_transcript_flags.py b/tests/test_render_cli_transcript_flags.py new file mode 100644 index 0000000..6e868d1 --- /dev/null +++ b/tests/test_render_cli_transcript_flags.py @@ -0,0 +1,119 @@ +import unittest +from unittest.mock import patch + +from typer.testing import CliRunner + +from pycaps.cli.cli import app + + +class _FakePipeline: + ran = False + + def run(self): + _FakePipeline.ran = True + + +class _FakeBuilder: + transcript_calls = [] + subtitle_data_calls = [] + build_calls = 0 + + @classmethod + def reset(cls): + cls.transcript_calls = [] + cls.subtitle_data_calls = [] + cls.build_calls = 0 + _FakePipeline.ran = False + + def with_output_video(self, *_args, **_kwargs): + return self + + def add_css_content(self, *_args, **_kwargs): + return self + + def with_whisper_config(self, *_args, **_kwargs): + return self + + def with_subtitle_data_path(self, path): + self.__class__.subtitle_data_calls.append(path) + return self + + def with_transcription_file(self, path, format): + self.__class__.transcript_calls.append((path, format)) + return self + + def should_preview_transcription(self, *_args, **_kwargs): + return self + + def with_video_quality(self, *_args, **_kwargs): + return self + + def with_layout_options(self, *_args, **_kwargs): + return self + + def build(self, *_args, **_kwargs): + self.__class__.build_calls += 1 + return _FakePipeline() + + +class _FakeTemplateLoader: + def __init__(self, _template): + pass + + def with_input_video(self, _input_video): + return self + + def load(self, _should_build_pipeline): + return _FakeBuilder() + + +class RenderCliTranscriptFlagsTests(unittest.TestCase): + def setUp(self): + self.runner = CliRunner() + _FakeBuilder.reset() + + def test_transcript_flag_is_accepted_and_forwarded(self): + with patch("pycaps.cli.render_cli.TemplateFactory") as template_factory: + template_factory.return_value.create.return_value = object() + with patch("pycaps.cli.render_cli.TemplateLoader", _FakeTemplateLoader): + result = self.runner.invoke( + app, + ["render", "--input", "video.mp4", "--template", "default", "--transcript", "transcript.json"], + ) + + self.assertEqual(result.exit_code, 0) + self.assertEqual(len(_FakeBuilder.transcript_calls), 1) + self.assertEqual(_FakeBuilder.transcript_calls[0][0], "transcript.json") + self.assertTrue(_FakePipeline.ran) + + def test_transcript_format_requires_transcript(self): + result = self.runner.invoke( + app, + ["render", "--input", "video.mp4", "--template", "default", "--transcript-format", "srt"], + ) + + self.assertEqual(result.exit_code, 0) + self.assertIn("--transcript-format requires --transcript", result.output) + + def test_transcript_and_subtitle_data_are_mutually_exclusive(self): + result = self.runner.invoke( + app, + [ + "render", + "--input", + "video.mp4", + "--template", + "default", + "--transcript", + "transcript.srt", + "--subtitle-data", + "already_processed.json", + ], + ) + + self.assertEqual(result.exit_code, 0) + self.assertIn("Only one of --subtitle-data or --transcript can be provided", result.output) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_transcript_loader.py b/tests/test_transcript_loader.py new file mode 100644 index 0000000..3992b38 --- /dev/null +++ b/tests/test_transcript_loader.py @@ -0,0 +1,133 @@ +import os +import tempfile +import unittest + +from pycaps.transcriber import TranscriptFormat, load_transcription + + +class TranscriptLoaderTests(unittest.TestCase): + def test_load_whisper_json_with_word_timestamps(self): + data = { + "segments": [ + { + "start": 0.0, + "end": 1.0, + "words": [ + {"word": "Hello", "start": 0.0, "end": 0.5}, + {"word": "world", "start": 0.5, "end": 1.0}, + ], + } + ] + } + + document = load_transcription(data, TranscriptFormat.WHISPER_JSON) + self.assertEqual(len(document.segments), 1) + self.assertEqual([word.text for word in document.get_words()], ["Hello", "world"]) + self.assertAlmostEqual(document.get_words()[0].time.start, 0.0) + self.assertAlmostEqual(document.get_words()[1].time.end, 1.0) + + def test_load_pycaps_document_json_shape(self): + data = { + "segments": [ + { + "lines": [ + { + "words": [ + {"text": "Hello", "time": {"start": 0.0, "end": 0.4}}, + {"text": "there", "time": {"start": 0.4, "end": 0.8}}, + ] + } + ] + } + ] + } + + document = load_transcription(data, TranscriptFormat.PYCAPS_JSON) + self.assertEqual([word.text for word in document.get_words()], ["Hello", "there"]) + self.assertEqual(len(document.segments[0].lines), 1) + + def test_load_pycaps_lightweight_json_shape(self): + data = { + "segments": [ + { + "words": [ + {"text": "One", "start": 1.0, "end": 1.5}, + {"word": "two", "start": 1.5, "end": 2.0}, + ] + } + ] + } + + document = load_transcription(data, TranscriptFormat.PYCAPS_JSON) + self.assertEqual([word.text for word in document.get_words()], ["One", "two"]) + self.assertAlmostEqual(document.get_words()[0].time.start, 1.0) + self.assertAlmostEqual(document.get_words()[1].time.end, 2.0) + + def test_parse_srt_single_word_preserves_cue_timing(self): + srt = "1\n00:00:01,000 --> 00:00:02,000\nHello\n" + path = self._write_temp_file(".srt", srt) + self.addCleanup(lambda: os.path.exists(path) and os.remove(path)) + + document = load_transcription(path, TranscriptFormat.AUTO) + word = document.get_words()[0] + self.assertEqual(word.text, "Hello") + self.assertAlmostEqual(word.time.start, 1.0) + self.assertAlmostEqual(word.time.end, 2.0) + + def test_parse_srt_multi_word_uses_monotonic_distribution(self): + srt = "1\n00:00:00,000 --> 00:00:02,000\nHi there friend\n" + path = self._write_temp_file(".srt", srt) + self.addCleanup(lambda: os.path.exists(path) and os.remove(path)) + + document = load_transcription(path, TranscriptFormat.SRT) + words = document.get_words() + self.assertEqual([word.text for word in words], ["Hi", "there", "friend"]) + self.assertAlmostEqual(words[0].time.start, 0.0) + self.assertAlmostEqual(words[-1].time.end, 2.0) + self.assertLess(words[0].time.end, words[1].time.end) + self.assertLess(words[1].time.end, words[2].time.end) + + def test_parse_vtt_with_inline_timestamp_tags(self): + vtt = ( + "WEBVTT\n\n" + "00:00:00.000 --> 00:00:02.000\n" + "<00:00:00.200>Hello <00:00:01.200>world\n" + ) + path = self._write_temp_file(".vtt", vtt) + self.addCleanup(lambda: os.path.exists(path) and os.remove(path)) + + document = load_transcription(path, TranscriptFormat.VTT) + words = document.get_words() + self.assertEqual([word.text for word in words], ["Hello", "world"]) + self.assertAlmostEqual(words[0].time.start, 0.2, places=3) + self.assertAlmostEqual(words[1].time.start, 1.2, places=3) + self.assertAlmostEqual(words[1].time.end, 2.0, places=3) + + def test_auto_detects_whisper_json_structure(self): + data = { + "language": "en", + "segments": [ + { + "id": 0, + "start": 0.0, + "end": 1.0, + "words": [{"word": "test", "start": 0.0, "end": 1.0}], + } + ], + } + + document = load_transcription(data, TranscriptFormat.AUTO) + self.assertEqual([word.text for word in document.get_words()], ["test"]) + + def test_invalid_format_raises_clear_error(self): + with self.assertRaisesRegex(ValueError, "Invalid transcript format"): + load_transcription({"segments": []}, "invalid") + + def _write_temp_file(self, suffix: str, content: str) -> str: + with tempfile.NamedTemporaryFile("w", suffix=suffix, delete=False) as temp: + temp.write(content) + return temp.name + + +if __name__ == "__main__": + unittest.main()