from __future__ import annotations import mimetypes import shutil import time from pathlib import Path from renderer.audio import AudioMixer from renderer.core.config import Settings from renderer.core.ingest import AssetIngestor from renderer.core.models import AIReelsRequest, RenderRequest, RenderResult from renderer.core.utils import cleanup_directory, temp_workdir from renderer.exports import ExportManager from renderer.ffmpeg.assets import AssetProbe from renderer.ffmpeg.command import FFmpegCommand from renderer.ffmpeg.normalize import Normalizer from renderer.ffmpeg.runner import FFmpegRunner from renderer.scenes import Timeline from renderer.subtitles import SubtitleGenerator from renderer.transcription import WhisperTranscriber from renderer.transitions import TransitionBuilder class RenderEngine: def __init__(self, settings: Settings | None = None, log=None) -> None: self.settings = settings or Settings() self.settings.ensure_dirs() self._commands: list[list[str]] = [] self._logs: list[str] = [] self.runner = FFmpegRunner(self.settings.ffmpeg_timeout_seconds, log=log, on_command=self._record_command) self.ingest = AssetIngestor(self.settings) self.assets = AssetProbe(self.settings.metadata_cache) self.normalizer = Normalizer(self.settings, self.runner) self.subtitles = SubtitleGenerator() self.transcriber = WhisperTranscriber(self.settings) self.transitions = TransitionBuilder() self.audio = AudioMixer(self.runner) self.exports = ExportManager(self.settings.exports_dir) def render(self, request: RenderRequest, job_id: str) -> RenderResult: self._commands = [] self._logs = [] started = time.time() timeline = Timeline(request.scenes) with temp_workdir(self.settings.temp_dir, job_id) as work: workdir = Path(work) resolved = self.ingest.resolve_render_request(request, workdir) timeline = Timeline(resolved.scenes) prepared = self._prepare_scene_media(resolved, workdir) subtitles = self._write_subtitles(resolved, timeline, workdir) video = self._compose_video(prepared, resolved, subtitles, workdir) mixed = self.audio.mix(video, resolved.background_music, resolved.voiceover, workdir / "mixed.mp4", normalize=resolved.audio_normalize) mixed = self._apply_watermark(mixed, resolved, workdir) output = self.exports.save(mixed, job_id, request.output_name) cleanup_directory(workdir, keep={mixed}) metrics = { "render_time_seconds": round(time.time() - started, 3), "output_size_bytes": output.stat().st_size, "scene_count": len(request.scenes), } return RenderResult(output_path=output, commands=list(self._commands), metrics=metrics, logs=list(self._logs)) def ai_reels(self, request: AIReelsRequest, job_id: str) -> RenderResult: if not request.voiceover: raise ValueError("AI reels v1 requires a provided voiceover path; TTS is pluggable but not bundled.") if not request.assets: raise ValueError("AI reels rendering requires at least one visual asset.") with temp_workdir(self.settings.temp_dir, job_id) as work: workdir = Path(work) resolved = self.ingest.resolve_ai_reels_request(request, workdir) voice_meta = self.assets.probe(resolved.voiceover) duration = max(voice_meta.duration, len(resolved.script.split()) * 0.35, 3.0) per_scene = duration / max(1, len(resolved.assets)) captions = _split_script(resolved.script, len(resolved.assets)) scenes = [ { "start": round(idx * per_scene, 3), "duration": round(per_scene, 3), "media": asset, "caption": captions[idx] if idx < len(captions) else "", } for idx, asset in enumerate(resolved.assets) ] render_request = RenderRequest( scenes=Timeline.request_from_payload({"scenes": scenes}).scenes, template=resolved.template, output_name=resolved.output_name, voiceover=resolved.voiceover, background_music=resolved.background_music, ) return self._render_resolved(render_request, job_id, workdir) def inspect_asset(self, path: str | Path) -> dict: return self.assets.probe(path).__dict__ def transcribe( self, audio_path: str | Path, *, model_size: str | None = None, language: str | None = None, task: str = "transcribe", beam_size: int = 5, vad_filter: bool = True, word_timestamps: bool = True, ) -> dict: return self.transcriber.transcribe( audio_path, model_size=model_size, language=language, task=task, beam_size=beam_size, vad_filter=vad_filter, word_timestamps=word_timestamps, ).as_dict() def _render_resolved(self, request: RenderRequest, job_id: str, workdir: Path) -> RenderResult: self._commands = [] self._logs = [] started = time.time() timeline = Timeline(request.scenes) prepared = self._prepare_scene_media(request, workdir) subtitles = self._write_subtitles(request, timeline, workdir) video = self._compose_video(prepared, request, subtitles, workdir) mixed = self.audio.mix(video, request.background_music, request.voiceover, workdir / "mixed.mp4", normalize=request.audio_normalize) mixed = self._apply_watermark(mixed, request, workdir) output = self.exports.save(mixed, job_id, request.output_name) metrics = { "render_time_seconds": round(time.time() - started, 3), "output_size_bytes": output.stat().st_size, "scene_count": len(request.scenes), } return RenderResult(output_path=output, commands=list(self._commands), metrics=metrics, logs=list(self._logs)) def _prepare_scene_media(self, request: RenderRequest, workdir: Path) -> list[Path]: prepared: list[Path] = [] for idx, scene in enumerate(request.scenes): source = Path(scene.media) metadata = self.assets.probe(source) target = workdir / f"scene_{idx:03d}.mp4" mime_type = metadata.mime_type or mimetypes.guess_type(str(source))[0] or "" if mime_type.startswith("image/") or source.suffix.lower() in {".jpg", ".jpeg", ".png", ".webp", ".gif"}: self.normalizer.image_to_video(source, target, scene.duration) elif request.normalize and self.normalizer.needs_normalization(metadata): self.normalizer.normalize(source, target, scene.duration) else: shutil.copy2(source, target) if request.preview: preview = workdir / f"scene_{idx:03d}_preview.mp4" self._scale_preview(target, preview) target = preview prepared.append(target) return prepared def _write_subtitles(self, request: RenderRequest, timeline: Timeline, workdir: Path) -> Path | None: if request.auto_subtitles and request.voiceover: transcript = self.transcriber.transcribe( request.voiceover, model_size=request.whisper_model_size, language=request.subtitle_language, word_timestamps=True, ) events = transcript.subtitle_events(prefer_words=True) else: events = self.subtitles.from_scenes(request.scenes, timeline.total_duration) if not events: return None if request.subtitle_format == "srt": return self.subtitles.write_srt(events, workdir / "captions.srt") return self.subtitles.write_ass(events, workdir / "captions.ass", request.template) def _compose_video(self, scenes: list[Path], request: RenderRequest, subtitle_path: Path | None, workdir: Path) -> Path: if len(scenes) == 1: composed = workdir / "composed.mp4" shutil.copy2(scenes[0], composed) else: composed = workdir / "composed.mp4" durations = [scene.duration for scene in request.scenes] transitions = [scene.transition for scene in request.scenes] filter_graph, final_stream = self.transitions.xfade_chain(len(scenes), durations, transitions) cmd = FFmpegCommand().add("-hide_banner") for scene in scenes: cmd.input(scene) cmd.add("-filter_complex", filter_graph) cmd.add("-map", final_stream, "-an", "-c:v", "libx264", "-preset", self.settings.preset, "-crf", self.settings.crf) command = cmd.overwrite().add(composed).build() self._run(command) if subtitle_path: subtitled = workdir / "subtitled.mp4" escaped = _ffmpeg_subtitle_path(subtitle_path) command = ( FFmpegCommand() .add("-hide_banner") .input(composed) .add("-vf", f"subtitles='{escaped}'", "-c:a", "copy", "-c:v", "libx264", "-preset", self.settings.preset, "-crf", self.settings.crf) .overwrite() .add(subtitled) .build() ) self._run(command) return subtitled return composed def _apply_watermark(self, video: Path, request: RenderRequest, workdir: Path) -> Path: if not request.watermark: return video watermark = Path(request.watermark) if not watermark.exists(): self._logs.append(f"Watermark skipped; file not found: {watermark}") return video output = workdir / "watermarked.mp4" position = { "top-left": "20:20", "top-right": "W-w-20:20", "bottom-left": "20:H-h-20", "bottom-right": "W-w-20:H-h-20", "center": "(W-w)/2:(H-h)/2", }.get(request.watermark_position, "W-w-20:H-h-20") command = ( FFmpegCommand() .add("-hide_banner") .input(video) .input(watermark) .add("-filter_complex", f"[1:v]scale=iw*0.22:-1[wm];[0:v][wm]overlay={position}") .add("-c:a", "copy", "-c:v", "libx264", "-preset", self.settings.preset, "-crf", self.settings.crf) .overwrite() .add(output) .build() ) self._run(command) return output def _scale_preview(self, source: Path, output: Path) -> None: command = ( FFmpegCommand() .add("-hide_banner") .input(source) .add("-vf", "scale=540:960:force_original_aspect_ratio=decrease,pad=540:960:(ow-iw)/2:(oh-ih)/2") .add("-c:v", "libx264", "-preset", "ultrafast", "-crf", "30", "-c:a", "aac") .overwrite() .add(output) .build() ) self._run(command) def _run(self, command: list[str]) -> None: result = self.runner.run(command) if result.stderr: self._logs.append(result.stderr[-4000:]) def _record_command(self, command: list[str]) -> None: self._commands.append(command) def _split_script(script: str, chunks: int) -> list[str]: words = script.split() if chunks <= 0: return [] size = max(1, round(len(words) / chunks)) return [" ".join(words[i : i + size]) for i in range(0, len(words), size)][:chunks] def _ffmpeg_subtitle_path(path: Path) -> str: return str(path).replace("\\", "/").replace(":", r"\:")