Add broken symlink diagnostics
This commit is contained in:
@@ -6,7 +6,7 @@ from serato_doctor.crate_parser import load_library_crates
|
||||
from serato_doctor.health import analyze_health
|
||||
from serato_doctor.logging import configure_logging
|
||||
from serato_doctor.models.library import Library
|
||||
from serato_doctor.scanner import scan_audio
|
||||
from serato_doctor.scanner import scan_filesystem
|
||||
from serato_doctor.report import write_csv, write_missing_report
|
||||
|
||||
|
||||
@@ -56,9 +56,11 @@ def main():
|
||||
logger.debug("Music directory: %s", config.music)
|
||||
|
||||
crates = load_library_crates(config.serato, config.reference_roots)
|
||||
filesystem = scan_filesystem(config.music)
|
||||
library = Library.from_crates(
|
||||
crates=crates,
|
||||
tracks=scan_audio(config.music),
|
||||
tracks=filesystem.tracks,
|
||||
broken_symlinks=filesystem.broken_symlinks,
|
||||
)
|
||||
results = library.reconcile_by_filename()
|
||||
missing_count = sum(1 for result in results if not result.exists_by_filename)
|
||||
@@ -90,6 +92,7 @@ def main():
|
||||
)
|
||||
print(f"Unused Tracks: {health.unused_tracks}")
|
||||
print(f"Suggested Matches: {health.suggested_matches}")
|
||||
print(f"Broken Symlinks: {health.broken_symlinks}")
|
||||
print(f"Static Crates: {health.static_crates}")
|
||||
print(f"Smart Crates: {health.smart_crates}")
|
||||
print(
|
||||
|
||||
@@ -62,6 +62,7 @@ def analyze_health(library: Library) -> HealthReport:
|
||||
),
|
||||
unused_tracks=unused_count,
|
||||
suggested_matches=suggested_count,
|
||||
broken_symlinks=len(library.broken_symlinks),
|
||||
static_crates=sum(
|
||||
crate.kind is CrateKind.STATIC for crate in library.crates
|
||||
),
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
from serato_doctor.models.crate import Crate, CrateKind
|
||||
from serato_doctor.models.duplicate import DuplicateGroup, DuplicateKind
|
||||
from serato_doctor.models.filesystem import BrokenSymlink, FilesystemScan
|
||||
from serato_doctor.models.health import HealthReport
|
||||
from serato_doctor.models.library import Library
|
||||
from serato_doctor.models.match import MatchEvidence, TrackMatch
|
||||
@@ -12,6 +13,8 @@ __all__ = [
|
||||
"DiskTrack",
|
||||
"DuplicateGroup",
|
||||
"DuplicateKind",
|
||||
"BrokenSymlink",
|
||||
"FilesystemScan",
|
||||
"HealthReport",
|
||||
"Library",
|
||||
"MatchEvidence",
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Optional, Tuple
|
||||
|
||||
from serato_doctor.models.track import DiskTrack
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class BrokenSymlink:
|
||||
"""A symbolic link whose target cannot be resolved."""
|
||||
|
||||
path: Path
|
||||
target: Optional[Path]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FilesystemScan:
|
||||
"""Read-only findings from one traversal of a music folder."""
|
||||
|
||||
tracks: Tuple[DiskTrack, ...]
|
||||
broken_symlinks: Tuple[BrokenSymlink, ...]
|
||||
@@ -19,6 +19,7 @@ class HealthReport:
|
||||
suspected_cloud_conflict_files: int
|
||||
unused_tracks: int
|
||||
suggested_matches: int
|
||||
broken_symlinks: int
|
||||
static_crates: int
|
||||
smart_crates: int
|
||||
unknown_crates: int
|
||||
|
||||
@@ -2,6 +2,7 @@ from dataclasses import dataclass
|
||||
from typing import Iterable, Tuple
|
||||
|
||||
from serato_doctor.models.crate import Crate
|
||||
from serato_doctor.models.filesystem import BrokenSymlink
|
||||
from serato_doctor.models.reference import ReferenceResult, TrackReference
|
||||
from serato_doctor.models.track import DiskTrack
|
||||
|
||||
@@ -13,18 +14,27 @@ class Library:
|
||||
references: Tuple[TrackReference, ...]
|
||||
tracks: Tuple[DiskTrack, ...]
|
||||
crates: Tuple[Crate, ...] = ()
|
||||
broken_symlinks: Tuple[BrokenSymlink, ...] = ()
|
||||
|
||||
@classmethod
|
||||
def build(
|
||||
cls,
|
||||
references: Iterable[TrackReference],
|
||||
tracks: Iterable[DiskTrack],
|
||||
broken_symlinks: Iterable[BrokenSymlink] = (),
|
||||
) -> "Library":
|
||||
return cls(tuple(references), tuple(tracks))
|
||||
return cls(
|
||||
tuple(references),
|
||||
tuple(tracks),
|
||||
broken_symlinks=tuple(broken_symlinks),
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_crates(
|
||||
cls, crates: Iterable[Crate], tracks: Iterable[DiskTrack]
|
||||
cls,
|
||||
crates: Iterable[Crate],
|
||||
tracks: Iterable[DiskTrack],
|
||||
broken_symlinks: Iterable[BrokenSymlink] = (),
|
||||
) -> "Library":
|
||||
crate_tuple = tuple(crates)
|
||||
references = tuple(
|
||||
@@ -32,7 +42,12 @@ class Library:
|
||||
for crate in crate_tuple
|
||||
for reference in crate.references
|
||||
)
|
||||
return cls(references, tuple(tracks), crate_tuple)
|
||||
return cls(
|
||||
references,
|
||||
tuple(tracks),
|
||||
crate_tuple,
|
||||
tuple(broken_symlinks),
|
||||
)
|
||||
|
||||
def reconcile_by_filename(self) -> Tuple[ReferenceResult, ...]:
|
||||
disk_names = {track.filename for track in self.tracks}
|
||||
|
||||
@@ -1,14 +1,23 @@
|
||||
from pathlib import Path
|
||||
|
||||
from serato_doctor.models.filesystem import BrokenSymlink, FilesystemScan
|
||||
from serato_doctor.models.track import DiskTrack
|
||||
|
||||
AUDIO_SUFFIXES = {".mp3", ".m4a", ".wav", ".aif", ".aiff", ".flac"}
|
||||
|
||||
|
||||
def scan_audio(folder: Path) -> list[DiskTrack]:
|
||||
def scan_filesystem(folder: Path) -> FilesystemScan:
|
||||
tracks = []
|
||||
broken_symlinks = []
|
||||
|
||||
for path in folder.rglob("*"):
|
||||
if path.is_symlink() and not path.exists():
|
||||
try:
|
||||
target = path.readlink()
|
||||
except OSError:
|
||||
target = None
|
||||
broken_symlinks.append(BrokenSymlink(path=path, target=target))
|
||||
continue
|
||||
if not path.is_file():
|
||||
continue
|
||||
if path.suffix.lower() not in AUDIO_SUFFIXES:
|
||||
@@ -28,4 +37,13 @@ def scan_audio(folder: Path) -> list[DiskTrack]:
|
||||
)
|
||||
)
|
||||
|
||||
return tracks
|
||||
return FilesystemScan(
|
||||
tracks=tuple(tracks),
|
||||
broken_symlinks=tuple(broken_symlinks),
|
||||
)
|
||||
|
||||
|
||||
def scan_audio(folder: Path) -> list[DiskTrack]:
|
||||
"""Scan audio files while preserving the prototype API."""
|
||||
|
||||
return list(scan_filesystem(folder).tracks)
|
||||
|
||||
Reference in New Issue
Block a user