1from __future__ import annotations
2 
3import json
4import shutil
5import subprocess
6from pathlib import Path
7 
8from video_timeline_editor.domain.model import ClipAnalysis, CropBox, FitSpec, MediaInfo, OverlaySpec
9 
10 
11DEFAULT_FONT_CANDIDATES = (
12    "/usr/share/fonts/TTF/DejaVuSans-Bold.ttf",
13    "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf",
14)
15 
16 
17def run_command(command: list[str], *, cwd: Path | None = None, check: bool = True) -> subprocess.CompletedProcess[str]:
18    result = subprocess.run(
19        command,
20        cwd=str(cwd) if cwd else None,
21        capture_output=True,
22        text=True,
23    )
24    if check and result.returncode != 0:
25        details = "\n".join(part for part in [result.stderr.strip(), result.stdout.strip()] if part)
26        raise RuntimeError(f"Command failed: {' '.join(command)}\n{details}".rstrip())
27    return result
28 
29 
30def probe_media(path: Path) -> MediaInfo:
31    result = run_command(
32        [
33            "ffprobe",
34            "-v",
35            "error",
36            "-print_format",
37            "json",
38            "-show_format",
39            "-show_streams",
40            str(path),
41        ]
42    )
43    payload = json.loads(result.stdout)
44    duration = float(payload["format"]["duration"])
45    has_audio = any(stream.get("codec_type") == "audio" for stream in payload.get("streams", []))
46    return MediaInfo(duration=duration, has_audio=has_audio)
47 
48 
49def escape_filter_value(value: str | Path) -> str:
50    text = str(value)
51    for src, dst in (
52        ("\\", "\\\\"),
53        (":", "\\:"),
54        (",", "\\,"),
55        ("'", "\\'"),
56    ):
57        text = text.replace(src, dst)
58    return text
59 
60 
61def quote_concat_path(path: Path) -> str:
62    return str(path).replace("'", "'\\''")
63 
64 
65def default_fontfile() -> str | None:
66    for candidate in DEFAULT_FONT_CANDIDATES:
67        if Path(candidate).exists():
68            return candidate
69    return None
70 
71 
72def atempo_filters(rate: float) -> list[str]:
73    if rate <= 0:
74        raise ValueError("Audio tempo must be positive")
75    parts: list[str] = []
76    remaining = rate
77    while remaining > 2.0:
78        parts.append("atempo=2.0")
79        remaining /= 2.0
80    while remaining < 0.5:
81        parts.append("atempo=0.5")
82        remaining /= 0.5
83    parts.append(f"atempo={remaining:.6f}".rstrip("0").rstrip("."))
84    return parts
85 
86 
87def crop_filters(crop: CropBox) -> list[str]:
88    return [
89        "crop="
90        f"floor(iw*{crop.width}/2)*2:"
91        f"floor(ih*{crop.height}/2)*2:"
92        f"floor(iw*{crop.x}/2)*2:"
93        f"floor(ih*{crop.y}/2)*2"
94    ]
95 
96 
97def anchor_crop(anchor: str, width: int, height: int) -> str:
98    anchor = anchor.lower()
99    center_x = f"(iw-{width})/2"
100    center_y = f"(ih-{height})/2"
101    x_map = {
102        "center": center_x,
103        "top": center_x,
104        "bottom": center_x,
105        "left": "0",
106        "right": f"iw-{width}",
107        "top_left": "0",
108        "top_right": f"iw-{width}",
109        "bottom_left": "0",
110        "bottom_right": f"iw-{width}",
111    }
112    y_map = {
113        "center": center_y,
114        "top": "0",
115        "bottom": f"ih-{height}",
116        "left": center_y,
117        "right": center_y,
118        "top_left": "0",
119        "top_right": "0",
120        "bottom_left": f"ih-{height}",
121        "bottom_right": f"ih-{height}",
122    }
123    if anchor not in x_map:
124        raise ValueError(f"Unsupported cover anchor: {anchor}")
125    return f"crop={width}:{height}:{x_map[anchor]}:{y_map[anchor]}"
126 
127 
128def fit_filters(fit: FitSpec, width: int, height: int, project_background: str) -> list[str]:
129    background = fit.background or project_background
130    if fit.mode == "contain":
131        return [
132            f"scale={width}:{height}:force_original_aspect_ratio=decrease",
133            f"pad={width}:{height}:(ow-iw)/2:(oh-ih)/2:{background}",
134        ]
135    if fit.mode == "cover":
136        return [
137            f"scale={width}:{height}:force_original_aspect_ratio=increase",
138            anchor_crop(fit.anchor, width, height),
139        ]
140    raise ValueError(f"Unsupported fit mode: {fit.mode}")
141 
142 
143def overlay_filters(overlays: tuple[OverlaySpec, ...], styles: dict[str, dict], work_dir: Path) -> list[str]:
144    filters: list[str] = []
145    text_dir = work_dir / "_overlay_text"
146    text_dir.mkdir(parents=True, exist_ok=True)
147    fontfile = default_fontfile()
148    allowed = (
149        "fontfile",
150        "fontsize",
151        "fontcolor",
152        "borderw",
153        "bordercolor",
154        "box",
155        "boxcolor",
156        "boxborderw",
157        "x",
158        "y",
159        "alpha",
160    )
161    for index, overlay in enumerate(overlays):
162        style = dict(styles.get(overlay.style, styles.get("default", {})))
163        style.update(overlay.options)
164        if fontfile and "fontfile" not in style:
165            style["fontfile"] = fontfile
166 
167        text_path = text_dir / f"overlay-{index:03d}.txt"
168        text_path.write_text(overlay.text)
169        parts = [f"drawtext=textfile={escape_filter_value(text_path)}"]
170        for key in allowed:
171            if key not in style:
172                continue
173            value = style[key]
174            if isinstance(value, bool):
175                value = int(value)
176            parts.append(f"{key}={escape_filter_value(value)}")
177        if overlay.start is not None or overlay.end is not None:
178            start = 0.0 if overlay.start is None else float(overlay.start)
179            end = overlay.end
180            if end is None:
181                parts.append(f"enable={escape_filter_value(f'gte(t,{start})')}")
182            else:
183                parts.append(f"enable={escape_filter_value(f'between(t,{start},{end})')}")
184        filters.append(":".join(parts))
185    return filters
186 
187 
188def render_clip_file(
189    analysis: ClipAnalysis,
190    *,
191    width: int,
192    height: int,
193    fps: int,
194    background: str,
195    overlay_styles: dict[str, dict],
196    work_dir: Path,
197    caption_ass_path: Path | None = None,
198) -> Path:
199    clip = analysis.clip
200    output_path = work_dir / f"{clip.id}.mp4"
201    video_filters: list[str] = []
202    audio_filters: list[str] = []
203 
204    if clip.crop:
205        video_filters.extend(crop_filters(clip.crop))
206 
207    transform = clip.transform
208    if transform.hflip:
209        video_filters.append("hflip")
210    if transform.vflip:
211        video_filters.append("vflip")
212    if transform.zoom != 1.0:
213        video_filters.append(f"scale=iw*{transform.zoom}:ih*{transform.zoom}")
214        video_filters.append(f"crop=iw/{transform.zoom}:ih/{transform.zoom}")
215    if transform.video_pts != 1.0:
216        video_filters.append(f"setpts={transform.video_pts}*PTS")
217    if transform.audio_tempo != 1.0:
218        audio_filters.extend(atempo_filters(transform.audio_tempo))
219    if transform.eq:
220        eq_parts = [f"{key}={value}" for key, value in transform.eq.items()]
221        if eq_parts:
222            video_filters.append("eq=" + ":".join(eq_parts))
223 
224    if clip.speed != 1.0:
225        video_filters.append(f"setpts={1 / clip.speed:.6f}*PTS")
226        audio_filters.extend(atempo_filters(clip.speed))
227 
228    video_filters.extend(fit_filters(clip.fit, width, height, background))
229    video_filters.extend(overlay_filters(clip.overlays, overlay_styles, work_dir))
230    if caption_ass_path:
231        video_filters.append(f"ass={escape_filter_value(caption_ass_path)}")
232 
233    use_silent_audio = clip.muted or not analysis.media.has_audio
234    command = [
235        "ffmpeg",
236        "-y",
237        "-hide_banner",
238        "-loglevel",
239        "error",
240        "-avoid_negative_ts",
241        "make_zero",
242        "-ss",
243        str(analysis.timing.actual_in),
244        "-to",
245        str(analysis.timing.actual_out),
246        "-i",
247        str(analysis.source_path),
248    ]
249    if use_silent_audio:
250        command.extend(["-f", "lavfi", "-i", "anullsrc=r=44100:cl=stereo"])
251    if video_filters:
252        command.extend(["-vf", ",".join(video_filters)])
253    if not use_silent_audio:
254        if clip.volume != 1.0:
255            audio_filters.append(f"volume={clip.volume}")
256        if audio_filters:
257            command.extend(["-af", ",".join(audio_filters)])
258    command.extend(
259        [
260            "-map",
261            "0:v:0",
262            "-map",
263            "1:a:0" if use_silent_audio else "0:a:0",
264            "-c:v",
265            "libx264",
266            "-pix_fmt",
267            "yuv420p",
268            "-r",
269            str(fps),
270            "-c:a",
271            "aac",
272            "-b:a",
273            "192k",
274            "-ac",
275            "2",
276            "-ar",
277            "44100",
278            "-movflags",
279            "+faststart",
280        ]
281    )
282    if use_silent_audio:
283        command.append("-shortest")
284    command.append(str(output_path))
285    run_command(command)
286    return output_path
287 
288 
289def concat_clips(rendered_paths: list[Path], output_path: Path, work_dir: Path) -> None:
290    concat_path = work_dir / "concat.txt"
291    concat_path.write_text("".join(f"file '{quote_concat_path(path)}'\n" for path in rendered_paths))
292    run_command(
293        [
294            "ffmpeg",
295            "-y",
296            "-hide_banner",
297            "-loglevel",
298            "error",
299            "-f",
300            "concat",
301            "-safe",
302            "0",
303            "-i",
304            str(concat_path),
305            "-c",
306            "copy",
307            str(output_path),
308        ]
309    )
310 
311 
312def maybe_open(path: Path) -> None:
313    opener = shutil.which("xdg-open") or shutil.which("open")
314    if opener:
315        subprocess.Popen([opener, str(path)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
316