Compare commits
8 Commits
e36c1d69ed
..
dev
| Author | SHA1 | Date | |
|---|---|---|---|
| 2ab85b64ff | |||
| cf0b9f62cb | |||
| 1b0767af90 | |||
| a8b0c538ff | |||
| 3a41bd2920 | |||
| 78ec4dd63a | |||
| 8b67de59a3 | |||
| 25e8c0293e |
+47
-19
@@ -7,6 +7,7 @@ import sys
|
||||
from typing import List, Dict, Any
|
||||
|
||||
from .constants import (
|
||||
CONTAINER_INFO,
|
||||
CONTAINER_NAMES,
|
||||
CODEC_TO_EXTENSION_MAP,
|
||||
CODEC_NAME_TO_FFMPEG,
|
||||
@@ -59,7 +60,7 @@ def split_audio(input_file: str, output_directory: str, tracks: List[Dict[str, A
|
||||
- merge_comments: bool
|
||||
- comment_separator: str
|
||||
- delete_original: bool
|
||||
- cover_image: str or None (optional, set by web backend)
|
||||
- cover_image: List[str] or None (optional, CLI list of cover image paths)
|
||||
|
||||
Raises:
|
||||
RuntimeError: If no audio stream is found.
|
||||
@@ -70,6 +71,17 @@ def split_audio(input_file: str, output_directory: str, tracks: List[Dict[str, A
|
||||
# --------------------------------------------------------------------------
|
||||
os.makedirs(output_directory, exist_ok=True)
|
||||
|
||||
# Check write permission on output directory.
|
||||
try:
|
||||
test_file = os.path.join(output_directory, f'.write_test_{os.getpid()}')
|
||||
with open(test_file, 'w') as f:
|
||||
f.write('test')
|
||||
os.remove(test_file)
|
||||
except PermissionError:
|
||||
raise RuntimeError(
|
||||
f"Error: No write permission for output directory: {output_directory}"
|
||||
)
|
||||
|
||||
total_duration = get_audio_duration(input_file)
|
||||
stream_info = get_stream_info(input_file)
|
||||
|
||||
@@ -117,12 +129,15 @@ def split_audio(input_file: str, output_directory: str, tracks: List[Dict[str, A
|
||||
else:
|
||||
output_audio_codec = stream_info.get('audio_codec', '')
|
||||
|
||||
# Choose extension based on audio codec if possible, otherwise fallback to container default
|
||||
if output_audio_codec in CODEC_TO_EXTENSION_MAP:
|
||||
# Choose extension based on the output container, not the audio codec.
|
||||
# When user specifies --container, the file extension must match the container.
|
||||
container_info = next((c for c in CONTAINER_INFO if c['name'] == output_container), None)
|
||||
if container_info:
|
||||
extension = container_info['extension']
|
||||
elif output_audio_codec in CODEC_TO_EXTENSION_MAP:
|
||||
extension = CODEC_TO_EXTENSION_MAP[output_audio_codec]
|
||||
else:
|
||||
container_info = next((c for c in CONTAINER_INFO if c['name'] == output_container), None)
|
||||
extension = container_info['extension'] if container_info else '.mkv'
|
||||
extension = '.mkv'
|
||||
|
||||
print(f"Output extension: {extension}")
|
||||
|
||||
@@ -152,24 +167,33 @@ def split_audio(input_file: str, output_directory: str, tracks: List[Dict[str, A
|
||||
# --------------------------------------------------------------------------
|
||||
# 7. 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 cover_image_path and not args.drop_video and stream_info.get('has_video'):
|
||||
user_cover_images = getattr(args, 'cover_image', None)
|
||||
# Resolve per-track cover images: single image reused, or one-per-track
|
||||
track_cover_images = []
|
||||
if user_cover_images:
|
||||
for img in user_cover_images:
|
||||
if os.path.exists(img):
|
||||
track_cover_images.append(img)
|
||||
else:
|
||||
print(f"Warning: Cover image not found, skipping: {img}")
|
||||
# Extract cover from input if no user cover and input has video
|
||||
if not track_cover_images 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):
|
||||
track_cover_images.append(cover_image_path)
|
||||
print("Extracted cover image for all tracks.")
|
||||
else:
|
||||
cover_image_path = None
|
||||
track_cover_images = []
|
||||
|
||||
# Check if we need special handling (e.g., Opus files need opustags)
|
||||
# Use output_audio_codec (not input_audio_codec) so the handler is resolved
|
||||
# against the actual output codec (e.g., transcoding aac→opus in ogg).
|
||||
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}")
|
||||
cover_handler = get_handler(output_container, output_audio_codec)
|
||||
needs_drop = needs_drop_video(output_container, output_audio_codec)
|
||||
if track_cover_images and needs_drop:
|
||||
print(f"Using special handler for {output_container} + {output_audio_codec}")
|
||||
print("Temporarily dropping video for opustags post-processing.")
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
@@ -219,10 +243,14 @@ def split_audio(input_file: str, output_directory: str, tracks: List[Dict[str, A
|
||||
|
||||
# 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)
|
||||
effective_drop_video = args.drop_video or (track_cover_images 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
|
||||
per_track_cover = (
|
||||
track_cover_images[idx - 1] if len(track_cover_images) == len(tracks)
|
||||
else (track_cover_images[0] if track_cover_images else None)
|
||||
)
|
||||
ffmpeg_cover_path = None if (per_track_cover and needs_drop) else per_track_cover
|
||||
cmd = build_ffmpeg_command(
|
||||
input_file=input_file,
|
||||
start_seconds=start_seconds,
|
||||
@@ -251,11 +279,11 @@ def split_audio(input_file: str, output_directory: str, tracks: List[Dict[str, A
|
||||
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:
|
||||
if per_track_cover 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,
|
||||
cover_image_path=per_track_cover,
|
||||
output_path=output_path,
|
||||
stream_info=stream_info,
|
||||
audio_codec=audio_enc,
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
"""FFmpeg/FFprobe interaction utilities."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
|
||||
@@ -201,6 +202,86 @@ def get_container_format(input_file: str) -> Optional[str]:
|
||||
return mapping.get(format_name, format_name)
|
||||
|
||||
|
||||
MAX_COVER_IMAGE_SIZE_MB = 4
|
||||
|
||||
|
||||
def has_cover_or_video(input_file: str) -> bool:
|
||||
"""
|
||||
Check if the input file contains a cover image (attached picture) or any video track.
|
||||
|
||||
Returns:
|
||||
True if a cover image or video stream is detected, False otherwise.
|
||||
"""
|
||||
cmd = [
|
||||
'ffprobe', '-v', 'quiet',
|
||||
'-print_format', 'json',
|
||||
'-select_streams', 'v',
|
||||
'-show_entries', 'stream=codec_type,codec_name,disposition',
|
||||
input_file
|
||||
]
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, check=False)
|
||||
if result.returncode != 0:
|
||||
return False
|
||||
|
||||
try:
|
||||
data = json.loads(result.stdout)
|
||||
streams = data.get('streams', [])
|
||||
return len(streams) > 0
|
||||
except (json.JSONDecodeError, KeyError):
|
||||
return False
|
||||
|
||||
|
||||
def validate_cover_images(cover_images: List[str], num_tracks: int) -> None:
|
||||
"""
|
||||
Validate cover image(s) before splitting.
|
||||
|
||||
Checks:
|
||||
- Each file exists
|
||||
- Only PNG and JPEG formats are accepted
|
||||
- Single image OR count matches number of tracks
|
||||
- File size warning for images > 4 MB
|
||||
|
||||
Args:
|
||||
cover_images: List of cover image paths (single image or multiple).
|
||||
num_tracks: Number of tracks in the tracklist.
|
||||
|
||||
Raises:
|
||||
ValueError: If validation fails.
|
||||
"""
|
||||
if len(cover_images) == 0:
|
||||
return
|
||||
|
||||
if len(cover_images) == 1 and num_tracks > 1:
|
||||
# Single image is fine - will be reused for all tracks
|
||||
pass
|
||||
elif len(cover_images) == num_tracks:
|
||||
pass
|
||||
else:
|
||||
raise ValueError(
|
||||
f"Cover image count ({len(cover_images)}) must be 1 or match "
|
||||
f"the number of tracks ({num_tracks})."
|
||||
)
|
||||
|
||||
for img_path in cover_images:
|
||||
if not os.path.exists(img_path):
|
||||
raise ValueError(f"Cover image not found: {img_path}")
|
||||
|
||||
ext = os.path.splitext(img_path)[1].lower()
|
||||
if ext not in ('.png', '.jpg', '.jpeg'):
|
||||
raise ValueError(
|
||||
f"Unsupported cover image format '{ext}' for '{img_path}'. "
|
||||
f"Only PNG and JPEG are supported."
|
||||
)
|
||||
|
||||
size_mb = os.path.getsize(img_path) / (1024 * 1024)
|
||||
if size_mb > MAX_COVER_IMAGE_SIZE_MB:
|
||||
print(
|
||||
f"Warning: Cover image '{img_path}' is {size_mb:.1f} MB "
|
||||
f"(exceeds {MAX_COVER_IMAGE_SIZE_MB} MB limit). "
|
||||
f"Large images may cause issues."
|
||||
)
|
||||
|
||||
|
||||
def is_attached_picture(input_file: str) -> bool:
|
||||
"""
|
||||
Check if the input file has a video stream that is an attached picture (cover art).
|
||||
|
||||
@@ -41,7 +41,18 @@ def build_filename(track, idx, extension, args):
|
||||
clean_filename = apply_replacement(raw_filename,
|
||||
args.bad_chars,
|
||||
args.replacement_char)
|
||||
clean_filename = cleanup_good_chars(clean_filename, args.replacement_char)
|
||||
# Split off the file extension before cleanup so trailing
|
||||
# replacement chars don't leak into the name portion.
|
||||
name_part, _, ext_part = clean_filename.rpartition('.')
|
||||
clean_filename = cleanup_good_chars(name_part, args.replacement_char)
|
||||
# Strip leading/trailing separator characters (replacement char,
|
||||
# hyphen, etc.) that may result from empty fields or bad-char
|
||||
# sequences adjacent to template separators.
|
||||
clean_filename = clean_filename.strip(f'{args.replacement_char}-')
|
||||
# Fallback: if cleanup leaves an empty name, use the track number.
|
||||
if not clean_filename:
|
||||
clean_filename = f"{idx:02d}"
|
||||
clean_filename = f"{clean_filename}.{ext_part}"
|
||||
else:
|
||||
clean_filename = raw_filename
|
||||
# Warn about unsafe characters.
|
||||
|
||||
@@ -32,8 +32,10 @@ from .defaults import (
|
||||
DEFAULT_SKIP_EXISTING,
|
||||
DEFAULT_DELETE_ORIGINAL,
|
||||
)
|
||||
from .constants import CONTAINER_NAMES
|
||||
from .core import split_audio
|
||||
from .tracklist import read_tracklist, parse_format
|
||||
from .ffmpeg import has_cover_or_video, validate_cover_images
|
||||
|
||||
|
||||
def main():
|
||||
@@ -156,6 +158,16 @@ def main():
|
||||
help="Tracklist format (default: %(default)s)"
|
||||
)
|
||||
|
||||
# Cover image
|
||||
parser.add_argument(
|
||||
'--cover-image',
|
||||
nargs='+',
|
||||
metavar='IMAGE',
|
||||
default=None,
|
||||
help="Cover image path(s). Single image applied to all tracks, "
|
||||
"or one per track (must match track count)."
|
||||
)
|
||||
|
||||
# Other
|
||||
parser.add_argument(
|
||||
'--delete-original',
|
||||
@@ -185,6 +197,12 @@ def main():
|
||||
print(f"Error: Tracklist file not found: {args.tracklist_file}")
|
||||
sys.exit(1)
|
||||
|
||||
# Validate container if provided
|
||||
if args.container and args.container not in CONTAINER_NAMES:
|
||||
print(f"Error: Invalid container '{args.container}'. "
|
||||
f"Valid containers: {', '.join(sorted(CONTAINER_NAMES))}.")
|
||||
sys.exit(1)
|
||||
|
||||
# Check FFmpeg
|
||||
try:
|
||||
subprocess.run(['ffmpeg', '-version'], capture_output=True, check=True)
|
||||
@@ -208,6 +226,21 @@ def main():
|
||||
|
||||
print(f"Found {len(tracks)} tracks.")
|
||||
|
||||
# Validate cover images if provided
|
||||
if args.cover_image:
|
||||
# Check if input already has cover image or video track
|
||||
# Allow if --drop-video is specified (user wants to discard existing video)
|
||||
if has_cover_or_video(args.input_file) and not args.drop_video:
|
||||
print("Error: Input file already contains a cover image or video track. "
|
||||
"Remove it first, or use --drop-video to discard video streams.")
|
||||
sys.exit(1)
|
||||
|
||||
try:
|
||||
validate_cover_images(args.cover_image, len(tracks))
|
||||
except ValueError as e:
|
||||
print(f"Error: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
if args.dry_run:
|
||||
print("\nParsed tracklist:")
|
||||
print("-" * 60)
|
||||
|
||||
@@ -73,4 +73,15 @@ def resolve_end_times(track_times, total_duration):
|
||||
if end_sec <= start_sec:
|
||||
raise ValueError(f"Track {idx+1}: end time ({end_sec}) is not after start ({start_sec})")
|
||||
resolved.append((start_sec, end_sec))
|
||||
|
||||
# Warn about overlapping intervals
|
||||
for idx in range(len(resolved) - 1):
|
||||
cur_end = resolved[idx][1]
|
||||
next_start = resolved[idx + 1][0]
|
||||
if cur_end > next_start:
|
||||
print(
|
||||
f"Warning: Track {idx+1} ends at {cur_end}s but Track {idx+2} starts at "
|
||||
f"{next_start}s — timestamps overlap by {cur_end - next_start:.1f}s."
|
||||
)
|
||||
|
||||
return resolved
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
|
||||
from fastapi import APIRouter
|
||||
|
||||
from audio_splitter.constants import CONTAINER_INFO, CODEC_INFO
|
||||
from audio_splitter.constants import CONTAINER_INFO, CODEC_INFO, COMPATIBILITY_MATRIX
|
||||
|
||||
router = APIRouter(prefix="/api", tags=["formats"])
|
||||
|
||||
@@ -11,9 +11,10 @@ router = APIRouter(prefix="/api", tags=["formats"])
|
||||
@router.get("/formats")
|
||||
async def get_formats():
|
||||
"""
|
||||
Return the list of supported containers and codecs.
|
||||
Return the list of supported containers, codecs, and compatibility matrix.
|
||||
"""
|
||||
return {
|
||||
"containers": CONTAINER_INFO,
|
||||
"codecs": CODEC_INFO,
|
||||
"compatibility": COMPATIBILITY_MATRIX,
|
||||
}
|
||||
|
||||
@@ -14,7 +14,8 @@ router = APIRouter(prefix="/api", tags=["upload"])
|
||||
|
||||
ALLOWED_EXTENSIONS = {
|
||||
".mp3", ".flac", ".wav", ".m4a", ".ogg", ".opus",
|
||||
".aac", ".wma", ".aiff", ".alac", ".ac3"
|
||||
".aac", ".wma", ".aiff", ".alac", ".ac3",
|
||||
".mp4", ".mkv", ".webm",
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -7,7 +7,7 @@ import { useOptionsStore } from '../stores/optionsStore' // NEW
|
||||
import { uploadFile, getTaskInfo, getRecommendedFormat } from '../api/client' // NEW
|
||||
import { useTaskStore } from '../stores/taskStore'
|
||||
|
||||
const ALLOWED_EXTENSIONS = ['.mp3', '.flac', '.wav', '.m4a', '.ogg', '.opus', '.aac', '.wma', '.aiff', '.alac', '.ac3']
|
||||
const ALLOWED_EXTENSIONS = ['.mp3', '.flac', '.wav', '.m4a', '.ogg', '.opus', '.aac', '.wma', '.aiff', '.alac', '.ac3', '.mp4', '.mkv', '.webm']
|
||||
|
||||
export const UploadZone: React.FC = () => {
|
||||
const {
|
||||
@@ -96,7 +96,7 @@ export const UploadZone: React.FC = () => {
|
||||
const { getRootProps, getInputProps, isDragActive } = useDropzone({
|
||||
onDrop,
|
||||
accept: {
|
||||
'audio/*': ALLOWED_EXTENSIONS,
|
||||
'media/*': ALLOWED_EXTENSIONS
|
||||
},
|
||||
multiple: false,
|
||||
disabled: isUploading,
|
||||
|
||||
Reference in New Issue
Block a user