diff --git a/.gsd/milestones/M024/slices/S04/S04-PLAN.md b/.gsd/milestones/M024/slices/S04/S04-PLAN.md index f2a9b2c..8fc392c 100644 --- a/.gsd/milestones/M024/slices/S04/S04-PLAN.md +++ b/.gsd/milestones/M024/slices/S04/S04-PLAN.md @@ -29,7 +29,7 @@ Steps: - Estimate: 2h - Files: backend/pipeline/caption_generator.py, backend/pipeline/shorts_generator.py, backend/pipeline/stages.py, backend/models.py, alembic/versions/027_add_captions_enabled.py, backend/pipeline/test_caption_generator.py - Verify: cd backend && python -m pytest pipeline/test_caption_generator.py -v && python -c "from pipeline.caption_generator import generate_ass_captions; print('import ok')" -- [ ] **T02: Build card renderer and concat pipeline for intro/outro templates** — Create `card_renderer.py` that generates intro/outro card video segments using ffmpeg lavfi (color + drawtext). Add `shorts_template` JSONB column to Creator model. Implement ffmpeg concat demuxer logic to assemble intro + main clip + outro into final short. Wire into `stage_generate_shorts`. Write unit tests for card renderer. +- [x] **T02: Built ffmpeg-based card renderer with concat demuxer pipeline and wired creator-configurable intro/outro cards into shorts generation** — Create `card_renderer.py` that generates intro/outro card video segments using ffmpeg lavfi (color + drawtext). Add `shorts_template` JSONB column to Creator model. Implement ffmpeg concat demuxer logic to assemble intro + main clip + outro into final short. Wire into `stage_generate_shorts`. Write unit tests for card renderer. Steps: 1. Read T01 outputs: `backend/pipeline/caption_generator.py`, modified `shorts_generator.py` and `stages.py`. diff --git a/.gsd/milestones/M024/slices/S04/tasks/T01-VERIFY.json b/.gsd/milestones/M024/slices/S04/tasks/T01-VERIFY.json new file mode 100644 index 0000000..e5f2eb8 --- /dev/null +++ b/.gsd/milestones/M024/slices/S04/tasks/T01-VERIFY.json @@ -0,0 +1,22 @@ +{ + "schemaVersion": 1, + "taskId": "T01", + "unitId": "M024/S04/T01", + "timestamp": 1775301139872, + "passed": true, + "discoverySource": "task-plan", + "checks": [ + { + "command": "cd backend", + "exitCode": 0, + "durationMs": 8, + "verdict": "pass" + }, + { + "command": "python -m pytest pipeline/test_caption_generator.py -v", + "exitCode": 0, + "durationMs": 253, + "verdict": "pass" + } + ] +} diff --git a/.gsd/milestones/M024/slices/S04/tasks/T02-SUMMARY.md b/.gsd/milestones/M024/slices/S04/tasks/T02-SUMMARY.md new file mode 100644 index 0000000..e605072 --- /dev/null +++ b/.gsd/milestones/M024/slices/S04/tasks/T02-SUMMARY.md @@ -0,0 +1,87 @@ +--- +id: T02 +parent: S04 +milestone: M024 +provides: [] +requires: [] +affects: [] +key_files: ["backend/pipeline/card_renderer.py", "backend/pipeline/shorts_generator.py", "backend/pipeline/stages.py", "backend/models.py", "alembic/versions/028_add_shorts_template.py", "backend/pipeline/test_card_renderer.py"] +key_decisions: ["Card rendering failures are non-blocking per preset — intro/outro render errors skip that card but don't fail the short", "Cards include silent audio track (anullsrc) so concat with audio segments works without stream mismatch"] +patterns_established: [] +drill_down_paths: [] +observability_surfaces: [] +duration: "" +verification_result: "28 unit tests pass covering render_card command construction, codec settings, validation, concat list format, concat_segments demuxer args, parse_template_config defaults/overrides/coercion, and extract_clip_with_template delegation. Import verification passes. T01 caption tests still pass (17/17)." +completed_at: 2026-04-04T11:17:28.338Z +blocker_discovered: false +--- + +# T02: Built ffmpeg-based card renderer with concat demuxer pipeline and wired creator-configurable intro/outro cards into shorts generation + +> Built ffmpeg-based card renderer with concat demuxer pipeline and wired creator-configurable intro/outro cards into shorts generation + +## What Happened +--- +id: T02 +parent: S04 +milestone: M024 +key_files: + - backend/pipeline/card_renderer.py + - backend/pipeline/shorts_generator.py + - backend/pipeline/stages.py + - backend/models.py + - alembic/versions/028_add_shorts_template.py + - backend/pipeline/test_card_renderer.py +key_decisions: + - Card rendering failures are non-blocking per preset — intro/outro render errors skip that card but don't fail the short + - Cards include silent audio track (anullsrc) so concat with audio segments works without stream mismatch +duration: "" +verification_result: passed +completed_at: 2026-04-04T11:17:28.359Z +blocker_discovered: false +--- + +# T02: Built ffmpeg-based card renderer with concat demuxer pipeline and wired creator-configurable intro/outro cards into shorts generation + +**Built ffmpeg-based card renderer with concat demuxer pipeline and wired creator-configurable intro/outro cards into shorts generation** + +## What Happened + +Created card_renderer.py with render_card (lavfi command builder), render_card_to_file (ffmpeg executor), build_concat_list and concat_segments (demuxer pipeline), and parse_template_config (JSONB normalizer). Added shorts_template JSONB column to Creator model with migration 028. Added extract_clip_with_template to shorts_generator.py for intro/main/outro concatenation. Modified stage_generate_shorts to load creator template, render per-preset cards, and pass to extract_clip_with_template with non-blocking card render failures. + +## Verification + +28 unit tests pass covering render_card command construction, codec settings, validation, concat list format, concat_segments demuxer args, parse_template_config defaults/overrides/coercion, and extract_clip_with_template delegation. Import verification passes. T01 caption tests still pass (17/17). + +## Verification Evidence + +| # | Command | Exit Code | Verdict | Duration | +|---|---------|-----------|---------|----------| +| 1 | `cd backend && python -m pytest pipeline/test_card_renderer.py -v` | 0 | ✅ pass | 320ms | +| 2 | `cd backend && python -c "from pipeline.card_renderer import render_card, concat_segments; print('import ok')"` | 0 | ✅ pass | 100ms | +| 3 | `cd backend && python -m pytest pipeline/test_caption_generator.py -v` | 0 | ✅ pass | 20ms | + + +## Deviations + +None. + +## Known Issues + +None. + +## Files Created/Modified + +- `backend/pipeline/card_renderer.py` +- `backend/pipeline/shorts_generator.py` +- `backend/pipeline/stages.py` +- `backend/models.py` +- `alembic/versions/028_add_shorts_template.py` +- `backend/pipeline/test_card_renderer.py` + + +## Deviations +None. + +## Known Issues +None. diff --git a/alembic/versions/028_add_shorts_template.py b/alembic/versions/028_add_shorts_template.py new file mode 100644 index 0000000..1cf3a42 --- /dev/null +++ b/alembic/versions/028_add_shorts_template.py @@ -0,0 +1,26 @@ +"""Add shorts_template JSONB column to creators. + +Revision ID: 028_add_shorts_template +Revises: 027_add_captions_enabled +""" + +import sqlalchemy as sa +from sqlalchemy.dialects.postgresql import JSONB + +from alembic import op + +revision = "028_add_shorts_template" +down_revision = "027_add_captions_enabled" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "creators", + sa.Column("shorts_template", JSONB, nullable=True), + ) + + +def downgrade() -> None: + op.drop_column("creators", "shorts_template") diff --git a/backend/models.py b/backend/models.py index 0d9d72c..6c8191b 100644 --- a/backend/models.py +++ b/backend/models.py @@ -132,6 +132,7 @@ class Creator(Base): bio: Mapped[str | None] = mapped_column(Text, nullable=True) social_links: Mapped[dict | None] = mapped_column(JSONB, nullable=True) personality_profile: Mapped[dict | None] = mapped_column(JSONB, nullable=True) + shorts_template: Mapped[dict | None] = mapped_column(JSONB, nullable=True) featured: Mapped[bool] = mapped_column(default=False, server_default="false") view_count: Mapped[int] = mapped_column(Integer, default=0, server_default="0") hidden: Mapped[bool] = mapped_column(default=False, server_default="false") diff --git a/backend/pipeline/card_renderer.py b/backend/pipeline/card_renderer.py new file mode 100644 index 0000000..22a4b2f --- /dev/null +++ b/backend/pipeline/card_renderer.py @@ -0,0 +1,298 @@ +"""FFmpeg-based intro/outro card video generation and segment concatenation. + +Generates solid-color card clips with centered text using ffmpeg lavfi +(color + drawtext filters). Provides concat demuxer logic to assemble +intro + main clip + outro into a final short. + +Pure functions — no DB access, no Celery dependency. +""" + +from __future__ import annotations + +import logging +import subprocess +import tempfile +from pathlib import Path + +logger = logging.getLogger(__name__) + +FFMPEG_TIMEOUT_SECS = 120 + +# Default template values +DEFAULT_ACCENT_COLOR = "#22d3ee" +DEFAULT_FONT_FAMILY = "Inter" +DEFAULT_INTRO_DURATION = 2.0 +DEFAULT_OUTRO_DURATION = 2.0 + + +def render_card( + text: str, + duration_secs: float, + width: int, + height: int, + accent_color: str = DEFAULT_ACCENT_COLOR, + font_family: str = DEFAULT_FONT_FAMILY, +) -> list[str]: + """Build ffmpeg command args that generate a card mp4 from lavfi input. + + Produces a solid black background with centered white text and a thin + accent-color underline bar at the bottom third. + + Args: + text: Display text (e.g., creator name or "Thanks for watching"). + duration_secs: Card duration in seconds. + width: Output width in pixels. + height: Output height in pixels. + accent_color: Hex color for the underline glow bar. + font_family: Font family for drawtext (must be available on system). + + Returns: + List of ffmpeg command arguments (without the output path — caller appends). + """ + if duration_secs <= 0: + raise ValueError(f"duration_secs must be positive, got {duration_secs}") + if width <= 0 or height <= 0: + raise ValueError(f"dimensions must be positive, got {width}x{height}") + + # Font size scales with height — ~5% of output height + font_size = max(24, int(height * 0.05)) + # Accent bar: thin horizontal line at ~65% down + bar_y = int(height * 0.65) + bar_height = max(2, int(height * 0.004)) + bar_margin = int(width * 0.2) + + # Escape text for ffmpeg drawtext (colons, backslashes, single quotes) + escaped_text = ( + text.replace("\\", "\\\\") + .replace(":", "\\:") + .replace("'", "'\\''") + ) + + # Build complex filtergraph: + # 1. color source for black background + # 2. drawtext for centered title + # 3. drawbox for accent underline bar + filtergraph = ( + f"color=c=black:s={width}x{height}:d={duration_secs}:r=30," + f"drawtext=text='{escaped_text}'" + f":fontcolor=white:fontsize={font_size}" + f":fontfile='':font='{font_family}'" + f":x=(w-text_w)/2:y=(h-text_h)/2-{font_size}," + f"drawbox=x={bar_margin}:y={bar_y}" + f":w={width - 2 * bar_margin}:h={bar_height}" + f":color='{accent_color}'@0.8:t=fill" + ) + + cmd = [ + "ffmpeg", + "-y", + "-f", "lavfi", + "-i", filtergraph, + "-c:v", "libx264", + "-preset", "fast", + "-crf", "23", + "-pix_fmt", "yuv420p", + "-t", str(duration_secs), + # Silent audio track so concat with audio segments works + "-f", "lavfi", + "-i", f"anullsrc=r=44100:cl=stereo:d={duration_secs}", + "-c:a", "aac", + "-b:a", "128k", + "-shortest", + "-movflags", "+faststart", + ] + + return cmd + + +def render_card_to_file( + text: str, + duration_secs: float, + width: int, + height: int, + output_path: Path, + accent_color: str = DEFAULT_ACCENT_COLOR, + font_family: str = DEFAULT_FONT_FAMILY, +) -> Path: + """Generate a card mp4 file via ffmpeg. + + Args: + text: Display text for the card. + duration_secs: Card duration in seconds. + width: Output width in pixels. + height: Output height in pixels. + output_path: Destination mp4 file. + accent_color: Hex color for accent elements. + font_family: Font family for text. + + Returns: + The output_path on success. + + Raises: + subprocess.CalledProcessError: If ffmpeg exits non-zero. + subprocess.TimeoutExpired: If ffmpeg exceeds timeout. + """ + cmd = render_card( + text=text, + duration_secs=duration_secs, + width=width, + height=height, + accent_color=accent_color, + font_family=font_family, + ) + cmd.append(str(output_path)) + + logger.info( + "Rendering card: text=%r duration=%.1fs size=%dx%d → %s", + text, duration_secs, width, height, output_path, + ) + + result = subprocess.run( + cmd, + capture_output=True, + timeout=FFMPEG_TIMEOUT_SECS, + ) + + if result.returncode != 0: + stderr_text = result.stderr.decode("utf-8", errors="replace")[-2000:] + logger.error("Card render failed (rc=%d): %s", result.returncode, stderr_text) + raise subprocess.CalledProcessError( + result.returncode, cmd, output=result.stdout, stderr=result.stderr, + ) + + logger.info("Card rendered: %s (%d bytes)", output_path, output_path.stat().st_size) + return output_path + + +def build_concat_list(segments: list[Path], list_path: Path) -> Path: + """Write an ffmpeg concat demuxer list file. + + Args: + segments: Ordered list of segment mp4 paths. + list_path: Where to write the concat list. + + Returns: + The list_path. + """ + lines = [f"file '{seg.resolve()}'" for seg in segments] + list_path.write_text("\n".join(lines) + "\n", encoding="utf-8") + return list_path + + +def concat_segments(segments: list[Path], output_path: Path) -> Path: + """Concatenate mp4 segments using ffmpeg concat demuxer. + + All segments must share the same codec settings (libx264/aac, same + resolution). Uses ``-c copy`` for fast stream-copy concatenation. + + Args: + segments: Ordered list of segment mp4 paths. + output_path: Destination mp4 file. + + Returns: + The output_path on success. + + Raises: + ValueError: If segments list is empty. + subprocess.CalledProcessError: If ffmpeg exits non-zero. + subprocess.TimeoutExpired: If ffmpeg exceeds timeout. + """ + if not segments: + raise ValueError("segments list cannot be empty") + + # Write concat list to a temp file + with tempfile.NamedTemporaryFile( + mode="w", suffix=".txt", delete=False, prefix="concat_", + ) as f: + for seg in segments: + f.write(f"file '{seg.resolve()}'\n") + list_path = Path(f.name) + + try: + cmd = [ + "ffmpeg", + "-y", + "-f", "concat", + "-safe", "0", + "-i", str(list_path), + "-c", "copy", + "-movflags", "+faststart", + str(output_path), + ] + + logger.info( + "Concatenating %d segments → %s", + len(segments), output_path, + ) + + result = subprocess.run( + cmd, + capture_output=True, + timeout=FFMPEG_TIMEOUT_SECS, + ) + + if result.returncode != 0: + stderr_text = result.stderr.decode("utf-8", errors="replace")[-2000:] + logger.error( + "Concat failed (rc=%d): %s", result.returncode, stderr_text, + ) + raise subprocess.CalledProcessError( + result.returncode, cmd, output=result.stdout, stderr=result.stderr, + ) + + logger.info( + "Concatenated %d segments: %s (%d bytes)", + len(segments), output_path, output_path.stat().st_size, + ) + return output_path + + finally: + # Clean up temp list file + try: + list_path.unlink() + except OSError: + pass + + +def parse_template_config( + shorts_template: dict | None, +) -> dict: + """Parse a creator's shorts_template JSONB into normalized config. + + Expected schema:: + + { + "show_intro": true, + "intro_text": "Creator Name Presents", + "intro_duration": 2.0, + "show_outro": true, + "outro_text": "Thanks for watching!", + "outro_duration": 2.0, + "accent_color": "#22d3ee", + "font_family": "Inter" + } + + Missing fields get defaults. Returns a dict with all keys guaranteed. + """ + if not shorts_template: + return { + "show_intro": False, + "intro_text": "", + "intro_duration": DEFAULT_INTRO_DURATION, + "show_outro": False, + "outro_text": "", + "outro_duration": DEFAULT_OUTRO_DURATION, + "accent_color": DEFAULT_ACCENT_COLOR, + "font_family": DEFAULT_FONT_FAMILY, + } + + return { + "show_intro": bool(shorts_template.get("show_intro", False)), + "intro_text": str(shorts_template.get("intro_text", "")), + "intro_duration": float(shorts_template.get("intro_duration", DEFAULT_INTRO_DURATION)), + "show_outro": bool(shorts_template.get("show_outro", False)), + "outro_text": str(shorts_template.get("outro_text", "")), + "outro_duration": float(shorts_template.get("outro_duration", DEFAULT_OUTRO_DURATION)), + "accent_color": str(shorts_template.get("accent_color", DEFAULT_ACCENT_COLOR)), + "font_family": str(shorts_template.get("font_family", DEFAULT_FONT_FAMILY)), + } diff --git a/backend/pipeline/shorts_generator.py b/backend/pipeline/shorts_generator.py index 0f7c711..3fd8cf8 100644 --- a/backend/pipeline/shorts_generator.py +++ b/backend/pipeline/shorts_generator.py @@ -143,3 +143,80 @@ def extract_clip( raise subprocess.CalledProcessError( result.returncode, cmd, output=result.stdout, stderr=result.stderr, ) + + +def extract_clip_with_template( + input_path: Path | str, + output_path: Path | str, + start_secs: float, + end_secs: float, + vf_filter: str, + ass_path: Path | str | None = None, + intro_path: Path | str | None = None, + outro_path: Path | str | None = None, +) -> None: + """Extract a clip and optionally prepend/append intro/outro cards. + + If neither intro nor outro is provided, delegates directly to + :func:`extract_clip`. When cards are provided, the main clip is + extracted to a temp file, then all segments are concatenated via + :func:`~pipeline.card_renderer.concat_segments`. + + Args: + input_path: Source video file. + output_path: Final destination mp4 file. + start_secs: Start time in seconds. + end_secs: End time in seconds. + vf_filter: ffmpeg ``-vf`` filter string. + ass_path: Optional ASS subtitle file path. + intro_path: Optional intro card mp4 path. + outro_path: Optional outro card mp4 path. + + Raises: + subprocess.CalledProcessError: If any ffmpeg command fails. + ValueError: If clip range is invalid. + """ + has_cards = intro_path is not None or outro_path is not None + + if not has_cards: + # No template cards — simple extraction + extract_clip( + input_path=input_path, + output_path=output_path, + start_secs=start_secs, + end_secs=end_secs, + vf_filter=vf_filter, + ass_path=ass_path, + ) + return + + # Extract main clip to a temp file for concatenation + main_clip_path = Path(str(output_path) + ".main.mp4") + try: + extract_clip( + input_path=input_path, + output_path=main_clip_path, + start_secs=start_secs, + end_secs=end_secs, + vf_filter=vf_filter, + ass_path=ass_path, + ) + + # Build segment list in order: intro → main → outro + segments: list[Path] = [] + if intro_path is not None: + segments.append(Path(intro_path)) + segments.append(main_clip_path) + if outro_path is not None: + segments.append(Path(outro_path)) + + from pipeline.card_renderer import concat_segments + concat_segments(segments=segments, output_path=Path(output_path)) + + finally: + # Clean up temp main clip + if main_clip_path.exists(): + try: + main_clip_path.unlink() + except OSError: + pass diff --git a/backend/pipeline/stages.py b/backend/pipeline/stages.py index 03c50d2..ff40603 100644 --- a/backend/pipeline/stages.py +++ b/backend/pipeline/stages.py @@ -2875,8 +2875,9 @@ def stage_generate_shorts(self, highlight_candidate_id: str) -> str: Returns the highlight_candidate_id on completion. """ - from pipeline.shorts_generator import PRESETS, extract_clip, resolve_video_path + from pipeline.shorts_generator import PRESETS, extract_clip_with_template, resolve_video_path from pipeline.caption_generator import generate_ass_captions, write_ass_file + from pipeline.card_renderer import parse_template_config, render_card_to_file from models import FormatPreset, GeneratedShort, ShortStatus, SourceVideo start = time.monotonic() @@ -3005,6 +3006,21 @@ def stage_generate_shorts(self, highlight_candidate_id: str) -> str: highlight_candidate_id, cap_exc, ) + # ── Load creator template config (if available) ───────────────── + intro_path: Path | None = None + outro_path: Path | None = None + try: + creator = source_video.creator + template_cfg = parse_template_config( + creator.shorts_template if creator else None, + ) + except Exception as tmpl_exc: + logger.warning( + "Template config load failed for highlight=%s: %s — proceeding without cards", + highlight_candidate_id, tmpl_exc, + ) + template_cfg = parse_template_config(None) + # ── Process each preset independently ─────────────────────────── for preset in FormatPreset: spec = PRESETS[preset] @@ -3027,14 +3043,62 @@ def stage_generate_shorts(self, highlight_candidate_id: str) -> str: minio_key = f"shorts/{highlight_candidate_id}/{preset.value}.mp4" try: - # Extract clip - extract_clip( + # Render intro/outro cards for this preset's resolution + preset_intro: Path | None = None + preset_outro: Path | None = None + + if template_cfg["show_intro"] and template_cfg["intro_text"]: + preset_intro = Path( + f"/tmp/intro_{short.id}_{preset.value}.mp4" + ) + try: + render_card_to_file( + text=template_cfg["intro_text"], + duration_secs=template_cfg["intro_duration"], + width=spec.width, + height=spec.height, + output_path=preset_intro, + accent_color=template_cfg["accent_color"], + font_family=template_cfg["font_family"], + ) + except Exception as intro_exc: + logger.warning( + "Intro card render failed for highlight=%s preset=%s: %s — skipping intro", + highlight_candidate_id, preset.value, intro_exc, + ) + preset_intro = None + + if template_cfg["show_outro"] and template_cfg["outro_text"]: + preset_outro = Path( + f"/tmp/outro_{short.id}_{preset.value}.mp4" + ) + try: + render_card_to_file( + text=template_cfg["outro_text"], + duration_secs=template_cfg["outro_duration"], + width=spec.width, + height=spec.height, + output_path=preset_outro, + accent_color=template_cfg["accent_color"], + font_family=template_cfg["font_family"], + ) + except Exception as outro_exc: + logger.warning( + "Outro card render failed for highlight=%s preset=%s: %s — skipping outro", + highlight_candidate_id, preset.value, outro_exc, + ) + preset_outro = None + + # Extract clip (with optional template cards) + extract_clip_with_template( input_path=video_path, output_path=tmp_path, start_secs=clip_start, end_secs=clip_end, vf_filter=spec.vf_filter, ass_path=ass_path, + intro_path=preset_intro, + outro_path=preset_outro, ) # Upload to MinIO @@ -3081,12 +3145,13 @@ def stage_generate_shorts(self, highlight_candidate_id: str) -> str: ) finally: - # Clean up temp file - if tmp_path.exists(): - try: - tmp_path.unlink() - except OSError: - pass + # Clean up temp files (main clip + intro/outro cards) + for tmp in [tmp_path, preset_intro, preset_outro]: + if tmp is not None and tmp.exists(): + try: + tmp.unlink() + except OSError: + pass # Clean up temp ASS caption file if ass_path is not None and ass_path.exists(): diff --git a/backend/pipeline/test_card_renderer.py b/backend/pipeline/test_card_renderer.py new file mode 100644 index 0000000..bb1a9f5 --- /dev/null +++ b/backend/pipeline/test_card_renderer.py @@ -0,0 +1,365 @@ +"""Tests for card_renderer: ffmpeg card generation and concat pipeline. + +Tests verify command construction, concat list file format, and template +config parsing — no actual ffmpeg execution required. +""" + +from __future__ import annotations + +import subprocess +import textwrap +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from pipeline.card_renderer import ( + DEFAULT_ACCENT_COLOR, + DEFAULT_FONT_FAMILY, + DEFAULT_INTRO_DURATION, + DEFAULT_OUTRO_DURATION, + build_concat_list, + concat_segments, + parse_template_config, + render_card, + render_card_to_file, +) + + +# ── render_card tests ──────────────────────────────────────────────────────── + +class TestRenderCard: + """Tests for render_card() ffmpeg command generation.""" + + def test_returns_list_of_strings(self): + cmd = render_card("Hello", 2.0, 1080, 1920) + assert isinstance(cmd, list) + assert all(isinstance(s, str) for s in cmd) + + def test_contains_ffmpeg_and_lavfi(self): + cmd = render_card("Test", 3.0, 1080, 1920) + assert cmd[0] == "ffmpeg" + assert "-f" in cmd + lavfi_idx = cmd.index("-f") + assert cmd[lavfi_idx + 1] == "lavfi" + + def test_contains_correct_dimensions_in_filtergraph(self): + cmd = render_card("Test", 2.0, 1920, 1080) + # The filtergraph is the arg after -i for lavfi + filtergraph = None + for i, arg in enumerate(cmd): + if arg == "-i" and i > 0 and cmd[i - 1] == "lavfi": + filtergraph = cmd[i + 1] + break + assert filtergraph is not None + assert "s=1920x1080" in filtergraph + + def test_contains_duration_in_filtergraph(self): + cmd = render_card("Test", 5.5, 1080, 1920) + filtergraph = None + for i, arg in enumerate(cmd): + if arg == "-i" and i > 0 and cmd[i - 1] == "lavfi": + filtergraph = cmd[i + 1] + break + assert "d=5.5" in filtergraph + + def test_contains_drawtext_with_text(self): + cmd = render_card("My Creator", 2.0, 1080, 1920) + filtergraph = None + for i, arg in enumerate(cmd): + if arg == "-i" and i > 0 and cmd[i - 1] == "lavfi": + filtergraph = cmd[i + 1] + break + assert "drawtext=" in filtergraph + assert "My Creator" in filtergraph + + def test_codec_settings(self): + cmd = render_card("Test", 2.0, 1080, 1920) + assert "-c:v" in cmd + assert "libx264" in cmd + assert "-c:a" in cmd + assert "aac" in cmd + + def test_silent_audio_track(self): + """Card includes anullsrc so concat with audio segments works.""" + cmd = render_card("Test", 2.0, 1080, 1920) + # Should have a second -f lavfi -i anullsrc input + cmd_str = " ".join(cmd) + assert "anullsrc" in cmd_str + + def test_rejects_zero_duration(self): + with pytest.raises(ValueError, match="positive"): + render_card("Test", 0, 1080, 1920) + + def test_rejects_negative_duration(self): + with pytest.raises(ValueError, match="positive"): + render_card("Test", -1.0, 1080, 1920) + + def test_rejects_zero_dimensions(self): + with pytest.raises(ValueError, match="positive"): + render_card("Test", 2.0, 0, 1920) + + def test_custom_accent_color(self): + cmd = render_card("Test", 2.0, 1080, 1920, accent_color="#ff0000") + filtergraph = None + for i, arg in enumerate(cmd): + if arg == "-i" and i > 0 and cmd[i - 1] == "lavfi": + filtergraph = cmd[i + 1] + break + assert "#ff0000" in filtergraph + + def test_escapes_colons_in_text(self): + cmd = render_card("Hello: World", 2.0, 1080, 1920) + filtergraph = None + for i, arg in enumerate(cmd): + if arg == "-i" and i > 0 and cmd[i - 1] == "lavfi": + filtergraph = cmd[i + 1] + break + # Colons should be escaped for ffmpeg + assert "Hello\\: World" in filtergraph + + +# ── render_card_to_file tests ──────────────────────────────────────────────── + +class TestRenderCardToFile: + """Tests for render_card_to_file() — mocked ffmpeg execution.""" + + @patch("pipeline.card_renderer.subprocess.run") + def test_calls_ffmpeg_and_returns_path(self, mock_run, tmp_path): + mock_run.return_value = MagicMock(returncode=0) + out = tmp_path / "card.mp4" + out.write_bytes(b"fake") # stat().st_size needs the file + + result = render_card_to_file("Hi", 2.0, 1080, 1920, out) + assert result == out + mock_run.assert_called_once() + # Output path should be the last arg + call_args = mock_run.call_args[0][0] + assert call_args[-1] == str(out) + + @patch("pipeline.card_renderer.subprocess.run") + def test_raises_on_ffmpeg_failure(self, mock_run, tmp_path): + mock_run.return_value = MagicMock( + returncode=1, + stderr=b"error: something failed", + ) + out = tmp_path / "card.mp4" + with pytest.raises(subprocess.CalledProcessError): + render_card_to_file("Hi", 2.0, 1080, 1920, out) + + +# ── build_concat_list tests ───────────────────────────────────────────────── + +class TestBuildConcatList: + """Tests for build_concat_list() file content.""" + + def test_writes_correct_format(self, tmp_path): + seg1 = tmp_path / "intro.mp4" + seg2 = tmp_path / "main.mp4" + seg1.touch() + seg2.touch() + + list_file = tmp_path / "concat.txt" + result = build_concat_list([seg1, seg2], list_file) + + assert result == list_file + content = list_file.read_text() + lines = content.strip().split("\n") + assert len(lines) == 2 + assert lines[0] == f"file '{seg1.resolve()}'" + assert lines[1] == f"file '{seg2.resolve()}'" + + def test_three_segments(self, tmp_path): + segs = [tmp_path / f"seg{i}.mp4" for i in range(3)] + for s in segs: + s.touch() + + list_file = tmp_path / "list.txt" + build_concat_list(segs, list_file) + + content = list_file.read_text() + lines = content.strip().split("\n") + assert len(lines) == 3 + + +# ── concat_segments tests ──────────────────────────────────────────────────── + +class TestConcatSegments: + """Tests for concat_segments() — mocked ffmpeg execution.""" + + @patch("pipeline.card_renderer.subprocess.run") + def test_calls_ffmpeg_concat_demuxer(self, mock_run, tmp_path): + mock_run.return_value = MagicMock(returncode=0) + seg1 = tmp_path / "seg1.mp4" + seg2 = tmp_path / "seg2.mp4" + seg1.touch() + seg2.touch() + out = tmp_path / "output.mp4" + out.write_bytes(b"fakemp4") + + result = concat_segments([seg1, seg2], out) + assert result == out + mock_run.assert_called_once() + + call_args = mock_run.call_args[0][0] + assert "concat" in call_args + assert "-safe" in call_args + assert "0" in call_args + assert "-c" in call_args + assert "copy" in call_args + + def test_rejects_empty_segments(self): + with pytest.raises(ValueError, match="empty"): + concat_segments([], Path("/tmp/out.mp4")) + + @patch("pipeline.card_renderer.subprocess.run") + def test_raises_on_ffmpeg_failure(self, mock_run, tmp_path): + mock_run.return_value = MagicMock( + returncode=1, stderr=b"concat error", + ) + seg1 = tmp_path / "s.mp4" + seg1.touch() + out = tmp_path / "out.mp4" + + with pytest.raises(subprocess.CalledProcessError): + concat_segments([seg1], out) + + +# ── parse_template_config tests ────────────────────────────────────────────── + +class TestParseTemplateConfig: + """Tests for parse_template_config() defaults and overrides.""" + + def test_none_returns_all_defaults_disabled(self): + cfg = parse_template_config(None) + assert cfg["show_intro"] is False + assert cfg["show_outro"] is False + assert cfg["accent_color"] == DEFAULT_ACCENT_COLOR + assert cfg["font_family"] == DEFAULT_FONT_FAMILY + assert cfg["intro_duration"] == DEFAULT_INTRO_DURATION + assert cfg["outro_duration"] == DEFAULT_OUTRO_DURATION + + def test_empty_dict_returns_defaults_disabled(self): + cfg = parse_template_config({}) + assert cfg["show_intro"] is False + assert cfg["show_outro"] is False + + def test_full_config_preserves_values(self): + raw = { + "show_intro": True, + "intro_text": "Welcome!", + "intro_duration": 3.0, + "show_outro": True, + "outro_text": "Bye!", + "outro_duration": 1.5, + "accent_color": "#ff0000", + "font_family": "Roboto", + } + cfg = parse_template_config(raw) + assert cfg["show_intro"] is True + assert cfg["intro_text"] == "Welcome!" + assert cfg["intro_duration"] == 3.0 + assert cfg["show_outro"] is True + assert cfg["outro_text"] == "Bye!" + assert cfg["outro_duration"] == 1.5 + assert cfg["accent_color"] == "#ff0000" + assert cfg["font_family"] == "Roboto" + + def test_partial_config_fills_defaults(self): + raw = {"show_intro": True, "intro_text": "Hi"} + cfg = parse_template_config(raw) + assert cfg["show_intro"] is True + assert cfg["intro_text"] == "Hi" + assert cfg["intro_duration"] == DEFAULT_INTRO_DURATION + assert cfg["show_outro"] is False + assert cfg["outro_text"] == "" + assert cfg["accent_color"] == DEFAULT_ACCENT_COLOR + + def test_truthy_coercion(self): + """Non-bool truthy values should coerce to bool.""" + cfg = parse_template_config({"show_intro": 1, "show_outro": 0}) + assert cfg["show_intro"] is True + assert cfg["show_outro"] is False + + def test_duration_coercion_from_int(self): + cfg = parse_template_config({"intro_duration": 5}) + assert cfg["intro_duration"] == 5.0 + assert isinstance(cfg["intro_duration"], float) + + +# ── extract_clip_with_template tests ───────────────────────────────────────── + +class TestExtractClipWithTemplate: + """Tests for the shorts_generator.extract_clip_with_template function.""" + + @patch("pipeline.shorts_generator.extract_clip") + def test_no_cards_delegates_to_extract_clip(self, mock_extract): + from pipeline.shorts_generator import extract_clip_with_template + extract_clip_with_template( + input_path=Path("/fake/input.mp4"), + output_path=Path("/fake/output.mp4"), + start_secs=10.0, + end_secs=20.0, + vf_filter="scale=1080:-2", + ) + mock_extract.assert_called_once() + + @patch("pipeline.card_renderer.concat_segments") + @patch("pipeline.shorts_generator.extract_clip") + def test_with_intro_concats_two_segments(self, mock_extract, mock_concat, tmp_path): + from pipeline.shorts_generator import extract_clip_with_template + + intro = tmp_path / "intro.mp4" + intro.touch() + out = tmp_path / "final.mp4" + main_tmp = Path(str(out) + ".main.mp4") + # Create the main clip temp file so cleanup doesn't error + main_tmp.touch() + + mock_concat.return_value = out + + extract_clip_with_template( + input_path=Path("/fake/input.mp4"), + output_path=out, + start_secs=10.0, + end_secs=20.0, + vf_filter="scale=1080:-2", + intro_path=intro, + ) + mock_extract.assert_called_once() + mock_concat.assert_called_once() + # Segments should be [intro, main_clip] + segments = mock_concat.call_args[1]["segments"] + assert len(segments) == 2 + assert segments[0] == intro + + @patch("pipeline.card_renderer.concat_segments") + @patch("pipeline.shorts_generator.extract_clip") + def test_with_intro_and_outro_concats_three_segments( + self, mock_extract, mock_concat, tmp_path, + ): + from pipeline.shorts_generator import extract_clip_with_template + + intro = tmp_path / "intro.mp4" + outro = tmp_path / "outro.mp4" + intro.touch() + outro.touch() + out = tmp_path / "final.mp4" + main_tmp = Path(str(out) + ".main.mp4") + main_tmp.touch() + + mock_concat.return_value = out + + extract_clip_with_template( + input_path=Path("/fake/input.mp4"), + output_path=out, + start_secs=10.0, + end_secs=20.0, + vf_filter="scale=1080:-2", + intro_path=intro, + outro_path=outro, + ) + segments = mock_concat.call_args[1]["segments"] + assert len(segments) == 3 + assert segments[0] == intro + assert segments[2] == outro