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. Codec != Container != File Extension.
Example: Opus (codec) → Ogg (container) → .opus (extension) Example: Opus (codec) → Ogg (container) → .opus (extension)
""" """
from typing import Dict, Optional
# ------------------------------------------------------------------------------ # ------------------------------------------------------------------------------
# Container information # Container information
# ------------------------------------------------------------------------------ # ------------------------------------------------------------------------------
CONTAINER_INFO = [ CONTAINER_INFO = [
# Audio-only containers # Audio-only containers
{'name': 'mp3', 'ffmpeg': 'mp3', 'extension': '.mp3', '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', 'audio_only': True, 'supports_video': False, 'supports_subs': False}, {'name': 'm4a', 'ffmpeg': 'mp4', 'extension': '.m4a', 'supports_video': False, 'supports_subs': False},
{'name': 'flac', 'ffmpeg': 'flac', 'extension': '.flac', 'audio_only': True, 'supports_video': False, 'supports_subs': False}, {'name': 'flac', 'ffmpeg': 'flac', 'extension': '.flac', 'supports_video': False, 'supports_subs': False},
{'name': 'wav', 'ffmpeg': 'wav', 'extension': '.wav', 'audio_only': True, 'supports_video': False, 'supports_subs': False}, {'name': 'wav', 'ffmpeg': 'wav', 'extension': '.wav', 'supports_video': False, 'supports_subs': False},
{'name': 'aac', 'ffmpeg': 'adts', 'extension': '.aac', 'audio_only': True, 'supports_video': False, 'supports_subs': False}, {'name': 'aac', 'ffmpeg': 'adts', 'extension': '.aac', 'supports_video': False, 'supports_subs': False},
# Containers that support video and subtitles # Containers that support video and subtitles
{'name': 'mp4', 'ffmpeg': 'mp4', 'extension': '.mp4', 'audio_only': False, 'supports_video': True, 'supports_subs': True}, {'name': 'mp4', 'ffmpeg': 'mp4', 'extension': '.mp4', 'supports_video': True, 'supports_subs': True},
{'name': 'mkv', 'ffmpeg': 'matroska', 'extension': '.mkv', 'audio_only': False, 'supports_video': True, 'supports_subs': True}, {'name': 'mkv', 'ffmpeg': 'matroska', 'extension': '.mkv', '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', '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', 'supports_video': True, 'supports_subs': False},
{'name': 'webm', 'ffmpeg': 'webm', 'extension': '.webm', 'audio_only': False, 'supports_video': True, 'supports_subs': False},
] ]
# Legacy FORMAT_INFO for backward compatibility # Container name lookup set for fast membership checks
FORMAT_INFO = { CONTAINER_NAMES = {container['name'] for container in CONTAINER_INFO}
container['name']: {
'ffmpeg': container['ffmpeg'],
'ext': container['extension'], def get_container_info(container_name: str) -> Optional[Dict]:
'audio_only': container['audio_only'], """Return container info dict or None if not found."""
} for container in CONTAINER_INFO:
for container in CONTAINER_INFO if container['name'] == container_name:
} return container
return None
# ------------------------------------------------------------------------------ # ------------------------------------------------------------------------------
# Codec information # Codec information
# ------------------------------------------------------------------------------ # ------------------------------------------------------------------------------
CODEC_INFO = [ CODEC_INFO = [
# Audio codecs # Audio codecs
{'name': 'opus', 'ffmpeg': 'libopus', 'recommended_container': 'ogg', 'recommended_extension': '.opus', 'supports_transcoding': True, 'supports_video': False}, {'name': 'opus', 'ffmpeg': 'libopus', 'recommended_container': 'ogg', 'recommended_extension': '.opus', 'codec_type': 'audio'},
{'name': 'vorbis', 'ffmpeg': 'libvorbis', 'recommended_container': 'ogg', 'recommended_extension': '.ogg', 'supports_transcoding': True, 'supports_video': False}, {'name': 'vorbis', 'ffmpeg': 'libvorbis', 'recommended_container': 'ogg', 'recommended_extension': '.ogg', 'codec_type': 'audio'},
{'name': 'flac', 'ffmpeg': 'flac', 'recommended_container': 'flac', 'recommended_extension': '.flac', 'supports_transcoding': True, 'supports_video': False}, {'name': 'flac', 'ffmpeg': 'flac', 'recommended_container': 'flac', 'recommended_extension': '.flac', 'codec_type': 'audio'},
{'name': 'aac', 'ffmpeg': 'aac', 'recommended_container': 'mp4', 'recommended_extension': '.m4a', 'supports_transcoding': True, 'supports_video': False}, {'name': 'aac', 'ffmpeg': 'aac', 'recommended_container': 'mp4', 'recommended_extension': '.m4a', 'codec_type': 'audio'},
{'name': 'mp3', 'ffmpeg': 'libmp3lame', 'recommended_container': 'mp3', 'recommended_extension': '.mp3', 'supports_transcoding': True, 'supports_video': False}, {'name': 'mp3', 'ffmpeg': 'libmp3lame', 'recommended_container': 'mp3', 'recommended_extension': '.mp3', 'codec_type': 'audio'},
{'name': 'alac', 'ffmpeg': 'alac', 'recommended_container': 'mp4', 'recommended_extension': '.m4a', 'supports_transcoding': True, 'supports_video': False}, {'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', 'supports_transcoding': True, 'supports_video': False}, {'name': 'pcm_s16le', 'ffmpeg': 'pcm_s16le', 'recommended_container': 'wav', 'recommended_extension': '.wav', 'codec_type': 'audio'},
# Video codecs # Video codecs (including cover images)
{'name': 'theora', 'ffmpeg': 'libtheora', 'recommended_container': 'ogg', 'recommended_extension': '.ogv', 'supports_transcoding': True, 'supports_video': True}, {'name': 'theora', 'ffmpeg': 'libtheora', 'recommended_container': 'ogg', 'recommended_extension': '.ogv', 'codec_type': 'video'},
{'name': 'vp8', 'ffmpeg': 'libvpx', 'recommended_container': 'webm', 'recommended_extension': '.webm', 'supports_transcoding': True, 'supports_video': True}, {'name': 'vp8', 'ffmpeg': 'libvpx', 'recommended_container': 'webm', 'recommended_extension': '.webm', 'codec_type': 'video'},
{'name': 'vp9', 'ffmpeg': 'libvpx-vp9', 'recommended_container': 'webm', 'recommended_extension': '.webm', 'supports_transcoding': True, 'supports_video': True}, {'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', 'supports_transcoding': True, 'supports_video': True}, {'name': 'av1', 'ffmpeg': 'libaom-av1', 'recommended_container': 'mkv', 'recommended_extension': '.mkv', 'codec_type': 'video'},
{'name': 'h264', 'ffmpeg': 'libx264', 'recommended_container': 'mp4', 'recommended_extension': '.mp4', 'supports_transcoding': True, 'supports_video': True}, {'name': 'h264', 'ffmpeg': 'libx264', 'recommended_container': 'mp4', 'recommended_extension': '.mp4', 'codec_type': 'video'},
{'name': 'h265', 'ffmpeg': 'libx265', 'recommended_container': 'mp4', 'recommended_extension': '.mp4', 'supports_transcoding': True, 'supports_video': True}, {'name': 'h265', 'ffmpeg': 'libx265', 'recommended_container': 'mp4', 'recommended_extension': '.mp4', 'codec_type': 'video'},
{'name': 'png', 'ffmpeg': 'png', 'recommended_container': 'ogg', 'recommended_extension': '.png', 'supports_transcoding': False, 'supports_video': True}, {'name': 'png', 'ffmpeg': 'png', 'recommended_container': 'ogg', 'recommended_extension': '.png', 'codec_type': 'video'},
{'name': 'mjpeg', 'ffmpeg': 'mjpeg', 'recommended_container': 'ogg', 'recommended_extension': '.jpg', 'supports_transcoding': False, 'supports_video': True}, {'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 # 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 # Map codec name → FFmpeg encoder name
CODEC_NAME_TO_FFMPEG = {codec['name']: codec['ffmpeg'] for codec in CODEC_INFO} 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 # 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. # Use None to indicate that any video codec is supported.
# Empty list means no video support (audio-only container). # Empty list means no video support (audio-only container).
COMPATIBILITY_MATRIX = { COMPATIBILITY_MATRIX = {
'mp3': {'audio': ['mp3'], 'video': []}, 'mp3': {'audio': ['mp3'], 'video': ['png', 'mjpeg']},
'm4a': {'audio': ['aac', 'alac', 'opus', 'flac'], 'video': []}, 'm4a': {'audio': ['aac', 'alac', 'opus', 'flac'], 'video': []},
'mp4': {'audio': ['aac', 'alac', 'opus', 'flac', 'mp3'], 'video': ['h264', 'h265', 'vp9', 'av1']}, 'mp4': {'audio': ['aac', 'alac', 'opus', 'flac', 'mp3'], 'video': ['h264', 'h265', 'vp9', 'av1']},
'mkv': {'audio': None, 'video': None}, 'mkv': {'audio': None, 'video': None},
'matroska': {'audio': None, 'video': None}, 'ogg': {'audio': ['opus', 'vorbis', 'flac'], 'video': ['theora', 'vp8', 'png', 'mjpeg']},
'ogg': {'audio': ['opus', 'vorbis', 'flac'], 'video': ['theora', 'vp8']},
'webm': {'audio': ['opus', 'vorbis'], 'video': ['vp8', 'vp9', 'av1']}, 'webm': {'audio': ['opus', 'vorbis'], 'video': ['vp8', 'vp9', 'av1']},
'flac': {'audio': ['flac'], 'video': []}, 'flac': {'audio': ['flac'], 'video': ['png', 'mjpeg']},
'wav': {'audio': ['pcm_s16le'], 'video': []}, 'wav': {'audio': ['pcm_s16le'], 'video': []},
'aac': {'audio': ['aac'], '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 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 typing import List, Dict, Any
from .constants import ( from .constants import (
FORMAT_INFO, CONTAINER_NAMES,
CODEC_TO_EXTENSION_MAP, CODEC_TO_EXTENSION_MAP,
CODEC_NAME_TO_FFMPEG, # <-- Add this CODEC_NAME_TO_FFMPEG,
) )
from .ffmpeg import ( from .ffmpeg import (
get_audio_duration, get_audio_duration,
@@ -23,6 +23,7 @@ from .formats import (
determine_output_format, determine_output_format,
validate_format_compatibility, validate_format_compatibility,
) )
from .handlers import get_handler, needs_drop_video
from .timestamp import parse_track_timestamps, resolve_end_times from .timestamp import parse_track_timestamps, resolve_end_times
from .filename import build_filename from .filename import build_filename
from .metadata import build_metadata_dict 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: if output_audio_codec in CODEC_TO_EXTENSION_MAP:
extension = CODEC_TO_EXTENSION_MAP[output_audio_codec] extension = CODEC_TO_EXTENSION_MAP[output_audio_codec]
else: 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}") print(f"Output extension: {extension}")
@@ -162,6 +164,14 @@ def split_audio(input_file: str, output_directory: str, tracks: List[Dict[str, A
else: else:
cover_image_path = None 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 # 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) subtitle_enc = CODEC_NAME_TO_FFMPEG.get(subtitle_enc, subtitle_enc)
# Then build command with these mapped encoders # 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( cmd = build_ffmpeg_command(
input_file=input_file, input_file=input_file,
start_seconds=start_seconds, 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, video_codec=video_enc,
subtitle_codec=subtitle_enc, subtitle_codec=subtitle_enc,
metadata=metadata, metadata=metadata,
cover_image_path=cover_image_path, cover_image_path=ffmpeg_cover_path,
video_quality=getattr(args, 'video_quality', None), video_quality=getattr(args, 'video_quality', None),
drop_video=args.drop_video, drop_video=effective_drop_video,
drop_subs=args.drop_subs, 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) print(result.stderr)
else: else:
print(f" -> Saved to: {output_path}") 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 # 9. Delete original file if requested
+124 -90
View File
@@ -5,7 +5,7 @@ import json
import subprocess import subprocess
from typing import Dict, List, Optional, Tuple from typing import Dict, List, Optional, Tuple
from .constants import FORMAT_INFO from .constants import CONTAINER_INFO
from .utils import format_time 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]: 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: Args:
input_file: Path to the media file. 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 format_name = result.stdout.strip().split(',')[0] # take first if multiple
if not format_name: if not format_name:
return None return None
# Normalize common aliases to names used in FORMAT_INFO # Normalize common aliases to names used in CONTAINER_INFO
mapping = { mapping = {
'mpeg': 'mp3', 'mpeg': 'mp3',
'mp2': 'mp3', 'mp2': 'mp3',
@@ -187,8 +187,8 @@ def get_container_format(input_file: str) -> Optional[str]:
'm4a': 'mp4', 'm4a': 'mp4',
'mov': 'mp4', 'mov': 'mp4',
'3gp': 'mp4', '3gp': 'mp4',
'matroska': 'matroska', 'matroska': 'mkv',
'webm': 'matroska', 'webm': 'mkv',
'ogg': 'ogg', 'ogg': 'ogg',
'flac': 'flac', 'flac': 'flac',
'wav': 'wav', 'wav': 'wav',
@@ -269,6 +269,112 @@ def extract_cover_image(input_file: str, output_path: str) -> bool:
return True 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( def build_ffmpeg_command(
input_file: str, input_file: str,
start_seconds: int, start_seconds: int,
@@ -309,93 +415,21 @@ def build_ffmpeg_command(
""" """
cmd = ['ffmpeg'] 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: if cover_image_path:
cmd.extend(['-i', cover_image_path]) _build_cover_image_mapping(cmd, audio_codec, video_codec, video_quality)
# 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])
else: 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 _build_format(cmd, format_opt)
if audio_codec and audio_codec != 'copy': _build_output(cmd, output_path)
cmd.extend(['-c:a', audio_codec])
else:
cmd.extend(['-c:a', 'copy'])
# 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 return cmd
+24 -14
View File
@@ -4,7 +4,7 @@
from typing import Dict, Optional from typing import Dict, Optional
from .constants import ( from .constants import (
FORMAT_INFO, CONTAINER_NAMES,
CONTAINER_INFO, CONTAINER_INFO,
CODEC_TO_CONTAINER_MAP, CODEC_TO_CONTAINER_MAP,
COMPATIBILITY_MATRIX, 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. If the codec is not found, it falls back to the container.
Args: 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(). codec: Audio codec name (e.g., 'opus', 'aac', 'mp3') as returned by get_audio_codec().
Returns: Returns:
@@ -33,10 +33,10 @@ def determine_default_format(container: Optional[str], codec: Optional[str]) ->
if codec and codec in CODEC_TO_CONTAINER_MAP: if codec and codec in CODEC_TO_CONTAINER_MAP:
fmt = CODEC_TO_CONTAINER_MAP[codec] fmt = CODEC_TO_CONTAINER_MAP[codec]
if fmt in FORMAT_INFO: if fmt in CONTAINER_NAMES:
return fmt return fmt
if container in FORMAT_INFO: if container in CONTAINER_NAMES:
return container return container
return None return None
@@ -47,7 +47,7 @@ def determine_output_format(stream_info: Dict, user_format: Optional[str],
""" """
Decide which container format to use. 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. 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: If detection fails or format is not supported, fallback to:
- MKV if video/subtitles exist - MKV if video/subtitles exist
@@ -55,6 +55,10 @@ def determine_output_format(stream_info: Dict, user_format: Optional[str],
- MP4 (M4A) otherwise - MP4 (M4A) otherwise
""" """
if user_format: 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 return user_format
if input_file: if input_file:
@@ -63,14 +67,14 @@ def determine_output_format(stream_info: Dict, user_format: Optional[str],
container = get_container_format(input_file) container = get_container_format(input_file)
audio_codec = get_audio_codec(input_file) audio_codec = get_audio_codec(input_file)
fmt = determine_default_format(container, audio_codec) fmt = determine_default_format(container, audio_codec)
if fmt in FORMAT_INFO: if fmt in CONTAINER_NAMES:
return fmt return fmt
except Exception: except Exception:
pass pass
# Fallback # Fallback
if stream_info.get('has_video') or stream_info.get('has_subtitle'): if stream_info.get('has_video') or stream_info.get('has_subtitle'):
return 'matroska' return 'mkv'
audio_codec = stream_info.get('audio_codec', '') audio_codec = stream_info.get('audio_codec', '')
if audio_codec == 'mp3': if audio_codec == 'mp3':
return 'mp3' return 'mp3'
@@ -93,23 +97,25 @@ def validate_format_compatibility(
Raises: Raises:
ValueError: If the combination is incompatible. 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: if not info:
print(f"Warning: Unknown container '{container}'. Proceeding, but may fail.") print(f"Warning: Unknown container '{container}'. Proceeding, but may fail.")
return return
# Check if container is audio-only and video is present (unless dropped) # 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( raise ValueError(
f"Container '{container}' does not support video streams. " f"Container '{container}' does not support video streams. "
"Please use --drop-video or choose a container that supports video (e.g., MKV, MP4)." "Please use --drop-video or choose a container that supports video (e.g., MKV, MP4)."
) )
# Check if audio codec is supported # Determine actual audio codec for validation
if audio_codec and audio_codec != 'copy': # When 'copy' is used, we must still validate the input codec against the container
if not is_audio_codec_supported(container, audio_codec): 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( 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." 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: if video_codec is None and input_file:
from .ffmpeg import get_video_codec from .ffmpeg import get_video_codec
video_codec = get_video_codec(input_file) 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): if not is_video_codec_supported(container, video_codec):
raise ValueError( raise ValueError(
f"Container '{container}' does not support video codec '{video_codec}'. " 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 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 && \ RUN apt-get update && \
apt-get install -y --no-install-recommends \ apt-get install -y --no-install-recommends \
ffmpeg \ ffmpeg \
opustags \
ca-certificates \ ca-certificates \
gosu \ gosu \
&& \ && \
+2 -2
View File
@@ -9,7 +9,7 @@ from backend.services.task_manager import task_manager
# Import core functions # Import core functions
from audio_splitter.ffmpeg import get_stream_info, get_container_format, get_audio_codec 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.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 from audio_splitter.defaults import DEFAULT_FORMAT
router = APIRouter(prefix="/api", tags=["info"]) 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 # Fallback if detection fails or format is unsupported
if fmt is None: if fmt is None:
fmt = DEFAULT_FORMAT fmt = DEFAULT_FORMAT
if fmt not in FORMAT_INFO: if fmt not in CONTAINER_NAMES:
fmt = "mp3" # ultimate fallback fmt = "mp3" # ultimate fallback
return {"format": fmt} 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 from audio_splitter.core import split_audio
# Detect attached picture and extract if applicable # Cover image handling is now done internally by split_audio()
cover_image_path = None # via the handler system (opustags for Opus, ffmpeg for others)
if not args.drop_video: args.cover_image = None
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
task_manager.update_task_with_progress( task_manager.update_task_with_progress(
task_id, progress=10, message="Starting split..." 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 { useValidationStore } from '../stores/validationStore'
import { getFormats } from '../api/client' import { getFormats } from '../api/client'
import { SplitOptions } from '../types' import { SplitOptions } from '../types'
import { useFilterCodecs } from '../hooks/useFilterCodecs'
import { useFormatValidation } from '../hooks/useFormatValidation'
// ------------------------------------------------------------------------------ // ------------------------------------------------------------------------------
// Section component (collapsible) // Section component (collapsible)
@@ -91,64 +93,12 @@ export const OptionsPanel: React.FC = () => {
// -------------------------------------------------------------- // --------------------------------------------------------------
// Filter codec options based on selected container // Filter codec options based on selected container
// -------------------------------------------------------------- // --------------------------------------------------------------
const filteredAudioCodecs = useMemo(() => { const { filteredAudioCodecs, filteredVideoCodecs } = useFilterCodecs(compatibility, options.container, codecs)
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])
// -------------------------------------------------------------- // --------------------------------------------------------------
// Validate compatibility // Validate compatibility
// -------------------------------------------------------------- // --------------------------------------------------------------
useEffect(() => { useFormatValidation(options, hasVideo, compatibility, containers, setFormatError)
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])
// -------------------------------------------------------------- // --------------------------------------------------------------
// Handlers // 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 name: string
ffmpeg: string ffmpeg: string
extension: string extension: string
audio_only: boolean
supports_video: boolean supports_video: boolean
supports_subs: boolean supports_subs: boolean
} }
@@ -58,8 +57,7 @@ export interface CodecInfo {
ffmpeg: string ffmpeg: string
recommended_container: string recommended_container: string
recommended_extension: string recommended_extension: string
supports_transcoding: boolean codec_type: 'audio' | 'video' | 'subtitle'
supports_video: boolean
} }
export interface CompatibilityEntry { export interface CompatibilityEntry {