From 842e3bf1b0d01f7306d0e7555ae891474a541f55 Mon Sep 17 00:00:00 2001 From: Maxim Vershinin Date: Tue, 28 Jul 2026 16:25:11 +0500 Subject: [PATCH] Initial commit. Basic featues are implemented --- .gitignore | 1 + __init__.py | 3 + constants.py | 20 +++++ core.py | 130 ++++++++++++++++++++++++++++++ ffmpeg.py | 223 +++++++++++++++++++++++++++++++++++++++++++++++++++ filename.py | 53 ++++++++++++ formats.py | 55 +++++++++++++ main.py | 204 ++++++++++++++++++++++++++++++++++++++++++++++ metadata.py | 99 +++++++++++++++++++++++ timestamp.py | 76 ++++++++++++++++++ tracklist.py | 155 +++++++++++++++++++++++++++++++++++ utils.py | 171 +++++++++++++++++++++++++++++++++++++++ 12 files changed, 1190 insertions(+) create mode 100644 .gitignore create mode 100644 __init__.py create mode 100644 constants.py create mode 100644 core.py create mode 100644 ffmpeg.py create mode 100644 filename.py create mode 100644 formats.py create mode 100644 main.py create mode 100644 metadata.py create mode 100644 timestamp.py create mode 100644 tracklist.py create mode 100644 utils.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..bee8a64 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +__pycache__ diff --git a/__init__.py b/__init__.py new file mode 100644 index 0000000..2f17147 --- /dev/null +++ b/__init__.py @@ -0,0 +1,3 @@ +""" +Audio splitter package – split an audio file using a tracklist. +""" diff --git a/constants.py b/constants.py new file mode 100644 index 0000000..f608d35 --- /dev/null +++ b/constants.py @@ -0,0 +1,20 @@ +"""Global constants and default values.""" + +# Mapping from user‑friendly format names to FFmpeg format identifiers, +# file extensions, and whether the container is audio‑only. +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}, +} + +# Default characters to replace when --replace-bad-chars is enabled. +# Includes common punctuation, single quote, and a trailing space. +DEFAULT_BAD_CHARS = r',!@#№$;:%^&?*(){}[]\/<>+=~`\' ' diff --git a/core.py b/core.py new file mode 100644 index 0000000..9839ed3 --- /dev/null +++ b/core.py @@ -0,0 +1,130 @@ +"""Core logic: orchestrates the splitting process.""" + +import os +import subprocess + +from .constants import FORMAT_INFO +from .ffmpeg import get_audio_duration, get_stream_info, get_metadata, build_ffmpeg_command +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, output_directory, tracks, args): + """ + 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. + args: Parsed command‑line arguments (namespace). + + Raises: + RuntimeError: If no audio stream is found. + ValueError: If timestamp parsing or format compatibility 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. Format decision + # -------------------------------------------------------------------------- + output_format = determine_output_format(stream_info, args.format, args.transcode_to) + print(f"Output container: {output_format}") + + validate_format_compatibility(output_format, stream_info, + args.drop_video, args.drop_subs) + + # Determine file extension. + extension_info = FORMAT_INFO.get(output_format, {}) + extension = extension_info.get('ext', '.mkv') + + # -------------------------------------------------------------------------- + # 3. Parse timestamps + # -------------------------------------------------------------------------- + track_times = parse_track_timestamps(tracks) + resolved_times = resolve_end_times(track_times, total_duration) + + # -------------------------------------------------------------------------- + # 4. 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}'") + + # -------------------------------------------------------------------------- + # 5. 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') + + # ---------------------------------------------------------------------- + # 5a. Build filename + # ---------------------------------------------------------------------- + clean_filename = build_filename(track, idx, extension, args) + output_path = os.path.join(output_directory, clean_filename) + + # ---------------------------------------------------------------------- + # 5b. Handle existing files + # ---------------------------------------------------------------------- + if args.skip_existing and os.path.exists(output_path): + print(f"Skipping track {idx}: {output_path} already exists.") + continue + + # ---------------------------------------------------------------------- + # 5c. Build metadata + # ---------------------------------------------------------------------- + metadata = build_metadata_dict(track, idx, input_metadata, args) + + # ---------------------------------------------------------------------- + # 5d. Build and execute FFmpeg command + # ---------------------------------------------------------------------- + cmd = build_ffmpeg_command( + input_file, start_seconds, duration_seconds, output_path, + stream_info, output_format, args.transcode_to, + args.drop_video, args.drop_subs, + metadata=metadata + ) + + 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}") diff --git a/ffmpeg.py b/ffmpeg.py new file mode 100644 index 0000000..653decc --- /dev/null +++ b/ffmpeg.py @@ -0,0 +1,223 @@ +"""FFmpeg / FFprobe interactions and command building.""" + +import subprocess +import json + +from .constants import FORMAT_INFO +from .utils import format_time + + +def get_audio_duration(input_file: str) -> float: + """ + Retrieve the total duration (in seconds) of the input file using ffprobe. + + Args: + input_file: Path to the media file. + + Returns: + Duration in seconds as a float. + """ + cmd = [ + 'ffprobe', '-v', 'error', + '-show_entries', 'format=duration', + '-of', 'default=noprint_wrappers=1:nokey=1', + input_file + ] + result = subprocess.run(cmd, capture_output=True, text=True, check=False) + return float(result.stdout.strip()) + + +def has_stream_type(input_file: str, stream_type: str) -> bool: + """ + Check whether the input file contains a stream of the given type. + + Args: + input_file: Path to the media file. + stream_type: 'a' for audio, 'v' for video, 's' for subtitle. + + Returns: + True if at least one such stream exists, False otherwise. + """ + cmd = [ + 'ffprobe', '-v', 'error', + '-select_streams', stream_type, + '-show_entries', 'stream=codec_type', + '-of', 'default=noprint_wrappers=1:nokey=1', + input_file + ] + result = subprocess.run(cmd, capture_output=True, text=True, check=False) + return bool(result.stdout.strip()) + + +def get_audio_codec(input_file: str) -> str: + """ + Return the codec name of the first audio stream. + + Args: + input_file: Path to the media file. + + Returns: + Codec name as a lowercase string, or None if no audio stream exists. + """ + cmd = [ + 'ffprobe', '-v', 'error', + '-select_streams', 'a:0', + '-show_entries', 'stream=codec_name', + '-of', 'default=noprint_wrappers=1:nokey=1', + input_file + ] + result = subprocess.run(cmd, capture_output=True, text=True, check=False) + codec = result.stdout.strip().lower() + return codec if codec else None + + +def get_stream_info(input_file: str): + """ + Collect information about the streams present in the input file. + + Args: + input_file: Path to the media file. + + Returns: + A dictionary with keys: has_audio, has_video, has_subtitle, audio_codec. + """ + return { + 'has_audio': has_stream_type(input_file, 'a'), + 'has_video': has_stream_type(input_file, 'v'), + 'has_subtitle': has_stream_type(input_file, 's'), + 'audio_codec': get_audio_codec(input_file) + } + +def build_ffmpeg_command(input_file, start_seconds, duration_seconds, output_path, + stream_info, format_opt, transcode_audio, + drop_video, drop_subs, metadata=None): + """ + Construct the FFmpeg command line as a list of arguments. + + Args: + input_file: Path to the input media file. + start_seconds: Start time for the segment (in seconds). + duration_seconds: Duration of the segment (in seconds). + output_path: Destination path for the output file. + stream_info: Dictionary from get_stream_info(). + format_opt: Output container format (e.g., 'mp3'). + transcode_audio: Audio codec to transcode to (or None). + drop_video: True to remove video streams. + drop_subs: True to remove subtitle streams. + metadata: Optional dict of metadata key/value pairs to write. + + Returns: + A list of command‑line arguments suitable for subprocess.run(). + """ + cmd = [ + 'ffmpeg', + '-i', input_file, + '-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 -------------------- + # Map the streams we want to keep. + 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 transcode_audio: + cmd.extend(['-c:a', transcode_audio]) + if transcode_audio in ('libmp3lame', 'mp3'): + cmd.extend(['-b:a', '192k']) + elif transcode_audio in ('libopus', 'opus'): + cmd.extend(['-b:a', '128k']) + else: + cmd.extend(['-c:a', 'copy']) + + # -------------------- Video codec -------------------- + if not drop_video and stream_info['has_video']: + cmd.extend(['-c:v', 'copy']) + else: + cmd.append('-vn') + + # -------------------- Subtitle codec -------------------- + if not drop_subs and stream_info['has_subtitle']: + 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]) + + # Overwrite output if it already exists. + cmd.extend(['-y', output_path]) + + return cmd + +def get_metadata(input_file: str) -> dict: + """ + Retrieve metadata from the input file using ffprobe with JSON output. + Returns a dict with: + - album: merged from all sources (last wins) + - title: merged from all sources (last wins) + - comments: list of (stream_index, comment) tuples + """ + cmd = [ + 'ffprobe', '-v', 'quiet', + '-print_format', 'json', + '-show_entries', 'format_tags:stream_tags', + input_file + ] + result = subprocess.run(cmd, capture_output=True, text=True, check=False) + + if result.returncode != 0: + return {'album': None, 'title': None, 'comments': []} + + try: + data = json.loads(result.stdout) + # Collect album and title (merged, last wins). + album = None + title = None + comments = [] # list of (stream_index, comment) + + # Format tags. + fmt_tags = data.get('format', {}).get('tags', {}) + album = fmt_tags.get('album') or album + title = fmt_tags.get('title') or title + # Format does not have a stream index; we'll treat it as -1 if needed. + + # Stream tags. + for idx, stream in enumerate(data.get('streams', [])): + stream_tags = stream.get('tags', {}) + # Album and title: update if present. + if 'album' in stream_tags: + album = stream_tags['album'] + if 'title' in stream_tags: + title = stream_tags['title'] + # Comment: collect all occurrences. + if 'comment' in stream_tags: + comments.append((idx, stream_tags['comment'])) + + return { + 'album': album, + 'title': title, + 'comments': comments + } + except (json.JSONDecodeError, KeyError): + return {'album': None, 'title': None, 'comments': []} + diff --git a/filename.py b/filename.py new file mode 100644 index 0000000..741be7b --- /dev/null +++ b/filename.py @@ -0,0 +1,53 @@ +"""Output filename generation.""" + +import os +import re + +from .utils import apply_replacement, cleanup_good_chars, apply_template + + +def build_filename(track, idx, extension, args): + """ + Build the complete output filename. + + Args: + track: Dict of parsed track data. + idx: Track index (1‑based). + extension: File extension with leading dot (e.g., '.mp3'). + args: Parsed command‑line arguments. + + Returns: + Sanitized filename. + """ + # Prepare placeholder values. + placeholders = { + 'tn': track.get('tn', ''), + 'an': track.get('an', ''), + 'al': track.get('al', ''), + 'date': track.get('date', ''), + 'ext': extension.lstrip('.'), # e.g., 'mp3' + 'num': f"{idx:02d}" + } + + # Apply the template. + raw_filename = apply_template(args.output_template, placeholders) + + # If --number-tracks is used, prepend the track number. + if args.number_tracks: + raw_filename = f"{idx:02d} - {raw_filename}" + + # Apply character replacement if requested. + if args.replace_bad_chars: + clean_filename = apply_replacement(raw_filename, + args.bad_chars, + args.replacement_char) + clean_filename = cleanup_good_chars(clean_filename, args.replacement_char) + else: + clean_filename = raw_filename + # Warn about unsafe characters. + forbidden = set(r'<>:"/\\|?*') + if any(c in forbidden for c in clean_filename): + print(f"Warning: Track {idx} filename contains unsafe characters " + f"({clean_filename}) – may cause filesystem errors.") + + return clean_filename diff --git a/formats.py b/formats.py new file mode 100644 index 0000000..26c0d3d --- /dev/null +++ b/formats.py @@ -0,0 +1,55 @@ +"""Container format decision and validation.""" + +from .constants import FORMAT_INFO + + +def determine_output_format(stream_info, user_format, transcode_audio): + """ + Decide which container format to use. + + Args: + stream_info: Dict from get_stream_info(). + user_format: User‑requested format (or None). + transcode_audio: Audio codec to transcode to (or None). + + Returns: + A format name that exists in FORMAT_INFO. + """ + if user_format: + return user_format + + # If video or subtitles exist, use MKV (which supports everything). + if stream_info['has_video'] or stream_info['has_subtitle']: + return 'matroska' + + # Audio‑only: choose based on the current audio codec. + audio_codec = stream_info.get('audio_codec', '') + if audio_codec == 'mp3': + return 'mp3' + else: + return 'mp4' # .m4a + + +def validate_format_compatibility(format_name, stream_info, drop_video, drop_subs): + """ + Ensure the chosen container can accommodate the streams we intend to keep. + + Raises: + ValueError: If the format is incompatible with the intended streams. + """ + info = FORMAT_INFO.get(format_name) + if not info: + print(f"Warning: Unknown format '{format_name}'. Proceeding, but may fail.") + return + + if info['audio_only']: + if stream_info['has_video'] and not drop_video: + raise ValueError( + f"Format '{format_name}' does not support video streams. " + "Please use --drop-video or choose a container that supports video." + ) + if stream_info['has_subtitle'] and not drop_subs: + raise ValueError( + f"Format '{format_name}' does not support subtitle streams. " + "Please use --drop-subs or choose a container that supports subtitles." + ) diff --git a/main.py b/main.py new file mode 100644 index 0000000..257cdb6 --- /dev/null +++ b/main.py @@ -0,0 +1,204 @@ +"""Command‑line interface and entry point.""" + +import os +import sys +import subprocess +import argparse + +from .constants import DEFAULT_BAD_CHARS +from .tracklist import read_tracklist, parse_format +from .core import split_audio + + +def main(): + """Parse arguments and start the splitting process.""" + parser = argparse.ArgumentParser( + description="Split an audio file into tracks using a tracklist.", + epilog="Tracklist format: mm:ss track_name - author_name" + ) + + # Positional arguments. + parser.add_argument('input_file', help='Input audio/video file') + parser.add_argument('tracklist_file', help='Tracklist file') + + # Optional arguments. + parser.add_argument( + '--output-dir', '-o', + help='Output directory for split tracks (default: _splits)' + ) + parser.add_argument( + '--format', + help='Output container format (e.g., mp3, m4a, mkv, mp4, ogg, opus)' + ) + parser.add_argument( + '--transcode-to', + help='Re-encode audio to this codec (e.g., libmp3lame, aac, libopus)' + ) + parser.add_argument( + '--drop-video', + action='store_true', + help='Remove video streams from output' + ) + parser.add_argument( + '--drop-subs', + action='store_true', + help='Remove subtitle streams from output' + ) + parser.add_argument( + '--number-tracks', + action='store_true', + help='Prepend track number to output filenames (convenience; use %%num in template for full control)' + ) + parser.add_argument( + '--replace-bad-chars', + action='store_true', + help='Replace problematic characters in filenames (default: off)' + ) + parser.add_argument( + '--replacement-char', + default='_', + help='Character used as replacement (default: "_")' + ) + parser.add_argument( + '--bad-chars', + default=DEFAULT_BAD_CHARS, + help='String of characters to replace (default includes space and single quote)' + ) + parser.add_argument( + '--skip-existing', + action='store_true', + help='Skip extraction if output file already exists (default: overwrite)' + ) + parser.add_argument( + '--tracklist-format', + default='%ts %tn - %an', + help='Format of each line in the tracklist using placeholders: ' + '%%ts (timestamp), %%tn (track name), %%an (author), %%al (album), ' + '%%date (date/year), %%ext (file extension). ' + 'Default: "%%ts %%tn - %%an"' + ) + parser.add_argument( + '--output-template', + default='%an-%tn.%ext', + help='Template for output filenames using placeholders: ' + '%%tn (track name), %%an (author), %%al (album), ' + '%%date (date/year), %%ext (file extension), %%num (track number). ' + 'Default: "%%an-%%tn.%%ext"' + ) + parser.add_argument( + '--dry-run', + action='store_true', + help='Parse and display the tracklist without splitting any files' + ) + + # ------------------------------------------------------------ + # Metadata options (new) + # ------------------------------------------------------------ + parser.add_argument( + '--album', + help='Set album name in output metadata (overrides parsed %%al and original album)' + ) + parser.add_argument( + '--comment', + help='Explicit comment text (overrides all other comment settings)' + ) + parser.add_argument( + '--no-comment', + action='store_true', + help='Explicitly ignore any comment (no comment written)' + ) + parser.add_argument( + '--comment-stream', + type=int, + default=None, + help='Select comment from a specific stream index (0‑based). Default: first stream with a comment.' + ) + parser.add_argument( + '--merge-comments', + action='store_true', + help='Merge all comments from all streams into one (separated by --comment-separator)' + ) + parser.add_argument( + '--comment-separator', + default='; ', + help='Separator used when merging comments (default: "; ")' + ) + + args = parser.parse_args() + + # -------------------------------------------------------------------------- + # Input validation + # -------------------------------------------------------------------------- + if not os.path.exists(args.input_file): + print(f"Error: Input file not found: {args.input_file}") + sys.exit(1) + + if not os.path.exists(args.tracklist_file): + print(f"Error: Tracklist file not found: {args.tracklist_file}") + sys.exit(1) + + # Ensure the replacement character is a single character. + if len(args.replacement_char) != 1: + print("Error: --replacement-char must be a single character.") + sys.exit(1) + + # Check that FFmpeg is installed. + try: + subprocess.run(['ffmpeg', '-version'], capture_output=True, check=True) + except (subprocess.CalledProcessError, FileNotFoundError): + print("Error: FFmpeg not found. Please install FFmpeg first.") + print(" - https://ffmpeg.org/download.html") + sys.exit(1) + + # Parse the tracklist using the user‑provided format. + try: + tokens = parse_format(args.tracklist_format) + except ValueError as error: + print(f"Error in --tracklist-format: {error}") + sys.exit(1) + + # Read and parse the tracklist file. + tracks = read_tracklist(args.tracklist_file, tokens) + if not tracks: + print("Error: No valid tracks found in tracklist file.") + sys.exit(1) + + print(f"Found {len(tracks)} tracks.") + + # Dry‑run mode: display parsed data and exit. + if args.dry_run: + print("\nParsed tracklist:") + print("-" * 60) + if tracks: + fields = sorted(tracks[0].keys()) + header = " | ".join(f"{f:>10}" for f in fields) + print(header) + print("-" * len(header)) + for idx, track in enumerate(tracks, 1): + values = [f"{track.get(f, ''):>10}" for f in fields] + print(f"{idx:3d} | " + " | ".join(values)) + print("-" * 60) + print("Dry‑run complete. No files were created.") + sys.exit(0) + + # Determine the output directory. + if args.output_dir: + output_dir = args.output_dir + print(f"Using custom output directory: {output_dir}") + else: + base_name = os.path.splitext(os.path.basename(args.input_file))[0] + output_dir = base_name + "_splits" + print(f"Using default output directory: {output_dir}") + + # Run the splitter. + try: + split_audio(args.input_file, output_dir, tracks, args) + except (RuntimeError, ValueError) as error: + print(f"Error: {error}") + sys.exit(1) + + print("Done!") + + +if __name__ == "__main__": + main() diff --git a/metadata.py b/metadata.py new file mode 100644 index 0000000..2da7fe6 --- /dev/null +++ b/metadata.py @@ -0,0 +1,99 @@ +"""Metadata selection and building.""" + + +def select_album_value(track, input_metadata, args): + """ + Determine the album value according to precedence. + + Precedence: --album > %al > original album > original title > None. + + Args: + track: Dict of parsed track data. + input_metadata: Dict from get_metadata(). + args: Parsed command‑line arguments. + + Returns: + Album string or None. + """ + original_album = input_metadata.get('album') + original_title = input_metadata.get('title') + + if args.album: + return args.album + elif track.get('al'): + return track.get('al') + elif original_album: + return original_album + elif original_title: + return original_title + else: + return None + + +def select_comment_value(input_metadata, args): + """ + Determine the comment value according to precedence. + + Precedence: --no-comment > --comment > --merge-comments > + --comment-stream > first comment from file. + + Args: + input_metadata: Dict from get_metadata(). + args: Parsed command‑line arguments. + + Returns: + Comment string or None. + """ + original_comments = input_metadata.get('comments', []) + + if args.no_comment: + return None + elif args.comment: + return args.comment + elif args.merge_comments and original_comments: + separator = args.comment_separator if args.comment_separator else '; ' + comments_text = separator.join(comment for _, comment in original_comments) + return comments_text if comments_text else None + elif args.comment_stream is not None: + if 0 <= args.comment_stream < len(original_comments): + return original_comments[args.comment_stream][1] + else: + print(f"Warning: Stream {args.comment_stream} not found in comments list. " + f"Using first comment (if any).") + if original_comments: + return original_comments[0][1] + return None + elif original_comments: + # Default: use the first comment. + return original_comments[0][1] + else: + return None + + +def build_metadata_dict(track, idx, input_metadata, args): + """ + Assemble the final metadata dictionary for FFmpeg. + + Args: + track: Dict of parsed track data. + idx: Track index (1‑based). + input_metadata: Dict from get_metadata(). + args: Parsed command‑line arguments. + + Returns: + Dict with metadata fields (empty values removed). + """ + album_value = select_album_value(track, input_metadata, args) + comment_value = select_comment_value(input_metadata, args) + + metadata = { + 'title': track.get('tn', ''), + 'artist': track.get('an', ''), + 'album': album_value, + 'date': track.get('date', ''), + 'track': f"{idx:02d}", + 'comment': comment_value + } + + # Remove empty fields. + return {k: v for k, v in metadata.items() if v} diff --git a/timestamp.py b/timestamp.py new file mode 100644 index 0000000..aaab1c5 --- /dev/null +++ b/timestamp.py @@ -0,0 +1,76 @@ +"""Timestamp parsing and resolution.""" + +from .utils import parse_timestamp + + +def parse_track_timestamps(tracks): + """ + Parse timestamps from each track. + + Args: + tracks: List of dicts from tracklist parsing. + + Returns: + List of (start_seconds, end_seconds_or_None) for each track. + + Raises: + ValueError: If timestamp is invalid or range is malformed. + """ + track_times = [] + for idx, track in enumerate(tracks, 1): + ts_str = track.get('ts', '').strip() + if not ts_str: + raise ValueError(f"Track {idx}: timestamp is empty") + + # Detect range: contains a dash. + if '-' in ts_str: + parts = ts_str.split('-', 1) + start_str = parts[0].strip() + end_str = parts[1].strip() + if not start_str or not end_str: + raise ValueError(f"Track {idx}: invalid range format '{ts_str}'. Expected 'start-end'.") + try: + start_sec = parse_timestamp(start_str) + end_sec = parse_timestamp(end_str) + except ValueError as e: + raise ValueError(f"Track {idx}: invalid timestamp in range '{ts_str}': {e}") + if end_sec <= start_sec: + raise ValueError(f"Track {idx}: end time ({end_str}) must be after start time ({start_str})") + track_times.append((start_sec, end_sec)) + else: + # Start-only: we will compute end later. + try: + start_sec = parse_timestamp(ts_str) + except ValueError as e: + raise ValueError(f"Track {idx}: invalid timestamp '{ts_str}': {e}") + track_times.append((start_sec, None)) + + return track_times + + +def resolve_end_times(track_times, total_duration): + """ + Resolve None ends to next start or total duration. + + Args: + track_times: List of (start_sec, end_sec_or_None). + total_duration: Total duration of the input file. + + Returns: + List of (start_sec, end_sec) with all ends resolved. + + Raises: + ValueError: If end time is not after start time. + """ + resolved = [] + for idx, (start_sec, end_sec) in enumerate(track_times): + if end_sec is None: + if idx + 1 < len(track_times): + next_start = track_times[idx + 1][0] + end_sec = next_start + else: + end_sec = int(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)) + return resolved diff --git a/tracklist.py b/tracklist.py new file mode 100644 index 0000000..cc6811e --- /dev/null +++ b/tracklist.py @@ -0,0 +1,155 @@ +"""Tracklist file parsing with flexible format support.""" + +import re + + +def parse_format(format_str: str): + """ + Convert a format string (e.g., "%ts %tn - %an") into a list of tokens. + + Each token is a dict with keys: + 'type': 'literal' or 'placeholder' + 'value': the literal text or the placeholder name (e.g., 'tn') + + Placeholders are identified by a '%' followed by alphabetic characters. + A literal '%%' is escaped to a single '%'. + + Args: + format_str: The user‑provided format string. + + Returns: + List of token dicts. + + Raises: + ValueError: If the format is invalid (e.g., unknown placeholder). + """ + # Known placeholders. + valid_placeholders = {'ts', 'tn', 'an', 'al', 'date', 'ext'} + + tokens = [] + i = 0 + while i < len(format_str): + ch = format_str[i] + if ch == '%': + # Check for escaped '%%'. + if i + 1 < len(format_str) and format_str[i + 1] == '%': + tokens.append({'type': 'literal', 'value': '%'}) + i += 2 + continue + # Must be a placeholder. + # Match '%' followed by letters. + match = re.match(r'%([a-zA-Z]+)', format_str[i:]) + if not match: + raise ValueError(f"Invalid placeholder at position {i}: '{format_str[i:]}'") + placeholder = match.group(1) + if placeholder not in valid_placeholders: + raise ValueError(f"Unknown placeholder '%{placeholder}'. " + f"Allowed: {', '.join(valid_placeholders)}") + tokens.append({'type': 'placeholder', 'value': placeholder}) + i += len(match.group(0)) + else: + # Literal character. + # Collect consecutive non‑'%' characters. + j = i + while j < len(format_str) and format_str[j] != '%': + j += 1 + tokens.append({'type': 'literal', 'value': format_str[i:j]}) + i = j + return tokens + + +def parse_line(line: str, tokens): + """ + Parse a single line of the tracklist using the token list. + + Returns a dict mapping placeholder names to extracted strings. + If a placeholder is not found, its value is None. + + Args: + line: The raw line from the tracklist file. + tokens: The token list from parse_format(). + + Returns: + A dict with keys for each placeholder present in the format. + + Raises: + ValueError: If the line cannot be parsed according to the format. + """ + # Strip leading/trailing whitespace but preserve internal spaces. + line = line.strip() + if not line: + raise ValueError("Empty line") + + result = {} + pos = 0 + # We need to match the tokens in order. + for token in tokens: + if token['type'] == 'literal': + literal = token['value'] + if not line.startswith(literal, pos): + raise ValueError(f"Expected literal '{literal}' at position {pos}, got '{line[pos:]}'") + pos += len(literal) + else: # placeholder + placeholder = token['value'] + # If this is the last token, capture the rest of the line. + if token is tokens[-1]: + result[placeholder] = line[pos:].strip() or None + pos = len(line) + else: + # Find the next literal to use as a delimiter. + # We need to look ahead to the next literal token. + next_literal = None + for next_token in tokens[tokens.index(token) + 1:]: + if next_token['type'] == 'literal': + next_literal = next_token['value'] + break + if next_literal is None: + # If no more literals, capture the rest. + result[placeholder] = line[pos:].strip() or None + pos = len(line) + else: + # Find the occurrence of the next literal in the line starting from pos. + next_pos = line.find(next_literal, pos) + if next_pos == -1: + raise ValueError(f"Could not find literal '{next_literal}' after placeholder '{placeholder}'") + result[placeholder] = line[pos:next_pos].strip() or None + pos = next_pos + return result + + +def read_tracklist(filepath: str, tokens): + """ + Read and parse the tracklist file using the provided token list. + + Each line is parsed into a dict of fields. The timestamp field ('ts') + is required; if missing, an error is raised. + + Args: + filepath: Path to the tracklist file. + tokens: Token list from parse_format(). + + Returns: + A list of dicts, one per track. + + Raises: + ValueError: If a line cannot be parsed or the timestamp is missing. + """ + tracks = [] + with open(filepath, 'r', encoding='utf-8') as file: + for line_num, raw_line in enumerate(file, 1): + line = raw_line.strip() + if not line: + continue + try: + parsed = parse_line(line, tokens) + except ValueError as error: + print(f"Warning: Skipping line {line_num}: {error}") + continue + + # Ensure timestamp is present. + if 'ts' not in parsed or parsed['ts'] is None: + raise ValueError(f"Line {line_num}: timestamp (%ts) is missing or empty") + + tracks.append(parsed) + + return tracks diff --git a/utils.py b/utils.py new file mode 100644 index 0000000..1cd097d --- /dev/null +++ b/utils.py @@ -0,0 +1,171 @@ +"""General‑purpose helper functions.""" + +import re + + +def parse_timestamp(timestamp: str) -> int: + """ + Convert a timestamp in 'mm:ss' or 'HH:MM:SS' format to total seconds. + + Args: + timestamp: String in the format 'mm:ss' or 'HH:MM:SS'. + + Returns: + Total number of seconds. + + Raises: + ValueError: If the format is unrecognized. + """ + parts = timestamp.strip().split(':') + if len(parts) == 2: + minutes, seconds = map(int, parts) + return minutes * 60 + seconds + elif len(parts) == 3: + hours, minutes, seconds = map(int, parts) + return hours * 3600 + minutes * 60 + seconds + else: + raise ValueError(f"Invalid timestamp format: {timestamp}") + + +def format_time(seconds: int) -> str: + """ + Convert seconds to 'HH:MM:SS' format for FFmpeg. + + Args: + seconds: Total number of seconds. + + Returns: + Time string in 'HH:MM:SS' format. + """ + hours = seconds // 3600 + minutes = (seconds % 3600) // 60 + secs = seconds % 60 + return f"{hours:02d}:{minutes:02d}:{secs:02d}" + + +def sanitize_filename(name: str) -> str: + """ + Remove characters that are problematic on common filesystems. + + This is a fallback sanitizer; the main replacement is handled by + apply_replacement() when --replace-bad-chars is used. + + Args: + name: Original filename candidate. + + Returns: + Sanitized string with dangerous characters replaced by '_'. + """ + # Replace characters that are forbidden on Windows/Linux/macOS. + return re.sub(r'[<>:"/\\|?*]', '_', name).strip() + + +def apply_replacement(name: str, bad_chars: str, replacement_char: str) -> str: + """ + Replace every occurrence of any character in bad_chars with replacement_char. + + Args: + name: The original string. + bad_chars: String containing all characters to replace. + replacement_char: The character to insert. + + Returns: + The transformed string. + """ + if not bad_chars: + return name + # Build a translation table mapping each bad character to the replacement. + translation_table = str.maketrans({c: replacement_char for c in set(bad_chars)}) + return name.translate(translation_table) + + +def cleanup_good_chars(name: str, good_char: str) -> str: + """ + Clean up the filename after replacement to avoid awkward sequences. + + Steps performed: + 1. Strip leading/trailing good_char characters. + 2. Collapse consecutive good_char characters into a single one. + 3. Remove good_char characters that are adjacent to a single non‑good_char + (i.e., pattern good_char + X + good_char becomes just X), + and repeat this step until no more such patterns exist. + 4. After each iteration, re‑apply stripping and collapsing. + + This makes filenames more human‑readable, e.g.: + "author_-_song" -> "author-song" + "hello__world" -> "hello_world" + "_hello_" -> "hello" + + Args: + name: The string to clean (after replacement). + good_char: The character used as replacement. + + Returns: + The cleaned string. + """ + # Escape the good_char for regex usage. + escaped = re.escape(good_char) + + # Helper to collapse consecutive good chars and strip edges. + def collapse_and_strip(s: str) -> str: + # Collapse multiple consecutive good chars into one. + s = re.sub(rf'{escaped}+', good_char, s) + # Remove leading/trailing good chars. + s = s.strip(good_char) + return s + + # Apply stripping and collapsing initially. + name = collapse_and_strip(name) + + # Remove patterns: good_char + X + good_char (where X is any char != good_char). + # Repeat until no further changes. + while True: + # Replace pattern: good_char (non-good-char) good_char -> just the non-good-char. + # The negative lookahead ensures X is not the good_char. + pattern = rf'{escaped}([^{escaped}]){escaped}' + new_name = re.sub(pattern, r'\1', name) + if new_name == name: + break + name = new_name + # Re‑apply stripping and collapsing after removal. + name = collapse_and_strip(name) + + return name + +def apply_template(template: str, placeholders: dict) -> str: + """ + Replace placeholders in a template string with values from a dict. + + Placeholders are of the form %key (e.g., %tn, %an). + Use %% to escape a literal percent sign. + + Args: + template: The template string. + placeholders: Dict mapping placeholder names to values. + + Returns: + The rendered string. + """ + import re + result = [] + i = 0 + while i < len(template): + ch = template[i] + if ch == '%': + if i + 1 < len(template) and template[i + 1] == '%': + result.append('%') + i += 2 + continue + match = re.match(r'%([a-zA-Z]+)', template[i:]) + if match: + key = match.group(1) + value = placeholders.get(key, '') + result.append(value) + i += len(match.group(0)) + else: + result.append(ch) + i += 1 + else: + result.append(ch) + i += 1 + return ''.join(result)