year_determination

Year determination logic for albums.

This module contains the core logic for determining album years from various sources: local data, cache, and external APIs.

YearDeterminator

YearDeterminator(
    *,
    cache_service,
    external_api,
    pending_verification,
    consistency_checker,
    fallback_handler,
    console_logger,
    error_logger,
    config
)

Determines album year from various sources.

Responsibilities: - Query cache for existing year data - Query external APIs when needed - Apply fallback logic for uncertain updates - Handle prerelease and suspicious albums - Skip albums that already have valid years

Initialize the YearDeterminator.

Parameters:

Name	Type	Description	Default
cache_service	CacheServiceProtocol	Cache service for storing/retrieving years	required
external_api	ExternalApiServiceProtocol	External API service for fetching years	required
pending_verification	PendingVerificationServiceProtocol	Service for managing pending verifications	required
consistency_checker	YearConsistencyChecker	Checker for year consistency	required
fallback_handler	YearFallbackHandler	Handler for fallback logic	required
console_logger	Logger	Logger for console output	required
error_logger	Logger	Logger for error messages	required
config	AppConfig	Typed application configuration	required

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

def __init__(
    self,
    *,
    cache_service: CacheServiceProtocol,
    external_api: ExternalApiServiceProtocol,
    pending_verification: PendingVerificationServiceProtocol,
    consistency_checker: YearConsistencyChecker,
    fallback_handler: YearFallbackHandler,
    console_logger: logging.Logger,
    error_logger: logging.Logger,
    config: AppConfig,
) -> None:
    """Initialize the YearDeterminator.

    Args:
        cache_service: Cache service for storing/retrieving years
        external_api: External API service for fetching years
        pending_verification: Service for managing pending verifications
        consistency_checker: Checker for year consistency
        fallback_handler: Handler for fallback logic
        console_logger: Logger for console output
        error_logger: Logger for error messages
        config: Typed application configuration

    """
    self.cache_service = cache_service
    self.external_api = external_api
    self.pending_verification = pending_verification
    self.consistency_checker = consistency_checker
    self.fallback_handler = fallback_handler
    self.console_logger = console_logger
    self.error_logger = error_logger
    self.config = config

    # Extract configuration from typed model
    processing = config.year_retrieval.processing

    self.skip_prerelease = processing.skip_prerelease
    self.future_year_threshold = processing.future_year_threshold
    self.prerelease_recheck_days = processing.prerelease_recheck_days

determine_album_year async

determine_album_year(
    artist, album, album_tracks, force=False
)

Determine year for album using simplified Linus approach.

Order: dominant year -> cache (high confidence) -> consensus release_year -> API -> None When force=True, skips local sources - always queries API.

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
force	bool	If True, bypass local data and always query API	False

Returns:

Type	Description
str | None	Year string if found, None otherwise

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

async def determine_album_year(
    self,
    artist: str,
    album: str,
    album_tracks: list[TrackDict],
    force: bool = False,
) -> str | None:
    """Determine year for album using simplified Linus approach.

    Order: dominant year -> cache (high confidence) -> consensus release_year -> API -> None
    When force=True, skips local sources - always queries API.

    Args:
        artist: Artist name
        album: Album name
        album_tracks: List of tracks in the album
        force: If True, bypass local data and always query API

    Returns:
        Year string if found, None otherwise

    """
    if debug.year:
        self.console_logger.info(
            "determine_album_year called: artist='%s' album='%s' force=%s",
            artist,
            album,
            force,
        )

    # Try local sources first (unless force mode)
    if not force and (local_year := await self._try_local_sources(artist, album, album_tracks)):
        return local_year

    # Fallback to API - pass current year for year-match comparison
    # Use get_most_common_year (not get_dominant_year) because:
    # - get_dominant_year returns None for "suspicious" years (old albums added recently)
    # - Orchestrator needs the actual year to detect year-match and skip verification
    current_year = self.consistency_checker.get_most_common_year(album_tracks)
    return await self._fetch_from_api(artist, album, album_tracks, current_year)

should_skip_album async

should_skip_album(album_tracks, artist, album, force=False)

Check pre-conditions BEFORE any API call.

Guard clauses prevent wasted API calls by checking inexpensive conditions first. Trusts cache (API data) over Music.app data.

Parameters:

Name	Type	Description	Default
album_tracks	list[TrackDict]	List of tracks in the album	required
artist	str	Artist name for logging	required
album	str	Album name for logging	required
force	bool	If True, never skip - always process the album	False

Returns:

Type	Description
tuple[bool, str]	Tuple of (should_skip, reason) where reason is empty if not skipping.

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

async def should_skip_album(
    self,
    album_tracks: list[TrackDict],
    artist: str,
    album: str,
    force: bool = False,
) -> tuple[bool, str]:
    """Check pre-conditions BEFORE any API call.

    Guard clauses prevent wasted API calls by checking inexpensive conditions first.
    Trusts cache (API data) over Music.app data.

    Args:
        album_tracks: List of tracks in the album
        artist: Artist name for logging
        album: Album name for logging
        force: If True, never skip - always process the album

    Returns:
        Tuple of (should_skip, reason) where reason is empty if not skipping.

    """
    # Force mode bypasses all skip logic
    if force:
        self.console_logger.debug(
            "Force mode: processing '%s - %s' regardless of cache state",
            artist,
            album,
        )
        return False, ""

    # Pre-check 1: Already processed via year_set_by_mgu tracking
    if result := self._check_already_processed(album_tracks, artist, album):
        return result

    # Pre-check 2: Recently rejected by FALLBACK
    rejection_reason = await self._check_recent_rejection(artist, album)
    if rejection_reason:
        self.console_logger.debug(
            "[PRE-CHECK] Skip %s - %s: recently rejected (%s)",
            artist,
            album,
            rejection_reason,
        )
        return True, f"recently_rejected:{rejection_reason}"

    # Check cache and year consistency - API data is more reliable than Music.app
    return await self._check_cache_skip(album_tracks, artist, album)

check_prerelease_status async

check_prerelease_status(artist, album, album_tracks)

Check if the album should be skipped due to prerelease status.

Parameters:

Name	Type	Description	Default
artist	str	Artist name	required
album	str	Album name	required
album_tracks	list[TrackDict]	List of tracks on the album	required

Returns:

Type	Description
bool	True if the album should be skipped, False otherwise

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

async def check_prerelease_status(
    self,
    artist: str,
    album: str,
    album_tracks: list[TrackDict],
) -> bool:
    """Check if the album should be skipped due to prerelease status.

    Args:
        artist: Artist name
        album: Album name
        album_tracks: List of tracks on the album

    Returns:
        True if the album should be skipped, False otherwise

    """
    if not self.skip_prerelease:
        return False

    prerelease_tracks = [
        track for track in album_tracks if is_prerelease_status(track.track_status if isinstance(track.track_status, str) else None)
    ]

    if prerelease_tracks:
        self.console_logger.info(
            "Skipping album '%s - %s': %d of %d tracks are prerelease (read-only)",
            artist,
            album,
            len(prerelease_tracks),
            len(album_tracks),
        )
        await self.pending_verification.mark_for_verification(
            artist,
            album,
            reason="prerelease",
            metadata={
                "track_count": str(len(album_tracks)),
                "prerelease_count": str(len(prerelease_tracks)),
            },
            recheck_days=self.prerelease_recheck_days,
        )
        return True

    return False

check_suspicious_album async

check_suspicious_album(artist, album, album_tracks)

Check if the album is suspicious and should be skipped.

Parameters:

Name	Type	Description	Default
artist	str	Artist name	required
album	str	Album name	required
album_tracks	list[TrackDict]	List of tracks	required

Returns:

Type	Description
bool	True if album should be skipped, False otherwise

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

async def check_suspicious_album(
    self,
    artist: str,
    album: str,
    album_tracks: list[TrackDict],
) -> bool:
    """Check if the album is suspicious and should be skipped.

    Args:
        artist: Artist name
        album: Album name
        album_tracks: List of tracks

    Returns:
        True if album should be skipped, False otherwise

    """
    try:
        album_str = album or ""
        non_empty_years = [str(track.get("year")) for track in album_tracks if track.get("year") and str(track.get("year")).strip()]
        unique_years = set(non_empty_years) if non_empty_years else set()

        if len(album_str) <= SUSPICIOUS_ALBUM_MIN_LEN and len(unique_years) >= SUSPICIOUS_MANY_YEARS:
            self.console_logger.warning(
                "Safety check: Suspicious album '%s - %s' detected (%d unique years, name length=%d). "
                "Skipping automatic updates and marking for verification.",
                artist,
                album,
                len(unique_years),
                len(album_str),
            )
            await self.pending_verification.mark_for_verification(
                artist,
                album,
                reason="suspicious_album_name",
                metadata={
                    "unique_years": str(len(unique_years)),
                    "album_name_length": str(len(album_str)),
                },
            )
            return True
    except (AttributeError, TypeError, ValueError) as e:
        self.error_logger.exception(
            "Error during suspicious album safety check for '%s - %s': %s",
            artist,
            album,
            e,
        )

    return False

handle_future_years async

handle_future_years(
    artist, album, album_tracks, future_years
)

Handle case when future years are found in tracks.

Parameters:

Name	Type	Description	Default
artist	str	Artist name	required
album	str	Album name	required
album_tracks	list[TrackDict]	Album tracks	required
future_years	list[int]	List of future years found	required

Returns:

Type	Description
bool	True if album should be skipped

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

async def handle_future_years(
    self,
    artist: str,
    album: str,
    album_tracks: list[TrackDict],
    future_years: list[int],
) -> bool:
    """Handle case when future years are found in tracks.

    Args:
        artist: Artist name
        album: Album name
        album_tracks: Album tracks
        future_years: List of future years found

    Returns:
        True if album should be skipped

    """
    if not self.skip_prerelease or not future_years:
        return False

    current_year = datetime.now(UTC).year
    max_future_year = max(future_years)

    if max_future_year - current_year <= self.future_year_threshold:
        self.console_logger.debug(
            "Detected future year %s for '%s - %s' but within configured threshold (%d year(s)); continuing processing",
            max_future_year,
            artist,
            album,
            self.future_year_threshold,
        )
        return False

    self.console_logger.info(
        "Skipping prerelease album '%s - %s' with future year(s): %s",
        artist,
        album,
        max_future_year,
    )
    metadata = {
        "expected_year": str(max_future_year),
        "track_count": str(len(album_tracks)),
    }
    await self.pending_verification.mark_for_verification(
        artist,
        album,
        reason="prerelease",
        metadata=metadata,
        recheck_days=self.prerelease_recheck_days,
    )
    return True

extract_future_years staticmethod

extract_future_years(album_tracks)

Extract future years from album tracks.

Parameters:

Name	Type	Description	Default
album_tracks	list[TrackDict]	List of tracks from the album	required

Returns:

Type	Description
list[int]	List of years that are in the future

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

@staticmethod
def extract_future_years(album_tracks: list[TrackDict]) -> list[int]:
    """Extract future years from album tracks.

    Args:
        album_tracks: List of tracks from the album

    Returns:
        List of years that are in the future

    """
    current_year = datetime.now(UTC).year
    future_years: list[int] = []

    for track in album_tracks:
        year_value = track.get("year")
        if year_value is not None:
            try:
                year_int = int(float(str(year_value)))
                if year_int > current_year:
                    future_years.append(year_int)
            except (ValueError, TypeError):
                continue

    return future_years

extract_release_years staticmethod

extract_release_years(album_tracks)

Extract valid release years from album tracks.

Parameters:

Name	Type	Description	Default
album_tracks	list[TrackDict]	List of tracks from the album	required

Returns:

Type	Description
list[str]	List of valid release years from track metadata

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

@staticmethod
def extract_release_years(album_tracks: list[TrackDict]) -> list[str]:
    """Extract valid release years from album tracks.

    Args:
        album_tracks: List of tracks from the album

    Returns:
        List of valid release years from track metadata

    """
    release_years: list[str] = []
    for track in album_tracks:
        release_year_value = track.get("release_year")
        if release_year_value and is_valid_year(release_year_value):
            release_years.append(str(release_year_value))
    return release_years