Add disc count validation before apply

Check audio file count vs JSON track count per disc before processing.
Aborts with a clear error showing which discs have discrepancies and
whether tracks are missing from the JSON or audio files are missing
from the directory.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Dieter Schlüter 2026-02-17 23:08:24 +01:00
commit f4e49a3df6
2 changed files with 80 additions and 1 deletions

View file

@ -13,7 +13,7 @@ from musiksammlung.cover import copy_covers
from musiksammlung.llm_parser import parse_tracklist from musiksammlung.llm_parser import parse_tracklist
from musiksammlung.models import Album from musiksammlung.models import Album
from musiksammlung.ocr import ocr_images from musiksammlung.ocr import ocr_images
from musiksammlung.organizer import apply_mapping, build_mapping from musiksammlung.organizer import apply_mapping, build_mapping, check_disc_counts
from musiksammlung.playlist import generate_playlist from musiksammlung.playlist import generate_playlist
from musiksammlung.ripper import RipperConfig, interactive_rip from musiksammlung.ripper import RipperConfig, interactive_rip
from musiksammlung.tagger import tag_album from musiksammlung.tagger import tag_album
@ -146,6 +146,36 @@ def apply(
raw = json.loads(album_json.read_text(encoding="utf-8")) raw = json.loads(album_json.read_text(encoding="utf-8"))
album = Album.model_validate(raw) album = Album.model_validate(raw)
# Prüfe Track-Anzahl pro Disc
checks = check_disc_counts(album, input_dir)
problems = [c for c in checks if not c.ok]
if problems:
typer.echo(
"\nFEHLER: Track-Diskrepanz zwischen gerippten Dateien und album.json:\n",
err=True,
)
for c in checks:
status = "OK" if c.ok else "!!"
typer.echo(
f" [{status}] Disc {c.disc_number}: "
f"{c.audio_file_count} Datei(en), {c.json_track_count} JSON-Track(s)",
err=True,
)
if c.surplus_files:
typer.echo(
f"{c.surplus_files} Track(s) fehlen im JSON "
f"(Tracks {c.json_track_count + 1}{c.audio_file_count} eintragen)",
err=True,
)
elif c.surplus_json:
typer.echo(
f"{c.surplus_json} JSON-Eintrag/Einträge ohne Audiodatei "
f"(Tracks {c.audio_file_count + 1}{c.json_track_count} prüfen)",
err=True,
)
typer.echo(f"\nBitte {album_json} korrigieren und erneut aufrufen.", err=True)
raise typer.Exit(1)
mapping = build_mapping(album, input_dir, output_dir) mapping = build_mapping(album, input_dir, output_dir)
typer.echo(f"Mapping: {len(mapping)} Dateien") typer.echo(f"Mapping: {len(mapping)} Dateien")
for src, dst in mapping.items(): for src, dst in mapping.items():

View file

@ -5,6 +5,7 @@ from __future__ import annotations
import logging import logging
import re import re
import shutil import shutil
from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from musiksammlung.config import AUDIO_EXTENSIONS from musiksammlung.config import AUDIO_EXTENSIONS
@ -13,6 +14,29 @@ from musiksammlung.models import Album
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@dataclass
class DiscCheck:
"""Ergebnis der Track-Zählung für eine einzelne Disc."""
disc_number: int
audio_file_count: int
json_track_count: int
@property
def ok(self) -> bool:
return self.audio_file_count == self.json_track_count
@property
def surplus_files(self) -> int:
"""Dateien ohne JSON-Eintrag (Tracks fehlen im JSON)."""
return max(0, self.audio_file_count - self.json_track_count)
@property
def surplus_json(self) -> int:
"""JSON-Einträge ohne Datei (Dateien fehlen im Verzeichnis)."""
return max(0, self.json_track_count - self.audio_file_count)
def _sanitize_filename(name: str) -> str: def _sanitize_filename(name: str) -> str:
"""Entfernt problematische Zeichen aus Dateinamen.""" """Entfernt problematische Zeichen aus Dateinamen."""
return re.sub(r'[<>:"/\\|?*]', "_", name).strip() return re.sub(r'[<>:"/\\|?*]', "_", name).strip()
@ -28,6 +52,31 @@ def discover_audio_files(directory: Path) -> list[Path]:
return sorted(files, key=extract_number) return sorted(files, key=extract_number)
def check_disc_counts(album: Album, input_dir: Path) -> list[DiscCheck]:
"""Vergleicht Dateianzahl und JSON-Track-Anzahl pro Disc.
Args:
album: Validiertes Album-Modell
input_dir: Verzeichnis mit gerippten Dateien (enthält CD1/, CD2/, ... bei Multi-Disc)
Returns:
Liste von DiscCheck-Objekten auch für korrekte Discs (ok=True).
"""
multi_disc = len(album.discs) > 1
results: list[DiscCheck] = []
for disc in album.discs:
source_dir = input_dir / f"CD{disc.disc_number}" if multi_disc else input_dir
file_count = len(discover_audio_files(source_dir)) if source_dir.exists() else 0
results.append(DiscCheck(
disc_number=disc.disc_number,
audio_file_count=file_count,
json_track_count=len(disc.tracks),
))
return results
def build_mapping( def build_mapping(
album: Album, album: Album,
input_dir: Path, input_dir: Path,