year_update

Year Update Service.

Handles year update operations for tracks including revert, update, and pipeline steps.

YearUpdateService

YearUpdateService(
    track_processor,
    year_retriever,
    snapshot_manager,
    config,
    console_logger,
    error_logger,
    cleaning_service=None,
    artist_renamer=None,
)

Service for year update operations.

Handles year updates, reverts, and pipeline steps for track metadata.

Initialize the year update service.

Parameters:

Name	Type	Description	Default
track_processor	TrackProcessor	Processor for fetching and updating tracks.	required
year_retriever	YearRetriever	Retriever for year operations.	required
snapshot_manager	PipelineSnapshotManager	Manager for pipeline snapshots.	required
config	AppConfig	Typed application configuration.	required
console_logger	Logger	Logger for console output.	required
error_logger	Logger	Logger for error output.	required
cleaning_service	TrackCleaningService | None	Optional service for cleaning track metadata.	None
artist_renamer	ArtistRenamer | None	Optional service for renaming artists.	None

Source code in src/app/year_update.py

def __init__(
    self,
    track_processor: TrackProcessor,
    year_retriever: YearRetriever,
    snapshot_manager: PipelineSnapshotManager,
    config: AppConfig,
    console_logger: logging.Logger,
    error_logger: logging.Logger,
    cleaning_service: TrackCleaningService | None = None,
    artist_renamer: ArtistRenamer | None = None,
) -> None:
    """Initialize the year update service.

    Args:
        track_processor: Processor for fetching and updating tracks.
        year_retriever: Retriever for year operations.
        snapshot_manager: Manager for pipeline snapshots.
        config: Typed application configuration.
        console_logger: Logger for console output.
        error_logger: Logger for error output.
        cleaning_service: Optional service for cleaning track metadata.
        artist_renamer: Optional service for renaming artists.
    """
    self._track_processor = track_processor
    self._year_retriever = year_retriever
    self._snapshot_manager = snapshot_manager
    self._config = config
    self._console_logger = console_logger
    self._error_logger = error_logger
    self._cleaning_service = cleaning_service
    self._artist_renamer = artist_renamer
    self._test_artists: set[str] | None = None

set_test_artists

set_test_artists(test_artists)

Set test artists for filtering.

Parameters:

Name	Type	Description	Default
test_artists	set[str] | None	Set of artist names to filter to, or None to process all.	required

Source code in src/app/year_update.py

def set_test_artists(self, test_artists: set[str] | None) -> None:
    """Set test artists for filtering.

    Args:
        test_artists: Set of artist names to filter to, or None to process all.
    """
    self._test_artists = test_artists

get_tracks_for_year_update async

get_tracks_for_year_update(artist)

Get tracks for year update based on artist filter.

Parameters:

Name	Type	Description	Default
artist	str | None	Optional artist filter.	required

Returns:

Type	Description
list[TrackDict] | None	List of tracks or None if not found.

Source code in src/app/year_update.py

async def get_tracks_for_year_update(self, artist: str | None) -> list[TrackDict] | None:
    """Get tracks for year update based on artist filter.

    Args:
        artist: Optional artist filter.

    Returns:
        List of tracks or None if not found.
    """
    # For full library (no artist filter), use batch fetcher to avoid AppleScript timeout
    # For specific artist, use direct fetch which is more efficient
    fetched_tracks: list[TrackDict] | None
    if artist is None:
        fetched_tracks = await self._track_processor.fetch_tracks_in_batches()
    else:
        fetched_tracks = await self._track_processor.fetch_tracks_async(artist=artist)

    # Filter by test_artists if in test mode
    if self._test_artists and fetched_tracks:
        fetched_tracks = [t for t in fetched_tracks if t.get("artist") in self._test_artists]
        self._console_logger.info(
            "Test mode: filtered to %d tracks for %d test artists",
            len(fetched_tracks),
            len(self._test_artists),
        )

    if not fetched_tracks:
        self._console_logger.warning(
            "No tracks found for year update (artist=%s, test_mode=%s)",
            artist or "all",
            bool(self._test_artists),
        )
        return None
    return fetched_tracks

run_update_years async

run_update_years(artist, force, fresh=False)

Update album years for all or specific artist.

Parameters:

Name	Type	Description	Default
artist	str | None	Optional artist filter.	required
force	bool	Force update even if year exists.	required
fresh	bool	Fresh mode - invalidate cache before processing, implies force.	False

Source code in src/app/year_update.py

async def run_update_years(self, artist: str | None, force: bool, fresh: bool = False) -> None:
    """Update album years for all or specific artist.

    Args:
        artist: Optional artist filter.
        force: Force update even if year exists.
        fresh: Fresh mode - invalidate cache before processing, implies force.
    """
    self._console_logger.info(
        "Starting year update operation%s",
        f" for artist: {artist}" if artist else " for all artists",
    )

    tracks = await self.get_tracks_for_year_update(artist)
    if not tracks:
        return

    # Preprocessing - clean metadata first
    if self._cleaning_service:
        self._console_logger.info("Preprocessing: Cleaning metadata...")
        await self._cleaning_service.clean_all_metadata_with_logs(tracks)

    # Preprocessing - rename artists
    if self._artist_renamer and self._artist_renamer.has_mapping:
        self._console_logger.info("Preprocessing: Renaming artists...")
        await self._artist_renamer.rename_tracks(tracks)

    # Process album years
    success = await self._year_retriever.process_album_years(tracks, force=force, fresh=fresh)

    if success:
        self._console_logger.info("Year update operation completed successfully")
    else:
        self._error_logger.error(
            "Year update operation failed (artist=%s, force=%s, fresh=%s, tracks_count=%d)",
            artist or "all",
            force,
            fresh,
            len(tracks) if tracks else 0,
        )

run_revert_years async

run_revert_years(artist, album, backup_csv=None)

Revert year updates for an artist (optionally per album).

Uses backup CSV if provided; otherwise uses the latest changes_report.csv.

Parameters:

Name	Type	Description	Default
artist	str	Artist name.	required
album	str | None	Optional album name filter.	required
backup_csv	str | None	Optional path to backup CSV file.	None

Source code in src/app/year_update.py

async def run_revert_years(self, artist: str, album: str | None, backup_csv: str | None = None) -> None:
    """Revert year updates for an artist (optionally per album).

    Uses backup CSV if provided; otherwise uses the latest changes_report.csv.

    Args:
        artist: Artist name.
        album: Optional album name filter.
        backup_csv: Optional path to backup CSV file.
    """
    self._console_logger.info(
        "Starting revert of year changes for '%s'%s",
        artist,
        f" - {album}" if album else " (all albums)",
    )

    targets = repair_utils.build_revert_targets(
        config=self._config,
        artist=artist,
        album=album,
        backup_csv_path=backup_csv,
    )

    if not targets:
        self._console_logger.warning(
            "No revert targets found for '%s'%s",
            artist,
            f" - {album}" if album else "",
        )
        return

    updated, missing, changes_log = await repair_utils.apply_year_reverts(
        track_processor=self._track_processor,
        artist=artist,
        targets=targets,
    )

    self._console_logger.info("Revert complete: %d tracks updated, %d not found", updated, missing)

    if changes_log:
        revert_path = get_full_log_path(self._config, "changes_report_file", "csv/changes_revert.csv")
        save_changes_report(changes=changes_log, file_path=revert_path, console_logger=self._console_logger, error_logger=self._error_logger)
        self._console_logger.info("Revert changes report saved to %s", revert_path)

run_restore_release_years async

run_restore_release_years(
    artist=None, album=None, threshold=5
)

Restore year from Apple Music's release_year field.

Finds albums where 'year' differs dramatically from 'release_year' and updates year to match release_year.

Parameters:

Name	Type	Description	Default
artist	str | None	Optional artist name filter.	None
album	str | None	Optional album name filter (requires artist).	None
threshold	int	Year difference threshold (default: 5 years).	5

Source code in src/app/year_update.py

async def run_restore_release_years(
    self,
    artist: str | None = None,
    album: str | None = None,
    threshold: int = 5,
) -> None:
    """Restore year from Apple Music's release_year field.

    Finds albums where 'year' differs dramatically from 'release_year'
    and updates year to match release_year.

    Args:
        artist: Optional artist name filter.
        album: Optional album name filter (requires artist).
        threshold: Year difference threshold (default: 5 years).

    """
    target_msg = self._format_restore_target(artist, album)
    self._console_logger.info(
        "Starting restore_release_years%s (threshold: %d years)",
        target_msg,
        threshold,
    )

    tracks = await self._get_filtered_tracks_for_restore(artist, album)
    if not tracks:
        return

    albums_to_restore = self._find_albums_needing_restoration(tracks, threshold)
    if not albums_to_restore:
        self._console_logger.info("No albums found needing year restoration")
        return

    self._console_logger.info("Found %d albums needing year restoration:", len(albums_to_restore))

    changes_log, updated_count, failed_count = await self._process_album_restorations(albums_to_restore)

    self._console_logger.info(
        "Restore complete: %d tracks updated, %d failed",
        updated_count,
        failed_count,
    )

    self._save_restore_changes_report(changes_log)

update_all_years_with_logs async

update_all_years_with_logs(tracks, force, fresh=False)

Update years for all tracks and return change logs (Step 4 of pipeline).

Parameters:

Name	Type	Description	Default
tracks	list[TrackDict]	List of tracks to process.	required
force	bool	Force update - bypass cache/skip checks and re-query API for all albums.	required
fresh	bool	Fresh mode - invalidate cache before processing, implies force.	False

Returns:

Type	Description
list[ChangeLogEntry]	List of change log entries.

Source code in src/app/year_update.py

async def update_all_years_with_logs(self, tracks: list[TrackDict], force: bool, fresh: bool = False) -> list[ChangeLogEntry]:
    """Update years for all tracks and return change logs (Step 4 of pipeline).

    Args:
        tracks: List of tracks to process.
        force: Force update - bypass cache/skip checks and re-query API for all albums.
        fresh: Fresh mode - invalidate cache before processing, implies force.

    Returns:
        List of change log entries.
    """
    self._console_logger.info("Step 4/4: Updating album years")
    changes_log: list[ChangeLogEntry] = []

    # fresh implies force
    if fresh:
        force = True
        self._console_logger.info("Fresh mode: invalidating album years cache")
        await self._track_processor.cache_service.invalidate_all_albums()

    try:
        updated_tracks, year_changes = await self._year_retriever.get_album_years_with_logs(tracks, force=force)
        self._year_retriever.set_last_updated_tracks(updated_tracks)

        self._snapshot_manager.update_tracks(updated_tracks)
        changes_log = year_changes
        self._console_logger.info("Step 4 completed successfully with %d changes", len(changes_log))
    except (OSError, ValueError, RuntimeError) as e:
        self._error_logger.exception(
            "Step 4 year retrieval failed: %s",
            type(e).__name__,
        )
        # Add error marker to ensure data consistency
        now = datetime.now(UTC)
        changes_log.append(
            ChangeLogEntry(
                timestamp=now.strftime("%Y-%m-%d %H:%M:%S"),
                change_type="year_update_error",
                track_id="",
                artist="ERROR",
                album_name=f"Year update failed: {type(e).__name__}",
                track_name=str(e)[:100],
                year_before_mgu="",
                year_set_by_mgu="",
            )
        )

    return changes_log