""" Result Ranker for Smart Scout Search. Scores, ranks, and deduplicates art results across museum sources. """ import logging import re import unicodedata from difflib import SequenceMatcher from typing import Dict, List logger = logging.getLogger("artwork-display-api.ranker") # Museum/aggregator titles often arrive with structured-data noise. Strip a leading # language label (e.g. "Dutch: Meisje met de parel") and Wikidata QS suffixes so the # placard reads cleanly. Intentionally conservative — no translation, no guessing # between bilingual halves (those stay editable in the Review Queue). _LANG_PREFIX_RE = re.compile( r"^(?:English|French|Dutch|German|Italian|Spanish|Latin|Greek|Russian|Japanese|Chinese|Title|Titre|Titel|Título|Titolo)\s*:\s*", re.IGNORECASE, ) def clean_title(title: str) -> str: """Light, safe normalization of a museum-supplied title for display.""" if not title: return title t = str(title).strip() t = re.split(r"\s+QS:P\d+", t)[0] # Wikidata structured suffix: '… QS:P1476,en:"…"' t = _LANG_PREFIX_RE.sub("", t) # leading 'Language: ' label t = re.sub(r"\s+", " ", t).strip() # collapse whitespace return t or title def _normalize_text(text: str) -> str: """Normalize text for comparison: lowercase, strip accents, remove articles.""" if not text: return "" # Lowercase text = text.lower().strip() # Strip accents text = ''.join( c for c in unicodedata.normalize('NFD', text) if unicodedata.category(c) != 'Mn' ) # Remove common articles for article in ("the ", "a ", "an ", "le ", "la ", "les ", "de ", "het ", "een "): if text.startswith(article): text = text[len(article):] return text.strip() def _fuzzy_match(a: str, b: str, threshold: float = 0.85) -> bool: """Check if two strings are fuzzy matches above threshold.""" if not a or not b: return False return SequenceMatcher(None, a, b).ratio() >= threshold class ResultRanker: """ Scores and deduplicates results from multiple museum scouts. """ def rank_and_deduplicate( self, results: List[Dict], intent=None ) -> List[Dict]: """ Score each result, deduplicate across sources, and return sorted by relevance. Args: results: List of art dicts from scouts intent: SearchIntent from the classifier (optional) Returns: Sorted, deduplicated list of art dicts with 'relevance_score' added """ if not results: return [] # 1. Score each result for item in results: item['relevance_score'] = self._score(item, intent) # 2. Sort by relevance (highest first) results.sort(key=lambda x: x.get('relevance_score', 0), reverse=True) # 3. Deduplicate deduped = self._deduplicate(results) logger.info( f"[Ranker] Scored {len(results)} results, " f"deduped to {len(deduped)} unique works" ) return deduped def _score(self, item: Dict, intent=None) -> float: """ Score a single result on a 0–100 scale. Signals: - Artist match (30 pts): proposed_artist matches intent.canonical_name - Title relevance (20 pts): query terms in proposed_title - Highlight flag (20 pts): museum curator pick - Image quality (15 pts): heuristic from URL patterns - Metadata completeness (15 pts): title + artist + context_hints present """ score = 0.0 proposed_artist = (item.get('proposed_artist') or '').lower() proposed_title = (item.get('proposed_title') or '').lower() # --- Artist Match (30 pts) --- if intent and intent.query_type == "artist": canonical_lower = intent.canonical_name.lower() # Check various matching strategies if canonical_lower in proposed_artist or proposed_artist in canonical_lower: score += 30 elif _fuzzy_match(_normalize_text(intent.canonical_name), _normalize_text(proposed_artist)): score += 25 # Partial last-name match (e.g., "Van Gogh" in "Vincent Willem van Gogh") elif intent.original_query.lower() in proposed_artist: score += 20 # --- Title Relevance (20 pts) --- if intent: query_terms = intent.original_query.lower().split() matched_terms = sum(1 for t in query_terms if t in proposed_title) if query_terms: score += 20 * (matched_terms / len(query_terms)) # --- Highlight / Featured Flag (20 pts) --- context = item.get('context_hints', '') if isinstance(context, str): ctx_lower = context.lower() # Various API highlight indicators if any(flag in ctx_lower for flag in ( '"ishighlight":true', '"ishighlight": true', '"highlight":1', '"highlight": 1', '"is_boosted":true', '"is_boosted": true', )): score += 20 # --- Image Quality Heuristic (15 pts) --- source_url = item.get('source_url', '') if source_url: # Full resolution indicators if '/full/max/' in source_url or '/full/full/' in source_url: score += 15 elif 'width=2000' in source_url or '/print/' in source_url: score += 12 elif '/full/' in source_url: score += 10 else: score += 5 # At least it has an image # --- Metadata Completeness (15 pts) --- completeness = 0 if item.get('proposed_title') and item['proposed_title'] != 'Unknown': completeness += 1 if item.get('proposed_artist') and item['proposed_artist'] != 'Unknown Artist': completeness += 1 if item.get('context_hints'): completeness += 1 score += 15 * (completeness / 3) return round(score, 1) def _deduplicate(self, results: List[Dict]) -> List[Dict]: """ Remove duplicate artworks across museum sources. Uses fuzzy matching on normalized title + artist. Keeps the higher-scoring copy. """ unique = [] seen_keys = [] # List of (normalized_title, normalized_artist) tuples for item in results: norm_title = _normalize_text(item.get('proposed_title', '')) norm_artist = _normalize_text(item.get('proposed_artist', '')) item_key = f"{norm_title} by {norm_artist}" is_duplicate = False for existing_key in seen_keys: if _fuzzy_match(item_key, existing_key): is_duplicate = True break if not is_duplicate: unique.append(item) seen_keys.append(item_key) return unique