year_score_resolver

Year score resolution logic extracted from ExternalApiOrchestrator.

This module handles the scoring and selection of the best release year from multiple API responses with potentially conflicting years.

YearScoreResolver

YearScoreResolver(
    *,
    console_logger,
    min_valid_year,
    current_year,
    definitive_score_threshold,
    definitive_score_diff,
    remaster_keywords=None
)

Resolves the best release year from scored API responses.

Handles: - Aggregating scores by year across multiple API sources - Selecting the best year considering future dates and reissues - Determining if the result is definitive

Initialize the year score resolver.

Parameters:

Name	Type	Description	Default
console_logger	Logger	Logger for console output	required
min_valid_year	int	Minimum year to consider valid (e.g., 1900)	required
current_year	int	The current calendar year	required
definitive_score_threshold	int	Min score to consider definitive	required
definitive_score_diff	int	Min difference to prefer one year over another	required
remaster_keywords	list[str] | None	Keywords indicating reissue/remaster editions	None

Source code in src/services/api/year_score_resolver.py

def __init__(
    self,
    *,
    console_logger: logging.Logger,
    min_valid_year: int,
    current_year: int,
    definitive_score_threshold: int,
    definitive_score_diff: int,
    remaster_keywords: list[str] | None = None,
) -> None:
    """Initialize the year score resolver.

    Args:
        console_logger: Logger for console output
        min_valid_year: Minimum year to consider valid (e.g., 1900)
        current_year: The current calendar year
        definitive_score_threshold: Min score to consider definitive
        definitive_score_diff: Min difference to prefer one year over another
        remaster_keywords: Keywords indicating reissue/remaster editions

    """
    self.console_logger = console_logger
    self.min_valid_year = min_valid_year
    self.current_year = current_year
    self.definitive_score_threshold = definitive_score_threshold
    self.definitive_score_diff = definitive_score_diff
    self.remaster_keywords = remaster_keywords or []

aggregate_year_scores

aggregate_year_scores(all_releases)

Aggregate release scores by year, filtering out invalid years.

Source code in src/services/api/year_score_resolver.py

def aggregate_year_scores(self, all_releases: list[ScoredRelease]) -> defaultdict[str, list[int]]:
    """Aggregate release scores by year, filtering out invalid years."""
    year_scores: defaultdict[str, list[int]] = defaultdict(list)

    for release in all_releases:
        year_value = release.get("year")
        year = str(year_value) if year_value is not None else None
        score = int(release.get("score", 0))
        if year and is_valid_year(year, self.min_valid_year, self.current_year):
            year_scores[year].append(score)

    return year_scores

select_best_year

select_best_year(
    year_scores, all_releases=None, existing_year=None
)

Select the best year from aggregated scores and determine if definitive.

Parameters:

Name	Type	Description	Default
year_scores	defaultdict[str, list[int]]	Mapping of year to list of scores	required
all_releases	list[ScoredRelease] | None	Optional list of all scored releases for keyword detection	None
existing_year	str | None	Current year from library (for stability boost)	None

Returns:

Type	Description
tuple[str, bool, int]	Tuple of (best_year, is_definitive, confidence_score)

Source code in src/services/api/year_score_resolver.py

def select_best_year(
    self,
    year_scores: defaultdict[str, list[int]],
    all_releases: list[ScoredRelease] | None = None,
    existing_year: str | None = None,
) -> tuple[str, bool, int]:
    """Select the best year from aggregated scores and determine if definitive.

    Args:
        year_scores: Mapping of year to list of scores
        all_releases: Optional list of all scored releases for keyword detection
        existing_year: Current year from library (for stability boost)

    Returns:
        Tuple of (best_year, is_definitive, confidence_score)
    """
    if not year_scores:
        self.console_logger.warning(
            "No year scores to evaluate (releases_count=%d, existing_year=%s)",
            len(all_releases) if all_releases else 0,
            existing_year or "none",
        )
        return "", False, 0

    final_year_scores = self._compute_final_year_scores(year_scores)
    sorted_years = self._sort_years_by_score(final_year_scores)

    if not sorted_years:
        self.console_logger.warning(
            "No valid years after score computation (year_candidates=%d, min_valid=%d, max_valid=%d)",
            len(year_scores),
            self.min_valid_year,
            self.current_year,
        )
        return "", False, 0

    self._log_ranked_years(sorted_years)

    if boost_result := self._apply_existing_year_boost(existing_year, final_year_scores, sorted_years):
        return boost_result

    best_year, best_score, best_year_is_future = self._determine_best_year_candidate(sorted_years, all_releases)

    # Initialize variables to avoid potential unbound variable errors
    score_thresholds = self._calculate_score_thresholds(best_score)
    has_score_conflict = self._check_score_conflicts(sorted_years, best_year_is_future)

    # Check for suspiciously old years when we have only one result
    if len(sorted_years) == 1:
        best_year, is_definitive = self._validate_single_result(best_year, best_score)
        if not is_definitive:
            self.console_logger.warning(
                "Single result validation failed for year %s - marking as non-definitive",
                best_year,
            )
            return best_year, False, best_score
    else:
        is_definitive = self._determine_definitiveness(score_thresholds, best_year_is_future, has_score_conflict)

    if not is_definitive:
        self._log_non_definitive_reasons(best_year_is_future, score_thresholds, has_score_conflict, best_score)

    return best_year, is_definitive, best_score