"""
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"""