232 lines
9.5 KiB
Python
232 lines
9.5 KiB
Python
# audio_splitter/core.py
|
||
"""Core splitting logic – orchestrates the entire split process."""
|
||
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
from typing import List, Dict, Any
|
||
|
||
from .constants import FORMAT_INFO, CODEC_TO_EXTENSION_MAP
|
||
from .ffmpeg import (
|
||
get_audio_duration,
|
||
get_stream_info,
|
||
get_metadata,
|
||
build_ffmpeg_command,
|
||
is_attached_picture,
|
||
extract_cover_image,
|
||
)
|
||
from .formats import (
|
||
determine_output_format,
|
||
validate_format_compatibility,
|
||
)
|
||
from .timestamp import parse_track_timestamps, resolve_end_times
|
||
from .filename import build_filename
|
||
from .metadata import build_metadata_dict
|
||
from .utils import format_time
|
||
|
||
|
||
def split_audio(input_file: str, output_directory: str, tracks: List[Dict[str, Any]], args) -> None:
|
||
"""
|
||
Main orchestration function: split the audio file into tracks.
|
||
|
||
Args:
|
||
input_file: Path to the input media file.
|
||
output_directory: Directory where output files will be saved.
|
||
tracks: List of dicts, each containing parsed fields (ts, tn, an, ...).
|
||
args: Parsed command‑line arguments (namespace) with attributes:
|
||
- container: output container name (or None for auto-detect)
|
||
- audio_codec: audio codec (copy or encoder)
|
||
- video_codec: video codec (copy or encoder)
|
||
- subtitle_codec: subtitle codec (copy or encoder)
|
||
- video_quality: integer or None
|
||
- drop_video: bool
|
||
- drop_subs: bool
|
||
- number_tracks: bool
|
||
- output_template: str
|
||
- replace_bad_chars: bool
|
||
- replacement_char: str
|
||
- bad_chars: str
|
||
- skip_existing: bool
|
||
- album: str or None
|
||
- comment: str or None
|
||
- no_comment: bool
|
||
- comment_stream: int or None
|
||
- merge_comments: bool
|
||
- comment_separator: str
|
||
- delete_original: bool
|
||
- cover_image: str or None (optional, set by web backend)
|
||
|
||
Raises:
|
||
RuntimeError: If no audio stream is found.
|
||
ValueError: If compatibility validation fails.
|
||
"""
|
||
# --------------------------------------------------------------------------
|
||
# 1. Setup
|
||
# --------------------------------------------------------------------------
|
||
os.makedirs(output_directory, exist_ok=True)
|
||
|
||
total_duration = get_audio_duration(input_file)
|
||
stream_info = get_stream_info(input_file)
|
||
|
||
print(f"Detected streams: audio={stream_info['has_audio']}, "
|
||
f"video={stream_info['has_video']}, subs={stream_info['has_subtitle']}")
|
||
print(f"Audio codec: {stream_info.get('audio_codec', 'unknown')}")
|
||
|
||
if not stream_info['has_audio']:
|
||
raise RuntimeError("No audio stream found in input file.")
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 2. Determine output container
|
||
# --------------------------------------------------------------------------
|
||
user_container = getattr(args, 'container', None)
|
||
output_container = determine_output_format(
|
||
stream_info,
|
||
user_format=user_container,
|
||
transcode_audio=args.audio_codec if args.audio_codec != 'copy' else None,
|
||
input_file=input_file
|
||
)
|
||
print(f"Output container: {output_container}")
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 3. Validate compatibility
|
||
# --------------------------------------------------------------------------
|
||
try:
|
||
validate_format_compatibility(
|
||
container=output_container,
|
||
stream_info=stream_info,
|
||
drop_video=args.drop_video,
|
||
drop_subs=args.drop_subs,
|
||
input_file=input_file,
|
||
audio_codec=args.audio_codec,
|
||
video_codec=args.video_codec,
|
||
)
|
||
except ValueError as e:
|
||
raise RuntimeError(f"Compatibility error: {e}")
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 4. Determine output audio codec and extension
|
||
# --------------------------------------------------------------------------
|
||
# Determine the audio codec that will be used in the output
|
||
if args.audio_codec != 'copy':
|
||
output_audio_codec = args.audio_codec
|
||
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:
|
||
extension = CODEC_TO_EXTENSION_MAP[output_audio_codec]
|
||
else:
|
||
extension = FORMAT_INFO.get(output_container, {}).get('ext', '.mkv')
|
||
|
||
print(f"Output extension: {extension}")
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 5. Parse timestamps
|
||
# --------------------------------------------------------------------------
|
||
track_times = parse_track_timestamps(tracks)
|
||
resolved_times = resolve_end_times(track_times, total_duration)
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 6. Fetch original metadata (for fallbacks)
|
||
# --------------------------------------------------------------------------
|
||
input_metadata = get_metadata(input_file)
|
||
original_album = input_metadata.get('album')
|
||
original_title = input_metadata.get('title')
|
||
original_comments = input_metadata.get('comments', [])
|
||
|
||
if original_album:
|
||
print(f"Original album: '{original_album}'")
|
||
if original_title:
|
||
print(f"Original title: '{original_title}'")
|
||
if original_comments:
|
||
print(f"Found {len(original_comments)} comment(s) in input file.")
|
||
for idx, comment in original_comments:
|
||
print(f" Stream {idx}: '{comment}'")
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 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'):
|
||
if is_attached_picture(input_file):
|
||
cover_image_path = os.path.join(output_directory, 'cover.png')
|
||
if extract_cover_image(input_file, cover_image_path):
|
||
print("Extracted cover image for all tracks.")
|
||
else:
|
||
cover_image_path = None
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 8. Process each track
|
||
# --------------------------------------------------------------------------
|
||
for idx, track in enumerate(tracks, start=1):
|
||
start_seconds, end_seconds = resolved_times[idx - 1]
|
||
duration_seconds = end_seconds - start_seconds
|
||
|
||
if duration_seconds <= 0:
|
||
print(f"Warning: Track {idx} has zero or negative duration, skipping.")
|
||
continue
|
||
|
||
track_name = track.get('tn', 'Unknown')
|
||
|
||
# ----------------------------------------------------------------------
|
||
# 8a. Build filename
|
||
# ----------------------------------------------------------------------
|
||
clean_filename = build_filename(track, idx, extension, args)
|
||
output_path = os.path.join(output_directory, clean_filename)
|
||
|
||
# ----------------------------------------------------------------------
|
||
# 8b. Handle existing files
|
||
# ----------------------------------------------------------------------
|
||
if args.skip_existing and os.path.exists(output_path):
|
||
print(f"Skipping track {idx}: {output_path} already exists.")
|
||
continue
|
||
|
||
# ----------------------------------------------------------------------
|
||
# 8c. Build metadata
|
||
# ----------------------------------------------------------------------
|
||
metadata = build_metadata_dict(track, idx, input_metadata, args)
|
||
|
||
# ----------------------------------------------------------------------
|
||
# 8d. Build and execute FFmpeg command
|
||
# ----------------------------------------------------------------------
|
||
cmd = build_ffmpeg_command(
|
||
input_file=input_file,
|
||
start_seconds=start_seconds,
|
||
duration_seconds=duration_seconds,
|
||
output_path=output_path,
|
||
stream_info=stream_info,
|
||
format_opt=output_container,
|
||
audio_codec=args.audio_codec,
|
||
video_codec=args.video_codec,
|
||
subtitle_codec=args.subtitle_codec,
|
||
metadata=metadata,
|
||
cover_image_path=cover_image_path,
|
||
video_quality=getattr(args, 'video_quality', None),
|
||
drop_video=args.drop_video,
|
||
drop_subs=args.drop_subs,
|
||
)
|
||
|
||
print(f"Extracting track {idx}: {track_name} "
|
||
f"({format_time(start_seconds)} - {format_time(end_seconds)})")
|
||
|
||
result = subprocess.run(cmd, capture_output=True, text=True, check=False)
|
||
|
||
if result.returncode != 0:
|
||
print(f"ERROR extracting track {idx}:")
|
||
print(result.stderr)
|
||
else:
|
||
print(f" -> Saved to: {output_path}")
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 9. Delete original file if requested
|
||
# --------------------------------------------------------------------------
|
||
if getattr(args, 'delete_original', False):
|
||
try:
|
||
os.remove(input_file)
|
||
print(f"Deleted original file: {input_file}")
|
||
except OSError as e:
|
||
print(f"Warning: Could not delete original file: {e}")
|