commit 842e3bf1b0d01f7306d0e7555ae891474a541f55 Author: Maxim Vershinin Date: Tue Jul 28 16:25:11 2026 +0500 Initial commit. Basic featues are implemented 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)