"""Add artist page - add new artists to roster tables.""" import streamlit as st from common import queries from common.db import get_session from common.services import add_artist_to_roster, search_artists, search_vendors from common.types import AddArtistForm from common.utils import init_session_state, search_countries, validate_country_code # Page config st.set_page_config( page_title="Artist Roster - Add Artist", page_icon="➕", layout="wide" ) st.title("➕ Add Artist to Roster") st.write("Add a new artist to vendor roster tables") # Get session session = get_session() # Show schema indicator if queries.FANSIFTER_SCHEMA == "prod": st.error("âš ī¸ PRODUCTION MODE - Writes are BLOCKED for safety") else: st.info(f"📊 Using schema: {queries.SCHEMA_NAME}") # Initialize session state init_session_state("add_success", False) init_session_state("add_error", None) init_session_state("selected_vendor_id", None) init_session_state("selected_artist_uuid", None) init_session_state("selected_roster_type", None) init_session_state("country_code", "") init_session_state("status", "ACTIVE") init_session_state("is_artist_team", False) st.divider() # Vendor search section (outside form for interactivity) st.subheader("Step 1: Select Vendor") col1, col2 = st.columns(2) with col1: vendor_id_search = st.text_input( "Vendor ID", value=st.session_state.selected_vendor_id or "", help="Enter exact vendor ID", key="vendor_id_search", ) with col2: vendor_name_search = st.text_input( "Or search by Vendor Name", help="Search for vendor by name", key="vendor_name_search", ) # Vendor search results with selection buttons if vendor_name_search: with st.spinner("Searching vendors..."): vendors = search_vendors(session, vendor_name=vendor_name_search) if vendors: st.write(f"Found {len(vendors)} vendor(s):") for vendor in vendors[:10]: col_a, col_b = st.columns([4, 1]) with col_a: st.code(f"ID: {vendor.vendor_id} | Name: {vendor.name}") with col_b: if st.button("Select", key=f"select_vendor_{vendor.vendor_id}"): st.session_state.selected_vendor_id = str(vendor.vendor_id) st.rerun() else: st.warning("No vendors found") # Detect vendor brand type and show vendor name when vendor ID is entered vendor_brand_type = None # 'SME' or 'NON_SME' or None available_roster_types = [] vendor_display_name = None # Use selected vendor ID from session state or direct input final_vendor_id = vendor_id_search or st.session_state.selected_vendor_id if final_vendor_id: try: vendor_id = int(final_vendor_id) from common.services import is_vendor_sme with st.spinner("Loading vendor information..."): # Get vendor details vendors = search_vendors(session, vendor_id=vendor_id) if vendors: vendor_display_name = vendors[0].name # Check brand type is_sme = is_vendor_sme(session, vendor_id) vendor_brand_type = "SME" if is_sme else "NON_SME" if is_sme: available_roster_types = ["MAIN_REP", "LOCAL_REP"] st.success( f"✅ **Vendor: {vendor_display_name}** (ID: {vendor_id})\n\n" "đŸĸ SME Brand (Sony Music) - can add to MAIN_REP or LOCAL_REP" ) else: available_roster_types = ["ARTIST_ROSTER"] st.success( f"✅ **Vendor: {vendor_display_name}** (ID: {vendor_id})\n\n" "đŸĸ Non-SME Brand - will be added to ARTIST_ROSTER" ) except ValueError: st.warning("Invalid Vendor ID format") st.divider() # Artist search section (outside form for interactivity) st.subheader("Step 2: Select Artist") col1, col2, col3 = st.columns(3) with col1: artist_uuid_search = st.text_input( "Artist UUID", value=st.session_state.selected_artist_uuid or "", help="Enter exact artist UUID", key="artist_uuid_search", ) with col2: artist_name_search = st.text_input( "Or search by Artist Name", help="Search for artist by name", key="artist_name_search", ) with col3: artist_spotify_search = st.text_input( "Or search by Spotify ID", help="Search for artist by Spotify ID (e.g., 6zyG7x3rPnqitv8h5FATI9)", key="artist_spotify_search", ) # Artist search results with selection buttons search_artists_list = [] if artist_name_search: with st.spinner("Searching artists by name..."): search_artists_list = search_artists(session, artist_name=artist_name_search) elif artist_spotify_search: with st.spinner("Searching artists by Spotify ID..."): search_artists_list = search_artists(session, spotify_id=artist_spotify_search) if search_artists_list: st.write(f"Found {len(search_artists_list)} artist(s):") for artist in search_artists_list[:10]: col_a, col_b = st.columns([4, 1]) with col_a: spotify_info = ( f" | Spotify: {artist.spotify_id}" if artist.spotify_id else "" ) st.code( f"UUID: {artist.artist_uuid} | Name: {artist.artist_name}{spotify_info}" ) with col_b: if st.button("Select", key=f"select_artist_{artist.artist_uuid}"): st.session_state.selected_artist_uuid = artist.artist_uuid st.rerun() else: if artist_name_search or artist_spotify_search: st.warning("No artists found") # Show artist details when UUID is entered artist_display_name = None artist_display_uuid = None # Use selected artist UUID from session state or direct input final_artist_uuid = artist_uuid_search or st.session_state.selected_artist_uuid if final_artist_uuid: with st.spinner("Loading artist information..."): artists = search_artists(session, artist_uuid=final_artist_uuid.strip()) if artists: artist = artists[0] artist_display_name = artist.artist_name artist_display_uuid = artist.artist_uuid spotify_info = ( f"Spotify ID: `{artist.spotify_id}`" if artist.spotify_id else "No Spotify ID" ) st.success( f"✅ **Artist: {artist_display_name}**\n\n" f"UUID: `{artist_display_uuid}`\n\n" f"{spotify_info}" ) else: st.warning("âš ī¸ Artist UUID not found in database") st.divider() # Roster type selection (outside form for interactivity) st.subheader("Step 3: Select Roster Type") roster_type: str | None if vendor_brand_type == "SME": # Get index for saved roster type if valid for current vendor default_index = 0 if ( st.session_state.selected_roster_type and st.session_state.selected_roster_type in available_roster_types ): default_index = available_roster_types.index( st.session_state.selected_roster_type ) roster_type = st.selectbox( "Target Roster *", options=available_roster_types, index=default_index, help="Select MAIN_REP or LOCAL_REP for SME brands", key="roster_type_select", ) # Save to session state if roster_type != st.session_state.selected_roster_type: st.session_state.selected_roster_type = roster_type elif vendor_brand_type == "NON_SME": roster_type = "ARTIST_ROSTER" st.text_input( "Target Roster *", value="ARTIST_ROSTER", disabled=True, help="Non-SME brands automatically use ARTIST_ROSTER", key="roster_type_display", ) # Save to session state if st.session_state.selected_roster_type != "ARTIST_ROSTER": st.session_state.selected_roster_type = "ARTIST_ROSTER" else: st.selectbox( "Target Roster *", options=[], help="Select Vendor first (Step 1) to see available roster types", key="roster_type_placeholder", disabled=True, ) roster_type = None st.divider() # Additional fields (outside form for validation feedback) st.subheader("Step 4: Additional Fields") country_code_input = None status_input = "ACTIVE" is_artist_team_input = False if roster_type == "LOCAL_REP": st.write("**LOCAL_REP Specific Fields**") col1, col2 = st.columns([2, 1]) with col1: # Country name search input country_search = st.text_input( "Search Country by Name", help="Type country name to search (e.g., 'United States', 'Ukraine', 'Japan')", key="country_search_input", ) with col2: # Direct country code input (for advanced users) country_code_raw = st.text_input( "Or enter Country Code *", value=st.session_state.country_code, max_chars=2, help="2-letter ISO code (e.g., US, UA, JP)", key="country_code_input", ) # Show search results with Select buttons if country_search and len(country_search) >= 2: countries = search_countries(country_search) if countries: st.write(f"Found {len(countries)} countrie(s):") for code, name in countries[:10]: col_a, col_b = st.columns([4, 1]) with col_a: st.code(f"{name} ({code})") with col_b: if st.button("Select", key=f"select_country_{code}"): st.session_state.country_code = code st.rerun() else: st.warning("No countries found") # Update session state and convert to uppercase country_code_input = country_code_raw.upper() if country_code_input != st.session_state.country_code: st.session_state.country_code = country_code_input # Use saved country code if available if st.session_state.country_code: country_code_input = st.session_state.country_code # Validate country code in real-time using pycountry if country_code_input: is_valid, country_name = validate_country_code(country_code_input) if is_valid and country_name: st.success( f"✅ Valid country code: **{country_code_input}** ({country_name})" ) else: if len(country_code_input) != 2: st.error("❌ Country code must be exactly 2 characters") elif not country_code_input.isalpha(): st.error("❌ Country code must contain only letters") elif not country_code_input.isupper(): st.error("❌ Country code must be uppercase") else: st.error( f"❌ Invalid ISO Alpha-2 country code: **{country_code_input}**" ) elif roster_type == "MAIN_REP": st.write("**MAIN_REP Specific Fields**") col1, col2 = st.columns(2) with col1: # Get index for current status value status_options = ["ACTIVE", "INACTIVE"] status_index = ( status_options.index(st.session_state.status) if st.session_state.status in status_options else 0 ) status_input = st.selectbox( "Status", options=status_options, index=status_index, help="Select status (defaults to ACTIVE)", key="status_select", ) # Update session state if status_input != st.session_state.status: st.session_state.status = status_input with col2: is_artist_team_input = st.checkbox( "Is Artist Team", value=st.session_state.is_artist_team, help="Check if this is an artist team entry", key="is_artist_team_checkbox", ) # Update session state if is_artist_team_input != st.session_state.is_artist_team: st.session_state.is_artist_team = is_artist_team_input # Show info about subaccount ID st.info("â„šī¸ Subaccount ID is automatically set to 0") st.divider() # Submit section st.subheader("Step 5: Submit") # Validate if form is ready to submit form_is_ready = False duplicate_exists = False if final_vendor_id and final_artist_uuid and roster_type: # Check roster-specific required fields if roster_type == "LOCAL_REP": # LOCAL_REP requires valid country code - use pycountry validation if country_code_input: is_valid, _ = validate_country_code(country_code_input) form_is_ready = is_valid else: form_is_ready = False else: # MAIN_REP and ARTIST_ROSTER don't require additional fields form_is_ready = True # Check for duplicates if all fields are ready if form_is_ready: from common.services import check_duplicate try: vendor_id = int(final_vendor_id) form_data = AddArtistForm( vendor_id=vendor_id, artist_uuid=final_artist_uuid.strip(), subaccount_id=0, # Always 0 roster_type=roster_type, country_code=country_code_input if country_code_input else None, status=status_input, is_artist_team=is_artist_team_input, ) with st.spinner("Checking for duplicates..."): duplicate_exists = check_duplicate(session, form_data) except Exception: # If check fails, assume no duplicate duplicate_exists = False # Show ready indicator if duplicate_exists: st.error("❌ This artist roster entry already exists! Cannot add duplicate.") elif form_is_ready: st.success("✅ All required fields are filled - ready to submit!") else: st.warning("âš ī¸ Please complete all required fields above") with st.form("add_artist_form", clear_on_submit=False): # Submit button - green when ready, grey when not ready or duplicate button_enabled = form_is_ready and not duplicate_exists submitted = st.form_submit_button( "✅ Add Artist to Roster", type="primary" if button_enabled else "secondary", disabled=not button_enabled, ) if submitted: # Reset previous messages st.session_state.add_success = False st.session_state.add_error = None # Validate required fields using session state values if not final_vendor_id: st.session_state.add_error = "Vendor ID is required (Step 1)" elif not final_artist_uuid: st.session_state.add_error = "Artist UUID is required (Step 2)" elif roster_type == "LOCAL_REP" and not country_code_input: st.session_state.add_error = "Country code is required for LOCAL_REP" else: # Try to add artist try: from common.services import check_duplicate vendor_id = int(final_vendor_id) subaccount_id = 0 # Always 0 form_data = AddArtistForm( vendor_id=vendor_id, artist_uuid=final_artist_uuid.strip(), subaccount_id=subaccount_id, # Always 0 roster_type=roster_type, country_code=country_code_input if country_code_input else None, status=status_input, is_artist_team=is_artist_team_input, ) # Check for duplicate before adding if check_duplicate(session, form_data): st.session_state.add_error = ( "This artist roster entry already exists! Cannot add duplicate." ) else: with st.spinner("Adding artist to roster..."): add_artist_to_roster(session, form_data) st.session_state.add_success = True # Conditional clearing based on roster type if roster_type == "LOCAL_REP": # For LOCAL_REP: Keep vendor, artist, and roster_type for adding to multiple countries # Only clear country_code to select new country st.session_state.country_code = "" else: # For MAIN_REP and ARTIST_ROSTER: Clear vendor and artist st.session_state.selected_vendor_id = None st.session_state.selected_artist_uuid = None except ValueError as e: st.session_state.add_error = str(e) except Exception as e: st.session_state.add_error = f"Unexpected error: {str(e)}" # Trigger rerun to show messages outside form st.rerun() # Display messages outside form if st.session_state.add_success: st.success("✅ Artist successfully added to roster!") st.balloons() # Clear success flag after displaying st.session_state.add_success = False if st.session_state.add_error: st.error(f"❌ Error: {st.session_state.add_error}") # Clear error flag after displaying st.session_state.add_error = None st.divider() # Help section with st.expander("â„šī¸ Help & Rules"): st.markdown( """ ### Adding Artists to Roster **Required Fields:** - **Vendor ID**: The vendor's unique identifier (determines brand type) - **Artist UUID**: The artist's unique identifier from GLOBAL_PARTICIPANT - **Subaccount ID**: The subaccount number - **Target Roster**: Automatically determined by brand type **Brand Types:** 1. **SME Brands (Sony Music)** - Detected automatically based on vendor's COMPANY_BRAND - Can add to: **MAIN_REP** or **LOCAL_REP** - Cannot have artist in both rosters simultaneously **MAIN_REP** - Main representative - Requires: Status (ACTIVE/INACTIVE) - Optional: Is Artist Team checkbox **LOCAL_REP** - Territory-specific representative - Requires: Country Code (2-letter ISO: US, GB, JP, etc.) - Can have multiple countries for same artist 2. **Non-SME Brands (All others)** - Can only add to: **ARTIST_ROSTER** - No additional fields required **Business Rules:** - **SME artists** cannot exist in both MAIN_REP and LOCAL_REP - If in MAIN_REP → cannot add to LOCAL_REP - If in LOCAL_REP → cannot add to MAIN_REP - Future update functionality will allow switching between rosters - **Duplicate prevention**: System uses MERGE to prevent duplicate entries - Match on: vendor_id + artist_uuid + subaccount_id - LOCAL_REP also matches on country_code (allows multiple countries) - **Cross-validation**: Checks opposite roster before allowing insert **Search Tips:** - Use exact IDs for fastest lookup - Name searches support partial matching - Search results show top 10 matches - Enter Vendor ID first to see available roster options """ )