""" SQL query builders for priority release matching. Ported from priority-release-check/src/queries/releases.ts """ import re from datetime import date as date_type DATE_RE = re.compile(r"^\d{4}-(0[1-9]|1[0-2])-(0[1-9]|[12]\d|3[01])$") VALID_TIERS = {"exact", "partial", "fuzzy"} # --------------------------------------------------------------------------- # SQL escaping helpers # --------------------------------------------------------------------------- def esc(s: str) -> str: """Escape single quotes in SQL strings.""" return s.replace("'", "''") def esc_like(s: str) -> str: """Escape LIKE wildcard characters in addition to single quotes.""" return esc(s).replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") def esc_prompt(s: str) -> str: """Escape values embedded in LLM prompt strings. Strips quotes, newlines, angle brackets, and SQL comment sequences. """ return re.sub(r'--|/\*|\*/|["\n\r<>]', " ", s).strip() def validate_params(release_date: str, date_window: int): if not DATE_RE.match(release_date): raise ValueError(f"Invalid release date format: {release_date}. Expected YYYY-MM-DD.") try: date_type.fromisoformat(release_date) except ValueError: raise ValueError(f"Invalid calendar date: {release_date}.") if not isinstance(date_window, int) or date_window < 1 or date_window > 30: raise ValueError(f"Invalid date window: {date_window}. Expected integer 1-30.") # --------------------------------------------------------------------------- # GRPS constants # --------------------------------------------------------------------------- GRPS = "GRPS_APP_REPORTING_NOCONF.GRPS_DRAS002" GRPS_TRACK_FROM = f"""{GRPS}.TRAS007 t JOIN {GRPS}.TRAS398 l ON t.CLIENT_KEY = l.CLIENT_KEY AND t.TRACK_EXT = l.TRACK_EXT AND t.TRACK_NO = l.TRACK_NO JOIN {GRPS}.TRAS011 tp ON t.CLIENT_KEY = tp.CLIENT_KEY AND t.TRACK_EXT = tp.TRACK_EXT AND t.TRACK_NO = tp.TRACK_NO AND tp.IS_MAIN_ARTIST = 'Y' JOIN {GRPS}.TRAS006 p ON tp.CLIENT_KEY = p.CLIENT_KEY AND tp.PARTICIP_NO = p.PARTICIP_NO""" GRPS_ALBUM_FROM = f"""{GRPS}.TRAS002 pv JOIN {GRPS}.TRAS001 prod ON pv.CLIENT_KEY = prod.CLIENT_KEY AND pv.PROD_VERS_NO = prod.PROD_VERS_NO JOIN {GRPS}.TRAS006 art ON pv.CLIENT_KEY = art.CLIENT_KEY AND pv.MAIN_ARTIST_NO = art.PARTICIP_NO""" GRPS_TRACK_DATE_EXPR = ( "SUBSTRING(t.ORIG_RELEASE_DATE,1,4)||'-'||SUBSTRING(t.ORIG_RELEASE_DATE,5,2)" "||'-'||SUBSTRING(t.ORIG_RELEASE_DATE,7,2)" ) GRPS_ALBUM_DATE_EXPR = ( "CASE WHEN prod.FIRST_RELEASE_DATE IS NOT NULL THEN " "SUBSTRING(prod.FIRST_RELEASE_DATE,1,4)||'-'||SUBSTRING(prod.FIRST_RELEASE_DATE,5,2)" "||'-'||SUBSTRING(prod.FIRST_RELEASE_DATE,7,2) ELSE NULL END" ) def _grps_artist_condition(col: str, artist: str) -> str: """Generate GRPS artist condition matching exact name or collaborator patterns.""" escaped = esc(artist).upper() escaped_like = esc_like(artist).upper() conditions = [ f"UPPER({col}) = '{escaped}'", f"UPPER({col}) LIKE '{escaped_like} &%' ESCAPE '\\\\'", f"UPPER({col}) LIKE '{escaped_like} X %' ESCAPE '\\\\'", ] primary = _strip_featuring(artist) if primary: primary_esc = esc(primary).upper() primary_like = esc_like(primary).upper() conditions.append(f"UPPER({col}) = '{primary_esc}'") conditions.append(f"UPPER({col}) LIKE '{primary_like} &%' ESCAPE '\\\\'") conditions.append(f"UPPER({col}) LIKE '{primary_like} X %' ESCAPE '\\\\'") return f"({' OR '.join(conditions)})" # --------------------------------------------------------------------------- # DIM Query Builders # --------------------------------------------------------------------------- FEAT_RE = re.compile(r'\s+(?:feat\.?|ft\.?|featuring)\s+', re.IGNORECASE) def _strip_featuring(artist: str) -> str | None: """Return the primary artist (before feat/ft/featuring), or None if no featuring.""" parts = FEAT_RE.split(artist, maxsplit=1) if len(parts) > 1: return parts[0].strip() return None def _dim_artist_condition(col: str, artist: str) -> str: """Generate DIM artist condition matching exact name or collaborator patterns. Handles both directions: DB may store 'BUNT.' while email has 'BUNT. & Malou', or DB may store 'BUNT. & Malou' while email has 'BUNT.'. Also handles featuring artists: email has 'Osé feat. Victony' but DB has 'Osé'. """ escaped = esc(artist).upper() escaped_like = esc_like(artist).upper() conditions = [ f"UPPER({col}) = '{escaped}'", f"UPPER({col}) LIKE '{escaped_like} &%' ESCAPE '\\\\'", f"UPPER({col}) LIKE '{escaped_like} X %' ESCAPE '\\\\'", f"'{escaped}' LIKE UPPER({col}) || ' &%' ESCAPE '\\\\'", f"'{escaped}' LIKE UPPER({col}) || ' X %' ESCAPE '\\\\'", ] primary = _strip_featuring(artist) if primary: primary_esc = esc(primary).upper() primary_like = esc_like(primary).upper() conditions.append(f"UPPER({col}) = '{primary_esc}'") conditions.append(f"UPPER({col}) LIKE '{primary_like} &%' ESCAPE '\\\\'") conditions.append(f"UPPER({col}) LIKE '{primary_like} X %' ESCAPE '\\\\'") return f"({' OR '.join(conditions)})" def build_track_query(tracks, release_date: str, date_window: int, tier: str) -> str: if not tracks: return "" if tier not in VALID_TIERS: raise ValueError(f"Invalid tier: {tier}. Expected one of {VALID_TIERS}.") validate_params(release_date, date_window) conditions = [] for t in tracks: artist = esc(t.artist).upper() title = esc(t.title).upper() artist_cond = _dim_artist_condition("a.ARTISTNAME", t.artist) if tier == "exact": conditions.append(f"({artist_cond} AND UPPER(t.TRACKNAME) = '{title}')") elif tier == "partial": conditions.append( f"({artist_cond}" f" AND UPPER(t.TRACKNAME) LIKE '%{esc_like(t.title).upper()}%' ESCAPE '\\\\')" ) elif tier == "fuzzy": primary = _strip_featuring(t.artist) if primary: primary_esc = esc(primary).upper() conditions.append( f"((JAROWINKLER_SIMILARITY(UPPER(a.ARTISTNAME), '{artist}') > 80" f" OR JAROWINKLER_SIMILARITY(UPPER(a.ARTISTNAME), '{primary_esc}') > 80)" f" AND JAROWINKLER_SIMILARITY(UPPER(t.TRACKNAME), '{title}') > 80)" ) else: conditions.append( f"(JAROWINKLER_SIMILARITY(UPPER(a.ARTISTNAME), '{artist}') > 80" f" AND JAROWINKLER_SIMILARITY(UPPER(t.TRACKNAME), '{title}') > 80)" ) return f"""SELECT DISTINCT a.ARTISTNAME, t.TRACKNAME, t.ISRC, TO_VARCHAR(r.RELEASEDATE::DATE, 'YYYY-MM-DD') AS RELEASE_DATE, r.RELEASENAME FROM FACTS.PROD.DIM_TRACK_CLEAN_MV t JOIN FACTS.PROD.DIM_RELEASE r ON t.UPC = r.RELEASEID JOIN FACTS.PROD.DIM_ARTIST a ON r.ARTISTID = a.ARTISTID WHERE r.RELEASEDATE::DATE BETWEEN '{release_date}'::DATE - {date_window} AND '{release_date}'::DATE + {date_window} AND ({chr(10) + ' OR '.join(conditions)}) LIMIT 500""" def build_album_query(albums, release_date: str, date_window: int, tier: str) -> str: if not albums: return "" if tier not in VALID_TIERS: raise ValueError(f"Invalid tier: {tier}. Expected one of {VALID_TIERS}.") validate_params(release_date, date_window) conditions = [] for al in albums: artist = esc(al.artist).upper() title = esc(al.title).upper() artist_cond = _dim_artist_condition("a.ARTISTNAME", al.artist) if tier == "exact": conditions.append(f"({artist_cond} AND UPPER(r.RELEASENAME) = '{title}')") elif tier == "partial": conditions.append( f"({artist_cond}" f" AND UPPER(r.RELEASENAME) LIKE '%{esc_like(al.title).upper()}%' ESCAPE '\\\\')" ) elif tier == "fuzzy": primary = _strip_featuring(al.artist) if primary: primary_esc = esc(primary).upper() conditions.append( f"((JAROWINKLER_SIMILARITY(UPPER(a.ARTISTNAME), '{artist}') > 80" f" OR JAROWINKLER_SIMILARITY(UPPER(a.ARTISTNAME), '{primary_esc}') > 80)" f" AND JAROWINKLER_SIMILARITY(UPPER(r.RELEASENAME), '{title}') > 80)" ) else: conditions.append( f"(JAROWINKLER_SIMILARITY(UPPER(a.ARTISTNAME), '{artist}') > 80" f" AND JAROWINKLER_SIMILARITY(UPPER(r.RELEASENAME), '{title}') > 80)" ) return f"""SELECT DISTINCT a.ARTISTNAME, r.RELEASENAME, TO_VARCHAR(r.RELEASEDATE::DATE, 'YYYY-MM-DD') AS RELEASE_DATE, r.FORMAT, r.MANUFACTURER_UPC FROM FACTS.PROD.DIM_RELEASE r JOIN FACTS.PROD.DIM_ARTIST a ON r.ARTISTID = a.ARTISTID WHERE r.RELEASEDATE::DATE BETWEEN '{release_date}'::DATE - {date_window} AND '{release_date}'::DATE + {date_window} AND ({chr(10) + ' OR '.join(conditions)}) LIMIT 500""" def build_ai_track_query(track, release_date: str, date_window: int) -> str: validate_params(release_date, date_window) safe_title = esc(esc_prompt(track.title)) safe_artist = esc(esc_prompt(track.artist)) artist_cond = _dim_artist_condition("a.ARTISTNAME", track.artist) return f"""SELECT a.ARTISTNAME, t.TRACKNAME, t.ISRC, TO_VARCHAR(r.RELEASEDATE::DATE, 'YYYY-MM-DD') AS RELEASE_DATE, r.RELEASENAME, SNOWFLAKE.CORTEX.COMPLETE('llama3.1-8b', 'Compare these two songs and answer only YES or NO. ' || '{safe_title}' || ' by ' || '{safe_artist}' || ' ' || t.TRACKNAME || ' by ' || a.ARTISTNAME || '' ) AS AI_MATCH FROM FACTS.PROD.DIM_TRACK_CLEAN_MV t JOIN FACTS.PROD.DIM_RELEASE r ON t.UPC = r.RELEASEID JOIN FACTS.PROD.DIM_ARTIST a ON r.ARTISTID = a.ARTISTID WHERE {artist_cond} AND r.RELEASEDATE::DATE BETWEEN '{release_date}'::DATE - {date_window} AND '{release_date}'::DATE + {date_window} LIMIT 20""" def build_ai_album_query(album, release_date: str, date_window: int) -> str: validate_params(release_date, date_window) safe_title = esc(esc_prompt(album.title)) safe_artist = esc(esc_prompt(album.artist)) artist_cond = _dim_artist_condition("a.ARTISTNAME", album.artist) return f"""SELECT a.ARTISTNAME, r.RELEASENAME, TO_VARCHAR(r.RELEASEDATE::DATE, 'YYYY-MM-DD') AS RELEASE_DATE, r.FORMAT, r.MANUFACTURER_UPC, SNOWFLAKE.CORTEX.COMPLETE('llama3.1-8b', 'Compare these two albums and answer only YES or NO. ' || '{safe_title}' || ' by ' || '{safe_artist}' || ' ' || r.RELEASENAME || ' by ' || a.ARTISTNAME || '' ) AS AI_MATCH FROM FACTS.PROD.DIM_RELEASE r JOIN FACTS.PROD.DIM_ARTIST a ON r.ARTISTID = a.ARTISTID WHERE {artist_cond} AND r.RELEASEDATE::DATE BETWEEN '{release_date}'::DATE - {date_window} AND '{release_date}'::DATE + {date_window} LIMIT 20""" def build_wide_track_query(tracks) -> str: if not tracks: return "" conditions = [] for t in tracks: title = esc(t.title).upper() artist_cond = _dim_artist_condition("a.ARTISTNAME", t.artist) conditions.append(f"({artist_cond} AND UPPER(t.TRACKNAME) = '{title}')") return f"""SELECT DISTINCT a.ARTISTNAME, t.TRACKNAME, t.ISRC, TO_VARCHAR(r.RELEASEDATE::DATE, 'YYYY-MM-DD') AS RELEASE_DATE, r.RELEASENAME FROM FACTS.PROD.DIM_TRACK_CLEAN_MV t JOIN FACTS.PROD.DIM_RELEASE r ON t.UPC = r.RELEASEID JOIN FACTS.PROD.DIM_ARTIST a ON r.ARTISTID = a.ARTISTID WHERE ({chr(10) + ' OR '.join(conditions)}) LIMIT 100""" def build_wide_album_query(albums) -> str: if not albums: return "" conditions = [] for al in albums: title = esc(al.title).upper() artist_cond = _dim_artist_condition("a.ARTISTNAME", al.artist) conditions.append(f"({artist_cond} AND UPPER(r.RELEASENAME) = '{title}')") return f"""SELECT DISTINCT a.ARTISTNAME, r.RELEASENAME, TO_VARCHAR(r.RELEASEDATE::DATE, 'YYYY-MM-DD') AS RELEASE_DATE, r.FORMAT, r.MANUFACTURER_UPC FROM FACTS.PROD.DIM_RELEASE r JOIN FACTS.PROD.DIM_ARTIST a ON r.ARTISTID = a.ARTISTID WHERE ({chr(10) + ' OR '.join(conditions)}) LIMIT 100""" # --------------------------------------------------------------------------- # GRPS Query Builders # --------------------------------------------------------------------------- def build_grps_track_query(tracks, release_date: str, date_window: int, tier: str) -> str: if not tracks: return "" if tier not in VALID_TIERS: raise ValueError(f"Invalid tier: {tier}. Expected one of {VALID_TIERS}.") validate_params(release_date, date_window) conditions = [] for t in tracks: artist_cond = _grps_artist_condition("p.PARTICIP_FULL_NAME", t.artist) title = esc(t.title).upper() if tier == "exact": conditions.append(f"({artist_cond} AND UPPER(l.LOCLAN_TITLE) = '{title}')") elif tier == "partial": conditions.append( f"({artist_cond}" f" AND UPPER(l.LOCLAN_TITLE) LIKE '%{esc_like(t.title).upper()}%' ESCAPE '\\\\')" ) elif tier == "fuzzy": artist = esc(t.artist).upper() primary = _strip_featuring(t.artist) if primary: primary_esc = esc(primary).upper() conditions.append( f"((JAROWINKLER_SIMILARITY(UPPER(p.PARTICIP_FULL_NAME), '{artist}') > 80" f" OR JAROWINKLER_SIMILARITY(UPPER(p.PARTICIP_FULL_NAME), '{primary_esc}') > 80)" f" AND JAROWINKLER_SIMILARITY(UPPER(l.LOCLAN_TITLE), '{title}') > 80)" ) else: conditions.append( f"(JAROWINKLER_SIMILARITY(UPPER(p.PARTICIP_FULL_NAME), '{artist}') > 80" f" AND JAROWINKLER_SIMILARITY(UPPER(l.LOCLAN_TITLE), '{title}') > 80)" ) return f"""SELECT DISTINCT p.PARTICIP_FULL_NAME AS ARTISTNAME, l.LOCLAN_TITLE AS TRACKNAME, t.ISRC, {GRPS_TRACK_DATE_EXPR} AS RELEASE_DATE, NULL AS RELEASENAME FROM {GRPS_TRACK_FROM} WHERE t.ORIG_RELEASE_DATE BETWEEN TO_VARCHAR('{release_date}'::DATE - {date_window}, 'YYYYMMDD') AND TO_VARCHAR('{release_date}'::DATE + {date_window}, 'YYYYMMDD') AND ({chr(10) + ' OR '.join(conditions)}) LIMIT 500""" def build_grps_album_query(albums, release_date: str, date_window: int, tier: str) -> str: if not albums: return "" if tier not in VALID_TIERS: raise ValueError(f"Invalid tier: {tier}. Expected one of {VALID_TIERS}.") validate_params(release_date, date_window) conditions = [] for a in albums: artist_cond = _grps_artist_condition("art.PARTICIP_FULL_NAME", a.artist) title = esc(a.title).upper() if tier == "exact": conditions.append(f"({artist_cond} AND UPPER(pv.PROD_TITLE) = '{title}')") elif tier == "partial": conditions.append( f"({artist_cond}" f" AND UPPER(pv.PROD_TITLE) LIKE '%{esc_like(a.title).upper()}%' ESCAPE '\\\\')" ) elif tier == "fuzzy": artist = esc(a.artist).upper() primary = _strip_featuring(a.artist) if primary: primary_esc = esc(primary).upper() conditions.append( f"((JAROWINKLER_SIMILARITY(UPPER(art.PARTICIP_FULL_NAME), '{artist}') > 80" f" OR JAROWINKLER_SIMILARITY(UPPER(art.PARTICIP_FULL_NAME), '{primary_esc}') > 80)" f" AND JAROWINKLER_SIMILARITY(UPPER(pv.PROD_TITLE), '{title}') > 80)" ) else: conditions.append( f"(JAROWINKLER_SIMILARITY(UPPER(art.PARTICIP_FULL_NAME), '{artist}') > 80" f" AND JAROWINKLER_SIMILARITY(UPPER(pv.PROD_TITLE), '{title}') > 80)" ) return f"""SELECT DISTINCT art.PARTICIP_FULL_NAME AS ARTISTNAME, pv.PROD_TITLE AS RELEASENAME, {GRPS_ALBUM_DATE_EXPR} AS RELEASE_DATE, prod.BARCODE AS MANUFACTURER_UPC FROM {GRPS_ALBUM_FROM} WHERE prod.FIRST_RELEASE_DATE BETWEEN TO_VARCHAR('{release_date}'::DATE - {date_window}, 'YYYYMMDD') AND TO_VARCHAR('{release_date}'::DATE + {date_window}, 'YYYYMMDD') AND ({chr(10) + ' OR '.join(conditions)}) LIMIT 500""" def build_grps_ai_track_query(track, release_date: str, date_window: int) -> str: validate_params(release_date, date_window) safe_title = esc(esc_prompt(track.title)) safe_artist = esc(esc_prompt(track.artist)) artist_cond = _grps_artist_condition("p.PARTICIP_FULL_NAME", track.artist) return f"""SELECT p.PARTICIP_FULL_NAME AS ARTISTNAME, l.LOCLAN_TITLE AS TRACKNAME, t.ISRC, {GRPS_TRACK_DATE_EXPR} AS RELEASE_DATE, NULL AS RELEASENAME, SNOWFLAKE.CORTEX.COMPLETE('llama3.1-8b', 'Compare these two songs and answer only YES or NO. ' || '{safe_title}' || ' by ' || '{safe_artist}' || ' ' || l.LOCLAN_TITLE || ' by ' || p.PARTICIP_FULL_NAME || '' ) AS AI_MATCH FROM {GRPS_TRACK_FROM} WHERE {artist_cond} AND t.ORIG_RELEASE_DATE BETWEEN TO_VARCHAR('{release_date}'::DATE - {date_window}, 'YYYYMMDD') AND TO_VARCHAR('{release_date}'::DATE + {date_window}, 'YYYYMMDD') LIMIT 20""" def build_grps_ai_album_query(album, release_date: str, date_window: int) -> str: validate_params(release_date, date_window) safe_title = esc(esc_prompt(album.title)) safe_artist = esc(esc_prompt(album.artist)) artist_cond = _grps_artist_condition("art.PARTICIP_FULL_NAME", album.artist) return f"""SELECT art.PARTICIP_FULL_NAME AS ARTISTNAME, pv.PROD_TITLE AS RELEASENAME, {GRPS_ALBUM_DATE_EXPR} AS RELEASE_DATE, prod.BARCODE AS MANUFACTURER_UPC, SNOWFLAKE.CORTEX.COMPLETE('llama3.1-8b', 'Compare these two albums and answer only YES or NO. ' || '{safe_title}' || ' by ' || '{safe_artist}' || ' ' || pv.PROD_TITLE || ' by ' || art.PARTICIP_FULL_NAME || '' ) AS AI_MATCH FROM {GRPS_ALBUM_FROM} WHERE {artist_cond} AND prod.FIRST_RELEASE_DATE BETWEEN TO_VARCHAR('{release_date}'::DATE - {date_window}, 'YYYYMMDD') AND TO_VARCHAR('{release_date}'::DATE + {date_window}, 'YYYYMMDD') LIMIT 20""" def build_grps_wide_track_query(tracks) -> str: if not tracks: return "" conditions = [] for t in tracks: artist_cond = _grps_artist_condition("p.PARTICIP_FULL_NAME", t.artist) title = esc(t.title).upper() conditions.append(f"({artist_cond} AND UPPER(l.LOCLAN_TITLE) = '{title}')") return f"""SELECT DISTINCT p.PARTICIP_FULL_NAME AS ARTISTNAME, l.LOCLAN_TITLE AS TRACKNAME, t.ISRC, {GRPS_TRACK_DATE_EXPR} AS RELEASE_DATE, NULL AS RELEASENAME FROM {GRPS_TRACK_FROM} WHERE ({chr(10) + ' OR '.join(conditions)}) LIMIT 100""" def build_grps_wide_album_query(albums) -> str: if not albums: return "" conditions = [] for a in albums: artist_cond = _grps_artist_condition("art.PARTICIP_FULL_NAME", a.artist) title = esc(a.title).upper() conditions.append(f"({artist_cond} AND UPPER(pv.PROD_TITLE) = '{title}')") return f"""SELECT DISTINCT art.PARTICIP_FULL_NAME AS ARTISTNAME, pv.PROD_TITLE AS RELEASENAME, {GRPS_ALBUM_DATE_EXPR} AS RELEASE_DATE, prod.BARCODE AS MANUFACTURER_UPC FROM {GRPS_ALBUM_FROM} WHERE ({chr(10) + ' OR '.join(conditions)}) LIMIT 100"""