xpltdco/chrysopedia
2
0
Fork 0
Code Issues 2 Pull requests Projects Releases Packages Wiki Activity Actions

feat: Built ffmpeg-based card renderer with concat demuxer pipeline and…

Browse source 
- "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"

GSD-Task: S04/T02
This commit is contained in:
jlightner 2026-04-04 11:17:38 +00:00
parent 125983588d
commit fa493e2640
9 changed files with 951 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.gsd/milestones/M024/slices/S04/S04-PLAN.md
View file

@ -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`.

22
.gsd/milestones/M024/slices/S04/tasks/T01-VERIFY.json Normal file
View file

@ -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"
}
]
}

87
.gsd/milestones/M024/slices/S04/tasks/T02-SUMMARY.md Normal file
View file

@ -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.

26
alembic/versions/028_add_shorts_template.py Normal file
View file

@ -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")

1
backend/models.py
View file

@ -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")

298
backend/pipeline/card_renderer.py Normal file
View file

@ -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)),
}

77
backend/pipeline/shorts_generator.py
View file

@ -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

83
backend/pipeline/stages.py
View file

@ -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():

365
backend/pipeline/test_card_renderer.py Normal file
View file

@ -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