Spaces:
Sleeping
Sleeping
File size: 3,362 Bytes
7d761b6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 | """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)
|