dschlueter/Musiksammlung
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Fix three ripper bugs: input cleanup, file move, recursive search

Browse source 
- Add _clean_input(): strips ANSI escape codes (^[[D from arrow keys),
  control characters and surrounding quotes from user input
- Add _write_abcde_config(): writes temp abcde config with OUTPUTDIR
  and flat OUTPUTFORMAT so files land in the right directory
- Add 'move' action to abcde so encoded files are actually placed in
  OUTPUTDIR instead of staying in abcde's internal temp directory
- Change _get_audio_files() to use rglob() (recursive) so files in
  abcde subdirectories are found
- Improve error messages: include abcde output on failure

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Dieter Schlüter 2026-02-17 19:58:59 +01:00
commit 096be97ba8
1 changed files with 124 additions and 89 deletions
Download patch file Download diff file Expand all files Collapse all files

199
src/musiksammlung/ripper.py
View file

@ -14,6 +14,9 @@ from musiksammlung.config import AudioFormat
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# ANSI escape sequence pattern (e.g. arrow keys from broken readline)
_ANSI_ESC = re.compile(r"(\x1b|\^)\[[\d;]*[A-Za-z@]?")
class TrackInfo(NamedTuple): class TrackInfo(NamedTuple):
"""Track information from abcde.""" """Track information from abcde."""
@ -35,6 +38,27 @@ class RipperConfig(BaseModel):
use_cddb: bool = True # Use CDDB lookup use_cddb: bool = True # Use CDDB lookup
def _clean_input(raw: str) -> str:
"""Strip ANSI escape codes, control characters and surrounding quotes.
Handles broken readline environments where arrow keys produce
literal escape sequences like ^[[D instead of moving the cursor.
Args:
raw: Raw string from input()
Returns:
Cleaned string
"""
# Remove ANSI escape sequences (\x1b[... and ^[[...)
cleaned = _ANSI_ESC.sub("", raw)
# Remove remaining control characters (backspace \x08, etc.)
cleaned = re.sub(r"[\x00-\x1f\x7f]", "", cleaned)
# Strip surrounding whitespace and quotes
cleaned = cleaned.strip().strip('"\'')
return cleaned
def _sanitize_name(name: str) -> str: def _sanitize_name(name: str) -> str:
"""Remove problematic characters and replace spaces. """Remove problematic characters and replace spaces.
@ -79,7 +103,7 @@ def _parse_cddb_response(output: str) -> list[TrackInfo]:
def _get_audio_files(output_dir: Path, audio_format: AudioFormat) -> list[Path]: def _get_audio_files(output_dir: Path, audio_format: AudioFormat) -> list[Path]:
"""Find all audio files in directory (case-insensitive). """Find all audio files in directory recursively (case-insensitive).
Args: Args:
output_dir: Target directory output_dir: Target directory
@ -93,13 +117,38 @@ def _get_audio_files(output_dir: Path, audio_format: AudioFormat) -> list[Path]:
pattern = re.compile(rf".*\.{ext}$", re.IGNORECASE) pattern = re.compile(rf".*\.{ext}$", re.IGNORECASE)
audio_files = [] audio_files = []
for file in output_dir.iterdir(): # rglob: search recursively so abcde subdirs are also covered
for file in output_dir.rglob("*"):
if file.is_file() and pattern.match(file.name): if file.is_file() and pattern.match(file.name):
audio_files.append(file) audio_files.append(file)
return sorted(audio_files) return sorted(audio_files)
def _write_abcde_config(output_dir: Path) -> Path:
"""Write a temporary abcde config file.
Sets OUTPUTDIR to output_dir and uses a flat filename format
(track number only) so we can rename files ourselves afterward.
Args:
output_dir: Directory where encoded files should be placed
Returns:
Path to the config file
"""
config = f"""\
OUTPUTDIR="{output_dir}"
OUTPUTFORMAT="${{TRACKNUM}}"
VAOUTPUTFORMAT="${{TRACKNUM}}"
ONETRACKOUTPUTFORMAT="${{TRACKNUM}}"
PLAYLISTFORMAT=""
"""
config_path = output_dir / ".abcde.conf"
config_path.write_text(config, encoding="utf-8")
return config_path
def _rename_files( def _rename_files(
output_dir: Path, output_dir: Path,
tracks: list[TrackInfo], tracks: list[TrackInfo],
@ -133,12 +182,11 @@ def _rename_files(
f"{artist_clean}{audio_format.extension}" f"{artist_clean}{audio_format.extension}"
) )
old_path = file
new_path = output_dir / new_name new_path = output_dir / new_name
if old_path != new_path: if file != new_path:
logger.info("Renaming: %s -> %s", old_path.name, new_name) logger.info("Renaming: %s -> %s", file.name, new_name)
old_path.rename(new_path) file.rename(new_path)
break break
@ -167,15 +215,20 @@ def _rip_with_abcde(
""" """
output_dir.mkdir(parents=True, exist_ok=True) output_dir.mkdir(parents=True, exist_ok=True)
# Write abcde config: controls OUTPUTDIR and flat filename format
config_path = _write_abcde_config(output_dir)
# abcde options: # abcde options:
# -a: cddb,read,encode,tag if use_cddb, else read,encode # -c config: use our config (OUTPUTDIR, OUTPUTFORMAT)
# -a actions: cddb+read+encode+tag+move, or read+encode+move
# -p: pad track numbers with zeros # -p: pad track numbers with zeros
# -o format: output format # -o format: output format
# -d device: CD drive # -d device: CD drive
# -x: eject CD after ripping # -x: eject CD after ripping
# -N: non-interactive (no prompts) # -N: non-interactive (no prompts, auto-select first CDDB match)
cmd = [ cmd = [
"abcde", "abcde",
"-c", str(config_path),
"-p", "-p",
"-o", audio_format.get_abcde_format(), "-o", audio_format.get_abcde_format(),
"-d", device, "-d", device,
@ -183,11 +236,11 @@ def _rip_with_abcde(
"-N", "-N",
] ]
# Actions # Actions — move is required so files land in OUTPUTDIR
if use_cddb: if use_cddb:
cmd.extend(["-a", "cddb,read,encode,tag"]) cmd.extend(["-a", "cddb,read,encode,tag,move"])
else: else:
cmd.extend(["-a", "read,encode"]) cmd.extend(["-a", "read,encode,move"])
# Parallel encodes # Parallel encodes
if parallel_jobs > 1: if parallel_jobs > 1:
@ -200,14 +253,16 @@ def _rip_with_abcde(
# Encoder options for quality # Encoder options for quality
encoder_opts = audio_format.get_encoder_options(quality) encoder_opts = audio_format.get_encoder_options(quality)
if encoder_opts: if encoder_opts:
# abcde accepts encoder options with colon # abcde accepts encoder options with colon: -o format:options
# Format: -o format:options
cmd[-2] = f"{audio_format.get_abcde_format()}:{encoder_opts}" cmd[-2] = f"{audio_format.get_abcde_format()}:{encoder_opts}"
logger.info("Starting abcde in %s (Format: %s, Quality: %s, CDDB: %s)", logger.info(
output_dir, audio_format.value, quality, use_cddb) "Starting abcde in %s (Format: %s, Quality: %s, CDDB: %s)",
output_dir, audio_format.value, quality, use_cddb,
)
logger.debug("Command: %s", " ".join(cmd))
# Run abcde non-interactively # Run abcde non-interactively, capture output for CDDB parsing
result = subprocess.run( result = subprocess.run(
cmd, cmd,
cwd=str(output_dir), cwd=str(output_dir),
@ -215,24 +270,34 @@ def _rip_with_abcde(
text=True, text=True,
) )
# Log output for debugging
if result.stdout:
logger.debug("abcde stdout:\n%s", result.stdout)
if result.stderr:
logger.debug("abcde stderr:\n%s", result.stderr)
if result.returncode != 0: if result.returncode != 0:
raise RuntimeError( raise RuntimeError(
f"abcde failed (exit {result.returncode}). " f"abcde failed (exit {result.returncode}).\n"
"Check if a CD is in the drive and readable." f"{result.stderr or result.stdout}"
) )
# Track information from CDDB parsing # Parse track info from CDDB output
tracks = None tracks = None
if use_cddb: if use_cddb:
tracks = _parse_cddb_response(result.stdout) combined = result.stdout + result.stderr
tracks = _parse_cddb_response(combined)
if tracks: if tracks:
logger.info("CDDB data found: %d tracks", len(tracks)) logger.info("CDDB data found: %d tracks", len(tracks))
# Find files (case-insensitive) # Find files (case-insensitive, recursive)
audio_files = _get_audio_files(output_dir, audio_format) audio_files = _get_audio_files(output_dir, audio_format)
if not audio_files: if not audio_files:
raise RuntimeError("No files created by abcde") raise RuntimeError(
"No audio files found after ripping.\n"
"abcde output:\n" + (result.stderr or result.stdout)
)
logger.info("Ripping completed: %d tracks in %s", len(audio_files), output_dir) logger.info("Ripping completed: %d tracks in %s", len(audio_files), output_dir)
return audio_files, tracks return audio_files, tracks
@ -265,34 +330,19 @@ def rip_disc(
device, output_dir, audio_format, quality, parallel_jobs, use_pipes, use_cddb device, output_dir, audio_format, quality, parallel_jobs, use_pipes, use_cddb
) )
# Extract album name from first track (artist part)
album_name = None album_name = None
if tracks and len(tracks) > 0:
# For Various Artists, this will be "Sampler" or similar
# For single artist, this will be the artist name
album_name = tracks[0].artist
# If CDDB data available, rename files
if tracks: if tracks:
album_name = tracks[0].artist
_rename_files(output_dir, tracks, audio_format) _rename_files(output_dir, tracks, audio_format)
return output_dir, album_name, tracks return output_dir, album_name, tracks
def interactive_rip( def interactive_rip(config: RipperConfig) -> None:
config: RipperConfig,
) -> None:
"""Interactive rip workflow for multiple CDs. """Interactive rip workflow for multiple CDs.
Prompts for each album/CD:
- Album name (or empty for default 'Album{N}')
- CD number (e.g., 1, 2, ...)
- Optional continuation
Files are placed under config.output_dir: Files are placed under config.output_dir:
temp/Album_Name/CD1/01_-_title_-_artist.flac, ... Album_Name/CD1/01_-_title_-_artist.flac, ...
If CDDB is available, files are automatically named.
Args: Args:
config: Ripper configuration config: Ripper configuration
@ -306,39 +356,35 @@ def interactive_rip(
print(f"CDDB Lookup: {config.use_cddb}") print(f"CDDB Lookup: {config.use_cddb}")
print(f"Parallel Encodes: {config.parallel_jobs}") print(f"Parallel Encodes: {config.parallel_jobs}")
print(f"Pipes: {config.use_pipes}") print(f"Pipes: {config.use_pipes}")
print(f"Output Directory: {config.output_dir.absolute()}\n") print(f"Output Directory: {config.output_dir.absolute()}")
print("\nNote: Do not use arrow keys while typing — press Enter to confirm.\n")
album_counter = 1 album_counter = 1
while True: while True:
print(f"\n--- Album {album_counter} ---") print(f"\n--- Album {album_counter} ---")
# Ask for album name (optional, overridden if CDDB available) raw = input("Album name (Enter = CDDB name / default 'Album{N}'): ")
album_name = input( album_name = _clean_input(raw)
"Enter album name (or Enter for CDDB/default 'Album{N}'): " if not album_name:
).strip() album_name = f"Album{album_counter}"
default_album_name = album_name if album_name else f"Album{album_counter}"
disc_counter = 1 disc_counter = 1
while True: while True:
print(f"\n Album: {default_album_name}") print(f"\n Album: {album_name}")
print(f" CD Drive: {config.device}") print(f" CD Drive: {config.device}")
# Ask for disc number raw_disc = input(" CD number [1]: ")
disc_input = input( disc_num = int(_clean_input(raw_disc)) if _clean_input(raw_disc) else 1
" CD number for this CD [1]: "
).strip()
disc_num = int(disc_input) if disc_input else 1
# Build target directory
disc_dir = ( disc_dir = (
config.output_dir config.output_dir
/ _sanitize_name(default_album_name) / _sanitize_name(album_name)
/ f"CD{disc_num}" / f"CD{disc_num}"
) )
print(f" Ripping CD to: {disc_dir.relative_to(config.output_dir)}") print(f" Ripping to: {disc_dir.relative_to(config.output_dir)}")
print(" (Ripping in progress, please wait...)") print(" (Ripping in progress, please wait...)")
try: try:
@ -352,55 +398,44 @@ def interactive_rip(
use_cddb=config.use_cddb, use_cddb=config.use_cddb,
) )
# Show detected information if tracks:
if tracks and detected_album: print(f" ✓ CD {disc_num} ripped successfully — {len(tracks)} tracks")
print(f" ✓ CD {disc_num} ripped successfully")
print(f" Detected: {detected_album}")
if len(tracks) > 0:
print(f" Tracks: {len(tracks)}")
# Show first and last track
first = tracks[0] first = tracks[0]
last = tracks[-1] if len(tracks) > 1 else None last = tracks[-1]
print(f" {first.track_number}. {first.title} ({first.artist})") print(f" {first.track_number:2d}. {first.title} — {first.artist}")
if last: if last != first:
print(f" ... {last.track_number}. {last.title} ({last.artist})") print(f" {last.track_number:2d}. {last.title} — {last.artist}")
else: else:
print(f" ✓ CD {disc_num} ripped successfully") print(f" ✓ CD {disc_num} ripped successfully")
except RuntimeError as e: except RuntimeError as e:
print(f" ✗ Ripping error: {e}") print(f" ✗ Ripping error: {e}")
retry = input(" Try again? (y/n): ").strip().lower() raw_retry = input(" Try again? (y/n): ")
if retry != "y": if _clean_input(raw_retry).lower() != "y":
print(" Aborting disc.") print(" Aborting disc.")
break break
continue continue
# Continue? raw_next = input(" Next CD for this album? (y/n): ")
next_disc = input( if _clean_input(raw_next).lower() != "y":
" Next CD for this album? (y/n): "
).strip().lower()
if next_disc != "y":
break break
disc_counter += 1 disc_counter += 1
# Next album? raw_album = input("\nNext album? (y/n): ")
next_album = input("\nNext album? (y/n): ").strip().lower() if _clean_input(raw_album).lower() != "y":
if next_album != "y":
break break
album_counter += 1 album_counter += 1
print("\n" + "=" * 60) print("\n" + "=" * 60)
print("Ripping completed!") print("Ripping completed!")
print(f"\nFiles are in: {config.output_dir.absolute()}") print(f"Files are in: {config.output_dir.absolute()}")
print("\nNext steps:") print("\nNext steps:")
print(" 1. Check filenames and tags") print(" 1. Check filenames and tags")
if config.use_cddb: if config.use_cddb:
print(" 2. Adjust tags and covers with 'musiksammlung apply'") print(" 2. Adjust tags/covers with 'musiksammlung apply'")
else: else:
print(" 2. Scan CD cover images") print(" 2. Run 'musiksammlung scan' to extract metadata")
print(" 3. 'musiksammlung scan' for album JSON") print(" 3. Run 'musiksammlung apply' to organize & tag")
print(" 4. 'musiksammlung apply' to organize & tag")
print("=" * 60 + "\n") print("=" * 60 + "\n")