import numpy as np import streamlit as st import pandas as pd from snowflake.snowpark.exceptions import SnowparkSessionException from streamlit.runtime.media_file_storage import MediaFileStorageError from common.statics_lists import acquisition_channels, dsps, input_file_columns from common.local_connection import get_connection_parameters from snowflake.snowpark.session import Session from tzlocal import get_localzone from common.user_input import create_layout from snowflake.snowpark.context import get_active_session from collections import defaultdict import pytz from common.functions import ( create_id, df_check_columns, df_quality_check, get_campaign_ids, get_filenames, prepare_fileupload_app, handle_file_upload, country_to_iso2, ) try: session = get_active_session() except SnowparkSessionException: connection_params = get_connection_parameters("fileupload") session = Session.builder.configs(connection_params).create() col1, col2 = st.columns([1, 3]) with col1: try: st.image("common/sme.svg", caption="Sony Music", width=150) except MediaFileStorageError: st.write("logo placeholder") with col2: st.title("Subscription Upload") st.write( "Contact crm.support@sonymusic.com if the mailing list you are looking for doesn't appear." ) st.divider() st.subheader("Please fill all fields to proceed!") result_df = prepare_fileupload_app(session) # Initialize session state for the selectbox if "territory" not in st.session_state: st.session_state.territory = None if "overwrite_data" not in st.session_state: st.session_state.overwrite_data = None if "upload_name" not in st.session_state: st.session_state.upload_name = None if "file_source_description" not in st.session_state: st.session_state.file_source_description = None if "dsp" not in st.session_state: st.session_state.dsp = None if "crm_generated" not in st.session_state: st.session_state.crm_generated = None if "acquisition_channels" not in st.session_state: st.session_state.acquisition_channels = None if "error_message" not in st.session_state: st.session_state.error_message = None if "tla_id_value" not in st.session_state: st.session_state.tla_id_value = None if "file_upload_completed" not in st.session_state: st.session_state.file_upload_completed = None ( file_source_description, select_acquisition_channels, select_is_crm_generated, select_dsp, select_mailing_list, select_label, select_artist, select_territory, overwrite_fan_data, tla_id_value, ) = create_layout(result_df, dsps, acquisition_channels) st.divider() required_fields = [ select_acquisition_channels, select_is_crm_generated, select_dsp, file_source_description, select_mailing_list, select_label, select_artist, select_territory, tla_id_value, ] upload_disabled = True if st.session_state.file_upload_completed: st.subheader("File upload completed succesfully!") if all([rf for rf in required_fields]): upload_disabled = False st.session_state["disable"] = False st.subheader("Upload an attachment") # multifile support # https://discuss.streamlit.io/t/how-to-use-multiple-files-loaded-with-file-uploader/25051 file_upload = st.file_uploader( "Choose a file", type="csv", disabled=upload_disabled ) if file_upload is not None: columns_to_use = input_file_columns dataframe_column_types = defaultdict(lambda: str) # type: ignore dataframe = pd.read_csv( file_upload, delimiter=",", dtype=dataframe_column_types ) existing_file_uploads = get_filenames(session) if file_upload.name in existing_file_uploads: st.write(":red[Error! File with same name already uploaded!!]") else: st.session_state.valid_filename = False dataframe, matching_columns = df_check_columns(dataframe, columns_to_use) optins_missing = dataframe["OPTIN_REQUIRED"].isna().sum() > 0 if not matching_columns: st.write( ":red[Error! Uploaded CSV file contains wrong column names or some columns are missing or extra.]" ) elif optins_missing: st.write( ":red[Error! Uploaded CSV file cannot contain missing values for opt-in column.]" ) else: dataframe = dataframe.drop( ["LABEL_REQUIRED", "TERRITORY_2_DIGIT_ISO_REQUIRED"], axis=1 ) # TODO: test in Snowflake env also local_tz = get_localzone() if local_tz is None: local_tz = pytz.UTC dataframe["DATE_CREATED"] = pd.to_datetime("today").tz_localize( local_tz ) dataframe["DATE_CREATED"] = dataframe["DATE_CREATED"].dt.tz_convert( "UTC" ) dataframe["DATE_CREATED"] = dataframe["DATE_CREATED"].astype(str) try: user = st.experimental_user.user_name except AttributeError: user = "local_session_user" dataframe["CAMPAIGN_CREATED_BY"] = user dataframe["FILE_NAME"] = file_upload.name dataframe["FILE_SOURCE_DESCRIPTION"] = file_source_description dataframe["TERRITORY"] = select_territory dataframe["LABEL"] = select_label dataframe["ARTIST"] = select_artist dataframe["MAILING_LIST"] = select_mailing_list dataframe["TLA_ID"] = tla_id_value[0] dataframe["ACQUISITION_CHANNEL"] = select_acquisition_channels dataframe["CRM_GENERATED"] = select_is_crm_generated dataframe["DSP"] = select_dsp dataframe["OVERWRITE_FAN_DATA"] = overwrite_fan_data dataframe["CAMPAIGN_NAME"] = ( dataframe["ARTIST"] + " - " + dataframe["FILE_SOURCE_DESCRIPTION"] + " - " + pd.to_datetime("today").strftime("%m/%d/%Y") ) # dataframe["MOBILE_PHONE"] = dataframe["MOBILE_PHONE"].apply(check_phone) dataframe = country_to_iso2(dataframe, session) allowed_genders = [ "female", "male", "prefer not to answer", "non-binary/other", ] dataframe["GENDER"] = dataframe["GENDER"].astype(str) dataframe["GENDER"] = dataframe["GENDER"].apply( lambda x: x if x.lower() in allowed_genders else np.nan ) dataframe = create_id(dataframe) existing_campaign_ids = get_campaign_ids(session) if dataframe["CAMPAIGN_ID"].unique()[0] in existing_campaign_ids: st.write( ":red[Error! This campaign already exists, please update the input values to create a new campaign!]" ) else: if select_acquisition_channels is not None: st.divider() upload_focus = acquisition_channels[select_acquisition_channels] if not upload_disabled: df_quality_check( dataframe, upload_focus, select_acquisition_channels ) st.divider() submit_button = st.button( label="Upload data", on_click=handle_file_upload, args=(dataframe, session, "EVENT_FILE_UPLOAD"), disabled=upload_disabled, ) if submit_button: upload_disabled = True # st.session_state['disable'] = True if st.session_state.error_message: st.write(":red[UPLOAD FAILED!!!]") st.write(f":red[{st.session_state.error_message}]") else: st.session_state.file_upload_completed = None st.write("Make sure all previous fields have been filled to enable file upload.")