""" 5-tier matching orchestration for priority release check. Ported from priority-release-check/src/pages/ReleaseCheck.tsx """ import logging import re from dataclasses import dataclass, field from typing import Optional from parse_email import Track, Album import queries logger = logging.getLogger(__name__) @dataclass class TrackMatch: email_artist: str email_title: str label: str = "" notes: str = "" db_artist: Optional[str] = None db_track_name: Optional[str] = None db_release_name: Optional[str] = None db_release_date: Optional[str] = None isrc: Optional[str] = None match_tier: str = "not_found" @dataclass class AlbumMatch: email_artist: str email_title: str label: str = "" notes: str = "" db_artist: Optional[str] = None db_release_name: Optional[str] = None db_release_date: Optional[str] = None db_format: Optional[str] = None upc: Optional[str] = None match_tier: str = "not_found" # --------------------------------------------------------------------------- # Client-side matching helpers # --------------------------------------------------------------------------- def bigram_similarity(a: str, b: str) -> float: """Simple bigram similarity score (0-1).""" if a == b: return 1.0 if len(a) < 2 or len(b) < 2: return 1.0 if a == b else 0.0 def bigrams(s): bg = {} for i in range(len(s) - 1): bi = s[i : i + 2] bg[bi] = bg.get(bi, 0) + 1 return bg a_bi = bigrams(a) b_bi = bigrams(b) intersection = sum(min(count, b_bi.get(bi, 0)) for bi, count in a_bi.items()) return (2 * intersection) / (len(a) - 1 + len(b) - 1) def artist_matches(db_artist: str, email_artist: str) -> bool: """Check if DB artist matches email artist, including collaborator patterns. Handles both directions: DB may have collaborator suffix, or email may have it. """ if db_artist == email_artist: return True # DB has collaborator: "BUNT. & MALOU" matches email "BUNT." if db_artist.startswith(email_artist + " & ") or db_artist.startswith(email_artist + " X "): return True # Email has collaborator: email "BUNT. & MALOU" matches DB "BUNT." if email_artist.startswith(db_artist + " & ") or email_artist.startswith(db_artist + " X "): return True return False def find_track_match(row: dict, tracks: list, tier: str) -> int: """Find matching email track index for a DB result row. Returns -1 if no match.""" db_artist = str(row.get("ARTISTNAME", "")).upper() db_track = str(row.get("TRACKNAME", "")).upper() for i, t in enumerate(tracks): email_artist = t.artist.upper() email_title = t.title.upper() if tier == "exact": if artist_matches(db_artist, email_artist) and db_track == email_title: return i elif tier == "partial": if artist_matches(db_artist, email_artist) and email_title in db_track: return i elif tier == "fuzzy": if bigram_similarity(db_artist, email_artist) > 0.6 and bigram_similarity(db_track, email_title) > 0.6: return i return -1 def find_album_match(row: dict, albums: list, tier: str) -> int: """Find matching email album index for a DB result row. Returns -1 if no match.""" db_artist = str(row.get("ARTISTNAME", "")).upper() db_release = str(row.get("RELEASENAME", "")).upper() for i, a in enumerate(albums): email_artist = a.artist.upper() email_title = a.title.upper() if tier == "exact": if artist_matches(db_artist, email_artist) and db_release == email_title: return i elif tier == "partial": if artist_matches(db_artist, email_artist) and email_title in db_release: return i elif tier == "fuzzy": if bigram_similarity(db_artist, email_artist) > 0.6 and bigram_similarity(db_release, email_title) > 0.6: return i return -1 # --------------------------------------------------------------------------- # Tier execution # --------------------------------------------------------------------------- def _execute_sql(session, sql: str) -> list: """Execute SQL via Snowpark and return list of dicts.""" if not sql: return [] df = session.sql(sql).to_pandas() return df.to_dict("records") def run_tier(session, tracks, albums, t_matches, a_matches, release_date, date_window, tier, source="dim"): """Run a matching tier (exact/partial/fuzzy) for unmatched items.""" track_builder = queries.build_grps_track_query if source == "grps" else queries.build_track_query album_builder = queries.build_grps_album_query if source == "grps" else queries.build_album_query unmatched_tracks = [tracks[i] for i in range(len(tracks)) if t_matches[i].match_tier == "not_found"] unmatched_track_indices = [i for i in range(len(tracks)) if t_matches[i].match_tier == "not_found"] unmatched_albums = [albums[i] for i in range(len(albums)) if a_matches[i].match_tier == "not_found"] unmatched_album_indices = [i for i in range(len(albums)) if a_matches[i].match_tier == "not_found"] if unmatched_tracks: sql = track_builder(unmatched_tracks, release_date, date_window, tier) rows = _execute_sql(session, sql) for row in rows: sub_idx = find_track_match(row, unmatched_tracks, tier) if sub_idx >= 0: orig_idx = unmatched_track_indices[sub_idx] if t_matches[orig_idx].match_tier == "not_found": t_matches[orig_idx].db_artist = str(row.get("ARTISTNAME", "")) t_matches[orig_idx].db_track_name = str(row.get("TRACKNAME", "")) t_matches[orig_idx].db_release_name = str(row.get("RELEASENAME", "")) t_matches[orig_idx].db_release_date = str(row.get("RELEASE_DATE", "")) t_matches[orig_idx].isrc = str(row.get("ISRC", "")) t_matches[orig_idx].match_tier = tier if unmatched_albums: sql = album_builder(unmatched_albums, release_date, date_window, tier) rows = _execute_sql(session, sql) for row in rows: sub_idx = find_album_match(row, unmatched_albums, tier) if sub_idx >= 0: orig_idx = unmatched_album_indices[sub_idx] if a_matches[orig_idx].match_tier == "not_found": a_matches[orig_idx].db_artist = str(row.get("ARTISTNAME", "")) a_matches[orig_idx].db_release_name = str(row.get("RELEASENAME", "")) a_matches[orig_idx].db_release_date = str(row.get("RELEASE_DATE", "")) a_matches[orig_idx].db_format = str(row.get("FORMAT", "")) a_matches[orig_idx].upc = str(row.get("MANUFACTURER_UPC", "")) a_matches[orig_idx].match_tier = tier def run_ai_tier(session, tracks, albums, t_matches, a_matches, release_date, date_window, source="dim"): """Run AI matching tier for remaining unmatched items.""" ai_track_builder = queries.build_grps_ai_track_query if source == "grps" else queries.build_ai_track_query ai_album_builder = queries.build_grps_ai_album_query if source == "grps" else queries.build_ai_album_query for i in range(len(tracks)): if t_matches[i].match_tier != "not_found": continue try: sql = ai_track_builder(tracks[i], release_date, date_window) rows = _execute_sql(session, sql) for row in rows: ai_response = str(row.get("AI_MATCH", "")).strip().upper() if re.search(r"\bYES\b", ai_response): t_matches[i].db_artist = str(row.get("ARTISTNAME", "")) t_matches[i].db_track_name = str(row.get("TRACKNAME", "")) t_matches[i].db_release_name = str(row.get("RELEASENAME", "")) t_matches[i].db_release_date = str(row.get("RELEASE_DATE", "")) t_matches[i].isrc = str(row.get("ISRC", "")) t_matches[i].match_tier = "ai" break except Exception as e: logger.warning("AI track match failed for '%s' by '%s': %s", tracks[i].title, tracks[i].artist, e) for i in range(len(albums)): if a_matches[i].match_tier != "not_found": continue try: sql = ai_album_builder(albums[i], release_date, date_window) rows = _execute_sql(session, sql) for row in rows: ai_response = str(row.get("AI_MATCH", "")).strip().upper() if re.search(r"\bYES\b", ai_response): a_matches[i].db_artist = str(row.get("ARTISTNAME", "")) a_matches[i].db_release_name = str(row.get("RELEASENAME", "")) a_matches[i].db_release_date = str(row.get("RELEASE_DATE", "")) a_matches[i].db_format = str(row.get("FORMAT", "")) a_matches[i].upc = str(row.get("MANUFACTURER_UPC", "")) a_matches[i].match_tier = "ai" break except Exception as e: logger.warning("AI album match failed for '%s' by '%s': %s", albums[i].title, albums[i].artist, e) def run_wide_tier(session, tracks, albums, t_matches, a_matches, source="dim"): """Run wide search (no date filter) for remaining unmatched items.""" wide_track_builder = queries.build_grps_wide_track_query if source == "grps" else queries.build_wide_track_query wide_album_builder = queries.build_grps_wide_album_query if source == "grps" else queries.build_wide_album_query unmatched_tracks = [tracks[i] for i in range(len(tracks)) if t_matches[i].match_tier == "not_found"] if unmatched_tracks: try: sql = wide_track_builder(unmatched_tracks) rows = _execute_sql(session, sql) for row in rows: idx = find_track_match(row, tracks, "exact") if idx >= 0 and t_matches[idx].match_tier == "not_found": t_matches[idx].db_artist = str(row.get("ARTISTNAME", "")) t_matches[idx].db_track_name = str(row.get("TRACKNAME", "")) t_matches[idx].db_release_name = str(row.get("RELEASENAME", "")) t_matches[idx].db_release_date = str(row.get("RELEASE_DATE", "")) t_matches[idx].isrc = str(row.get("ISRC", "")) t_matches[idx].match_tier = "wide" except Exception as e: logger.warning("Wide track search failed (%s): %s", source, e) unmatched_albums = [albums[i] for i in range(len(albums)) if a_matches[i].match_tier == "not_found"] if unmatched_albums: try: sql = wide_album_builder(unmatched_albums) rows = _execute_sql(session, sql) for row in rows: idx = find_album_match(row, albums, "exact") if idx >= 0 and a_matches[idx].match_tier == "not_found": a_matches[idx].db_artist = str(row.get("ARTISTNAME", "")) a_matches[idx].db_release_name = str(row.get("RELEASENAME", "")) a_matches[idx].db_release_date = str(row.get("RELEASE_DATE", "")) a_matches[idx].db_format = str(row.get("FORMAT", "")) a_matches[idx].upc = str(row.get("MANUFACTURER_UPC", "")) a_matches[idx].match_tier = "wide" except Exception as e: logger.warning("Wide album search failed (%s): %s", source, e) def run_all_tiers(session, parsed_email, date_window, progress_callback=None): """Run all 5 matching tiers and return results. Args: session: Snowpark session parsed_email: ParsedEmail instance date_window: int days window progress_callback: optional callable(phase_name, detail) for UI updates Returns: dict with keys: dim_tracks, dim_albums, grps_tracks, grps_albums """ tracks = parsed_email.tracks albums = parsed_email.albums release_date = parsed_email.release_date def _init_track(t): return TrackMatch(email_artist=t.artist, email_title=t.title, label=t.label, notes=t.notes) def _init_album(a): return AlbumMatch(email_artist=a.artist, email_title=a.title, label=a.label, notes=a.notes) dim_t = [_init_track(t) for t in tracks] dim_a = [_init_album(a) for a in albums] grps_t = [_init_track(t) for t in tracks] grps_a = [_init_album(a) for a in albums] def notify(phase, detail=""): if progress_callback: progress_callback(phase, detail) # Tier 1: Exact notify("exact", "Running exact match...") run_tier(session, tracks, albums, dim_t, dim_a, release_date, date_window, "exact", "dim") run_tier(session, tracks, albums, grps_t, grps_a, release_date, date_window, "exact", "grps") # Tier 2: Partial notify("partial", "Running partial match...") run_tier(session, tracks, albums, dim_t, dim_a, release_date, date_window, "partial", "dim") run_tier(session, tracks, albums, grps_t, grps_a, release_date, date_window, "partial", "grps") # Tier 3: Fuzzy notify("fuzzy", "Running fuzzy match...") run_tier(session, tracks, albums, dim_t, dim_a, release_date, date_window, "fuzzy", "dim") run_tier(session, tracks, albums, grps_t, grps_a, release_date, date_window, "fuzzy", "grps") # Tier 4: AI notify("ai", "Running AI match (may take a moment)...") run_ai_tier(session, tracks, albums, dim_t, dim_a, release_date, date_window, "dim") run_ai_tier(session, tracks, albums, grps_t, grps_a, release_date, date_window, "grps") # Tier 5: Wide notify("wide", "Running wide search (no date filter)...") run_wide_tier(session, tracks, albums, dim_t, dim_a, "dim") run_wide_tier(session, tracks, albums, grps_t, grps_a, "grps") notify("done", "Matching complete.") return { "dim_tracks": dim_t, "dim_albums": dim_a, "grps_tracks": grps_t, "grps_albums": grps_a, }