""" Priority Release Check — Streamlit on Snowflake Matches weekly priority release emails against DIM and GRPS data sources. """ import re from datetime import date as date_type import streamlit as st import pandas as pd from parse_email import parse_email from matching import run_all_tiers, TrackMatch, AlbumMatch # --------------------------------------------------------------------------- # Page config # --------------------------------------------------------------------------- st.set_page_config( page_title="Priority Release Check", page_icon=":musical_note:", layout="wide", ) # --------------------------------------------------------------------------- # Spotify-inspired dark theme via CSS # --------------------------------------------------------------------------- st.markdown(""" """, unsafe_allow_html=True) # --------------------------------------------------------------------------- # Session / connection # --------------------------------------------------------------------------- @st.cache_resource def get_session(): return st.connection("snowflake").session() # --------------------------------------------------------------------------- # Styling # --------------------------------------------------------------------------- TIER_STYLES = { "exact": {"label": "Exact", "color": "#49a22f", "bg": "#eaf5e6"}, "partial": {"label": "Partial", "color": "#0670db", "bg": "#e6f0fb"}, "fuzzy": {"label": "Fuzzy", "color": "#0670db", "bg": "#e6f0fb"}, "ai": {"label": "AI", "color": "#7c3aed", "bg": "#f0ebff"}, "wide": {"label": "Outside Window", "color": "#eb8100", "bg": "#fef3e6"}, "not_found": {"label": "Not Found", "color": "#cf2617", "bg": "#fce8e6"}, } # --------------------------------------------------------------------------- # Helper functions # --------------------------------------------------------------------------- def _sort_key(pair): """Sort: not_found in both first (0), found in one (1), found in both (2).""" dim_match, grps_match = pair score = 0 if dim_match.match_tier != "not_found": score += 1 if grps_match.match_tier != "not_found": score += 1 return score def _color_match(val): """Style function for match tier cells (dark theme friendly).""" colors = { "Exact": "background-color: #1a3d1a; color: #1DB954; font-weight: 600", "Partial": "background-color: #1a2d4d; color: #4da6ff; font-weight: 600", "Fuzzy": "background-color: #1a2d4d; color: #4da6ff; font-weight: 600", "AI": "background-color: #2d1a4d; color: #b388ff; font-weight: 600", "Outside Window": "background-color: #3d2d1a; color: #ffb347; font-weight: 600", "Not Found": "background-color: #3d1a1a; color: #ff6b6b; font-weight: 600", } return colors.get(val, "") def render_match_table(dim_matches, grps_matches, item_type="track"): """Render a combined DIM + GRPS match table.""" rows = [] paired = list(zip(dim_matches, grps_matches)) paired.sort(key=_sort_key) for dim, grps in paired: row = { "Artist": dim.email_artist, "Title": dim.email_title, "Label": dim.label, "DIM": TIER_STYLES.get(dim.match_tier, TIER_STYLES["not_found"])["label"], "GRPS": TIER_STYLES.get(grps.match_tier, TIER_STYLES["not_found"])["label"], "Date": dim.db_release_date or grps.db_release_date or "", } if item_type == "track": row["ISRC"] = dim.isrc or (getattr(grps, "isrc", None) or "") else: row["UPC"] = dim.upc or (getattr(grps, "upc", None) or "") if dim.notes: row["Notes"] = dim.notes rows.append(row) df = pd.DataFrame(rows) df.index = range(1, len(df) + 1) styled = df.style.map(_color_match, subset=["DIM", "GRPS"]) st.dataframe(styled, use_container_width=True) def display_results(results): """Render track and album result tables.""" dim_tracks = results["dim_tracks"] dim_albums = results["dim_albums"] grps_tracks = results["grps_tracks"] grps_albums = results["grps_albums"] # Summary metrics dim_t_found = sum(1 for m in dim_tracks if m.match_tier != "not_found") grps_t_found = sum(1 for m in grps_tracks if m.match_tier != "not_found") dim_a_found = sum(1 for m in dim_albums if m.match_tier != "not_found") grps_a_found = sum(1 for m in grps_albums if m.match_tier != "not_found") c1, c2, c3, c4 = st.columns(4) c1.metric("DIM Tracks", f"{dim_t_found}/{len(dim_tracks)}") c2.metric("GRPS Tracks", f"{grps_t_found}/{len(grps_tracks)}") c3.metric("DIM Albums", f"{dim_a_found}/{len(dim_albums)}") c4.metric("GRPS Albums", f"{grps_a_found}/{len(grps_albums)}") if dim_tracks: st.subheader("Tracks") render_match_table(dim_tracks, grps_tracks, item_type="track") if dim_albums: st.subheader("Albums") render_match_table(dim_albums, grps_albums, item_type="album") st.caption("\\* ISRC/UPC and Date values are sourced from whichever database matched first (DIM preferred, GRPS as fallback).") def render_parse_sidebar(parsed): """Show parse results in the sidebar.""" with st.sidebar: st.markdown("---") st.subheader("Parse Results") st.markdown(f"**Release Date:** {parsed.release_date or 'Not detected'}") st.markdown(f"**Tracks:** {len(parsed.tracks)}") st.markdown(f"**Albums:** {len(parsed.albums)}") if parsed.tracks: st.markdown("---") st.caption("TRACKS") for t in parsed.tracks: label_tag = f" _({t.label})_" if t.label else "" notes_tag = f" — {t.notes}" if t.notes else "" st.markdown(f"- **{t.artist}** — {t.title}{label_tag}{notes_tag}") if parsed.albums: st.markdown("---") st.caption("ALBUMS") for a in parsed.albums: label_tag = f" _({a.label})_" if a.label else "" notes_tag = f" — {a.notes}" if a.notes else "" st.markdown(f"- **{a.artist}** — {a.title}{label_tag}{notes_tag}") # --------------------------------------------------------------------------- # Header # --------------------------------------------------------------------------- st.title("Priority Release Check") st.caption("Paste a weekly priority release email to match tracks and albums against DIM and GRPS databases.") # --------------------------------------------------------------------------- # Sidebar (base) # --------------------------------------------------------------------------- with st.sidebar: st.header("Settings") date_window = st.slider("Date Window (days)", min_value=3, max_value=14, value=7, step=1) # --------------------------------------------------------------------------- # State management # --------------------------------------------------------------------------- if "parsed" not in st.session_state: st.session_state.parsed = None if "results" not in st.session_state: st.session_state.results = None if "show_results" not in st.session_state: st.session_state.show_results = False # --------------------------------------------------------------------------- # Main UI # --------------------------------------------------------------------------- if not st.session_state.show_results: # --- Email input view --- email_text = st.text_area( "Paste email text", height=350, placeholder="Paste the weekly priority release email here...", key="email_input", ) col1, col2 = st.columns([1, 3]) with col1: date_override = st.text_input( "Date override (YYYY-MM-DD)", placeholder="Auto-detected", key="date_override", ) if st.button("Parse Email"): if email_text: parsed = parse_email(email_text) if date_override: try: date_type.fromisoformat(date_override) parsed.release_date = date_override except ValueError: st.error("Invalid date. Use YYYY-MM-DD format with a valid calendar date (e.g. 2026-04-03).") st.session_state.parsed = parsed # Show parse results in sidebar if available if st.session_state.parsed: p = st.session_state.parsed render_parse_sidebar(p) if not p.release_date: st.warning("Could not detect release date. Please enter a date override.") else: col_a, col_b, col_c = st.columns(3) col_a.metric("Release Date", p.release_date) col_b.metric("Tracks", len(p.tracks)) col_c.metric("Albums", len(p.albums)) if st.button("Check Releases"): session = get_session() progress_bar = st.progress(0, text="Starting matching...") phase_pcts = {"exact": 10, "partial": 30, "fuzzy": 50, "ai": 70, "wide": 90, "done": 100} def progress(phase, detail): pct = phase_pcts.get(phase, 0) progress_bar.progress(pct, text=detail) results = run_all_tiers(session, p, date_window, progress_callback=progress) progress_bar.empty() st.session_state.results = results st.session_state.show_results = True st.rerun() else: # --- Results view --- # Show parse info in sidebar on results page too if st.session_state.parsed: render_parse_sidebar(st.session_state.parsed) if st.button("< Check Another"): st.session_state.parsed = None st.session_state.results = None st.session_state.show_results = False st.rerun() if st.session_state.results: display_results(st.session_state.results)