import json from typing import Tuple import re import pandas as pd import streamlit as st from snowflake.snowpark import Session from streamlit.runtime.state import SessionStateProxy from common.constants_general import ( filter_columns_dict, selection_columns, main_filter_columns, advanced_filter_columns_dict, advanced_filter_operators_dict, DATABASE_NAME, SCHEMA_NAME, FORMS, FORM_RESPONSES, FANS, MAILING_LISTS, SUBSCRIPTIONS, PROMOTIONS, desired_column_order, ) def convert_df_to_csv(df: pd.DataFrame, ignore: bool) -> bytes: """ Function to convert dataframe to csv for user download :param df: dataframe :param ignore: whether to exclude EMAIL column from final export :return: Object representing csv file for download """ ordered_cols = [col for col in desired_column_order if col in df.columns] remaining_cols = [col for col in df.columns if col not in ordered_cols] df = df[ordered_cols + remaining_cols] if ignore: df = df.drop(columns="EMAIL") return df.to_csv(index=False).encode("utf-8") else: return df.to_csv(index=False).encode("utf-8") def prepare_filter_selections(streamlit_session: SessionStateProxy) -> str: """ Function to iterate over active Streamlit app variables and construct output for audit :param streamlit_session: Streamlit session containing active filters :return: String representing values from Python dictionary """ selection = {} for m in main_filter_columns: if m in streamlit_session: selection[m] = streamlit_session[m] for k, v in filter_columns_dict.values(): if k in streamlit_session: selection[k] = streamlit_session[k] for k2, v2 in advanced_filter_columns_dict.values(): if k2 in streamlit_session: selection[k2] = streamlit_session[k2] return json.dumps(selection, indent=4) def prepare_export_data( snowpark_session: Session, streamlit_session: SessionStateProxy, selected_report_type: str | None, ) -> Tuple[pd.DataFrame, str, bool]: """ Function to generate and execute SQL query based on user input. :param snowpark_session: Snowpark session used to execute query :param streamlit_session: Streamlit session containing active values/filters :param selected_report_type: Whether it's mailing list or forms based query :return: String representing values from Python dictionary """ query_columns = "" filter_columns = "" custom_query_columns = "" for session_key in streamlit_session: if session_key == "custom_fields_1_20_key" and streamlit_session[session_key]: custom_query_columns = """ , RESP.CUSTOM_FORM_FIELD_1_C, RESP.CUSTOM_FORM_FIELD_2_C, RESP.CUSTOM_FORM_FIELD_3_C, RESP.CUSTOM_FORM_FIELD_4_C, RESP.CUSTOM_FORM_FIELD_5_C, RESP.CUSTOM_FORM_FIELD_6_C, RESP.CUSTOM_FORM_FIELD_7_C, RESP.CUSTOM_FORM_FIELD_8_C, RESP.CUSTOM_FORM_FIELD_9_C, RESP.CUSTOM_FORM_FIELD_10_C, RESP.CUSTOM_FORM_FIELD_11_C, RESP.CUSTOM_FORM_FIELD_12_C, RESP.CUSTOM_FORM_FIELD_13_C, RESP.CUSTOM_FORM_FIELD_14_C, RESP.CUSTOM_FORM_FIELD_15_C, RESP.CUSTOM_FORM_FIELD_16_C, RESP.CUSTOM_FORM_FIELD_17_C, RESP.CUSTOM_FORM_FIELD_18_C, RESP.CUSTOM_FORM_FIELD_19_C, RESP.CUSTOM_FORM_FIELD_20_C """ # generate filter columns if streamlit_session[session_key] and any( session_key in key for key in filter_columns_dict ): if ( session_key == "mailing_list_status_key" and streamlit_session[session_key] == "Both" ): # we ignore this filter if both values are selected pass elif session_key == "form_id_keys": filter_columns += f"AND {filter_columns_dict[session_key]['column']} {filter_columns_dict[session_key]['operator']} '{streamlit_session[session_key][1]}' " else: filter_columns += f"AND {filter_columns_dict[session_key]['column']} {filter_columns_dict[session_key]['operator']} '{streamlit_session[session_key]}' " # generate advanced filter columns if streamlit_session[session_key] and any( session_key in key for key in advanced_filter_columns_dict ): session_value = advanced_filter_columns_dict[session_key]["key_value"] if streamlit_session[session_key] == "Contains": filter_columns += f"AND CONTAINS(lower({advanced_filter_columns_dict[session_key]['column']}),lower('{streamlit_session[session_value]}')) " elif streamlit_session[session_key] == "Starts with": filter_columns += f"AND STARTSWITH(lower({advanced_filter_columns_dict[session_key]['column']}),lower('{streamlit_session[session_value]}')) " else: operator_value = advanced_filter_operators_dict[streamlit_session[session_key]] filter_columns += f"AND lower({advanced_filter_columns_dict[session_key]['column']}){operator_value}lower('{streamlit_session[session_value]}') " # generate select columns if streamlit_session[session_key] and any(session_key in key for key in selection_columns): query_columns += f"{selection_columns[session_key]}, " # add address 2 column also if 1 was selected if ( session_key in ["address_column_key", "ml_address_column_key"] and streamlit_session[session_key] ): if selected_report_type == "Form Response": query_columns += f"{selection_columns['address_column_key2']}, " else: query_columns += f"{selection_columns['ml_address_column_key2']}, " if ( "FAN.EMAIL_C".lower() in query_columns.lower() or "RESP.EMAIL_C".lower() in query_columns.lower() ): drop_cols = ["EXPORT_DATA", "FAN_ID"] exclude_email_col = False else: drop_cols = ["EXPORT_DATA", "FAN_ID", "EMAIL"] if selected_report_type == "Form Response": query_columns += "RESP.EMAIL_C AS EMAIL, " else: query_columns += "FAN.EMAIL_C AS EMAIL, " exclude_email_col = True if selected_report_type == "Form Response": query = f""" SELECT FAN.ID AS FAN_ID, {query_columns.rstrip(", ")} {custom_query_columns} FROM {DATABASE_NAME}.{SCHEMA_NAME}.{FORMS} FORM INNER JOIN {DATABASE_NAME}.{SCHEMA_NAME}.{FORM_RESPONSES} RESP ON FORM.ID=RESP.FORM_ID_C INNER JOIN {DATABASE_NAME}.{SCHEMA_NAME}.{FANS} FAN ON RESP.FAN_ID_C=FAN.ID WHERE FAN.CCPA_DO_NOT_SELL_C = False {filter_columns} """ else: query = f""" SELECT DISTINCT FAN.ID AS FAN_ID, {query_columns.rstrip(", ")} FROM {DATABASE_NAME}.{SCHEMA_NAME}.{MAILING_LISTS} ML INNER JOIN {DATABASE_NAME}.{SCHEMA_NAME}.{SUBSCRIPTIONS} SUB ON ML.ID=SUB.MAILING_LIST_ID_C LEFT JOIN {DATABASE_NAME}.{SCHEMA_NAME}.{PROMOTIONS} PROMO ON SUB.PROMO_ID_C = PROMO.ID INNER JOIN {DATABASE_NAME}.{SCHEMA_NAME}.{FANS} FAN ON SUB.FAN_ID_C=FAN.ID WHERE FAN.CCPA_DO_NOT_SELL_C = False {filter_columns} """ # print(f"Generated query: {query}") df = snowpark_session.sql(query).to_pandas() df["EXPORT_DATA"] = df.apply(lambda row: row.to_dict(), axis=1) st.write( f"Total of {df.shape[0]} records. Following columns will be exported: {df.drop(columns=drop_cols).columns.tolist()}." ) return df, query, exclude_email_col def validate_email(email: str) -> bool: """ Function to validate email address format :param email: email to validate :return: True in case of valid email, otherwise False """ pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$" if not re.match(pattern, email): st.write("Please enter a valid email address!") return False return True