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 typing import List, Dict, Any
|
||||||
|
|
||||||
from .constants import (
|
from .constants import (
|
||||||
|
CONTAINER_INFO,
|
||||||
CONTAINER_NAMES,
|
CONTAINER_NAMES,
|
||||||
CODEC_TO_EXTENSION_MAP,
|
CODEC_TO_EXTENSION_MAP,
|
||||||
CODEC_NAME_TO_FFMPEG,
|
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
|
- merge_comments: bool
|
||||||
- comment_separator: str
|
- comment_separator: str
|
||||||
- delete_original: bool
|
- 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:
|
Raises:
|
||||||
RuntimeError: If no audio stream is found.
|
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)
|
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)
|
total_duration = get_audio_duration(input_file)
|
||||||
stream_info = get_stream_info(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:
|
else:
|
||||||
output_audio_codec = stream_info.get('audio_codec', '')
|
output_audio_codec = stream_info.get('audio_codec', '')
|
||||||
|
|
||||||
# Choose extension based on audio codec if possible, otherwise fallback to container default
|
# Choose extension based on the output container, not the audio codec.
|
||||||
if output_audio_codec in CODEC_TO_EXTENSION_MAP:
|
# 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]
|
extension = CODEC_TO_EXTENSION_MAP[output_audio_codec]
|
||||||
else:
|
else:
|
||||||
container_info = next((c for c in CONTAINER_INFO if c['name'] == output_container), None)
|
extension = '.mkv'
|
||||||
extension = container_info['extension'] if container_info else '.mkv'
|
|
||||||
|
|
||||||
print(f"Output extension: {extension}")
|
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)
|
# 7. Handle attached picture (cover art)
|
||||||
# --------------------------------------------------------------------------
|
# --------------------------------------------------------------------------
|
||||||
cover_image_path = getattr(args, 'cover_image', None)
|
user_cover_images = getattr(args, 'cover_image', None)
|
||||||
if cover_image_path and not os.path.exists(cover_image_path):
|
# Resolve per-track cover images: single image reused, or one-per-track
|
||||||
cover_image_path = None
|
track_cover_images = []
|
||||||
|
if user_cover_images:
|
||||||
if not cover_image_path and not args.drop_video and stream_info.get('has_video'):
|
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):
|
if is_attached_picture(input_file):
|
||||||
cover_image_path = os.path.join(output_directory, 'cover.png')
|
cover_image_path = os.path.join(output_directory, 'cover.png')
|
||||||
if extract_cover_image(input_file, cover_image_path):
|
if extract_cover_image(input_file, cover_image_path):
|
||||||
|
track_cover_images.append(cover_image_path)
|
||||||
print("Extracted cover image for all tracks.")
|
print("Extracted cover image for all tracks.")
|
||||||
else:
|
else:
|
||||||
cover_image_path = None
|
track_cover_images = []
|
||||||
|
|
||||||
# Check if we need special handling (e.g., Opus files need opustags)
|
# 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')
|
input_audio_codec = stream_info.get('audio_codec')
|
||||||
cover_handler = get_handler(output_container, input_audio_codec)
|
cover_handler = get_handler(output_container, output_audio_codec)
|
||||||
needs_drop = needs_drop_video(output_container, input_audio_codec)
|
needs_drop = needs_drop_video(output_container, output_audio_codec)
|
||||||
if cover_image_path and needs_drop:
|
if track_cover_images and needs_drop:
|
||||||
print(f"Using special handler for {output_container} + {input_audio_codec}")
|
print(f"Using special handler for {output_container} + {output_audio_codec}")
|
||||||
print("Temporarily dropping video for opustags post-processing.")
|
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
|
# Then build command with these mapped encoders
|
||||||
# Use temporary drop_video for handlers that need it
|
# 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
|
# 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
|
# 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(
|
cmd = build_ffmpeg_command(
|
||||||
input_file=input_file,
|
input_file=input_file,
|
||||||
start_seconds=start_seconds,
|
start_seconds=start_seconds,
|
||||||
@@ -251,11 +279,11 @@ def split_audio(input_file: str, output_directory: str, tracks: List[Dict[str, A
|
|||||||
else:
|
else:
|
||||||
print(f" -> Saved to: {output_path}")
|
print(f" -> Saved to: {output_path}")
|
||||||
# Apply cover image handler if needed
|
# 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...")
|
print(f"Applying cover image via {output_container} handler...")
|
||||||
if not cover_handler(
|
if not cover_handler(
|
||||||
input_file=input_file,
|
input_file=input_file,
|
||||||
cover_image_path=cover_image_path,
|
cover_image_path=per_track_cover,
|
||||||
output_path=output_path,
|
output_path=output_path,
|
||||||
stream_info=stream_info,
|
stream_info=stream_info,
|
||||||
audio_codec=audio_enc,
|
audio_codec=audio_enc,
|
||||||
|
|||||||
@@ -2,6 +2,7 @@
|
|||||||
"""FFmpeg/FFprobe interaction utilities."""
|
"""FFmpeg/FFprobe interaction utilities."""
|
||||||
|
|
||||||
import json
|
import json
|
||||||
|
import os
|
||||||
import subprocess
|
import subprocess
|
||||||
from typing import Dict, List, Optional, Tuple
|
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)
|
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:
|
def is_attached_picture(input_file: str) -> bool:
|
||||||
"""
|
"""
|
||||||
Check if the input file has a video stream that is an attached picture (cover art).
|
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,
|
clean_filename = apply_replacement(raw_filename,
|
||||||
args.bad_chars,
|
args.bad_chars,
|
||||||
args.replacement_char)
|
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:
|
else:
|
||||||
clean_filename = raw_filename
|
clean_filename = raw_filename
|
||||||
# Warn about unsafe characters.
|
# Warn about unsafe characters.
|
||||||
|
|||||||
@@ -32,8 +32,10 @@ from .defaults import (
|
|||||||
DEFAULT_SKIP_EXISTING,
|
DEFAULT_SKIP_EXISTING,
|
||||||
DEFAULT_DELETE_ORIGINAL,
|
DEFAULT_DELETE_ORIGINAL,
|
||||||
)
|
)
|
||||||
|
from .constants import CONTAINER_NAMES
|
||||||
from .core import split_audio
|
from .core import split_audio
|
||||||
from .tracklist import read_tracklist, parse_format
|
from .tracklist import read_tracklist, parse_format
|
||||||
|
from .ffmpeg import has_cover_or_video, validate_cover_images
|
||||||
|
|
||||||
|
|
||||||
def main():
|
def main():
|
||||||
@@ -156,6 +158,16 @@ def main():
|
|||||||
help="Tracklist format (default: %(default)s)"
|
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
|
# Other
|
||||||
parser.add_argument(
|
parser.add_argument(
|
||||||
'--delete-original',
|
'--delete-original',
|
||||||
@@ -185,6 +197,12 @@ def main():
|
|||||||
print(f"Error: Tracklist file not found: {args.tracklist_file}")
|
print(f"Error: Tracklist file not found: {args.tracklist_file}")
|
||||||
sys.exit(1)
|
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
|
# Check FFmpeg
|
||||||
try:
|
try:
|
||||||
subprocess.run(['ffmpeg', '-version'], capture_output=True, check=True)
|
subprocess.run(['ffmpeg', '-version'], capture_output=True, check=True)
|
||||||
@@ -208,6 +226,21 @@ def main():
|
|||||||
|
|
||||||
print(f"Found {len(tracks)} tracks.")
|
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:
|
if args.dry_run:
|
||||||
print("\nParsed tracklist:")
|
print("\nParsed tracklist:")
|
||||||
print("-" * 60)
|
print("-" * 60)
|
||||||
|
|||||||
@@ -73,4 +73,15 @@ def resolve_end_times(track_times, total_duration):
|
|||||||
if end_sec <= start_sec:
|
if end_sec <= start_sec:
|
||||||
raise ValueError(f"Track {idx+1}: end time ({end_sec}) is not after start ({start_sec})")
|
raise ValueError(f"Track {idx+1}: end time ({end_sec}) is not after start ({start_sec})")
|
||||||
resolved.append((start_sec, end_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
|
return resolved
|
||||||
|
|||||||
@@ -3,7 +3,7 @@
|
|||||||
|
|
||||||
from fastapi import APIRouter
|
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"])
|
router = APIRouter(prefix="/api", tags=["formats"])
|
||||||
|
|
||||||
@@ -11,9 +11,10 @@ router = APIRouter(prefix="/api", tags=["formats"])
|
|||||||
@router.get("/formats")
|
@router.get("/formats")
|
||||||
async def 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 {
|
return {
|
||||||
"containers": CONTAINER_INFO,
|
"containers": CONTAINER_INFO,
|
||||||
"codecs": CODEC_INFO,
|
"codecs": CODEC_INFO,
|
||||||
|
"compatibility": COMPATIBILITY_MATRIX,
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -14,7 +14,8 @@ router = APIRouter(prefix="/api", tags=["upload"])
|
|||||||
|
|
||||||
ALLOWED_EXTENSIONS = {
|
ALLOWED_EXTENSIONS = {
|
||||||
".mp3", ".flac", ".wav", ".m4a", ".ogg", ".opus",
|
".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 { uploadFile, getTaskInfo, getRecommendedFormat } from '../api/client' // NEW
|
||||||
import { useTaskStore } from '../stores/taskStore'
|
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 = () => {
|
export const UploadZone: React.FC = () => {
|
||||||
const {
|
const {
|
||||||
@@ -96,7 +96,7 @@ export const UploadZone: React.FC = () => {
|
|||||||
const { getRootProps, getInputProps, isDragActive } = useDropzone({
|
const { getRootProps, getInputProps, isDragActive } = useDropzone({
|
||||||
onDrop,
|
onDrop,
|
||||||
accept: {
|
accept: {
|
||||||
'audio/*': ALLOWED_EXTENSIONS,
|
'media/*': ALLOWED_EXTENSIONS
|
||||||
},
|
},
|
||||||
multiple: false,
|
multiple: false,
|
||||||
disabled: isUploading,
|
disabled: isUploading,
|
||||||
|
|||||||
Reference in New Issue
Block a user