track_updater

Track-level year update operations.

Handles applying year values to tracks with validation, deduplication, retry logic, and bulk update batching.

TrackUpdater

TrackUpdater(
    *,
    track_processor,
    retry_handler,
    console_logger,
    error_logger,
    config
)

Applies year values to tracks with validation and retry.

Responsibilities: - Collect and validate tracks that need year updates - Execute bulk updates with batching and retry - Record successful updates in change log

Source code in src/core/tracks/track_updater.py

def __init__(
    self,
    *,
    track_processor: TrackProcessor,
    retry_handler: DatabaseRetryHandler,
    console_logger: logging.Logger,
    error_logger: logging.Logger,
    config: AppConfig,
) -> None:
    self.track_processor = track_processor
    self.retry_handler = retry_handler
    self.console_logger = console_logger
    self.error_logger = error_logger
    self.config = config

update_tracks_for_album async

update_tracks_for_album(
    artist,
    album,
    album_tracks,
    year,
    updated_tracks,
    changes_log,
)

Update tracks for a specific album and record changes.

Parameters:

Name	Type	Description	Default
artist	str	Artist name	required
album	str	Album name	required
album_tracks	list[TrackDict]	List of tracks in the album	required
year	str	Year to set	required
updated_tracks	list[TrackDict]	List to append updated tracks to	required
changes_log	list[ChangeLogEntry]	List to append change entries to	required

Source code in src/core/tracks/track_updater.py

async def update_tracks_for_album(
    self,
    artist: str,
    album: str,
    album_tracks: list[TrackDict],
    year: str,
    updated_tracks: list[TrackDict],
    changes_log: list[ChangeLogEntry],
) -> None:
    """Update tracks for a specific album and record changes.

    Args:
        artist: Artist name
        album: Album name
        album_tracks: List of tracks in the album
        year: Year to set
        updated_tracks: List to append updated tracks to
        changes_log: List to append change entries to

    """
    track_ids, tracks_needing_update = self._collect_tracks_for_update(album_tracks, year)

    if not track_ids:
        self.console_logger.info(
            "All tracks for '%s - %s' already have year %s, skipping update",
            artist,
            album,
            year,
        )
        return

    successful, _ = await self.update_album_tracks_bulk_async(
        tracks=tracks_needing_update,
        year=year,
        artist=artist,
        album=album,
    )

    if successful > 0:
        self.record_successful_updates(tracks_needing_update, year, artist, album, updated_tracks, changes_log)

record_successful_updates staticmethod

record_successful_updates(
    tracks, year, artist, album, updated_tracks, changes_log
)

Record successful track updates.

Parameters:

Name	Type	Description	Default
tracks	list[TrackDict]	Tracks that were updated	required
year	str	New year value	required
artist	str	Artist name	required
album	str	Album name	required
updated_tracks	list[TrackDict]	List to append updated tracks to	required
changes_log	list[ChangeLogEntry]	List to append change entries to	required

Source code in src/core/tracks/track_updater.py

@staticmethod
def record_successful_updates(
    tracks: list[TrackDict],
    year: str,
    artist: str,
    album: str,
    updated_tracks: list[TrackDict],
    changes_log: list[ChangeLogEntry],
) -> None:
    """Record successful track updates.

    Args:
        tracks: Tracks that were updated
        year: New year value
        artist: Artist name
        album: Album name
        updated_tracks: List to append updated tracks to
        changes_log: List to append change entries to

    """
    for track in tracks:
        updated_tracks.append(track.copy(year=year))

        old_year_value = track.get("year")
        changes_log.append(
            ChangeLogEntry(
                timestamp=datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S"),
                change_type="year_update",
                track_id=str(track.get("id", "")),
                artist=artist,
                album_name=album,
                track_name=str(track.get("name", "")),
                year_before_mgu=str(old_year_value) if old_year_value is not None else "",
                year_set_by_mgu=year,
            )
        )

        # Preserve original year in year_before_mgu (only if not already set)
        if not track.year_before_mgu:
            track.year_before_mgu = str(old_year_value) if old_year_value else ""

        # Keep the in-memory snapshot aligned
        track.year = year
        track.year_set_by_mgu = year

update_album_tracks_bulk_async async

update_album_tracks_bulk_async(tracks, year, artist, album)

Update year for multiple tracks in bulk.

Parameters:

Name	Type	Description	Default
tracks	list[TrackDict]	List of tracks to update	required
year	str	Year to set	required
artist	str	Artist name for contextual logging	required
album	str	Album name for contextual logging	required

Returns:

Type	Description
tuple[int, int]	Tuple of (successful_count, failed_count)

Source code in src/core/tracks/track_updater.py

async def update_album_tracks_bulk_async(
    self,
    tracks: list[TrackDict],
    year: str,
    artist: str,
    album: str,
) -> tuple[int, int]:
    """Update year for multiple tracks in bulk.

    Args:
        tracks: List of tracks to update
        year: Year to set
        artist: Artist name for contextual logging
        album: Album name for contextual logging

    Returns:
        Tuple of (successful_count, failed_count)

    """
    # Extract and validate track IDs
    track_ids = [str(track.get("id", "")) for track in tracks if track.get("id")]
    valid_track_ids = self._validate_track_ids(track_ids, artist=artist, album=album)
    if not valid_track_ids:
        self.console_logger.warning(
            "No valid track IDs to update for %s - %s (input: %d tracks, all IDs empty or invalid)",
            artist,
            album,
            len(tracks),
        )
        return 0, len(tracks)

    # Build mapping from track_id to track name for logging
    track_names: dict[str, str] = {str(track.get("id", "")): str(track.get("name", "")) for track in tracks if track.get("id")}

    # Process in batches
    batch_size = self.config.apple_script_concurrency
    successful = 0
    failed = 0

    for i in range(0, len(valid_track_ids), batch_size):
        batch = valid_track_ids[i : i + batch_size]

        # Create update tasks with retry logic
        tasks: list[Coroutine[Any, Any, bool]] = []
        for track_id in batch:
            task = self._update_track_with_retry(
                track_id=track_id,
                new_year=year,
                original_artist=artist,
                original_album=album,
                original_track=track_names.get(track_id),
            )
            tasks.append(task)

        # Execute batch
        results = await asyncio.gather(*tasks, return_exceptions=True)

        # Count results
        for index, result in enumerate(results):
            if isinstance(result, Exception):
                failed += 1
                track_id_in_batch = batch[index] if index < len(batch) else "unknown"
                self.error_logger.error(
                    "Failed to update track %s (artist=%s, album=%s, year=%s): %s",
                    track_id_in_batch,
                    artist,
                    album,
                    year,
                    result,
                )
            elif result:
                successful += 1
            else:
                failed += 1

    # Log summary
    self.console_logger.info(
        "Year update results: %d successful, %d failed",
        successful,
        failed,
    )

    return successful, failed