year_repair

Generic repair and revert utilities.

Provides helpers to revert year changes using either a changes report or a backup CSV of the track list. Designed to be artist/album agnostic.

YearRevertProcessorProtocol

Bases: Protocol

Protocol for track processor operations needed by year revert logic.

fetch_tracks_async async

fetch_tracks_async(
    artist=None,
    force_refresh=False,
    dry_run_test_tracks=None,
    ignore_test_filter=False,
)

Fetch tracks from the music library.

Source code in src/core/models/year_repair.py

async def fetch_tracks_async(
    self,
    artist: str | None = None,
    force_refresh: bool = False,
    dry_run_test_tracks: list[TrackDict] | None = None,
    ignore_test_filter: bool = False,
) -> list[TrackDict]:
    """Fetch tracks from the music library."""
    ...

update_track_async async

update_track_async(
    track_id,
    new_track_name=None,
    new_album_name=None,
    new_genre=None,
    new_year=None,
    track_status=None,
    original_artist=None,
    original_album=None,
    original_track=None,
)

Update a track's metadata in the music library.

Source code in src/core/models/year_repair.py

async def update_track_async(
    self,
    track_id: str,
    new_track_name: str | None = None,
    new_album_name: str | None = None,
    new_genre: str | None = None,
    new_year: str | None = None,
    track_status: str | None = None,
    original_artist: str | None = None,
    original_album: str | None = None,
    original_track: str | None = None,
) -> bool:
    """Update a track's metadata in the music library."""
    ...

RevertTarget dataclass

RevertTarget(track_id, track_name, album, year_before_mgu)

Describes a track that should have its year reverted.

build_revert_targets

build_revert_targets(
    *, config, artist, album, backup_csv_path=None
)

Build revert targets from either a backup CSV or the changes report.

If backup_csv_path is provided and readable, it is used; otherwise the changes_report.csv is used.

Source code in src/core/models/year_repair.py

def build_revert_targets(
    *,
    config: AppConfig,
    artist: str,
    album: str | None,
    backup_csv_path: str | None = None,
) -> list[RevertTarget]:
    """Build revert targets from either a backup CSV or the changes report.

    If backup_csv_path is provided and readable, it is used; otherwise
    the changes_report.csv is used.
    """
    if backup_csv_path and (targets := _read_backup_csv(backup_csv_path, artist, album)):
        return targets
    return _read_changes_report(config, artist, album)

apply_year_reverts async

apply_year_reverts(*, track_processor, artist, targets)

Apply year reverts to Music.app given revert targets.

Returns (updated_count, missing_count, change_log_entries).

Source code in src/core/models/year_repair.py

async def apply_year_reverts(
    *,
    track_processor: YearRevertProcessorProtocol,
    artist: str,
    targets: Iterable[RevertTarget],
) -> tuple[int, int, list[dict[str, str]]]:
    """Apply year reverts to Music.app given revert targets.

    Returns (updated_count, missing_count, change_log_entries).
    """
    current_tracks = await track_processor.fetch_tracks_async(artist=artist, ignore_test_filter=True)
    by_id, by_album_track, by_name = _build_track_lookups(current_tracks)

    updated = 0
    missing = 0
    change_log: list[dict[str, str]] = []

    for target in targets:
        track = _find_track_for_target(
            target,
            by_id=by_id,
            by_album_track=by_album_track,
            by_name=by_name,
        )
        if track is None:
            missing += 1
            continue

        track_id = _normalize_text(track.get("id"))
        if not track_id:
            missing += 1
            continue

        if not await _revert_track_year(track_processor, track=track, track_id=track_id, reverted_year=target.year_before_mgu):
            continue

        updated += 1
        change_log.append(_build_year_change_entry(track, target.year_before_mgu))

    return updated, missing, change_log