From 02b730b0821f565dedacc506cfa88c29df419c05 Mon Sep 17 00:00:00 2001 From: Maxim Vershinin Date: Wed, 19 Aug 2026 11:12:48 +0500 Subject: [PATCH] CHANGE: maps codec/container/file_extension more accurate --- audio_splitter/core.py | 2 +- audio_splitter/ffmpeg.py | 178 +++++++++++++++++++++++--------------- audio_splitter/formats.py | 93 ++++++++++++++++++-- 3 files changed, 192 insertions(+), 81 deletions(-) diff --git a/audio_splitter/core.py b/audio_splitter/core.py index 9839ed3..de5b51e 100644 --- a/audio_splitter/core.py +++ b/audio_splitter/core.py @@ -44,7 +44,7 @@ def split_audio(input_file, output_directory, tracks, args): # -------------------------------------------------------------------------- # 2. Format decision # -------------------------------------------------------------------------- - output_format = determine_output_format(stream_info, args.format, args.transcode_to) + output_format = determine_output_format(stream_info, args.format, args.transcode_to, input_file=input_file) print(f"Output container: {output_format}") validate_format_compatibility(output_format, stream_info, diff --git a/audio_splitter/ffmpeg.py b/audio_splitter/ffmpeg.py index 653decc..e6570ca 100644 --- a/audio_splitter/ffmpeg.py +++ b/audio_splitter/ffmpeg.py @@ -1,7 +1,8 @@ -"""FFmpeg / FFprobe interactions and command building.""" +"""FFmpeg/FFprobe interaction utilities for the CLI and web backend.""" -import subprocess import json +import subprocess +from typing import Dict, List, Optional, Tuple from .constants import FORMAT_INFO from .utils import format_time @@ -24,7 +25,10 @@ def get_audio_duration(input_file: str) -> float: input_file ] result = subprocess.run(cmd, capture_output=True, text=True, check=False) - return float(result.stdout.strip()) + try: + return float(result.stdout.strip()) + except ValueError: + return 0.0 def has_stream_type(input_file: str, stream_type: str) -> bool: @@ -49,7 +53,7 @@ def has_stream_type(input_file: str, stream_type: str) -> bool: return bool(result.stdout.strip()) -def get_audio_codec(input_file: str) -> str: +def get_audio_codec(input_file: str) -> Optional[str]: """ Return the codec name of the first audio stream. @@ -71,7 +75,7 @@ def get_audio_codec(input_file: str) -> str: return codec if codec else None -def get_stream_info(input_file: str): +def get_stream_info(input_file: str) -> Dict[str, any]: """ Collect information about the streams present in the input file. @@ -88,9 +92,97 @@ def get_stream_info(input_file: str): 'audio_codec': get_audio_codec(input_file) } -def build_ffmpeg_command(input_file, start_seconds, duration_seconds, output_path, - stream_info, format_opt, transcode_audio, - drop_video, drop_subs, metadata=None): + +def get_metadata(input_file: str) -> Dict[str, any]: + """ + Retrieve metadata from the input file using ffprobe with JSON output. + + Returns a dict with keys: album, title, comments (list of (stream_index, comment)). + """ + cmd = [ + 'ffprobe', '-v', 'quiet', + '-print_format', 'json', + '-show_entries', 'format_tags:stream_tags', + input_file + ] + result = subprocess.run(cmd, capture_output=True, text=True, check=False) + + if result.returncode != 0: + return {'album': None, 'title': None, 'comments': []} + + try: + data = json.loads(result.stdout) + album = None + title = None + comments = [] + + fmt_tags = data.get('format', {}).get('tags', {}) + album = fmt_tags.get('album') or album + title = fmt_tags.get('title') or title + + for idx, stream in enumerate(data.get('streams', [])): + stream_tags = stream.get('tags', {}) + if 'album' in stream_tags: + album = stream_tags['album'] + if 'title' in stream_tags: + title = stream_tags['title'] + if 'comment' in stream_tags: + comments.append((idx, stream_tags['comment'])) + + return {'album': album, 'title': title, 'comments': comments} + except (json.JSONDecodeError, KeyError): + return {'album': None, 'title': None, 'comments': []} + + +def get_container_format(input_file: str) -> Optional[str]: + """ + Retrieve the container format name (e.g., 'mp4', 'mp3', 'matroska') from the input file. + + Args: + input_file: Path to the media file. + + Returns: + Container format name (normalized) or None if detection fails. + """ + cmd = [ + 'ffprobe', '-v', 'error', + '-show_entries', 'format=format_name', + '-of', 'default=noprint_wrappers=1:nokey=1', + input_file + ] + result = subprocess.run(cmd, capture_output=True, text=True, check=False) + if result.returncode != 0: + return None + format_name = result.stdout.strip().split(',')[0] # take first if multiple + if not format_name: + return None + # Normalize common aliases to names used in FORMAT_INFO (container only, not codec-specific) + # This mapping is purely for container identification. + mapping = { + 'mpeg': 'mp3', # MPEG-1/2 audio (MP3) container + 'mp2': 'mp3', + 'mp4': 'mp4', + 'm4a': 'mp4', # M4A is MP4 container + 'mov': 'mp4', # QuickTime is MP4-like + '3gp': 'mp4', + 'matroska': 'matroska', + 'webm': 'matroska', # WebM uses Matroska container + 'ogg': 'ogg', + 'flac': 'flac', + 'wav': 'wav', + 'aac': 'aac', + 'opus': 'opus', + 'mp3': 'mp3', + 'adts': 'aac', # raw AAC in ADTS container + 'amr': 'amr', # AMR container (rare) + } + return mapping.get(format_name, format_name) + + +def build_ffmpeg_command(input_file: str, start_seconds: int, duration_seconds: int, + output_path: str, stream_info: Dict, format_opt: Optional[str], + transcode_audio: Optional[str], drop_video: bool, drop_subs: bool, + metadata: Optional[Dict] = None) -> List[str]: """ Construct the FFmpeg command line as a list of arguments. @@ -116,18 +208,17 @@ def build_ffmpeg_command(input_file, start_seconds, duration_seconds, output_pat '-t', format_time(duration_seconds) ] - # -------------------- Clear all original metadata -------------------- + # Clear all original metadata. cmd.append('-map_metadata') cmd.append('-1') - # -------------------- Apply custom metadata -------------------- + # Apply custom metadata. if metadata: for key, value in metadata.items(): if value is not None and value != '': cmd.extend(['-metadata', f"{key}={value}"]) - # -------------------- Stream mapping -------------------- - # Map the streams we want to keep. + # Stream mapping. if drop_video and drop_subs: cmd.extend(['-map', '0:a:0']) elif drop_video: @@ -137,7 +228,7 @@ def build_ffmpeg_command(input_file, start_seconds, duration_seconds, output_pat else: cmd.extend(['-map', '0']) - # -------------------- Audio codec -------------------- + # Audio codec. if transcode_audio: cmd.extend(['-c:a', transcode_audio]) if transcode_audio in ('libmp3lame', 'mp3'): @@ -147,77 +238,22 @@ def build_ffmpeg_command(input_file, start_seconds, duration_seconds, output_pat else: cmd.extend(['-c:a', 'copy']) - # -------------------- Video codec -------------------- + # Video codec. if not drop_video and stream_info['has_video']: cmd.extend(['-c:v', 'copy']) else: cmd.append('-vn') - # -------------------- Subtitle codec -------------------- + # Subtitle codec. if not drop_subs and stream_info['has_subtitle']: cmd.extend(['-c:s', 'copy']) else: cmd.append('-sn') - # -------------------- Output format -------------------- + # Output format. if format_opt: ffmpeg_format = FORMAT_INFO.get(format_opt, {}).get('ffmpeg', format_opt) cmd.extend(['-f', ffmpeg_format]) - # Overwrite output if it already exists. cmd.extend(['-y', output_path]) - return cmd - -def get_metadata(input_file: str) -> dict: - """ - Retrieve metadata from the input file using ffprobe with JSON output. - Returns a dict with: - - album: merged from all sources (last wins) - - title: merged from all sources (last wins) - - comments: list of (stream_index, comment) tuples - """ - cmd = [ - 'ffprobe', '-v', 'quiet', - '-print_format', 'json', - '-show_entries', 'format_tags:stream_tags', - input_file - ] - result = subprocess.run(cmd, capture_output=True, text=True, check=False) - - if result.returncode != 0: - return {'album': None, 'title': None, 'comments': []} - - try: - data = json.loads(result.stdout) - # Collect album and title (merged, last wins). - album = None - title = None - comments = [] # list of (stream_index, comment) - - # Format tags. - fmt_tags = data.get('format', {}).get('tags', {}) - album = fmt_tags.get('album') or album - title = fmt_tags.get('title') or title - # Format does not have a stream index; we'll treat it as -1 if needed. - - # Stream tags. - for idx, stream in enumerate(data.get('streams', [])): - stream_tags = stream.get('tags', {}) - # Album and title: update if present. - if 'album' in stream_tags: - album = stream_tags['album'] - if 'title' in stream_tags: - title = stream_tags['title'] - # Comment: collect all occurrences. - if 'comment' in stream_tags: - comments.append((idx, stream_tags['comment'])) - - return { - 'album': album, - 'title': title, - 'comments': comments - } - except (json.JSONDecodeError, KeyError): - return {'album': None, 'title': None, 'comments': []} - diff --git a/audio_splitter/formats.py b/audio_splitter/formats.py index 26c0d3d..6aaae08 100644 --- a/audio_splitter/formats.py +++ b/audio_splitter/formats.py @@ -1,16 +1,79 @@ """Container format decision and validation.""" +from typing import Dict, Optional + from .constants import FORMAT_INFO -def determine_output_format(stream_info, user_format, transcode_audio): +def determine_default_format(container: Optional[str], codec: Optional[str]) -> Optional[str]: + """ + Given the container format and audio codec, determine the recommended output format. + + This is used when the user has not explicitly specified a format. + It prioritizes the codec to choose the most appropriate container/extension. + + Args: + container: Container name (e.g., 'ogg', 'mp4', 'matroska') as returned by get_container_format(). + codec: Audio codec name (e.g., 'opus', 'aac', 'mp3') as returned by get_audio_codec(). + + Returns: + Format name (e.g., 'opus', 'm4a', 'mp3') or None if unknown. + """ + if not container: + return None + + # Codec-based decisions (highest priority) + if codec == 'opus': + return 'opus' + if codec in ('aac', 'alac', 'he-aac'): + return 'm4a' + if codec == 'mp3': + return 'mp3' + if codec == 'vorbis': + return 'ogg' + if codec == 'flac': + return 'flac' + + # Container-based fallback (lower priority) + if container in ('mp4', 'm4a', 'mov', '3gp'): + return 'mp4' + if container in ('matroska', 'webm'): + return 'matroska' + if container in ('ogg',): + return 'ogg' + if container in ('mp3', 'mpeg'): + return 'mp3' + if container == 'flac': + return 'flac' + if container == 'wav': + return 'wav' + if container == 'aac': + return 'aac' + if container == 'opus': + return 'opus' + if container == 'amr': + return 'amr' + + return None + + +def determine_output_format(stream_info: Dict, user_format: Optional[str], + transcode_audio: Optional[str], input_file: Optional[str] = None) -> str: """ Decide which container format to use. + If user_format is provided, use it. + Else, try to detect the input file's container and codec, and use the recommended format. + If detection fails or format is not supported, fallback to: + - MKV if video/subtitles exist + - MP3 if the audio codec is MP3 + - MP4 (M4A) otherwise + Args: stream_info: Dict from get_stream_info(). user_format: User‑requested format (or None). - transcode_audio: Audio codec to transcode to (or None). + transcode_audio: Audio codec to transcode to (or None) (unused in this function). + input_file: Path to the input file (optional, used to detect container and codec). Returns: A format name that exists in FORMAT_INFO. @@ -18,11 +81,22 @@ def determine_output_format(stream_info, user_format, transcode_audio): if user_format: return user_format - # If video or subtitles exist, use MKV (which supports everything). - if stream_info['has_video'] or stream_info['has_subtitle']: - return 'matroska' + # If input_file is provided, try to detect container and codec + if input_file: + try: + from .ffmpeg import get_container_format, get_audio_codec + container = get_container_format(input_file) + codec = get_audio_codec(input_file) + fmt = determine_default_format(container, codec) + if fmt in FORMAT_INFO: + return fmt + except Exception: + # If detection fails, fall through to legacy logic + pass - # Audio‑only: choose based on the current audio codec. + # Fallback: legacy behavior + if stream_info.get('has_video') or stream_info.get('has_subtitle'): + return 'matroska' audio_codec = stream_info.get('audio_codec', '') if audio_codec == 'mp3': return 'mp3' @@ -30,7 +104,8 @@ def determine_output_format(stream_info, user_format, transcode_audio): return 'mp4' # .m4a -def validate_format_compatibility(format_name, stream_info, drop_video, drop_subs): +def validate_format_compatibility(format_name: str, stream_info: Dict, + drop_video: bool, drop_subs: bool) -> None: """ Ensure the chosen container can accommodate the streams we intend to keep. @@ -43,12 +118,12 @@ def validate_format_compatibility(format_name, stream_info, drop_video, drop_sub return if info['audio_only']: - if stream_info['has_video'] and not drop_video: + if stream_info.get('has_video') and not drop_video: raise ValueError( f"Format '{format_name}' does not support video streams. " "Please use --drop-video or choose a container that supports video." ) - if stream_info['has_subtitle'] and not drop_subs: + if stream_info.get('has_subtitle') and not drop_subs: raise ValueError( f"Format '{format_name}' does not support subtitle streams. " "Please use --drop-subs or choose a container that supports subtitles."