CHANGE: maps codec/container/file_extension more accurate

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,

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,
+def get_metadata(input_file: str) -> Dict[str, any]:
-                         drop_video, drop_subs, metadata=None):
+    """
+    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 --------------------
+    # Stream mapping.
-    # Map the streams we want to keep.
     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': []}
-

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 input_file is provided, try to detect container and codec
-    if stream_info['has_video'] or stream_info['has_subtitle']:
+    if input_file:
-        return 'matroska'
+        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."