"""Lossless A/V shift export. Ports the ffmpeg logic already proven out in the `ffdelay` bash script (see PROJECT_BRIEF.md) -- reuse it rather than reinvent it. """ import subprocess from dataclasses import dataclass from typing import Optional class FfmpegExportError(RuntimeError): """ffmpeg exited non-zero. Carries stderr so the caller can surface the real reason (e.g. a codec/container that can't stream-copy) instead of silently falling back to a re-encode -- that fallback is an explicit non-goal for v1. """ def __init__(self, message: str, stderr: str): super().__init__(message) self.stderr = stderr @dataclass(frozen=True) class ExportResult: command: list stderr: str def export_synced( input_path: str, delay_ms: float, output_path: str, clip_start: Optional[str] = None, clip_duration: Optional[float] = None, ) -> ExportResult: """Stream-copy `input_path` to `output_path` with the audio and video tracks re-aligned by `delay_ms` (see sync_logic.py for the sign convention). `clip_start`/`clip_duration` extract a short segment instead of processing the whole file, for fast iteration -- mirrors ffdelay's optional clip mode. ffmpeg's `-itsoffset` doesn't accept a negative value the way this needs, so a negative delay_ms is applied by putting the offset on the *other* input (video) instead of trying to negate it -- these are genuinely two different command shapes, not one parametrized by sign. """ abs_delay_sec = abs(delay_ms) / 1000.0 # Python's float formatting always includes the leading zero (unlike # `bc`, which emits ".500" for sub-1 values and made ffmpeg reject the # duration outright) -- no special-casing needed here. offset_str = f"{abs_delay_sec:.3f}" clip_args = [] if clip_start is not None and clip_duration is not None: # Applied to BOTH inputs so the extracted clip covers the same # real-world time window in each, regardless of which one also # carries -itsoffset. clip_args = ["-ss", clip_start, "-t", str(clip_duration)] if delay_ms < 0: video_input_args = ["-itsoffset", offset_str, *clip_args, "-i", input_path] audio_input_args = [*clip_args, "-i", input_path] else: video_input_args = [*clip_args, "-i", input_path] audio_input_args = ["-itsoffset", offset_str, *clip_args, "-i", input_path] command = [ "ffmpeg", "-y", *video_input_args, *audio_input_args, "-map", "0:v", "-map", "1:a", "-c:v", "copy", "-c:a", "copy", output_path, ] proc = subprocess.run(command, capture_output=True, text=True) if proc.returncode != 0: raise FfmpegExportError( f"ffmpeg exited with code {proc.returncode}", proc.stderr ) # stderr is returned even on success: ffmpeg logs things like # "sps_id out of range" / "Invalid NAL unit size" / "missing picture in # access unit" on stream-copy operations from screen recordings/live # captures. These are harmless source-file quirks, not failures -- the # caller should log them, not treat them as errors. return ExportResult(command=command, stderr=proc.stderr)