From 690c5f98c6aec4374da2cfce1bed35bcebaa4136 Mon Sep 17 00:00:00 2001 From: Maxim Vershinin Date: Tue, 25 Aug 2026 10:40:48 +0500 Subject: [PATCH] FEATURE: adds strict codec/container/file_extension verification, renames ambiguous options, adds flexible transcoding --- audio_splitter/constants.py | 148 +++++++++++---------- audio_splitter/core.py | 146 +++++++++++++++++---- audio_splitter/ffmpeg.py | 254 ++++++++++++++++++++++++------------ audio_splitter/formats.py | 141 ++++++++++---------- audio_splitter/main.py | 124 ++++++++++-------- 5 files changed, 506 insertions(+), 307 deletions(-) diff --git a/audio_splitter/constants.py b/audio_splitter/constants.py index b49765f..941c788 100644 --- a/audio_splitter/constants.py +++ b/audio_splitter/constants.py @@ -4,7 +4,9 @@ This file serves as the single source of truth for: - Container formats and their properties - Audio codecs and their recommended containers +- Video codec support per container - File extensions for each codec/container combination +- Compatibility matrix for codec/container validation Codec != Container != File Extension. Example: Opus (codec) → Ogg (container) → .opus (extension) @@ -13,31 +15,22 @@ Example: Opus (codec) → Ogg (container) → .opus (extension) # ------------------------------------------------------------------------------ # Container information # ------------------------------------------------------------------------------ -# Each container entry: -# - name: internal identifier used in the code -# - ffmpeg: name passed to FFmpeg's -f option -# - extension: default file extension -# - audio_only: whether the container supports video/subtitle streams -# - supports_video: whether video streams can be stored -# - supports_subs: whether subtitle streams can be stored CONTAINER_INFO = [ # Audio-only containers {'name': 'mp3', 'ffmpeg': 'mp3', 'extension': '.mp3', 'audio_only': True, 'supports_video': False, 'supports_subs': False}, - {'name': 'm4a', 'ffmpeg': 'mp4', 'extension': '.m4a', 'audio_only': True, 'supports_video': False, 'supports_subs': False}, # MP4 container, .m4a extension for audio-only + {'name': 'm4a', 'ffmpeg': 'mp4', 'extension': '.m4a', 'audio_only': True, 'supports_video': False, 'supports_subs': False}, {'name': 'flac', 'ffmpeg': 'flac', 'extension': '.flac', 'audio_only': True, 'supports_video': False, 'supports_subs': False}, {'name': 'wav', 'ffmpeg': 'wav', 'extension': '.wav', 'audio_only': True, 'supports_video': False, 'supports_subs': False}, {'name': 'aac', 'ffmpeg': 'adts', 'extension': '.aac', 'audio_only': True, 'supports_video': False, 'supports_subs': False}, - # Containers that support video and subtitles {'name': 'mp4', 'ffmpeg': 'mp4', 'extension': '.mp4', 'audio_only': False, 'supports_video': True, 'supports_subs': True}, {'name': 'mkv', 'ffmpeg': 'matroska', 'extension': '.mkv', 'audio_only': False, 'supports_video': True, 'supports_subs': True}, {'name': 'matroska', 'ffmpeg': 'matroska', 'extension': '.mkv', 'audio_only': False, 'supports_video': True, 'supports_subs': True}, - {'name': 'ogg', 'ffmpeg': 'ogg', 'extension': '.ogg', 'audio_only': False, 'supports_video': True, 'supports_subs': True}, # Ogg supports video (Theora, Dirac) and subtitles - {'name': 'webm', 'ffmpeg': 'webm', 'extension': '.webm', 'audio_only': False, 'supports_video': True, 'supports_subs': False}, # WebM is a subset of Matroska + {'name': 'ogg', 'ffmpeg': 'ogg', 'extension': '.ogg', 'audio_only': False, 'supports_video': True, 'supports_subs': True}, + {'name': 'webm', 'ffmpeg': 'webm', 'extension': '.webm', 'audio_only': False, 'supports_video': True, 'supports_subs': False}, ] -# Legacy FORMAT_INFO for backward compatibility with existing code -# Maps container name → FFmpeg format name, extension, and audio_only flag +# Legacy FORMAT_INFO for backward compatibility FORMAT_INFO = { container['name']: { 'ffmpeg': container['ffmpeg'], @@ -50,13 +43,6 @@ FORMAT_INFO = { # ------------------------------------------------------------------------------ # Codec information # ------------------------------------------------------------------------------ -# Each codec entry: -# - name: codec name (used in code) -# - ffmpeg: encoder name passed to FFmpeg's -c:a option -# - recommended_container: the container format recommended for this codec -# - recommended_extension: the recommended file extension for this codec -# - supports_transcoding: whether this codec can be used as output via FFmpeg -# - supports_video: whether this codec is for video (True) or audio (False) CODEC_INFO = [ # Audio codecs {'name': 'opus', 'ffmpeg': 'libopus', 'recommended_container': 'ogg', 'recommended_extension': '.opus', 'supports_transcoding': True, 'supports_video': False}, @@ -66,59 +52,81 @@ CODEC_INFO = [ {'name': 'mp3', 'ffmpeg': 'libmp3lame', 'recommended_container': 'mp3', 'recommended_extension': '.mp3', 'supports_transcoding': True, 'supports_video': False}, {'name': 'alac', 'ffmpeg': 'alac', 'recommended_container': 'mp4', 'recommended_extension': '.m4a', 'supports_transcoding': True, 'supports_video': False}, {'name': 'pcm_s16le', 'ffmpeg': 'pcm_s16le', 'recommended_container': 'wav', 'recommended_extension': '.wav', 'supports_transcoding': True, 'supports_video': False}, - - # Video codecs (for reference, not used for transcoding selection in the UI) - {'name': 'theora', 'ffmpeg': 'libtheora', 'recommended_container': 'ogg', 'recommended_extension': '.ogv', 'supports_transcoding': False, 'supports_video': True}, - {'name': 'vp8', 'ffmpeg': 'libvpx', 'recommended_container': 'webm', 'recommended_extension': '.webm', 'supports_transcoding': False, 'supports_video': True}, - {'name': 'vp9', 'ffmpeg': 'libvpx-vp9', 'recommended_container': 'webm', 'recommended_extension': '.webm', 'supports_transcoding': False, 'supports_video': True}, - {'name': 'av1', 'ffmpeg': 'libaom-av1', 'recommended_container': 'mkv', 'recommended_extension': '.mkv', 'supports_transcoding': False, 'supports_video': True}, - {'name': 'h264', 'ffmpeg': 'libx264', 'recommended_container': 'mp4', 'recommended_extension': '.mp4', 'supports_transcoding': False, 'supports_video': True}, - {'name': 'h265', 'ffmpeg': 'libx265', 'recommended_container': 'mp4', 'recommended_extension': '.mp4', 'supports_transcoding': False, 'supports_video': True}, + # Video codecs + {'name': 'theora', 'ffmpeg': 'libtheora', 'recommended_container': 'ogg', 'recommended_extension': '.ogv', 'supports_transcoding': True, 'supports_video': True}, + {'name': 'vp8', 'ffmpeg': 'libvpx', 'recommended_container': 'webm', 'recommended_extension': '.webm', 'supports_transcoding': True, 'supports_video': True}, + {'name': 'vp9', 'ffmpeg': 'libvpx-vp9', 'recommended_container': 'webm', 'recommended_extension': '.webm', 'supports_transcoding': True, 'supports_video': True}, + {'name': 'av1', 'ffmpeg': 'libaom-av1', 'recommended_container': 'mkv', 'recommended_extension': '.mkv', 'supports_transcoding': True, 'supports_video': True}, + {'name': 'h264', 'ffmpeg': 'libx264', 'recommended_container': 'mp4', 'recommended_extension': '.mp4', 'supports_transcoding': True, 'supports_video': True}, + {'name': 'h265', 'ffmpeg': 'libx265', 'recommended_container': 'mp4', 'recommended_extension': '.mp4', 'supports_transcoding': True, 'supports_video': True}, + {'name': 'png', 'ffmpeg': 'png', 'recommended_container': 'ogg', 'recommended_extension': '.png', 'supports_transcoding': False, 'supports_video': True}, + {'name': 'mjpeg', 'ffmpeg': 'mjpeg', 'recommended_container': 'ogg', 'recommended_extension': '.jpg', 'supports_transcoding': False, 'supports_video': True}, ] -# ------------------------------------------------------------------------------ -# Video codec support per container -# ------------------------------------------------------------------------------ -# For each container, list of video codecs it supports. -# Use '*' to indicate that the container supports all video codecs (e.g., Matroska). -CONTAINER_VIDEO_CODEC_SUPPORT = { - 'mp4': ['h264', 'h265', 'vp9', 'av1', 'mpeg4', 'hevc'], # MP4 supports these via ISO BMFF - 'mkv': ['*'], # Matroska supports virtually all video codecs - 'matroska': ['*'], - 'ogg': ['theora', 'dirac', 'vp8'], # Ogg supports Theora, Dirac, VP8 - 'webm': ['vp8', 'vp9', 'av1'], # WebM is a subset of Matroska with VP8/VP9/AV1 - 'mp3': [], # Audio-only, no video support - 'm4a': [], # Audio-only, no video support - 'flac': [], # Audio-only - 'wav': [], # Audio-only - 'aac': [], # Audio-only -} - -# Map codec name to its type (audio/video) -# This is used to determine if a stream is audio or video -CODEC_TYPE_MAP = { - 'opus': 'audio', - 'vorbis': 'audio', - 'flac': 'audio', - 'aac': 'audio', - 'mp3': 'audio', - 'alac': 'audio', - 'pcm_s16le': 'audio', - 'theora': 'video', - 'vp8': 'video', - 'vp9': 'video', - 'av1': 'video', - 'h264': 'video', - 'h265': 'video', - 'png': 'video', # PNG is often used as attached picture (cover art) - 'mjpeg': 'video', # MJPEG also used for attached pictures -} - -# Map codec name → recommended container name +# Map codec name → recommended container CODEC_TO_CONTAINER_MAP = {codec['name']: codec['recommended_container'] for codec in CODEC_INFO} - -# Map codec name → recommended file extension CODEC_TO_EXTENSION_MAP = {codec['name']: codec['recommended_extension'] for codec in CODEC_INFO} -# Default bad characters (unchanged) +# ------------------------------------------------------------------------------ +# Video codec support per container (legacy, will be superseded by compatibility matrix) +# ------------------------------------------------------------------------------ +CONTAINER_VIDEO_CODEC_SUPPORT = { + 'mp4': ['h264', 'h265', 'vp9', 'av1', 'mpeg4', 'hevc'], + 'mkv': ['*'], + 'matroska': ['*'], + 'ogg': ['theora', 'dirac', 'vp8', 'png', 'mjpeg'], + 'webm': ['vp8', 'vp9', 'av1'], + 'mp3': [], + 'm4a': [], + 'flac': [], + 'wav': [], + 'aac': [], +} + +# ------------------------------------------------------------------------------ +# Compatibility matrix: container → supported audio and video codecs +# ------------------------------------------------------------------------------ +# Each container entry maps to a dict with 'audio' and 'video' keys. +# - 'audio': list of audio codec names that are supported in this container. +# Use None to indicate that any audio codec is supported. +# - 'video': list of video codec names that are supported in this container. +# Use None to indicate that any video codec is supported. +# Empty list means no video support (audio-only container). +COMPATIBILITY_MATRIX = { + 'mp3': {'audio': ['mp3'], 'video': []}, + 'm4a': {'audio': ['aac', 'alac', 'opus', 'flac'], 'video': []}, + 'mp4': {'audio': ['aac', 'alac', 'opus', 'flac', 'mp3'], 'video': ['h264', 'h265', 'vp9', 'av1']}, + 'mkv': {'audio': None, 'video': None}, + 'matroska': {'audio': None, 'video': None}, + 'ogg': {'audio': ['opus', 'vorbis', 'flac'], 'video': ['theora', 'vp8']}, + 'webm': {'audio': ['opus', 'vorbis'], 'video': ['vp8', 'vp9', 'av1']}, + 'flac': {'audio': ['flac'], 'video': []}, + 'wav': {'audio': ['pcm_s16le'], 'video': []}, + 'aac': {'audio': ['aac'], 'video': []}, +} + +# ------------------------------------------------------------------------------ +# Helper functions for compatibility checking +# ------------------------------------------------------------------------------ +def is_audio_codec_supported(container: str, audio_codec: str) -> bool: + """Check if an audio codec is supported in the given container.""" + entry = COMPATIBILITY_MATRIX.get(container, {}) + supported = entry.get('audio') + if supported is None: + return True + return audio_codec in supported + + +def is_video_codec_supported(container: str, video_codec: str) -> bool: + """Check if a video codec is supported in the given container.""" + entry = COMPATIBILITY_MATRIX.get(container, {}) + supported = entry.get('video') + if supported is None: + return True + return video_codec in supported + + +# ------------------------------------------------------------------------------ +# Default bad characters (for filename sanitization) +# ------------------------------------------------------------------------------ DEFAULT_BAD_CHARS = r'!@#№$;:%^&?*(){}[]\/<>+=~`\' ' diff --git a/audio_splitter/core.py b/audio_splitter/core.py index c3b4b94..08a4a12 100644 --- a/audio_splitter/core.py +++ b/audio_splitter/core.py @@ -1,30 +1,63 @@ -"""Core logic: orchestrates the splitting process.""" +# audio_splitter/core.py +"""Core splitting logic – orchestrates the entire split process.""" import os import subprocess +import sys +from typing import List, Dict, Any from .constants import FORMAT_INFO -from .ffmpeg import get_audio_duration, get_stream_info, get_metadata, build_ffmpeg_command -from .formats import determine_output_format, validate_format_compatibility +from .ffmpeg import ( + get_audio_duration, + get_stream_info, + get_metadata, + build_ffmpeg_command, + is_attached_picture, + extract_cover_image, +) +from .formats import ( + determine_output_format, + validate_format_compatibility, +) from .timestamp import parse_track_timestamps, resolve_end_times from .filename import build_filename from .metadata import build_metadata_dict from .utils import format_time -def split_audio(input_file, output_directory, tracks, args): +def split_audio(input_file: str, output_directory: str, tracks: List[Dict[str, Any]], args) -> None: """ Main orchestration function: split the audio file into tracks. Args: input_file: Path to the input media file. output_directory: Directory where output files will be saved. - tracks: List of dicts, each containing parsed fields. - args: Parsed command‑line arguments (namespace). + tracks: List of dicts, each containing parsed fields (ts, tn, an, ...). + args: Parsed command‑line arguments (namespace) with attributes: + - container: output container name + - audio_codec: audio codec (copy or encoder) + - video_codec: video codec (copy or encoder) + - subtitle_codec: subtitle codec (copy or encoder) + - drop_video: bool + - drop_subs: bool + - number_tracks: bool + - output_template: str + - replace_bad_chars: bool + - replacement_char: str + - bad_chars: str + - skip_existing: bool + - album: str or None + - comment: str or None + - no_comment: bool + - comment_stream: int or None + - merge_comments: bool + - comment_separator: str + - delete_original: bool + - cover_image: str or None (optional, set by web backend) Raises: RuntimeError: If no audio stream is found. - ValueError: If timestamp parsing or format compatibility fails. + ValueError: If compatibility validation fails. """ # -------------------------------------------------------------------------- # 1. Setup @@ -42,26 +75,42 @@ def split_audio(input_file, output_directory, tracks, args): raise RuntimeError("No audio stream found in input file.") # -------------------------------------------------------------------------- - # 2. Format decision + # 2. Determine output container # -------------------------------------------------------------------------- - 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, - args.drop_video, args.drop_subs, input_file) - - # Determine file extension. - extension_info = FORMAT_INFO.get(output_format, {}) - extension = extension_info.get('ext', '.mkv') + # Use args.container if provided, otherwise auto-detect + user_container = getattr(args, 'container', None) + output_container = determine_output_format( + stream_info, + user_format=user_container, + transcode_audio=args.audio_codec if args.audio_codec != 'copy' else None, + input_file=input_file + ) + print(f"Output container: {output_container}") # -------------------------------------------------------------------------- - # 3. Parse timestamps + # 3. Validate compatibility + # -------------------------------------------------------------------------- + try: + validate_format_compatibility( + container=output_container, + stream_info=stream_info, + drop_video=args.drop_video, + drop_subs=args.drop_subs, + input_file=input_file, + audio_codec=args.audio_codec, + video_codec=args.video_codec, + ) + except ValueError as e: + raise RuntimeError(f"Compatibility error: {e}") + + # -------------------------------------------------------------------------- + # 4. Parse timestamps # -------------------------------------------------------------------------- track_times = parse_track_timestamps(tracks) resolved_times = resolve_end_times(track_times, total_duration) # -------------------------------------------------------------------------- - # 4. Fetch original metadata (for fallbacks) + # 5. Fetch original metadata (for fallbacks) # -------------------------------------------------------------------------- input_metadata = get_metadata(input_file) original_album = input_metadata.get('album') @@ -78,7 +127,29 @@ def split_audio(input_file, output_directory, tracks, args): print(f" Stream {idx}: '{comment}'") # -------------------------------------------------------------------------- - # 5. Process each track + # 6. Handle attached picture (cover art) + # -------------------------------------------------------------------------- + cover_image_path = getattr(args, 'cover_image', None) + if cover_image_path and not os.path.exists(cover_image_path): + cover_image_path = None + + # If not provided via args (CLI case), try to detect and extract + if not cover_image_path and not args.drop_video and stream_info.get('has_video'): + if is_attached_picture(input_file): + cover_image_path = os.path.join(output_directory, 'cover.png') + if extract_cover_image(input_file, cover_image_path): + print("Extracted cover image for all tracks.") + else: + cover_image_path = None + + # -------------------------------------------------------------------------- + # 7. Determine file extension + # -------------------------------------------------------------------------- + extension_info = FORMAT_INFO.get(output_container, {}) + extension = extension_info.get('ext', '.mkv') + + # -------------------------------------------------------------------------- + # 8. Process each track # -------------------------------------------------------------------------- for idx, track in enumerate(tracks, start=1): start_seconds, end_seconds = resolved_times[idx - 1] @@ -91,33 +162,41 @@ def split_audio(input_file, output_directory, tracks, args): track_name = track.get('tn', 'Unknown') # ---------------------------------------------------------------------- - # 5a. Build filename + # 8a. Build filename # ---------------------------------------------------------------------- clean_filename = build_filename(track, idx, extension, args) output_path = os.path.join(output_directory, clean_filename) # ---------------------------------------------------------------------- - # 5b. Handle existing files + # 8b. Handle existing files # ---------------------------------------------------------------------- if args.skip_existing and os.path.exists(output_path): print(f"Skipping track {idx}: {output_path} already exists.") continue # ---------------------------------------------------------------------- - # 5c. Build metadata + # 8c. Build metadata # ---------------------------------------------------------------------- metadata = build_metadata_dict(track, idx, input_metadata, args) # ---------------------------------------------------------------------- - # 5d. Build and execute FFmpeg command + # 8d. Build and execute FFmpeg command # ---------------------------------------------------------------------- cmd = build_ffmpeg_command( - input_file, start_seconds, duration_seconds, output_path, - stream_info, output_format, args.transcode_to, - args.drop_video, args.drop_subs, - metadata=metadata + input_file=input_file, + start_seconds=start_seconds, + duration_seconds=duration_seconds, + output_path=output_path, + stream_info=stream_info, + format_opt=output_container, + audio_codec=args.audio_codec, + video_codec=args.video_codec, + subtitle_codec=args.subtitle_codec, + metadata=metadata, + cover_image_path=cover_image_path, ) + print(cmd) print(f"Extracting track {idx}: {track_name} " f"({format_time(start_seconds)} - {format_time(end_seconds)})") @@ -126,5 +205,16 @@ def split_audio(input_file, output_directory, tracks, args): if result.returncode != 0: print(f"ERROR extracting track {idx}:") print(result.stderr) + # Optionally stop on first error? We'll continue. else: print(f" -> Saved to: {output_path}") + + # -------------------------------------------------------------------------- + # 9. Delete original file if requested + # -------------------------------------------------------------------------- + if getattr(args, 'delete_original', False): + try: + os.remove(input_file) + print(f"Deleted original file: {input_file}") + except OSError as e: + print(f"Warning: Could not delete original file: {e}") diff --git a/audio_splitter/ffmpeg.py b/audio_splitter/ffmpeg.py index fab1d67..76b6463 100644 --- a/audio_splitter/ffmpeg.py +++ b/audio_splitter/ffmpeg.py @@ -1,4 +1,5 @@ -"""FFmpeg/FFprobe interaction utilities for the CLI and web backend.""" +# audio_splitter/ffmpeg.py +"""FFmpeg/FFprobe interaction utilities.""" import json import subprocess @@ -75,6 +76,28 @@ def get_audio_codec(input_file: str) -> Optional[str]: return codec if codec else None +def get_video_codec(input_file: str) -> Optional[str]: + """ + Return the codec name of the first video stream. + + Args: + input_file: Path to the media file. + + Returns: + Codec name as a lowercase string, or None if no video stream exists. + """ + cmd = [ + 'ffprobe', '-v', 'error', + '-select_streams', 'v:0', + '-show_entries', 'stream=codec_name', + '-of', 'default=noprint_wrappers=1:nokey=1', + input_file + ] + result = subprocess.run(cmd, capture_output=True, text=True, check=False) + codec = result.stdout.strip().lower() + return codec if codec else None + + def get_stream_info(input_file: str) -> Dict[str, any]: """ Collect information about the streams present in the input file. @@ -156,125 +179,194 @@ def get_container_format(input_file: str) -> Optional[str]: 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. + # Normalize common aliases to names used in FORMAT_INFO mapping = { - 'mpeg': 'mp3', # MPEG-1/2 audio (MP3) container + 'mpeg': 'mp3', 'mp2': 'mp3', 'mp4': 'mp4', - 'm4a': 'mp4', # M4A is MP4 container - 'mov': 'mp4', # QuickTime is MP4-like + 'm4a': 'mp4', + 'mov': 'mp4', '3gp': 'mp4', 'matroska': 'matroska', - 'webm': 'matroska', # WebM uses Matroska container + 'webm': 'matroska', 'ogg': 'ogg', 'flac': 'flac', 'wav': 'wav', 'aac': 'aac', 'opus': 'opus', 'mp3': 'mp3', - 'adts': 'aac', # raw AAC in ADTS container - 'amr': 'amr', # AMR container (rare) + 'adts': 'aac', + 'amr': 'amr', } 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]: +def is_attached_picture(input_file: str) -> bool: """ - Construct the FFmpeg command line as a list of arguments. + Check if the input file has a video stream that is an attached picture (cover art). - Args: - input_file: Path to the input media file. - start_seconds: Start time for the segment (in seconds). - duration_seconds: Duration of the segment (in seconds). - output_path: Destination path for the output file. - stream_info: Dictionary from get_stream_info(). - format_opt: Output container format (e.g., 'mp3'). - transcode_audio: Audio codec to transcode to (or None). - drop_video: True to remove video streams. - drop_subs: True to remove subtitle streams. - metadata: Optional dict of metadata key/value pairs to write. + Detection logic: + 1. If there is a video stream with disposition.attached_pic == 1, return True. + 2. Otherwise, if there is exactly one video stream and its codec is an image + format (PNG, MJPEG, JPEG, GIF, BMP), return True. Returns: - A list of command‑line arguments suitable for subprocess.run(). + True if a cover image is detected, False otherwise. """ cmd = [ - 'ffmpeg', - '-i', input_file, - '-ss', format_time(start_seconds), - '-t', format_time(duration_seconds) + 'ffprobe', '-v', 'quiet', + '-print_format', 'json', + '-select_streams', 'v', + '-show_entries', 'stream=codec_name,disposition', + input_file ] + result = subprocess.run(cmd, capture_output=True, text=True, check=False) + if result.returncode != 0: + return False - # Clear all original metadata. + try: + data = json.loads(result.stdout) + streams = data.get('streams', []) + if not streams: + return False + + # Check each stream + for stream in streams: + codec = stream.get('codec_name', '').lower() + disposition = stream.get('disposition', {}) + # If attached_pic is set, it's a cover image + if disposition.get('attached_pic') == 1: + return True + # If not, check if it's an image codec and we have exactly one video stream + if codec in ('png', 'mjpeg', 'jpeg', 'gif', 'bmp'): + # If there is exactly one video stream, treat it as cover + if len(streams) == 1: + return True + return False + except (json.JSONDecodeError, KeyError): + return False + + +def extract_cover_image(input_file: str, output_path: str) -> bool: + """ + Extract the first frame of the video stream (assumed to be an attached picture) + and save it to output_path. + + Returns: + True if extraction succeeded, False otherwise. + """ + cmd = [ + 'ffmpeg', '-i', input_file, + '-map', '0:v:0', + '-frames:v', '1', + '-y', + output_path + ] + result = subprocess.run(cmd, capture_output=True, text=True, check=False) + if result.returncode != 0: + print(f"Failed to extract cover image: {result.stderr}") + return False + return True + + +def build_ffmpeg_command( + input_file: str, + start_seconds: int, + duration_seconds: int, + output_path: str, + stream_info: Dict, + format_opt: Optional[str], + audio_codec: Optional[str] = 'copy', + video_codec: Optional[str] = 'copy', + subtitle_codec: Optional[str] = 'copy', + metadata: Optional[Dict] = None, + cover_image_path: Optional[str] = None, +) -> List[str]: + """ + Construct the FFmpeg command line. + + Args: + audio_codec: 'copy' or encoder name (e.g., 'libopus') + video_codec: 'copy' or encoder name (e.g., 'libx264') + subtitle_codec: 'copy' or encoder name (e.g., 'srt') + """ + cmd = ['ffmpeg'] + + # Add cover image if provided + if cover_image_path: + cmd.extend(['-i', cover_image_path]) + cmd.extend(['-i', input_file]) + + # Time options + cmd.extend(['-ss', format_time(start_seconds), '-t', format_time(duration_seconds)]) + + # Clear 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. - if drop_video and drop_subs: - cmd.extend(['-map', '0:a:0']) - elif drop_video: - cmd.extend(['-map', '0:a:0', '-map', '0:s?']) - elif drop_subs: - cmd.extend(['-map', '0:a:0', '-map', '0:v:0']) - else: - cmd.extend(['-map', '0']) + # ---------- Stream mapping and codecs ---------- + if cover_image_path: + # We have two inputs: index 0 = cover image, index 1 = main input + # Map audio from main input (index 1) and video from cover image (index 0) + cmd.extend(['-map', '1:a:0', '-map', '0:v:0']) - # Audio codec. - if transcode_audio: - cmd.extend(['-c:a', transcode_audio]) - if transcode_audio in ('libmp3lame', 'mp3'): - cmd.extend(['-b:a', '192k']) - elif transcode_audio in ('libopus', 'opus'): - cmd.extend(['-b:a', '128k']) - else: - cmd.extend(['-c:a', 'copy']) + # Video codec: use user-specified codec if provided, otherwise fallback to png + if video_codec and video_codec != 'copy': + cmd.extend(['-c:v', video_codec]) + else: + cmd.extend(['-c:v', 'png']) - # Video codec. - if not drop_video and stream_info['has_video']: - cmd.extend(['-c:v', 'copy']) - else: - cmd.append('-vn') + # Audio codec + if audio_codec and audio_codec != 'copy': + cmd.extend(['-c:a', audio_codec]) + else: + cmd.extend(['-c:a', 'copy']) - # Subtitle codec. - if not drop_subs and stream_info['has_subtitle']: - cmd.extend(['-c:s', 'copy']) - else: + # Subtitle: none (we don't copy from original when using cover image) cmd.append('-sn') - # Output format. + else: + # Standard mapping: copy all streams by default, then filter based on options + if not stream_info.get('has_video') or drop_video: + cmd.extend(['-map', '0:a:0']) + else: + # Map all streams + cmd.extend(['-map', '0']) + + # Audio codec + if audio_codec and audio_codec != 'copy': + cmd.extend(['-c:a', audio_codec]) + else: + cmd.extend(['-c:a', 'copy']) + + # Video codec (if video present and not dropped) + if stream_info.get('has_video') and not drop_video: + if video_codec and video_codec != 'copy': + cmd.extend(['-c:v', video_codec]) + else: + cmd.extend(['-c:v', 'copy']) + else: + cmd.append('-vn') + + # Subtitle codec + if stream_info.get('has_subtitle') and not drop_subs: + if subtitle_codec and subtitle_codec != 'copy': + cmd.extend(['-c:s', subtitle_codec]) + else: + cmd.extend(['-c:s', 'copy']) + else: + cmd.append('-sn') + + # Output format if format_opt: ffmpeg_format = FORMAT_INFO.get(format_opt, {}).get('ffmpeg', format_opt) cmd.extend(['-f', ffmpeg_format]) cmd.extend(['-y', output_path]) return cmd - -def get_video_codec(input_file: str) -> Optional[str]: - """ - Return the codec name of the first video stream. - - Args: - input_file: Path to the media file. - - Returns: - Codec name as a lowercase string, or None if no video stream exists. - """ - cmd = [ - 'ffprobe', '-v', 'error', - '-select_streams', 'v:0', - '-show_entries', 'stream=codec_name', - '-of', 'default=noprint_wrappers=1:nokey=1', - input_file - ] - result = subprocess.run(cmd, capture_output=True, text=True, check=False) - codec = result.stdout.strip().lower() - return codec if codec else None diff --git a/audio_splitter/formats.py b/audio_splitter/formats.py index 87ff45e..3af08d9 100644 --- a/audio_splitter/formats.py +++ b/audio_splitter/formats.py @@ -3,13 +3,22 @@ from typing import Dict, Optional -from .constants import FORMAT_INFO, CODEC_TO_CONTAINER_MAP, CONTAINER_INFO, CONTAINER_VIDEO_CODEC_SUPPORT, CODEC_TYPE_MAP +from .constants import ( + FORMAT_INFO, + CONTAINER_INFO, + CODEC_TO_CONTAINER_MAP, + COMPATIBILITY_MATRIX, + is_audio_codec_supported, + is_video_codec_supported, +) +from .defaults import DEFAULT_FORMAT + 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 uses the CODEC_TO_CONTAINER_MAP to map codec → container. + Uses the CODEC_TO_CONTAINER_MAP to map codec → container. If the codec is not found, it falls back to the container. Args: @@ -22,69 +31,17 @@ def determine_default_format(container: Optional[str], codec: Optional[str]) -> if not container: return None - # Codec-based decision (highest priority) if codec and codec in CODEC_TO_CONTAINER_MAP: - return CODEC_TO_CONTAINER_MAP[codec] + fmt = CODEC_TO_CONTAINER_MAP[codec] + if fmt in FORMAT_INFO: + return fmt - # Container-based fallback (lowest priority) - # Ensure the container is in FORMAT_INFO if container in FORMAT_INFO: return container return None -def validate_format_compatibility(format_name: str, stream_info: Dict, - drop_video: bool, drop_subs: bool, input_file: Optional[str] = None) -> None: - """ - Ensure the chosen container can accommodate the streams we intend to keep. - """ - info = FORMAT_INFO.get(format_name) - if not info: - print(f"Warning: Unknown format '{format_name}'. Proceeding, but may fail.") - return - - if info['audio_only']: - 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.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." - ) - else: - # If video is present and not dropped, check if the container supports the video codec - if stream_info.get('has_video') and not drop_video and input_file: - from .ffmpeg import get_video_codec - video_codec = get_video_codec(input_file) - if video_codec and not is_video_codec_supported(format_name, video_codec): - raise ValueError( - f"Container '{format_name}' does not support video codec '{video_codec}'. " - "Please use --drop-video or choose a container that supports this video codec (e.g., MKV)." - ) - -def is_video_codec_supported(container_name: str, video_codec: str) -> bool: - """ - Check if a given container supports a specific video codec. - - Args: - container_name: Name of the container (e.g., 'mkv', 'ogg') - video_codec: Video codec name (e.g., 'theora', 'h264', 'png') - - Returns: - True if supported, False otherwise. - """ - support = CONTAINER_VIDEO_CODEC_SUPPORT.get(container_name, []) - if not support: - return False - if '*' in support: - return True - return video_codec in support - - def determine_output_format(stream_info: Dict, user_format: Optional[str], transcode_audio: Optional[str], input_file: Optional[str] = None) -> str: """ @@ -93,42 +50,78 @@ def determine_output_format(stream_info: Dict, user_format: Optional[str], 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 and the video codec is not supported by the recommended container + - MKV if video/subtitles exist - MP3 if the audio codec is MP3 - MP4 (M4A) otherwise """ if user_format: return user_format - # Try to detect container and codec from input file if input_file: try: - from .ffmpeg import get_container_format, get_audio_codec, get_video_codec + from .ffmpeg import get_container_format, get_audio_codec container = get_container_format(input_file) audio_codec = get_audio_codec(input_file) - video_codec = get_video_codec(input_file) # NEW: get video codec - - # Determine recommended format based on audio codec fmt = determine_default_format(container, audio_codec) - - # If video is present and not dropped, check if the recommended container supports the video codec - if stream_info.get('has_video') and not transcode_audio: # transcode_audio doesn't affect video - # If the recommended container doesn't support the video codec, fallback to MKV - if fmt and not is_video_codec_supported(fmt, video_codec): - fmt = 'matroska' # MKV supports virtually all video codecs - # Optionally log a warning - # print(f"Warning: Container '{fmt}' does not support video codec '{video_codec}'. Falling back to MKV.") - if fmt in FORMAT_INFO: return fmt except Exception: pass - # Fallback: legacy behavior + # Fallback 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' else: - return 'mp4' # .m4a + return 'mp4' + + +def validate_format_compatibility( + container: str, + stream_info: Dict, + drop_video: bool, + drop_subs: bool, + input_file: Optional[str] = None, + audio_codec: Optional[str] = None, + video_codec: Optional[str] = None, +) -> None: + """ + Ensure the chosen container and codec combination is valid. + + Raises: + ValueError: If the combination is incompatible. + """ + info = FORMAT_INFO.get(container) + if not info: + print(f"Warning: Unknown container '{container}'. Proceeding, but may fail.") + return + + # Check if container is audio-only and video is present (unless dropped) + if info['audio_only'] and stream_info.get('has_video') and not drop_video: + raise ValueError( + f"Container '{container}' does not support video streams. " + "Please use --drop-video or choose a container that supports video (e.g., MKV, MP4)." + ) + + # Check if audio codec is supported + if audio_codec and audio_codec != 'copy': + if not is_audio_codec_supported(container, audio_codec): + raise ValueError( + f"Container '{container}' does not support audio codec '{audio_codec}'. " + f"Please choose a different container or audio codec." + ) + + # Check if video codec is supported (if video is present and not dropped) + if stream_info.get('has_video') and not drop_video: + # Determine the video codec from input file if not provided + if video_codec is None and input_file: + from .ffmpeg import get_video_codec + video_codec = get_video_codec(input_file) + if video_codec and video_codec != 'copy': + if not is_video_codec_supported(container, video_codec): + raise ValueError( + f"Container '{container}' does not support video codec '{video_codec}'. " + f"Please choose a different container, drop video, or transcode video to a supported codec." + ) diff --git a/audio_splitter/main.py b/audio_splitter/main.py index 4c83bc9..9d8cd39 100644 --- a/audio_splitter/main.py +++ b/audio_splitter/main.py @@ -1,11 +1,18 @@ -"""Command‑line interface and entry point.""" +#!/usr/bin/env python3 +""" +Audio Splitter – Command‑Line Interface + +Split an audio file into tracks using a tracklist file. + +Usage: + audio_splitter input.mp3 tracklist.txt [OPTIONS] +""" -import argparse import os import sys import subprocess +import argparse -from .constants import DEFAULT_BAD_CHARS from .defaults import ( DEFAULT_FORMAT, DEFAULT_OUTPUT_TEMPLATE, @@ -29,45 +36,72 @@ from .defaults import ( from .core import split_audio from .tracklist import read_tracklist, parse_format + def main(): - parser = argparse.ArgumentParser(description="Split audio file using a tracklist.") + parser = argparse.ArgumentParser( + description="Split an audio file into tracks using a tracklist.", + epilog="Tracklist format: mm:ss track_name - author_name (or custom with --tracklist-format)" + ) + + # Positional parser.add_argument('input_file', help='Input audio file') parser.add_argument('tracklist_file', help='Tracklist file') - # Output options - parser.add_argument('--format', default=DEFAULT_FORMAT, help=f"Output container format (default: {DEFAULT_FORMAT})") - parser.add_argument('--transcode-to', default=DEFAULT_TRANSCODE_TO, help="Audio codec to transcode to (default: copy)") - parser.add_argument('--drop-video', action='store_true', default=DEFAULT_DROP_VIDEO, help="Drop video streams") - parser.add_argument('--drop-subs', action='store_true', default=DEFAULT_DROP_SUBS, help="Drop subtitle streams") + # Container and codec options + parser.add_argument('--container', default=DEFAULT_FORMAT, + help="Output container format (default: %(default)s)") + parser.add_argument('--audio-codec', default='copy', + help="Audio codec (copy or encoder name, e.g., libopus)") + parser.add_argument('--video-codec', default='copy', + help="Video codec (copy or encoder name, e.g., libx264)") + parser.add_argument('--subtitle-codec', default='copy', + help="Subtitle codec (copy or encoder name, e.g., srt)") + parser.add_argument('--drop-video', action='store_true', default=DEFAULT_DROP_VIDEO, + help="Drop video streams") + parser.add_argument('--drop-subs', action='store_true', default=DEFAULT_DROP_SUBS, + help="Drop subtitle streams") # Filename options - parser.add_argument('--number-tracks', action='store_true', default=DEFAULT_NUMBER_TRACKS, help="Prepend track numbers") - parser.add_argument('--output-template', default=DEFAULT_OUTPUT_TEMPLATE, help=f"Output filename template (default: {DEFAULT_OUTPUT_TEMPLATE})") - parser.add_argument('--replace-bad-chars', action='store_true', default=DEFAULT_REPLACE_BAD_CHARS, help="Replace bad characters") - parser.add_argument('--replacement-char', default=DEFAULT_REPLACEMENT_CHAR, help=f"Replacement character (default: {DEFAULT_REPLACEMENT_CHAR})") - parser.add_argument('--bad-chars', default=DEFAULT_BAD_CHARS, help=f"Bad characters to replace (default: {DEFAULT_BAD_CHARS})") - parser.add_argument('--skip-existing', action='store_true', default=DEFAULT_SKIP_EXISTING, help="Skip existing output files") + parser.add_argument('--number-tracks', action='store_true', default=DEFAULT_NUMBER_TRACKS, + help="Prepend track numbers") + parser.add_argument('--output-template', default=DEFAULT_OUTPUT_TEMPLATE, + help="Output filename template (default: %(default)s)") + parser.add_argument('--replace-bad-chars', action='store_true', default=DEFAULT_REPLACE_BAD_CHARS, + help="Replace bad characters") + parser.add_argument('--replacement-char', default=DEFAULT_REPLACEMENT_CHAR, + help="Replacement character (default: %(default)s)") + parser.add_argument('--bad-chars', default=DEFAULT_BAD_CHARS, + help="Bad characters to replace (default: %(default)s)") + parser.add_argument('--skip-existing', action='store_true', default=DEFAULT_SKIP_EXISTING, + help="Skip existing output files") - # Metadata options + # Metadata parser.add_argument('--album', default=DEFAULT_ALBUM, help="Album name") parser.add_argument('--comment', default=DEFAULT_COMMENT, help="Comment") - parser.add_argument('--no-comment', action='store_true', default=DEFAULT_NO_COMMENT, help="Ignore comment") - parser.add_argument('--comment-stream', type=int, default=DEFAULT_COMMENT_STREAM, help="Comment stream index") - parser.add_argument('--merge-comments', action='store_true', default=DEFAULT_MERGE_COMMENTS, help="Merge all comments") - parser.add_argument('--comment-separator', default=DEFAULT_COMMENT_SEPARATOR, help=f"Separator for merged comments (default: {DEFAULT_COMMENT_SEPARATOR})") + parser.add_argument('--no-comment', action='store_true', default=DEFAULT_NO_COMMENT, + help="Ignore comment") + parser.add_argument('--comment-stream', type=int, default=DEFAULT_COMMENT_STREAM, + help="Comment stream index") + parser.add_argument('--merge-comments', action='store_true', default=DEFAULT_MERGE_COMMENTS, + help="Merge all comments") + parser.add_argument('--comment-separator', default=DEFAULT_COMMENT_SEPARATOR, + help="Separator for merged comments (default: %(default)s)") # Tracklist format - parser.add_argument('--tracklist-format', default=DEFAULT_TRACKLIST_FORMAT, help=f"Tracklist format (default: {DEFAULT_TRACKLIST_FORMAT})") + parser.add_argument('--tracklist-format', default=DEFAULT_TRACKLIST_FORMAT, + help="Tracklist format (default: %(default)s)") # Other - parser.add_argument('--delete-original', action='store_true', default=DEFAULT_DELETE_ORIGINAL, help="Delete original file after split") - parser.add_argument('--dry-run', action='store_true', help="Parse and display tracklist without splitting") + parser.add_argument('--delete-original', action='store_true', default=DEFAULT_DELETE_ORIGINAL, + help="Delete original file after split") + parser.add_argument('--dry-run', action='store_true', + help="Parse and display tracklist without splitting") + parser.add_argument('--output-dir', '-o', default=None, + help="Output directory (default: _splits)") args = parser.parse_args() - # -------------------------------------------------------------------------- - # Input validation - # -------------------------------------------------------------------------- + # Validate input if not os.path.exists(args.input_file): print(f"Error: Input file not found: {args.input_file}") sys.exit(1) @@ -76,12 +110,7 @@ def main(): print(f"Error: Tracklist file not found: {args.tracklist_file}") sys.exit(1) - # Ensure the replacement character is a single character. - if len(args.replacement_char) != 1: - print("Error: --replacement-char must be a single character.") - sys.exit(1) - - # Check that FFmpeg is installed. + # Check FFmpeg try: subprocess.run(['ffmpeg', '-version'], capture_output=True, check=True) except (subprocess.CalledProcessError, FileNotFoundError): @@ -89,14 +118,14 @@ def main(): print(" - https://ffmpeg.org/download.html") sys.exit(1) - # Parse the tracklist using the user‑provided format. + # Parse tracklist format try: tokens = parse_format(args.tracklist_format) - except ValueError as error: - print(f"Error in --tracklist-format: {error}") + except ValueError as e: + print(f"Error in --tracklist-format: {e}") sys.exit(1) - # Read and parse the tracklist file. + # Read tracklist tracks = read_tracklist(args.tracklist_file, tokens) if not tracks: print("Error: No valid tracks found in tracklist file.") @@ -104,7 +133,6 @@ def main(): print(f"Found {len(tracks)} tracks.") - # Dry‑run mode: display parsed data and exit. if args.dry_run: print("\nParsed tracklist:") print("-" * 60) @@ -118,34 +146,22 @@ def main(): print(f"{idx:3d} | " + " | ".join(values)) print("-" * 60) print("Dry‑run complete. No files were created.") - sys.exit(0) + return - # Determine the output directory. + # Determine output directory if args.output_dir: output_dir = args.output_dir - print(f"Using custom output directory: {output_dir}") else: base_name = os.path.splitext(os.path.basename(args.input_file))[0] output_dir = base_name + "_splits" - print(f"Using default output directory: {output_dir}") - # Run the splitter. + # Run split try: split_audio(args.input_file, output_dir, tracks, args) - except (RuntimeError, ValueError) as error: - print(f"Error: {error}") + except Exception as e: + print(f"Error during split: {e}") sys.exit(1) - # -------------------------------------------------------------------------- - # Delete original file if requested and successful. - # -------------------------------------------------------------------------- - if args.delete_original: - try: - os.remove(args.input_file) - print(f"Deleted original file: {args.input_file}") - except OSError as e: - print(f"Warning: Could not delete original file: {e}") - print("Done!")