Files
Nat20-Notes/backend/app/pipeline/audio.py
KansaiGaijin 6c760b5e2e
Some checks failed
Build and Push / build (push) Has been cancelled
Preload audio as Blob for instant seeking; generate Opus playback file; delete original upload after extraction
- Download full audio as Blob on mount so seeking is instant (no network range-requests)
- Remove readyState guard from playAt() that was silently blocking seek attempts
- Add disabled/ready state to SessionReel for reliable click-to-seek
- Generate 48kbps Opus playback file during extraction (~65MB for 3hrs vs 1.5GB lossless)
- Serve Opus first in audio endpoint, then original upload, then extracted WAV
- Delete original upload after successful transcription to reclaim storage
- Reorder extract_audio checks so retry works without the original file
- Fix Back links on Speakers and Notes pages to go to home
2026-07-30 10:38:10 +12:00

63 lines
2.4 KiB
Python

import subprocess
from pathlib import Path
from ..errors import AudioExtractionError
from ..logging_config import get_logger
log = get_logger(__name__)
def extract_audio(video_path: Path, audio_path: Path) -> Path:
"""Extract 16kHz mono wav from any video/audio container."""
if audio_path.exists():
log.info("Audio already extracted at %s, skipping", audio_path)
return audio_path
if not video_path.exists():
raise AudioExtractionError(f"Uploaded file not found on disk at {video_path}. It may not have finished uploading, or the upload volume isn't mounted correctly.")
log.info("Extracting audio: %s -> %s", video_path, audio_path)
result = subprocess.run(
[
"ffmpeg", "-y", "-i", str(video_path),
"-vn", "-acodec", "pcm_s16le", "-ar", "16000", "-ac", "1",
str(audio_path),
],
capture_output=True, text=True,
)
if result.returncode != 0:
# surface ffmpeg's own error line rather than the whole verbose log
stderr_tail = "\n".join(result.stderr.strip().splitlines()[-5:])
raise AudioExtractionError(
f"ffmpeg failed to extract audio (exit code {result.returncode}). "
f"This usually means the file is corrupt or not a supported format. Last output:\n{stderr_tail}"
)
return audio_path
def generate_playback_audio(video_path: Path, playback_path: Path) -> Path | None:
"""Generate a compressed Opus file for browser playback (~48 kbps).
This is done alongside the WAV extraction so the frontend can download a
small file for instant-seeking Blob playback. Existing sessions that lack
the Opus file will fall back to the original upload or the extracted WAV.
"""
if playback_path.exists():
log.info("Playback audio already exists at %s, skipping", playback_path)
return playback_path
log.info("Generating playback audio: %s -> %s", video_path, playback_path)
result = subprocess.run(
[
"ffmpeg", "-y", "-i", str(video_path),
"-vn", "-c:a", "libopus", "-b:a", "48k",
str(playback_path),
],
capture_output=True, text=True,
)
if result.returncode != 0:
stderr_tail = "\n".join(result.stderr.strip().splitlines()[-5:])
log.warning("Failed to generate playback audio (non-fatal): %s", stderr_tail)
return None
return playback_path