refact (core): split large code blocks, remove useless files/code #12

Merged
max merged 12 commits from formats_consistency into dev 2026-09-01 15:19:34 +04:00
13 changed files with 494 additions and 252 deletions
+45 -56
View File
@@ -11,56 +11,62 @@ This file serves as the single source of truth for:
Codec != Container != File Extension.
Example: Opus (codec) → Ogg (container) → .opus (extension)
"""
from typing import Dict, Optional
# ------------------------------------------------------------------------------
# Container information
# ------------------------------------------------------------------------------
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},
{'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},
{'name': 'mp3', 'ffmpeg': 'mp3', 'extension': '.mp3', 'supports_video': True, 'supports_subs': False},
{'name': 'm4a', 'ffmpeg': 'mp4', 'extension': '.m4a', 'supports_video': False, 'supports_subs': False},
{'name': 'flac', 'ffmpeg': 'flac', 'extension': '.flac', 'supports_video': False, 'supports_subs': False},
{'name': 'wav', 'ffmpeg': 'wav', 'extension': '.wav', 'supports_video': False, 'supports_subs': False},
{'name': 'aac', 'ffmpeg': 'adts', 'extension': '.aac', '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},
{'name': 'webm', 'ffmpeg': 'webm', 'extension': '.webm', 'audio_only': False, 'supports_video': True, 'supports_subs': False},
{'name': 'mp4', 'ffmpeg': 'mp4', 'extension': '.mp4', 'supports_video': True, 'supports_subs': True},
{'name': 'mkv', 'ffmpeg': 'matroska', 'extension': '.mkv', 'supports_video': True, 'supports_subs': True},
{'name': 'ogg', 'ffmpeg': 'ogg', 'extension': '.ogg', 'supports_video': True, 'supports_subs': True},
{'name': 'webm', 'ffmpeg': 'webm', 'extension': '.webm', 'supports_video': True, 'supports_subs': False},
]
# Legacy FORMAT_INFO for backward compatibility
FORMAT_INFO = {
container['name']: {
'ffmpeg': container['ffmpeg'],
'ext': container['extension'],
'audio_only': container['audio_only'],
}
for container in CONTAINER_INFO
}
# Container name lookup set for fast membership checks
CONTAINER_NAMES = {container['name'] for container in CONTAINER_INFO}
def get_container_info(container_name: str) -> Optional[Dict]:
"""Return container info dict or None if not found."""
for container in CONTAINER_INFO:
if container['name'] == container_name:
return container
return None
# ------------------------------------------------------------------------------
# Codec information
# ------------------------------------------------------------------------------
CODEC_INFO = [
# Audio codecs
{'name': 'opus', 'ffmpeg': 'libopus', 'recommended_container': 'ogg', 'recommended_extension': '.opus', 'supports_transcoding': True, 'supports_video': False},
{'name': 'vorbis', 'ffmpeg': 'libvorbis', 'recommended_container': 'ogg', 'recommended_extension': '.ogg', 'supports_transcoding': True, 'supports_video': False},
{'name': 'flac', 'ffmpeg': 'flac', 'recommended_container': 'flac', 'recommended_extension': '.flac', 'supports_transcoding': True, 'supports_video': False},
{'name': 'aac', 'ffmpeg': 'aac', 'recommended_container': 'mp4', 'recommended_extension': '.m4a', 'supports_transcoding': True, 'supports_video': False},
{'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
{'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},
{'name': 'opus', 'ffmpeg': 'libopus', 'recommended_container': 'ogg', 'recommended_extension': '.opus', 'codec_type': 'audio'},
{'name': 'vorbis', 'ffmpeg': 'libvorbis', 'recommended_container': 'ogg', 'recommended_extension': '.ogg', 'codec_type': 'audio'},
{'name': 'flac', 'ffmpeg': 'flac', 'recommended_container': 'flac', 'recommended_extension': '.flac', 'codec_type': 'audio'},
{'name': 'aac', 'ffmpeg': 'aac', 'recommended_container': 'mp4', 'recommended_extension': '.m4a', 'codec_type': 'audio'},
{'name': 'mp3', 'ffmpeg': 'libmp3lame', 'recommended_container': 'mp3', 'recommended_extension': '.mp3', 'codec_type': 'audio'},
{'name': 'alac', 'ffmpeg': 'alac', 'recommended_container': 'mp4', 'recommended_extension': '.m4a', 'codec_type': 'audio'},
{'name': 'pcm_s16le', 'ffmpeg': 'pcm_s16le', 'recommended_container': 'wav', 'recommended_extension': '.wav', 'codec_type': 'audio'},
# Video codecs (including cover images)
{'name': 'theora', 'ffmpeg': 'libtheora', 'recommended_container': 'ogg', 'recommended_extension': '.ogv', 'codec_type': 'video'},
{'name': 'vp8', 'ffmpeg': 'libvpx', 'recommended_container': 'webm', 'recommended_extension': '.webm', 'codec_type': 'video'},
{'name': 'vp9', 'ffmpeg': 'libvpx-vp9', 'recommended_container': 'webm', 'recommended_extension': '.webm', 'codec_type': 'video'},
{'name': 'av1', 'ffmpeg': 'libaom-av1', 'recommended_container': 'mkv', 'recommended_extension': '.mkv', 'codec_type': 'video'},
{'name': 'h264', 'ffmpeg': 'libx264', 'recommended_container': 'mp4', 'recommended_extension': '.mp4', 'codec_type': 'video'},
{'name': 'h265', 'ffmpeg': 'libx265', 'recommended_container': 'mp4', 'recommended_extension': '.mp4', 'codec_type': 'video'},
{'name': 'png', 'ffmpeg': 'png', 'recommended_container': 'ogg', 'recommended_extension': '.png', 'codec_type': 'video'},
{'name': 'mjpeg', 'ffmpeg': 'mjpeg', 'recommended_container': 'ogg', 'recommended_extension': '.jpg', 'codec_type': 'video'},
# Subtitle codecs
{'name': 'srt', 'ffmpeg': 'srt', 'recommended_container': 'mkv', 'recommended_extension': '.srt', 'codec_type': 'subtitle'},
{'name': 'ass', 'ffmpeg': 'ass', 'recommended_container': 'mkv', 'recommended_extension': '.ass', 'codec_type': 'subtitle'},
{'name': 'vtt', 'ffmpeg': 'webvtt', 'recommended_container': 'webm', 'recommended_extension': '.vtt', 'codec_type': 'subtitle'},
]
# Map codec name → recommended container
@@ -70,22 +76,6 @@ CODEC_TO_EXTENSION_MAP = {codec['name']: codec['recommended_extension'] for code
# Map codec name → FFmpeg encoder name
CODEC_NAME_TO_FFMPEG = {codec['name']: codec['ffmpeg'] for codec in CODEC_INFO}
# ------------------------------------------------------------------------------
# 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
# ------------------------------------------------------------------------------
@@ -96,14 +86,13 @@ CONTAINER_VIDEO_CODEC_SUPPORT = {
# 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': []},
'mp3': {'audio': ['mp3'], 'video': ['png', 'mjpeg']},
'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']},
'ogg': {'audio': ['opus', 'vorbis', 'flac'], 'video': ['theora', 'vp8', 'png', 'mjpeg']},
'webm': {'audio': ['opus', 'vorbis'], 'video': ['vp8', 'vp9', 'av1']},
'flac': {'audio': ['flac'], 'video': []},
'flac': {'audio': ['flac'], 'video': ['png', 'mjpeg']},
'wav': {'audio': ['pcm_s16le'], 'video': []},
'aac': {'audio': ['aac'], 'video': []},
}
@@ -132,4 +121,4 @@ def is_video_codec_supported(container: str, video_codec: str) -> bool:
# ------------------------------------------------------------------------------
# Default bad characters (for filename sanitization)
# ------------------------------------------------------------------------------
DEFAULT_BAD_CHARS = r'!@#№$;:%^&?*(){}[]\/<>+=~`\' '
DEFAULT_BAD_CHARS = r'!@#№$;:%^&?*(){}[]\/<>+=~`\'|," '
+35 -5
View File
@@ -7,9 +7,9 @@ import sys
from typing import List, Dict, Any
from .constants import (
FORMAT_INFO,
CONTAINER_NAMES,
CODEC_TO_EXTENSION_MAP,
CODEC_NAME_TO_FFMPEG, # <-- Add this
CODEC_NAME_TO_FFMPEG,
)
from .ffmpeg import (
get_audio_duration,
@@ -23,6 +23,7 @@ from .formats import (
determine_output_format,
validate_format_compatibility,
)
from .handlers import get_handler, needs_drop_video
from .timestamp import parse_track_timestamps, resolve_end_times
from .filename import build_filename
from .metadata import build_metadata_dict
@@ -120,7 +121,8 @@ def split_audio(input_file: str, output_directory: str, tracks: List[Dict[str, A
if output_audio_codec in CODEC_TO_EXTENSION_MAP:
extension = CODEC_TO_EXTENSION_MAP[output_audio_codec]
else:
extension = FORMAT_INFO.get(output_container, {}).get('ext', '.mkv')
container_info = next((c for c in CONTAINER_INFO if c['name'] == output_container), None)
extension = container_info['extension'] if container_info else '.mkv'
print(f"Output extension: {extension}")
@@ -162,6 +164,14 @@ def split_audio(input_file: str, output_directory: str, tracks: List[Dict[str, A
else:
cover_image_path = None
# Check if we need special handling (e.g., Opus files need opustags)
input_audio_codec = stream_info.get('audio_codec')
cover_handler = get_handler(output_container, input_audio_codec)
needs_drop = needs_drop_video(output_container, input_audio_codec)
if cover_image_path and needs_drop:
print(f"Using special handler for {output_container} + {input_audio_codec}")
print("Temporarily dropping video for opustags post-processing.")
# --------------------------------------------------------------------------
# 8. Process each track
# --------------------------------------------------------------------------
@@ -208,6 +218,11 @@ def split_audio(input_file: str, output_directory: str, tracks: List[Dict[str, A
subtitle_enc = CODEC_NAME_TO_FFMPEG.get(subtitle_enc, subtitle_enc)
# Then build command with these mapped encoders
# Use temporary drop_video for handlers that need it
effective_drop_video = args.drop_video or (cover_image_path and needs_drop)
# For handlers that need opustags post-processing, don't pass cover_image_path
# to ffmpeg - it will process audio-only, then handler adds cover afterward
ffmpeg_cover_path = None if (cover_image_path and needs_drop) else cover_image_path
cmd = build_ffmpeg_command(
input_file=input_file,
start_seconds=start_seconds,
@@ -219,9 +234,9 @@ def split_audio(input_file: str, output_directory: str, tracks: List[Dict[str, A
video_codec=video_enc,
subtitle_codec=subtitle_enc,
metadata=metadata,
cover_image_path=cover_image_path,
cover_image_path=ffmpeg_cover_path,
video_quality=getattr(args, 'video_quality', None),
drop_video=args.drop_video,
drop_video=effective_drop_video,
drop_subs=args.drop_subs,
)
@@ -235,6 +250,21 @@ def split_audio(input_file: str, output_directory: str, tracks: List[Dict[str, A
print(result.stderr)
else:
print(f" -> Saved to: {output_path}")
# Apply cover image handler if needed
if cover_image_path and not args.drop_video and needs_drop:
print(f"Applying cover image via {output_container} handler...")
if not cover_handler(
input_file=input_file,
cover_image_path=cover_image_path,
output_path=output_path,
stream_info=stream_info,
audio_codec=audio_enc,
video_codec=video_enc,
video_quality=getattr(args, 'video_quality', None),
drop_video=args.drop_video,
drop_subs=args.drop_subs,
):
print(f"Warning: Failed to attach cover image to {output_path}")
# --------------------------------------------------------------------------
# 9. Delete original file if requested
+124 -90
View File
@@ -5,7 +5,7 @@ import json
import subprocess
from typing import Dict, List, Optional, Tuple
from .constants import FORMAT_INFO
from .constants import CONTAINER_INFO
from .utils import format_time
@@ -159,7 +159,7 @@ def get_metadata(input_file: str) -> Dict[str, any]:
def get_container_format(input_file: str) -> Optional[str]:
"""
Retrieve the container format name (e.g., 'mp4', 'mp3', 'matroska') from the input file.
Retrieve the container format name (e.g., 'mp4', 'mp3', 'mkv') from the input file.
Args:
input_file: Path to the media file.
@@ -179,7 +179,7 @@ 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
# Normalize common aliases to names used in CONTAINER_INFO
mapping = {
'mpeg': 'mp3',
'mp2': 'mp3',
@@ -187,8 +187,8 @@ def get_container_format(input_file: str) -> Optional[str]:
'm4a': 'mp4',
'mov': 'mp4',
'3gp': 'mp4',
'matroska': 'matroska',
'webm': 'matroska',
'matroska': 'mkv',
'webm': 'mkv',
'ogg': 'ogg',
'flac': 'flac',
'wav': 'wav',
@@ -269,6 +269,112 @@ def extract_cover_image(input_file: str, output_path: str) -> bool:
return True
def _build_inputs(cmd: List[str], input_file: str, cover_image_path: Optional[str]) -> None:
"""Add input files to the command."""
if cover_image_path:
cmd.extend(['-i', cover_image_path])
cmd.extend(['-i', input_file])
def _build_time_options(cmd: List[str], start_seconds: int, duration_seconds: int) -> None:
"""Add time-based options to the command."""
cmd.extend(['-ss', format_time(start_seconds), '-t', format_time(duration_seconds)])
def _build_metadata(cmd: List[str], metadata: Optional[Dict]) -> None:
"""Add metadata options to the command."""
cmd.append('-map_metadata')
cmd.append('-1')
if metadata:
for key, value in metadata.items():
if value is not None and value != '':
cmd.extend(['-metadata', f"{key}={value}"])
def _build_cover_image_mapping(
cmd: List[str],
audio_codec: Optional[str],
video_codec: Optional[str],
video_quality: Optional[int],
) -> None:
"""Build stream mapping for cover image extraction (audio + cover video)."""
cmd.extend(['-map', '1:a:0', '-map', '0:v:0'])
if video_codec and video_codec != 'copy':
cmd.extend(['-c:v', video_codec])
else:
cmd.extend(['-c:v', 'png'])
if audio_codec and audio_codec != 'copy':
cmd.extend(['-c:a', audio_codec])
else:
cmd.extend(['-c:a', 'copy'])
# Preserve attached_pic disposition for cover image
cmd.extend(['-disposition', 'attached_pic'])
cmd.append('-sn')
if video_quality is not None:
cmd.extend(['-q:v', str(video_quality)])
def _build_standard_mapping(
cmd: List[str],
stream_info: Dict,
audio_codec: Optional[str],
video_codec: Optional[str],
subtitle_codec: Optional[str],
video_quality: Optional[int],
drop_video: bool,
drop_subs: bool,
) -> None:
"""Build stream mapping for standard extraction (no cover image)."""
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'])
if audio_codec and audio_codec != 'copy':
cmd.extend(['-c:a', audio_codec])
else:
cmd.extend(['-c:a', 'copy'])
if not drop_video and stream_info.get('has_video'):
if video_codec and video_codec != 'copy':
cmd.extend(['-c:v', video_codec])
if video_quality is not None:
cmd.extend(['-q:v', str(video_quality)])
else:
cmd.extend(['-c:v', 'copy'])
else:
cmd.append('-vn')
if not drop_subs and stream_info.get('has_subtitle'):
if subtitle_codec and subtitle_codec != 'copy':
cmd.extend(['-c:s', subtitle_codec])
else:
cmd.extend(['-c:s', 'copy'])
else:
cmd.append('-sn')
def _build_format(cmd: List[str], format_opt: Optional[str]) -> None:
"""Add output format option if specified."""
if format_opt:
ffmpeg_format = next((c['ffmpeg'] for c in CONTAINER_INFO if c['name'] == format_opt), format_opt)
cmd.extend(['-f', ffmpeg_format])
def _build_output(cmd: List[str], output_path: str) -> None:
"""Add output file to the command."""
cmd.extend(['-y', output_path])
def build_ffmpeg_command(
input_file: str,
start_seconds: int,
@@ -309,93 +415,21 @@ def build_ffmpeg_command(
"""
cmd = ['ffmpeg']
# Add cover image as first input if provided
_build_inputs(cmd, input_file, cover_image_path)
_build_time_options(cmd, start_seconds, duration_seconds)
_build_metadata(cmd, metadata)
if cover_image_path:
cmd.extend(['-i', cover_image_path])
# Add main input file
cmd.extend(['-i', input_file])
# Time options
cmd.extend(['-ss', format_time(start_seconds), '-t', format_time(duration_seconds)])
# Clear all original metadata.
cmd.append('-map_metadata')
cmd.append('-1')
# 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 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'])
# 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])
_build_cover_image_mapping(cmd, audio_codec, video_codec, video_quality)
else:
cmd.extend(['-c:v', 'png'])
_build_standard_mapping(
cmd, stream_info, audio_codec, video_codec, subtitle_codec,
video_quality, drop_video, drop_subs
)
# Audio codec
if audio_codec and audio_codec != 'copy':
cmd.extend(['-c:a', audio_codec])
else:
cmd.extend(['-c:a', 'copy'])
_build_format(cmd, format_opt)
_build_output(cmd, output_path)
# Subtitle: we don't copy subtitles when using cover image (they would be from main input)
cmd.append('-sn')
# Video quality if specified
if video_quality is not None:
cmd.extend(['-q:v', str(video_quality)])
else:
# Standard mapping (no cover image)
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'])
# Audio codec
if audio_codec and audio_codec != 'copy':
cmd.extend(['-c:a', audio_codec])
else:
cmd.extend(['-c:a', 'copy'])
# Video codec
if not drop_video and stream_info.get('has_video'):
if video_codec and video_codec != 'copy':
cmd.extend(['-c:v', video_codec])
# Add video quality if specified (only when re-encoding)
if video_quality is not None:
cmd.extend(['-q:v', str(video_quality)])
else:
cmd.extend(['-c:v', 'copy'])
else:
cmd.append('-vn')
# Subtitle codec
if not drop_subs and stream_info.get('has_subtitle'):
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
+24 -14
View File
@@ -4,7 +4,7 @@
from typing import Dict, Optional
from .constants import (
FORMAT_INFO,
CONTAINER_NAMES,
CONTAINER_INFO,
CODEC_TO_CONTAINER_MAP,
COMPATIBILITY_MATRIX,
@@ -22,7 +22,7 @@ def determine_default_format(container: Optional[str], codec: Optional[str]) ->
If the codec is not found, it falls back to the container.
Args:
container: Container name (e.g., 'ogg', 'mp4', 'matroska') as returned by get_container_format().
container: Container name (e.g., 'ogg', 'mp4', 'mkv') as returned by get_container_format().
codec: Audio codec name (e.g., 'opus', 'aac', 'mp3') as returned by get_audio_codec().
Returns:
@@ -33,10 +33,10 @@ def determine_default_format(container: Optional[str], codec: Optional[str]) ->
if codec and codec in CODEC_TO_CONTAINER_MAP:
fmt = CODEC_TO_CONTAINER_MAP[codec]
if fmt in FORMAT_INFO:
if fmt in CONTAINER_NAMES:
return fmt
if container in FORMAT_INFO:
if container in CONTAINER_NAMES:
return container
return None
@@ -47,7 +47,7 @@ def determine_output_format(stream_info: Dict, user_format: Optional[str],
"""
Decide which container format to use.
If user_format is provided, use it.
If user_format is provided, map it to the correct container if it's a codec name.
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
@@ -55,6 +55,10 @@ def determine_output_format(stream_info: Dict, user_format: Optional[str],
- MP4 (M4A) otherwise
"""
if user_format:
# Map codec names to their correct containers
# e.g., 'opus' -> 'ogg', 'aac' -> 'mp4', etc.
if user_format in CODEC_TO_CONTAINER_MAP:
return CODEC_TO_CONTAINER_MAP[user_format]
return user_format
if input_file:
@@ -63,14 +67,14 @@ def determine_output_format(stream_info: Dict, user_format: Optional[str],
container = get_container_format(input_file)
audio_codec = get_audio_codec(input_file)
fmt = determine_default_format(container, audio_codec)
if fmt in FORMAT_INFO:
if fmt in CONTAINER_NAMES:
return fmt
except Exception:
pass
# Fallback
if stream_info.get('has_video') or stream_info.get('has_subtitle'):
return 'matroska'
return 'mkv'
audio_codec = stream_info.get('audio_codec', '')
if audio_codec == 'mp3':
return 'mp3'
@@ -93,23 +97,25 @@ def validate_format_compatibility(
Raises:
ValueError: If the combination is incompatible.
"""
info = FORMAT_INFO.get(container)
info = next((c for c in CONTAINER_INFO if c['name'] == container), None)
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:
if not info['supports_video'] 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):
# Determine actual audio codec for validation
# When 'copy' is used, we must still validate the input codec against the container
actual_audio_codec = audio_codec if audio_codec and audio_codec != 'copy' else stream_info.get('audio_codec')
if actual_audio_codec:
if not is_audio_codec_supported(container, actual_audio_codec):
raise ValueError(
f"Container '{container}' does not support audio codec '{audio_codec}'. "
f"Container '{container}' does not support audio codec '{actual_audio_codec}'. "
f"Please choose a different container or audio codec."
)
@@ -119,7 +125,11 @@ def validate_format_compatibility(
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':
# When 'copy' is used, validate the input video codec against the container
if video_codec == 'copy' and input_file:
from .ffmpeg import get_video_codec
video_codec = get_video_codec(input_file)
if video_codec:
if not is_video_codec_supported(container, video_codec):
raise ValueError(
f"Container '{container}' does not support video codec '{video_codec}'. "
+167
View File
@@ -0,0 +1,167 @@
# audio_splitter/handlers.py
"""Handler system for cover image attachment.
This module provides a registry of handlers for different
codec + image combinations. Each handler is responsible for
attaching the cover image to the output file in the appropriate way.
"""
import os
import subprocess
from typing import Dict, Optional
def _run_opustags(input_file: str, cover_image_path: str, output_path: Optional[str] = None) -> bool:
"""
Run opustags to attach a cover image to an Opus file.
Args:
input_file: Path to the input Opus file.
cover_image_path: Path to the cover image.
output_path: Optional output path (if None, overwrites input).
Returns:
True if successful, False otherwise.
"""
cmd = ['opustags', '--set-cover', cover_image_path, input_file]
if output_path:
cmd.extend(['-o', output_path, '-y'])
else:
cmd.extend(['-i'])
result = subprocess.run(cmd, capture_output=True, text=True, check=False)
if result.returncode != 0:
print(f"opustags failed: {result.stderr}")
return False
return True
def _handler_default(
input_file: str,
cover_image_path: str,
output_path: str,
stream_info: Dict,
audio_codec: str,
video_codec: str,
video_quality: Optional[int],
drop_video: bool,
drop_subs: bool,
) -> bool:
"""
Default handler: use ffmpeg with -disposition attached_pic.
This is the standard approach for formats like MP3, FLAC, etc.
"""
# This handler doesn't do anything - the cover is already handled
# by the ffmpeg command in core.py via build_ffmpeg_command()
# We just return True to indicate success.
return True
def _handler_opus_with_cover(
input_file: str,
cover_image_path: str,
output_path: str,
stream_info: Dict,
audio_codec: str,
video_codec: str,
video_quality: Optional[int],
drop_video: bool,
drop_subs: bool,
) -> bool:
"""
Handler for Opus files with cover image.
Uses opustags to attach the cover image after the file is processed.
"""
return _run_opustags(output_path, cover_image_path)
def _handler_mp3_with_cover(
input_file: str,
cover_image_path: str,
output_path: str,
stream_info: Dict,
audio_codec: str,
video_codec: str,
video_quality: Optional[int],
drop_video: bool,
drop_subs: bool,
) -> bool:
"""
Handler for MP3 files with cover image.
Uses ffmpeg with -disposition attached_pic.
"""
# MP3 handler - the cover is already attached by ffmpeg in core.py
return True
def _handler_flac_with_cover(
input_file: str,
cover_image_path: str,
output_path: str,
stream_info: Dict,
audio_codec: str,
video_codec: str,
video_quality: Optional[int],
drop_video: bool,
drop_subs: bool,
) -> bool:
"""
Handler for FLAC files with cover image.
Uses ffmpeg with -disposition attached_pic.
"""
# FLAC handler - the cover is already attached by ffmpeg in core.py
return True
# Handler registry: maps (output_container, input_codec) to handler function
HANDLER_REGISTRY: Dict[tuple, callable] = {
('opus', 'opus'): _handler_opus_with_cover,
('ogg', 'opus'): _handler_opus_with_cover,
('mp3', 'mp3'): _handler_mp3_with_cover,
('flac', 'flac'): _handler_flac_with_cover,
}
def get_handler(output_container: str, input_audio_codec: Optional[str] = None):
"""
Get the appropriate handler for the given container and input codec.
Args:
output_container: Output container format (e.g., 'opus', 'mp3', 'flac').
input_audio_codec: Input audio codec (e.g., 'opus', 'mp3', 'flac').
Returns:
Handler function, or the default handler if no specific handler is found.
"""
if input_audio_codec:
key = (output_container, input_audio_codec)
if key in HANDLER_REGISTRY:
return HANDLER_REGISTRY[key]
# Fall back to container-only key
if (output_container, None) in HANDLER_REGISTRY:
return HANDLER_REGISTRY[(output_container, None)]
# Default handler
return _handler_default
def needs_drop_video(output_container: str, input_audio_codec: Optional[str] = None) -> bool:
"""
Check if the handler requires dropping video even if user didn't specify --drop-video.
Args:
output_container: Output container format.
input_audio_codec: Input audio codec.
Returns:
True if video should be dropped, False otherwise.
"""
handler = get_handler(output_container, input_audio_codec)
# The opus handler requires dropping video
return handler == _handler_opus_with_cover
+2 -1
View File
@@ -1,9 +1,10 @@
FROM python:3.13-slim
# Install FFmpeg, system dependencies, and gosu from APT
# Install FFmpeg, opustags, system dependencies, and gosu from APT
RUN apt-get update && \
apt-get install -y --no-install-recommends \
ffmpeg \
opustags \
ca-certificates \
gosu \
&& \
+2 -2
View File
@@ -9,7 +9,7 @@ from backend.services.task_manager import task_manager
# Import core functions
from audio_splitter.ffmpeg import get_stream_info, get_container_format, get_audio_codec
from audio_splitter.formats import determine_default_format
from audio_splitter.constants import FORMAT_INFO
from audio_splitter.constants import CONTAINER_NAMES
from audio_splitter.defaults import DEFAULT_FORMAT
router = APIRouter(prefix="/api", tags=["info"])
@@ -63,7 +63,7 @@ async def get_recommended_format(task_id: str):
# Fallback if detection fails or format is unsupported
if fmt is None:
fmt = DEFAULT_FORMAT
if fmt not in FORMAT_INFO:
if fmt not in CONTAINER_NAMES:
fmt = "mp3" # ultimate fallback
return {"format": fmt}
-12
View File
@@ -1,12 +0,0 @@
FORMAT_INFO = {
'mp3': {'ffmpeg': 'mp3', 'ext': '.mp3', 'audio_only': True},
'm4a': {'ffmpeg': 'mp4', 'ext': '.m4a', 'audio_only': True},
'mp4': {'ffmpeg': 'mp4', 'ext': '.mp4', 'audio_only': False},
'mkv': {'ffmpeg': 'matroska', 'ext': '.mkv', 'audio_only': False},
'matroska': {'ffmpeg': 'matroska', 'ext': '.mkv', 'audio_only': False},
'ogg': {'ffmpeg': 'ogg', 'ext': '.ogg', 'audio_only': True},
'opus': {'ffmpeg': 'ogg', 'ext': '.opus', 'audio_only': True},
'flac': {'ffmpeg': 'flac', 'ext': '.flac', 'audio_only': True},
'wav': {'ffmpeg': 'wav', 'ext': '.wav', 'audio_only': True},
'aac': {'ffmpeg': 'adts', 'ext': '.aac', 'audio_only': True},
}
+3 -15
View File
@@ -70,21 +70,9 @@ def run_split_task(task_id: str, tracklist: List[TracklistEntry], options: Dict[
from audio_splitter.core import split_audio
# Detect attached picture and extract if applicable
cover_image_path = None
if not args.drop_video:
from audio_splitter.ffmpeg import is_attached_picture, extract_cover_image, get_stream_info
input_path_str = str(input_path)
stream_info = get_stream_info(input_path_str)
if stream_info.get('has_video') and is_attached_picture(input_path_str):
cover_image_path = output_dir / 'cover.png'
if extract_cover_image(input_path_str, str(cover_image_path)):
print(f"Extracted cover image for task {task_id}")
else:
cover_image_path = None
# Attach cover image to args
args.cover_image = str(cover_image_path) if cover_image_path else None
# Cover image handling is now done internally by split_audio()
# via the handler system (opustags for Opus, ffmpeg for others)
args.cover_image = None
task_manager.update_task_with_progress(
task_id, progress=10, message="Starting split..."
+4 -54
View File
@@ -18,6 +18,8 @@ import { useUploadStore } from '../stores/uploadStore'
import { useValidationStore } from '../stores/validationStore'
import { getFormats } from '../api/client'
import { SplitOptions } from '../types'
import { useFilterCodecs } from '../hooks/useFilterCodecs'
import { useFormatValidation } from '../hooks/useFormatValidation'
// ------------------------------------------------------------------------------
// Section component (collapsible)
@@ -91,64 +93,12 @@ export const OptionsPanel: React.FC = () => {
// --------------------------------------------------------------
// Filter codec options based on selected container
// --------------------------------------------------------------
const filteredAudioCodecs = useMemo(() => {
const entry = compatibility?.[options.container || '']
if (!entry) return codecs.filter(c => c.supports_transcoding)
const audioList = entry.audio
if (audioList === null) return codecs.filter(c => c.supports_transcoding)
return codecs.filter(c => c.supports_transcoding && audioList.includes(c.name))
}, [compatibility, options.container, codecs])
const filteredVideoCodecs = useMemo(() => {
const entry = compatibility?.[options.container || '']
if (!entry) return codecs.filter(c => c.supports_video)
const videoList = entry.video
if (videoList === null) return codecs.filter(c => c.supports_video)
return codecs.filter(c => c.supports_video && videoList.includes(c.name))
}, [compatibility, options.container, codecs])
const { filteredAudioCodecs, filteredVideoCodecs } = useFilterCodecs(compatibility, options.container, codecs)
// --------------------------------------------------------------
// Validate compatibility
// --------------------------------------------------------------
useEffect(() => {
const selectedContainer = containers.find(c => c.name === options.container)
if (selectedContainer?.audio_only && hasVideo && !options.drop_video) {
setFormatError(
`Container '${options.container}' does not support video streams. Please enable "Drop video" or choose a container that supports video.`
)
return
}
// Check audio codec compatibility
if (options.audio_codec && options.audio_codec !== 'copy' && options.container) {
const entry = compatibility?.[options.container]
if (entry) {
const audioList = entry.audio
if (audioList !== null && !audioList.includes(options.audio_codec)) {
setFormatError(
`Audio codec '${options.audio_codec}' is not supported by container '${options.container}'.`
)
return
}
}
}
// Check video codec compatibility
if (options.video_codec && options.video_codec !== 'copy' && options.container && hasVideo && !options.drop_video) {
const entry = compatibility?.[options.container]
if (entry) {
const videoList = entry.video
if (videoList !== null && !videoList.includes(options.video_codec)) {
setFormatError(
`Video codec '${options.video_codec}' is not supported by container '${options.container}'.`
)
return
}
}
}
setFormatError(null)
}, [options.container, options.audio_codec, options.video_codec, options.drop_video, hasVideo, compatibility, containers, setFormatError])
useFormatValidation(options, hasVideo, compatibility, containers, setFormatError)
// --------------------------------------------------------------
// Handlers
+31
View File
@@ -0,0 +1,31 @@
import { useMemo } from 'react'
import { CodecInfo } from '../types'
export interface CompatibilityEntry {
audio: string[] | null
video: string[] | null
}
export function useFilterCodecs(
compatibility: Record<string, CompatibilityEntry>,
container: string | null,
codecs: CodecInfo[]
) {
const filteredAudioCodecs = useMemo(() => {
const entry = compatibility?.[container || '']
if (!entry) return codecs.filter(c => c.codec_type === 'audio')
const audioList = entry.audio
if (audioList === null) return codecs.filter(c => c.codec_type === 'audio')
return codecs.filter(c => c.codec_type === 'audio' && audioList.includes(c.name))
}, [compatibility, container, codecs])
const filteredVideoCodecs = useMemo(() => {
const entry = compatibility?.[container || '']
if (!entry) return codecs.filter(c => c.codec_type === 'video')
const videoList = entry.video
if (videoList === null) return codecs.filter(c => c.codec_type === 'video')
return codecs.filter(c => c.codec_type === 'video' && videoList.includes(c.name))
}, [compatibility, container, codecs])
return { filteredAudioCodecs, filteredVideoCodecs }
}
@@ -0,0 +1,56 @@
import { useEffect } from 'react'
import { ContainerInfo } from '../types'
import { CompatibilityEntry } from './useFilterCodecs'
export function useFormatValidation(
options: {
container: string | null
audio_codec: string
video_codec: string
drop_video: boolean
},
hasVideo: boolean,
compatibility: Record<string, CompatibilityEntry>,
containers: ContainerInfo[],
setFormatError: (error: string | null) => void
) {
useEffect(() => {
const selectedContainer = containers.find(c => c.name === options.container)
if (!selectedContainer?.supports_video && hasVideo && !options.drop_video) {
setFormatError(
`Container '${options.container}' does not support video streams. Please enable "Drop video" or choose a container that supports video.`
)
return
}
// Check audio codec compatibility
if (options.audio_codec && options.audio_codec !== 'copy' && options.container) {
const entry = compatibility?.[options.container]
if (entry) {
const audioList = entry.audio
if (audioList !== null && !audioList.includes(options.audio_codec)) {
setFormatError(
`Audio codec '${options.audio_codec}' is not supported by container '${options.container}'.`
)
return
}
}
}
// Check video codec compatibility
if (options.video_codec && options.video_codec !== 'copy' && options.container && hasVideo && !options.drop_video) {
const entry = compatibility?.[options.container]
if (entry) {
const videoList = entry.video
if (videoList !== null && !videoList.includes(options.video_codec)) {
setFormatError(
`Video codec '${options.video_codec}' is not supported by container '${options.container}'.`
)
return
}
}
}
setFormatError(null)
}, [options.container, options.audio_codec, options.video_codec, options.drop_video, hasVideo, compatibility, containers, setFormatError])
}
+1 -3
View File
@@ -48,7 +48,6 @@ export interface ContainerInfo {
name: string
ffmpeg: string
extension: string
audio_only: boolean
supports_video: boolean
supports_subs: boolean
}
@@ -58,8 +57,7 @@ export interface CodecInfo {
ffmpeg: string
recommended_container: string
recommended_extension: string
supports_transcoding: boolean
supports_video: boolean
codec_type: 'audio' | 'video' | 'subtitle'
}
export interface CompatibilityEntry {