import subprocess import os import logging from ..config import settings logger = logging.getLogger(__name__) class FFmpegService: def split_audio(self, input_path: str, meeting_id: int) -> list: """ Splits audio into 30min chunks using ffmpeg segment muxer. Returns a list of paths to the created chunks. """ output_dir = os.path.join(settings.UPLOAD_DIR, "chunks", str(meeting_id)) os.makedirs(output_dir, exist_ok=True) output_pattern = os.path.join(output_dir, "chunk_%03d.mp3") # ffmpeg -i input.mp3 -f segment -segment_time 1800 -c copy chunk_%03d.mp3 cmd = [ "ffmpeg", "-i", input_path, "-f", "segment", "-segment_time", str(settings.CHUNK_DURATION_SEC), "-c:a", "libmp3lame", # Re-encode to MP3 for compatibility "-b:a", "256k", # Set a reasonable bitrate for quality output_pattern ] logger.info(f"Splitting {input_path} into segments...") result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: logger.error(f"FFmpeg Error: {result.stderr}") raise Exception("Failed to split audio file.") chunks = sorted([os.path.join(output_dir, f) for f in os.listdir(output_dir) if f.endswith(".mp3")]) logger.info(f"Created {len(chunks)} chunks.") return chunks def get_duration(self, input_path: str) -> float: """Gets duration of audio file in seconds.""" cmd = [ "ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", input_path ] result = subprocess.run(cmd, capture_output=True, text=True) try: return float(result.stdout.strip()) except: return 0.0