"""Audio + subtitle final-preview planning and FFmpeg rendering boundary.""" from __future__ import annotations from collections.abc import Callable from dataclasses import dataclass from pathlib import Path import subprocess from autoshorts.audio.subtitles import SubtitleTrack, validate_subtitle_track from autoshorts.ideas.candidate import ValidationResult Runner = Callable[..., subprocess.CompletedProcess[str]] @dataclass(frozen=True) class AudioCaptionPreviewPlan: video_path: str audio_path: str output_path: str subtitle_file_path: str subtitle_file_content: str subtitle_track: SubtitleTrack duration_mode: str = "preserve_video" side_effects: tuple[str, ...] = () external_calls: tuple[str, ...] = () @dataclass(frozen=True) class AudioCaptionPreviewResult: output_path: str subtitle_file_path: str returncode: int stdout: str stderr: str duration_mode: str = "preserve_video" side_effects: tuple[str, ...] = ("write_subtitles", "call_ffmpeg", "write_mp4") external_calls: tuple[str, ...] = ("ffmpeg",) def _is_blank(value: str) -> bool: return not (value or "").strip() def _ass_timestamp(ms: int) -> str: centiseconds = round(ms / 10) hours, remainder = divmod(centiseconds, 360000) minutes, remainder = divmod(remainder, 6000) seconds, centis = divmod(remainder, 100) return f"{hours}:{minutes:02d}:{seconds:02d}.{centis:02d}" def _ass_color(hex_color: str) -> str: # ASS uses &HAABBGGRR. We keep alpha at 00 and convert #RRGGBB to BGR. value = hex_color.removeprefix("#") if len(value) != 6: return "&H00FFFFFF" red, green, blue = value[0:2], value[2:4], value[4:6] return f"&H00{blue}{green}{red}" def _escape_ass_text(text: str) -> str: return text.replace("{", "").replace("}", "").replace("\n", " ") def render_ass_subtitle_file(track: SubtitleTrack) -> str: validation = validate_subtitle_track(track) if not validation.is_valid: raise ValueError("; ".join(validation.errors)) style = track.style primary = _ass_color(style.primary_color) accent = _ass_color(style.accent_color) outline = _ass_color(style.stroke_color) lines = [ "[Script Info]", "ScriptType: v4.00+", "PlayResX: 1080", "PlayResY: 1920", "ScaledBorderAndShadow: yes", "", "[V4+ Styles]", "Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding", f"Style: TrueTraceCaption,Arial,{style.font_size},{primary},{accent},{outline},&H80000000,-1,0,0,0,100,100,0,0,1,5,2,2,80,80,{style.margin_v},1", "", "[Events]", "Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text", ] for cue in track.cues: cue_text = _escape_ass_text(cue.text) lines.append( "Dialogue: 0," f"{_ass_timestamp(cue.start_ms)},{_ass_timestamp(cue.end_ms)}," f"TrueTraceCaption,,0,0,0,,{cue_text}" ) return "\n".join(lines) + "\n" def build_audio_caption_preview_plan( *, video_path: str | Path, audio_path: str | Path, subtitle_track: SubtitleTrack, output_path: str | Path, duration_mode: str = "preserve_video", ) -> AudioCaptionPreviewPlan: out = Path(output_path) plan = AudioCaptionPreviewPlan( video_path=str(video_path), audio_path=str(audio_path), output_path=str(out), subtitle_file_path=str(out.with_suffix(".ass")), subtitle_file_content=render_ass_subtitle_file(subtitle_track), subtitle_track=subtitle_track, duration_mode=duration_mode, side_effects=(), external_calls=(), ) validation = validate_audio_caption_preview_plan(plan) if not validation.is_valid: raise ValueError("; ".join(validation.errors)) return plan def validate_audio_caption_preview_plan(plan: AudioCaptionPreviewPlan) -> ValidationResult: errors: list[str] = [] for field_name, value in { "video_path": plan.video_path, "audio_path": plan.audio_path, "output_path": plan.output_path, "subtitle_file_path": plan.subtitle_file_path, "subtitle_file_content": plan.subtitle_file_content, }.items(): if _is_blank(value): errors.append(f"{field_name} is required") if plan.video_path and not plan.video_path.endswith(".mp4"): errors.append("video_path must end with .mp4") if plan.audio_path and not plan.audio_path.endswith(".mp3"): errors.append("audio_path must end with .mp3") if plan.output_path and not plan.output_path.endswith(".mp4"): errors.append("output_path must end with .mp4") if plan.subtitle_file_path and not plan.subtitle_file_path.endswith(".ass"): errors.append("subtitle_file_path must end with .ass") if plan.duration_mode not in {"preserve_video", "trim_to_audio"}: errors.append("duration_mode must be preserve_video or trim_to_audio") if plan.side_effects: errors.append("audio caption preview plans must not have side effects") if plan.external_calls: errors.append("audio caption preview plans must not have external calls") errors.extend(validate_subtitle_track(plan.subtitle_track).errors) return ValidationResult(is_valid=not errors, errors=tuple(errors)) def _ass_filter_path(path: str) -> str: # Avoid shell quoting entirely by passing argv; escape colon/backslash for FFmpeg filter parser. return path.replace("\\", "\\\\").replace(":", "\\:").replace("'", "\\'") def _command(plan: AudioCaptionPreviewPlan) -> list[str]: command = [ "ffmpeg", "-y", "-i", plan.video_path, "-i", plan.audio_path, "-vf", f"ass='{_ass_filter_path(plan.subtitle_file_path)}'", "-map", "0:v:0", "-map", "1:a:0", "-c:v", "libx264", "-c:a", "aac", ] if plan.duration_mode == "trim_to_audio": command.append("-shortest") command.extend([ "-movflags", "+faststart", plan.output_path, ]) return command def render_audio_caption_preview(plan: AudioCaptionPreviewPlan, *, runner: Runner = subprocess.run) -> AudioCaptionPreviewResult: """Write ASS subtitles and call FFmpeg to mux voiceover + burn captions into an MP4.""" validation = validate_audio_caption_preview_plan(plan) if not validation.is_valid: raise ValueError("; ".join(validation.errors)) if not Path(plan.video_path).exists(): raise FileNotFoundError(plan.video_path) if not Path(plan.audio_path).exists(): raise FileNotFoundError(plan.audio_path) subtitle_path = Path(plan.subtitle_file_path) output_path = Path(plan.output_path) subtitle_path.parent.mkdir(parents=True, exist_ok=True) output_path.parent.mkdir(parents=True, exist_ok=True) subtitle_path.write_text(plan.subtitle_file_content) completed = runner(_command(plan), capture_output=True, text=True, check=False) if completed.returncode != 0: raise RuntimeError(completed.stderr or "ffmpeg audio/caption preview failed") if not output_path.exists(): raise RuntimeError("ffmpeg completed but did not create output video") return AudioCaptionPreviewResult( output_path=str(output_path), subtitle_file_path=str(subtitle_path), returncode=completed.returncode, stdout=completed.stdout, stderr=completed.stderr, duration_mode=plan.duration_mode, side_effects=("write_subtitles", "call_ffmpeg", "write_mp4"), external_calls=("ffmpeg",), )