serato-doctor/serato_doctor/health.py

from collections import Counter

from serato_doctor.duplicates import find_duplicate_groups
from serato_doctor.matching import MatchingEngine, normalize
from serato_doctor.models.crate import CrateKind
from serato_doctor.models.duplicate import DuplicateKind
from serato_doctor.models.health import HealthReport
from serato_doctor.models.library import Library


def analyze_health(library: Library) -> HealthReport:
    """Calculate defensible health metrics without changing the library."""

    dynamic_sources = {
        crate.path for crate in library.crates if crate.kind is CrateKind.SMART
    }
    results = tuple(
        result
        for result in library.reconcile_by_filename()
        if result.reference.source not in dynamic_sources
    )
    missing = [result for result in results if not result.exists_by_filename]
    healthy_count = len(results) - len(missing)
    score = (
        round(healthy_count / len(results) * 100, 1) if results else None
    )

    duplicate_groups = find_duplicate_groups(library.tracks)
    exact_duplicates = [
        group
        for group in duplicate_groups
        if group.kind is DuplicateKind.EXACT_NAME
    ]
    cloud_conflicts = [
        group
        for group in duplicate_groups
        if group.kind is DuplicateKind.CLOUD_CONFLICT
    ]
    referenced_names = {
        normalize(reference.filename) for reference in library.references
    }
    unused_count = sum(
        1
        for track in library.tracks
        if normalize(track.filename) not in referenced_names
    )

    matcher = MatchingEngine(library.tracks)
    suggested_count = sum(
        bool(matcher.candidates_for(result.reference)) for result in missing
    )

    database_tracks = library.database.tracks if library.database else ()
    database_names = {normalize(track.filename) for track in database_tracks}
    library_names = {normalize(track.filename) for track in library.tracks}
    database_path_counts = Counter(
        normalize(str(track.path)) for track in database_tracks
    )
    missing_database_tracks = [
        track for track in database_tracks if not track.path.exists()
    ]

    return HealthReport(
        score=score,
        total_references=len(library.references),
        scored_references=len(results),
        healthy_references=healthy_count,
        missing_references=len(missing),
        unique_missing_filenames=len(
            {result.reference.filename for result in missing}
        ),
        disk_tracks=len(library.tracks),
        duplicate_filename_groups=len(exact_duplicates),
        duplicate_files=sum(group.extra_files for group in exact_duplicates),
        suspected_cloud_conflict_groups=len(cloud_conflicts),
        suspected_cloud_conflict_files=sum(
            group.extra_files for group in cloud_conflicts
        ),
        unused_tracks=unused_count,
        orphan_candidates=(
            sum(
                normalize(track.filename) not in referenced_names
                and normalize(track.filename) not in database_names
                for track in library.tracks
            )
            if library.database
            else None
        ),
        suggested_matches=suggested_count,
        broken_symlinks=len(library.broken_symlinks),
        static_crates=sum(
            crate.kind is CrateKind.STATIC for crate in library.crates
        ),
        smart_crates=sum(
            crate.kind is CrateKind.SMART and crate.is_smart_definition
            for crate in library.crates
        ),
        smart_crate_containers=sum(
            crate.kind is CrateKind.SMART and not crate.is_smart_definition
            for crate in library.crates
        ),
        unknown_crates=sum(
            crate.kind is CrateKind.UNKNOWN for crate in library.crates
        ),
        dynamic_references_excluded=sum(
            len(crate.references)
            for crate in library.crates
            if crate.kind is CrateKind.SMART
        ),
        database_present=library.database is not None,
        database_entries=len(database_tracks),
        database_library_matches=sum(
            normalize(track.filename) in library_names for track in database_tracks
        ),
        database_unmatched_entries=sum(
            normalize(track.filename) not in library_names for track in database_tracks
        ),
        database_missing_paths=len(missing_database_tracks),
        database_missing_unique_filenames=len(
            {normalize(track.filename) for track in missing_database_tracks}
        ),
        tracks_missing_from_database=sum(
            normalize(track.filename) not in database_names for track in library.tracks
        ) if library.database else 0,
        duplicate_database_paths=sum(
            count - 1 for count in database_path_counts.values() if count > 1
        ),
    )