Initial commit. Basic featues are implemented

This commit is contained in:
2026-07-28 16:25:11 +05:00
commit 842e3bf1b0
12 changed files with 1190 additions and 0 deletions
+1
View File
@@ -0,0 +1 @@
__pycache__
+3
View File
@@ -0,0 +1,3 @@
"""
Audio splitter package split an audio file using a tracklist.
"""
+20
View File
@@ -0,0 +1,20 @@
"""Global constants and default values."""
# Mapping from userfriendly format names to FFmpeg format identifiers,
# file extensions, and whether the container is audioonly.
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',!@#№$;:%^&?*(){}[]\/<>+=~`\' '
+130
View File
@@ -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 commandline 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}")
+223
View File
@@ -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 commandline 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': []}
+53
View File
@@ -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 (1based).
extension: File extension with leading dot (e.g., '.mp3').
args: Parsed commandline 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
+55
View File
@@ -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: Userrequested 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'
# Audioonly: 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."
)
+204
View File
@@ -0,0 +1,204 @@
"""Commandline 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: <input_basename>_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 (0based). 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 userprovided 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.")
# Dryrun 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("Dryrun 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()
+99
View File
@@ -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 commandline 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 commandline 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 (1based).
input_metadata: Dict from get_metadata().
args: Parsed commandline 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}
+76
View File
@@ -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
+155
View File
@@ -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 userprovided 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
+171
View File
@@ -0,0 +1,171 @@
"""Generalpurpose 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 nongood_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, reapply stripping and collapsing.
This makes filenames more humanreadable, 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
# Reapply 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)