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 fan_id_sources, custom_columns from common.data_utilities import execute_audit_flow, initiate_data_export from common.streamlit_utilities import reset_form2 from common.layout.fan_export_fields import selection_layout from common.utility_functions import convert_df_to_csv try: user = st.user.user_name # type: ignore[attr-defined] except AttributeError: user = "user" st.markdown( "On this page you will be able to upload your fan id values and export related data!", help="IDs must match CRM fan id format", ) if "fan_export_upload_completed" not in st.session_state: st.session_state.fan_export_upload_completed = False if st.session_state.fan_export_upload_completed: reset_form2() st.subheader("All successfully completed!") fan_id_source = st.selectbox( placeholder="select fan id source", label="Source*", options=fan_id_sources, index=None ) if fan_id_source is not None: fan_id_source_index = fan_id_sources.index(fan_id_source) approval_file_upload = None if fan_id_source_index == 0: with st.container(border=True): st.write("Fill in values to proceed:") a_, b_, c_ = st.columns(3) with a_: form_id_value = st.text_input("Form ID or Migration ID*", key="contest_form_id_key") today = datetime.now() two_years_ago = datetime(today.year - 2, 1, 1).date() contest_time_frame = st.date_input( "Select date range for contest time (DD.MM.YYYY)*", (two_years_ago, today), format="DD.MM.YYYY", ) nr_of_winners = st.number_input( "Number of winners*", min_value=1, key="nr_of_winners_key" ) bonus_entry_selector = st.checkbox( label="Bonus entry", key="bonus_entry_selector_key", value=True ) custom_field_id_1, custom_field_operator_1, custom_filed_value_1 = st.columns(3) with custom_field_id_1: selected_custom_field = st.selectbox( "Custom Field Name 1", [""] + custom_columns, key="custom_field_id_1_key" ) with custom_field_operator_1: st.selectbox( label="Field Operator", options=["Equal to", "Contains", "Not Equal to", "Starts with"], key="custom_field_operator_1_key", ) with custom_filed_value_1: st.text_input("Custom Field Value", key="custom_field_value_1_key") custom_field_id_2, custom_field_operator_2, custom_filed_value_2 = st.columns(3) with custom_field_id_2: selected_custom_field = st.selectbox( "Custom Field Name 2", [""] + custom_columns, key="custom_field_id_2_key" ) with custom_field_operator_2: st.selectbox( label="Field Operator", options=["Equal to", "Contains", "Not Equal to", "Starts with"], key="custom_field_operator_2_key", ) with custom_filed_value_2: st.text_input("Custom Field Value", key="custom_field_value_2_key") try: session_2 = get_active_session() except SnowparkSessionException: from common.local_connection import get_connection_parameters connection_params = get_connection_parameters("fan_data_export") session_2 = Session.builder.configs(connection_params).create() # if "fan_export_upload_completed" not in st.session_state: # st.session_state.fan_export_upload_completed = False # # if st.session_state.fan_export_upload_completed: # reset_form2() # st.subheader("All successfully completed!") optional_export_fields, fan_export_justification, fan_export_det_justification = ( selection_layout(fan_id_source_index) ) st.button("Clear fields", on_click=reset_form2) st.divider() # using your own file for getting fans if fan_id_source_index == 1: if not any(optional_export_fields) or fan_export_justification is None: upload_disabled = True st.session_state.fan_export_upload_completed = False else: upload_disabled = False file_upload = st.file_uploader( "Upload list of fans", type="csv", disabled=upload_disabled, help="Upload CSV file containing one column of fan id values", ) if file_upload is None: approval_upload_disabled = True else: csv_header = st.checkbox("First row of CSV file contains column header", True) st.divider() approval_upload_disabled = False approval_file_upload = st.file_uploader( "Upload approval", type="pdf", disabled=approval_upload_disabled ) if approval_file_upload is None: st.write("Approval missing") else: with st.spinner("Finding relevant fan data...."): ( main_fan_export_df, returned_fan_df, query, ignore_email_col, fan_data_export_id, ) = initiate_data_export( fan_id_source_index, session_2, file_upload, csv_header ) # using contest winners else: continue_ = all( [any(optional_export_fields), fan_export_justification is not None, form_id_value != ""] ) if continue_ is False: st.error( "Make sure you have selected all required fields: form id, justification and minimum 1 fan related column." ) st.session_state.fan_export_upload_completed = False else: st.divider() approval_upload_disabled = False approval_file_upload = st.file_uploader( "Upload approval", type="pdf", disabled=approval_upload_disabled ) if approval_file_upload is None: st.write("Approval missing") else: with st.spinner("Finding relevant fan data...."): ( main_fan_export_df, returned_fan_df, query, ignore_email_col, fan_data_export_id, ) = initiate_data_export( fan_id_source_index, session_2, form_id_value=form_id_value, contest_time_frame=contest_time_frame, nr_of_winners=nr_of_winners, bonus_entry_selected=bonus_entry_selector, custom_field_params=[ st.session_state.custom_field_id_1_key, st.session_state.custom_field_operator_1_key, st.session_state.custom_field_value_1_key, st.session_state.custom_field_id_2_key, st.session_state.custom_field_operator_2_key, st.session_state.custom_field_value_2_key, ], ) st.divider() if approval_file_upload is not None: if returned_fan_df.shape[0] > 0: if fan_export_det_justification is not None: justification_input = "Other: " + fan_export_det_justification else: justification_input = fan_export_justification # user filters define here if fan_id_source_index == 0: query_filters = { "justification": justification_input, "form_id": form_id_value, "date_range": contest_time_frame, } else: query_filters = {"justification": justification_input} download_btn = st.download_button( label="Click to Download Data and Upload Approval Document", file_name=f"{fan_data_export_id}.csv", mime="text/csv", data=convert_df_to_csv(main_fan_export_df, False), key="download-fan-export-csv", type="primary", on_click=execute_audit_flow, args=( query, query_filters, user, fan_data_export_id, returned_fan_df, session_2, approval_file_upload, "fan-id contest sub page", ), ) else: st.write("No matching fan data found!") else: st.error("In order to continue please select proper method for getting list of fan ids")