year_fallback

Year fallback logic extracted from YearRetriever.

This module handles the decision logic for when to apply, skip, or preserve year values based on confidence levels and existing data.

YearFallbackHandler

YearFallbackHandler(
    *,
    console_logger,
    pending_verification,
    fallback_enabled,
    absurd_year_threshold,
    year_difference_threshold,
    trust_api_score_threshold=DEFAULT_TRUST_API_SCORE_THRESHOLD,
    min_confidence_for_new_year=DEFAULT_MIN_CONFIDENCE_FOR_NEW_YEAR,
    api_orchestrator=None
)

Handles fallback logic for year decisions.

Decision Tree: 1. IF is_definitive=True → APPLY proposed year (high confidence from API) 2. IF proposed_year < absurd_threshold AND no existing year → MARK and SKIP 3. IF existing year is EMPTY → APPLY proposed year (nothing to preserve) 4. IF is_special_album_type → MARK and PROPAGATE existing year to all tracks 5. IF |proposed - existing| > THRESHOLD → MARK and PROPAGATE existing year to all tracks 6. ELSE → APPLY proposed year

Key principle: When we trust existing year over proposed year, we PROPAGATE the existing year to ALL tracks (including empty ones), not just preserve it.

Initialize the year fallback handler.

Parameters:

Name	Type	Description	Default
console_logger	Logger	Logger for console output	required
pending_verification	PendingVerificationServiceProtocol	Service for marking albums for verification	required
fallback_enabled	bool	Whether fallback logic is enabled	required
absurd_year_threshold	int	Years below this are considered absurd	required
year_difference_threshold	int	Max allowed year difference before dramatic change	required
trust_api_score_threshold	int	Trust API if confidence >= this value	DEFAULT_TRUST_API_SCORE_THRESHOLD
min_confidence_for_new_year	int	Minimum confidence to apply year when no existing year	DEFAULT_MIN_CONFIDENCE_FOR_NEW_YEAR
api_orchestrator	ExternalApiServiceProtocol | None	API orchestrator for artist data lookups (optional)	None

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

def __init__(
    self,
    *,
    console_logger: logging.Logger,
    pending_verification: PendingVerificationServiceProtocol,
    fallback_enabled: bool,
    absurd_year_threshold: int,
    year_difference_threshold: int,
    trust_api_score_threshold: int = DEFAULT_TRUST_API_SCORE_THRESHOLD,
    min_confidence_for_new_year: int = DEFAULT_MIN_CONFIDENCE_FOR_NEW_YEAR,
    api_orchestrator: ExternalApiServiceProtocol | None = None,
) -> None:
    """Initialize the year fallback handler.

    Args:
        console_logger: Logger for console output
        pending_verification: Service for marking albums for verification
        fallback_enabled: Whether fallback logic is enabled
        absurd_year_threshold: Years below this are considered absurd
        year_difference_threshold: Max allowed year difference before dramatic change
        trust_api_score_threshold: Trust API if confidence >= this value
        min_confidence_for_new_year: Minimum confidence to apply year when no existing year
        api_orchestrator: API orchestrator for artist data lookups (optional)

    """
    self.console_logger = console_logger
    self.pending_verification = pending_verification
    self.fallback_enabled = fallback_enabled
    self.absurd_year_threshold = absurd_year_threshold
    self.year_difference_threshold = year_difference_threshold
    self.trust_api_score_threshold = trust_api_score_threshold
    self.min_confidence_for_new_year = min_confidence_for_new_year
    self.api_orchestrator = api_orchestrator

apply_year_fallback async

apply_year_fallback(
    proposed_year,
    album_tracks,
    is_definitive,
    confidence_score,
    artist,
    album,
    year_scores=None,
    release_year=None,
)

Apply fallback logic for year decisions.

Parameters:

Name	Type	Description	Default
proposed_year	str	Year from API	required
album_tracks	list[TrackDict]	List of tracks in the album	required
is_definitive	bool	Whether API is confident in the year	required
confidence_score	int	API confidence score (0-100)	required
artist	str	Artist name	required
album	str	Album name	required
year_scores	dict[str, int] | None	Mapping of year to max score from API results (Issue #93)	None
release_year	str | None	Original release year from Apple Music (read-only, more authoritative)	None

Returns:

Type	Description
str | None	Year to apply (proposed or existing), or None to skip update

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

async def apply_year_fallback(
    self,
    proposed_year: str,
    album_tracks: list[TrackDict],
    is_definitive: bool,
    confidence_score: int,
    artist: str,
    album: str,
    year_scores: dict[str, int] | None = None,
    release_year: str | None = None,
) -> str | None:
    """Apply fallback logic for year decisions.

    Args:
        proposed_year: Year from API
        album_tracks: List of tracks in the album
        is_definitive: Whether API is confident in the year
        confidence_score: API confidence score (0-100)
        artist: Artist name
        album: Album name
        year_scores: Mapping of year to max score from API results (Issue #93)
        release_year: Original release year from Apple Music (read-only, more authoritative)

    Returns:
        Year to apply (proposed or existing), or None to skip update

    """
    # Fallback disabled - use original behavior (always apply, mark low confidence)
    if not self.fallback_enabled:
        if not is_definitive:
            await self.pending_verification.mark_for_verification(artist, album)
        return proposed_year

    # Rule 1: High confidence from API - apply directly
    if is_definitive:
        self.console_logger.debug(
            "[FALLBACK] Applying year %s for %s - %s (high confidence)",
            proposed_year,
            artist,
            album,
        )
        return proposed_year

    # Get existing year from tracks (needed for later rules)
    existing_year = self.get_existing_year_from_tracks(album_tracks)

    # Early exit: No change needed if existing year equals proposed year
    # Delegate to helper to keep main function return count under limit
    if existing_year and existing_year == proposed_year:
        return await self._handle_matching_year(existing_year, proposed_year, artist, album)

    # Rule 2: Absurd year detection (when no existing year to compare)
    if await self._handle_absurd_year(proposed_year, existing_year, artist, album):
        return None

    # Rule 2.5: Very low confidence with no existing year (Issue #105)
    # Delegate to helper to keep main function return count under limit
    if not existing_year and confidence_score < self.min_confidence_for_new_year:
        return await self._handle_low_confidence_no_existing_year(proposed_year, confidence_score, artist, album)

    # Rule 2.6: Fresh album detection (applies BEFORE Rule 3 - even when no existing year)
    # If release_year == current year and API returns older year, reject API year as stale
    fresh_album_result = await self._check_fresh_album_stale_api(
        proposed_year=proposed_year,
        release_year=release_year,
        artist=artist,
        album=album,
    )
    if fresh_album_result is not None:
        return fresh_album_result

    # Rule 3: No existing year - nothing to preserve (year passed absurd check and confidence check)
    if not existing_year:
        self.console_logger.debug(
            "[FALLBACK] Applying year %s for %s - %s (no existing year to preserve)",
            proposed_year,
            artist,
            album,
        )
        return proposed_year

    # Delegate remaining rules to helper method
    return await self._apply_existing_year_rules(proposed_year, existing_year, confidence_score, artist, album, year_scores, release_year)

get_existing_year_from_tracks staticmethod

get_existing_year_from_tracks(tracks)

Extract the most common existing year from tracks.

Uses Counter to find the most frequently occurring year among tracks.

Parameters:

Name	Type	Description	Default
tracks	list[TrackDict]	List of tracks to analyze	required

Returns:

Type	Description
str | None	Most common year string, or None if no valid years found

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

@staticmethod
def get_existing_year_from_tracks(tracks: list[TrackDict]) -> str | None:
    """Extract the most common existing year from tracks.

    Uses Counter to find the most frequently occurring year among tracks.

    Args:
        tracks: List of tracks to analyze

    Returns:
        Most common year string, or None if no valid years found

    """
    years = [str(track.get("year")) for track in tracks if track.get("year") and not is_empty_year(track.get("year"))]
    if not years:
        return None
    counter = Counter(years)
    most_common = counter.most_common(1)
    return most_common[0][0] if most_common else None

is_year_change_dramatic

is_year_change_dramatic(existing, proposed)

Check if year change exceeds the threshold.

A dramatic change (e.g., 2018→1998) suggests the API returned a reissue/compilation year rather than the original.

Parameters:

Name	Type	Description	Default
existing	str	Current year value	required
proposed	str	Proposed new year value	required

Returns:

Type	Description
bool	True if difference exceeds year_difference_threshold

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

def is_year_change_dramatic(self, existing: str, proposed: str) -> bool:
    """Check if year change exceeds the threshold.

    A dramatic change (e.g., 2018→1998) suggests the API returned
    a reissue/compilation year rather than the original.

    Args:
        existing: Current year value
        proposed: Proposed new year value

    Returns:
        True if difference exceeds year_difference_threshold

    """
    try:
        existing_int = int(existing)
        proposed_int = int(proposed)
        difference = abs(existing_int - proposed_int)
        return difference > self.year_difference_threshold
    except (ValueError, TypeError):
        return False