from random import randint from datetime import datetime import streamlit as st from snowflake.snowpark import Session from snowflake.snowpark.context import get_active_session from snowflake.snowpark.exceptions import SnowparkSessionException from common.constants_general import selection_columns from common.layout.main_fields import main_selection_layout from common.layout.selection_fields import display_mailing_list_columns, display_form_response_columns from common.utility_functions import prepare_export_data, prepare_filter_selections, convert_df_to_csv, validate_email from common.data_utilities import prepare_streamlit_app, execute_audit_flow from common.streamlit_utilities import reset_form st.set_page_config(layout="wide") if "dialog_shown" not in st.session_state: st.session_state["dialog_shown"] = False try: user = st.user.user_name # type: ignore[attr-defined] except AttributeError: user = "user" st.write(f"Hello {user}!") st.subheader("Data Export Streamlit App") file_prefix = 'EXP' if "file_upload_completed" not in st.session_state: st.session_state.file_upload_completed = None if "upload_completed" not in st.session_state: st.session_state.upload_completed = False mailing_list_status = None if st.session_state.file_upload_completed: reset_form() st.session_state.file_upload_completed = False try: session = get_active_session() except SnowparkSessionException: from common.local_connection import get_connection_parameters connection_params = get_connection_parameters("fan_data_export") session = Session.builder.configs(connection_params).create() available_forms, available_mailing_lists = prepare_streamlit_app(session) report_type, recepient_name, recepient_co, recepient_email, recepient_phone, justification, requester, approver = main_selection_layout() with st.container(border=True): if report_type == "Mailing List" or report_type == 'Form Response': if report_type == "Mailing List": result_df = available_mailing_lists else: result_df = available_forms report_builder, field_selector = st.columns(2) with report_builder: st.write('REPORT BUILDER') territory_values = sorted(result_df["TERRITORY_C"].unique()) select_territory = st.selectbox( label="Territory*", options=territory_values, index=None, key="territory_key", ) label_values = sorted(result_df[result_df["TERRITORY_C"] == select_territory][ "LABEL_NAME" ].unique()) select_label = st.selectbox( label="Label*", options=label_values, index=None, key="label_key", ) artist_values = sorted(result_df[ (result_df["TERRITORY_C"] == select_territory) & (result_df["LABEL_NAME"] == select_label) ]["ARTIST_C"].unique()) select_artist = st.selectbox( label="Artist*", options=artist_values, index=None, key="artist_key", ) if report_type == "Mailing List": mailing_list_values = result_df[ (result_df["TERRITORY_C"] == select_territory) & (result_df["LABEL_NAME"] == select_label) & (result_df["ARTIST_C"] == select_artist) ][["MAILING_LIST_NAME_C", "MAILING_LIST_ID"]].drop_duplicates().sort_values(by="MAILING_LIST_NAME_C") mailing_list_name = st.selectbox( label="Mailing List*", options=mailing_list_values, index=None, key="mailing_list_keys", ) mailing_list_status = st.selectbox( label="Active*", options=['Yes', 'No', 'Both'], index=0, key="mailing_list_status_key", ) mailing_list_form_id = st.text_input( label="Form ID", key="mailing_list_form_id_key", disabled=False, help="This field should be used to pull new fans only, not returning fans. Use 18-digit Salesforce ID" ) else: form_id_values = result_df[ (result_df["TERRITORY_C"] == select_territory) & (result_df["LABEL_NAME"] == select_label) & (result_df["ARTIST_C"] == select_artist) ][["FORM_NAME_C", "FORM_ID", "FORM_LAST_MODIFIED_DATE", "MIGRATION_ID_FORM_NAME"]].drop_duplicates().sort_values(by="FORM_LAST_MODIFIED_DATE", ascending=False) form_id_values = form_id_values.fillna("") form_options = [(row['MIGRATION_ID_FORM_NAME'], row['FORM_ID']) for index, row in form_id_values.iterrows()] select_form_name = st.selectbox( label="Form ID*", options=form_options, format_func=lambda x: x[0], index=None, key="form_id_keys", ) ccpa_do_not_sell_selecton = st.checkbox( label="Exclude CCPA Do Not Sell", value=False, key="ccpa_do_not_sell_selection_key", disabled=True ) if report_type == "Mailing List": fan_filter_type, fan_filter_value = st.columns(2) with fan_filter_type: fan_city_selection = st.selectbox( label="Fan City", options=['Equal to', 'Contains', 'Not Equal to', 'Starts with'], index=None, key="fan_city_selection_key", ) fan_state_selection = st.selectbox( label="Fan State", options=['Equal to', 'Contains', 'Not Equal to', 'Starts with'], index=None, key="fan_state_selection_key", ) fan_country_selection = st.selectbox( label="Fan Country", options=['Equal to', 'Contains', 'Not Equal to', 'Starts with'], index=None, key="fan_country_selection_key", ) fan_postal_code_selection = st.selectbox( label="Fan Postal Code", options=['Equal to', 'Contains', 'Not Equal to', 'Starts with'], index=None, key="fan_postal_code_selection_key", ) with fan_filter_value: fan_city_value = st.text_input( label="fan_city_label", key="fan_city_value_key", label_visibility='hidden' ) fan_state_value = st.text_input( label="fan_state_label", key="fan_state_value_key", label_visibility='hidden' ) fan_country_value = st.text_input( label="fan_country_label", key="fan_country_value_key", label_visibility='hidden' ) fan_postal_code_value = st.text_input( label="fan_postal_code_label", key="fan_postal_code_value_key", label_visibility='hidden' ) if report_type == "Mailing List": st.write('Subscription Created Date Range') else: st.write('Form Response Created Date Range') from_, to_ = st.columns(2) with from_: d = st.date_input("From", key="start_date_key", value=None) with to_: to = st.date_input("To", key="end_date_key", value=None) with field_selector: st.markdown('Choose field(s)', help='Choose at least one field') if report_type == "Mailing List": display_mailing_list_columns(report_type) else: display_form_response_columns(report_type) st.button('Clear fields', on_click=reset_form) st.divider() # if st.session_state.upload_completed: # st.subheader('All successfully finished!') try: if report_type == 'Mailing List': form_selection_mandatory_field = mailing_list_name else: form_selection_mandatory_field = select_form_name mailing_list_status = 'Both' except NameError: form_selection_mandatory_field = None mandatory_fields = [ form_selection_mandatory_field, recepient_name, recepient_co, recepient_email, recepient_phone, justification, requester, approver, mailing_list_status ] if form_selection_mandatory_field is not None: optional_export_fields = [key for key, value in st.session_state.items() if value and key in selection_columns.keys()] else: optional_export_fields = [] upload_disabled = True if all([rf for rf in mandatory_fields]) and any(optional_export_fields) and validate_email(recepient_email): st.session_state.upload_completed = False st.subheader('Upload approval document!') upload_disabled = False file_upload = st.file_uploader( "Choose a file", type="pdf", disabled=upload_disabled ) if file_upload: with st.spinner("Generating data preview..."): if report_type == 'Mailing List': export_id = (select_label or "") + "-" + (mailing_list_name or "") + "-" + datetime.now().strftime("%Y-%m-%d_%H.%M.%S") else: file_form_name = select_form_name[0] if select_form_name else "" export_id = (select_label or "") + "-" + file_form_name + "-" + datetime.now().strftime( "%Y-%m-%d_%H.%M.%S") st.divider() st.markdown(f'New export created: {export_id}', help="Export id follows this format: EXPORT TYPE-YYYY-MM-DD_HH.MM.SS") filter_columns = prepare_filter_selections(st.session_state) export_df, executed_query, ignore_email_col = prepare_export_data(session, st.session_state, report_type) export_df['EXPORT_ID'] = export_id main_export_df = export_df.drop(columns=["EXPORT_DATA", "FAN_ID", "EXPORT_ID"]) export_df = export_df[["EXPORT_ID", "FAN_ID", "EXPORT_DATA"]] final_btn = st.download_button( label="Click to Download Data and Upload Approval Document", file_name=f"{export_id}.csv", mime="text/csv", data=convert_df_to_csv(main_export_df, ignore_email_col), key='download-csv', type="primary", on_click=execute_audit_flow, args=(executed_query, str(filter_columns), user, export_id, export_df, session, file_upload, 'main page') )