services/audio.py

"""Audio input handling: extension validation and optional time-window trim.

This is the only service allowed to touch the raw audio file. Once a
Transcript exists, nothing else needs the audio again.
"""

from __future__ import annotations

import re
import subprocess
import tempfile
from pathlib import Path
from typing import Optional

SUPPORTED_EXTENSIONS = {".mp3", ".wav", ".m4a", ".flac"}

# Accepts "HH:MM:SS", "MM:SS", with optional fractional seconds.
_TIME_RE = re.compile(
    r"^(?:(?P<hours>\d{1,2}):)?(?P<minutes>\d{1,2}):(?P<seconds>\d{1,2}(?:\.\d+)?)$"
)


class AudioError(ValueError):
    """Raised for invalid audio files or invalid/illogical time windows."""


def validate_extension(filename: str) -> None:
    suffix = Path(filename).suffix.lower()
    if suffix not in SUPPORTED_EXTENSIONS:
        raise AudioError(
            f"Unsupported file type '{suffix}'. "
            f"Supported: {', '.join(sorted(SUPPORTED_EXTENSIONS))}"
        )


def parse_timestamp(value: Optional[str]) -> Optional[float]:
    """Parse 'HH:MM:SS' or 'MM:SS' into seconds. Blank/None -> None."""
    if value is None:
        return None
    value = value.strip()
    if not value:
        return None

    match = _TIME_RE.match(value)
    if not match:
        raise AudioError(f"Invalid timestamp '{value}'. Expected HH:MM:SS.")

    hours = int(match.group("hours") or 0)
    minutes = int(match.group("minutes"))
    seconds = float(match.group("seconds"))
    return hours * 3600 + minutes * 60 + seconds


def resolve_window(
    start_value: Optional[str],
    end_value: Optional[str],
) -> tuple[Optional[float], Optional[float]]:
    """Parse and sanity-check a start/end window.

    Returns (start_seconds, end_seconds); either or both may be None,
    meaning "entire file" on that side.
    """
    start = parse_timestamp(start_value)
    end = parse_timestamp(end_value)

    if start is not None and end is not None and end <= start:
        raise AudioError("End time must be after start time.")

    return start, end


def extract_window(
    audio_path: str,
    start: Optional[float],
    end: Optional[float],
    workdir: Optional[Path] = None,
) -> str:
    """Trim `audio_path` to [start, end] seconds using ffmpeg.

    Returns a path to the trimmed file. If both start and end are None,
    returns the original path unchanged ("entire file").
    """
    if start is None and end is None:
        return audio_path

    workdir = workdir or Path(tempfile.mkdtemp(prefix="echoscript_audio_"))
    workdir.mkdir(parents=True, exist_ok=True)

    suffix = Path(audio_path).suffix
    out_path = workdir / f"window{suffix}"

    cmd = ["ffmpeg", "-y", "-i", str(audio_path)]
    if start is not None:
        cmd += ["-ss", str(start)]
    if end is not None:
        cmd += ["-t", str(end - (start or 0.0))]
    cmd += ["-c", "copy", str(out_path)]

    result = subprocess.run(cmd, capture_output=True, text=True)
    if result.returncode != 0:
        # Stream-copy can fail if the cut point isn't on a keyframe;
        # fall back to re-encoding.
        cmd[-2:] = ["-c:a", "pcm_s16le", str(out_path)]
        result = subprocess.run(cmd, capture_output=True, text=True)
        if result.returncode != 0:
            raise AudioError(f"ffmpeg failed to trim audio: {result.stderr.strip()}")

    return str(out_path)