import hashlib
import os
from typing import Dict, List, Tuple

import pandas as pd
import snowflake.snowpark.session as ses
import streamlit as st
from snowflake.snowpark.functions import col, count, call_udf
from snowflake.connector.errors import ProgrammingError, DatabaseError

from common.constants_general import (
    DATABASE_NAME,
    SCHEMA_NAME,
    MAILING_LIST,
    TLAS,
    TLS,
    LABELS,
    TERRITORIES,
    ARTISTS,
    VALIDATION_TABLE,
    ERROR_TABLE,
)

env = os.getenv("target", "prod")
if env == "prod":
    from common.constants_prod import TARGET_TABLE, CRM_FANS_SCHEMA

    iso2_function = "PROD_CRM_FANS.country_to_iso2"
else:
    from common.constants_qa import TARGET_TABLE, CRM_FANS_SCHEMA

    iso2_function = "QA_CRM_FANS.country_to_iso2"


@st.cache_data(show_spinner="preparing application...")
def prepare_fileupload_app(_snowpark_session: ses.Session) -> pd.DataFrame:
    """
    Function to query data used for populating user input fields
    :param _snowpark_session: Active Snowpark session
    :return: Pandas dataframe containing related data
    """
    MAILING_LIST_DF = (
        _snowpark_session.table(f"{DATABASE_NAME}.{SCHEMA_NAME}.{MAILING_LIST}")
        .select(
            col("id").alias("mailing_list_id"),
            col("mailing_list_name_c"),
            col("tla_id_c"),
        )
        .filter(col("active_c") == True)  # noqa
    )
    TLAS_DF = _snowpark_session.table(f"{DATABASE_NAME}.{SCHEMA_NAME}.{TLAS}").select(
        col("id").alias("tla_id"), col("tl_id_c"), col("label_id_c"), col("artist_id_c")
    )
    TLS_DF = _snowpark_session.table(f"{DATABASE_NAME}.{SCHEMA_NAME}.{TLS}")
    LABELS_DF = _snowpark_session.table(
        f"{DATABASE_NAME}.{SCHEMA_NAME}.{LABELS}"
    ).select(col("name").alias("label_name"), col("id").alias("label_id"))
    TERRITORIES_DF = _snowpark_session.table(
        f"{DATABASE_NAME}.{SCHEMA_NAME}.{TERRITORIES}"
    )
    ARTISTS_DF = _snowpark_session.table(
        f"{DATABASE_NAME}.{SCHEMA_NAME}.{ARTISTS}"
    ).select(col("id").alias("artist_id"), col("artist_c"))

    query_result: pd.DataFrame = (
        MAILING_LIST_DF.join(TLAS_DF, MAILING_LIST_DF.tla_id_c == TLAS_DF.tla_id)
        .join(TLS_DF, TLAS_DF.tl_id_c == TLS_DF.id)
        .join(LABELS_DF, TLAS_DF.label_id_c == LABELS_DF.label_id)
        .join(TERRITORIES_DF, TLS_DF.territory_id_c == TERRITORIES_DF.id)
        .join(ARTISTS_DF, TLAS_DF.artist_id_c == ARTISTS_DF.artist_id)
        .select(
            "mailing_list_id",
            "tla_id",
            "territory_c",
            "label_name",
            "artist_c",
            "mailing_list_name_c",
        )
        .to_pandas()
    )

    return query_result


def get_filenames(snowpark_session: ses.Session) -> List[str]:
    """
    Function to query filenames previously uploaded to database
    :param snowpark_session: Active Snowpark session
    :return: List of existing filnames
    """
    try:
        FILE_NAMES_DF = (
            snowpark_session.table(f"{TARGET_TABLE}")
            .select(col("FILE_NAME"))
            .distinct()
            .to_pandas()
        )
        filename_list: List[str] = FILE_NAMES_DF.values.flatten().tolist()
    except (ProgrammingError, DatabaseError, ValueError) as e:
        st.write(f"An error occurred: {e}. Returning empty list of files.")
        filename_list = []
    return filename_list


def get_campaign_ids(snowpark_session: ses.Session) -> List[str]:
    """
    Function to query campaign ids previously uploaded to database
    :param snowpark_session: Active Snowpark session
    :return: List of existing campaign ids
    """
    try:
        CAMPAIGN_ID_DF = (
            snowpark_session.table(f"{TARGET_TABLE}")
            .select(col("CAMPAIGN_ID"))
            .distinct()
            .to_pandas()
        )
        id_list: List[str] = CAMPAIGN_ID_DF.values.flatten().tolist()
    except (ProgrammingError, DatabaseError, ValueError) as e:
        st.write(f"An error occurred: {e}. Returning empty list of ids.")
        id_list = []
    return id_list


def upload_csv_content(
    snowpark_session: ses.Session, dataframe: pd.DataFrame, snowflake_table_name: str
) -> str:
    """
    Function to upload csv file to snowflake database
    :param snowpark_session: Active Snowpark session
    :param dataframe: Pandas datafarme containing data to be uploaded
    :param snowflake_table_name: Name of snowflake table where to load data
    :return: String containing error message or empty string.
    """
    try:
        snowpark_session.write_pandas(
            dataframe,
            snowflake_table_name,
            database="FANSIFTER_APP_REPORTING",
            schema=CRM_FANS_SCHEMA,
            auto_create_table=False,
            overwrite=False,
            use_logical_type=True,
        )
    except Exception as e:
        # print(e)
        # st.write("Upload failed")
        st.write(f"Upload failed with error: {e}")
    return ""


def df_quality_check(
    dataframe: pd.DataFrame,
    upload_focus: str,
    select_acquisition_channels: Dict[str, str],
) -> None:
    """
    Function to check the quality of uploaded CSV file
    :param dataframe: Pandas dataframe containing data from CSV
    :param upload_focus: Is the related acquisition channel email or SMS focused
    :param select_acquisition_channels: Dictionary of acquisition channels to use for validation
    :return: Nothing. Prints out data quality issues if find any.
    """
    missing_count = dataframe[upload_focus].isna().sum()
    duplicate_count = dataframe.duplicated(subset=upload_focus, keep="first").sum()
    total_rows = len(dataframe)

    st.write(f"Total number of rows {total_rows}.")
    st.write("Preview of first 3 rows.")
    if duplicate_count > 0:
        st.write(
            f":orange[Warning! Number of duplicate {upload_focus} values: {duplicate_count}.]"
        )
    if missing_count > 0:
        st.write(
            f":orange[Warning! Column {upload_focus} is missing for {missing_count} rows out of {total_rows} rows.]"
        )
        if missing_count == total_rows:
            st.write(
                f":red[ERROR! Acquisition channel {select_acquisition_channels} requires at least one row filled for column {upload_focus}.]"
            )
    st.write(dataframe.head(10))


def df_check_columns(
    dataframe: pd.DataFrame, columns_to_use: List[str]
) -> Tuple[pd.DataFrame, bool]:
    """
    Function that standardizes column names and compares if all required columns are present and nothing else
    :param dataframe: Pandas dataframe that contains column names
    :param columns_to_use: List of required columns
    :return: Pandas dataframe with correct column names and result if all required columns are present
    """
    dataframe.columns = dataframe.columns.str.strip()
    dataframe.columns = (
        dataframe.columns.str.replace("(", "_")
        .str.replace(")", "")
        .str.replace("-", "")
        .str.replace("/", "_")
        .str.replace(" ", "_")
        .str.replace("__", "_")
    )
    dataframe.columns = dataframe.columns.str.upper()

    matching_columns = all(e in columns_to_use for e in dataframe.columns) & all(
        e in dataframe.columns for e in columns_to_use
    )
    return dataframe, matching_columns


def create_id(df: pd.DataFrame) -> pd.DataFrame:
    """
    Function to create hash out of columns
    :param df: Pandas dataframe where to add hash based id column
    :return: Pandas dataframe with hash based id column included
    """
    df["CAMPAIGN_ID"] = (
        df["TERRITORY"] + df["LABEL"] + df["ARTIST"] + df["FILE_SOURCE_DESCRIPTION"]
    )
    df["CAMPAIGN_ID"] = df["CAMPAIGN_ID"].apply(
        lambda x: hashlib.sha256(x.encode()).hexdigest()
    )
    return df


def handle_file_upload(
    dataframe: pd.DataFrame, session: ses.Session, snowflake_table: str
) -> None:
    """
    Function to send CSV file for upload and then clean session keys/ input fields in order to start new file upload process
    :param dataframe: Pandas dataframe containing data from CSV file
    :param session: Active Snowflake session
    :param snowflake_table: Name of the Snowflake table where data will be stored
    :return: Nnne
    """
    snowflake_table_name = snowflake_table
    upload_message = upload_csv_content(session, dataframe, snowflake_table_name)
    if upload_message == "":
        st.session_state.upload_name = None
        st.session_state.territory = None
        st.session_state.overwrite_data = None
        st.session_state.upload_name = None
        st.session_state.file_source_description = None
        st.session_state.dsp = None
        st.session_state.crm_generated = None
        st.session_state.acquisition_channels = None
        st.session_state.valid_filename = True
        st.session_state.file_upload_completed = True
        st.session_state.error_message = None
        st.session_state.tla_id_value = None
    else:
        st.session_state.error_message = upload_message


def check_phone(phone: str) -> str:
    """
    Function to make sure every phone number contains + sign
    :param phone: number to check
    :return: number with + added
    """
    if not phone.startswith("+"):
        return "+" + phone
    return phone


def country_to_iso2(df: pd.DataFrame, session: ses.Session) -> pd.DataFrame:
    """
    Function to convert long country name to iso2
    :param df: Pandas dataframe that contains uncleaned country values
    :param session: Active Snowflake session
    :return: Pandas dataframe that contains iso2 country values
    """
    snowpark_df = session.create_dataframe(df)
    pandas_df: pd.DataFrame = snowpark_df.withColumn(
        "COUNTRY_REGION", call_udf(iso2_function, col("COUNTRY_REGION"))
    ).to_pandas()

    return pandas_df


def call_deletion_procedure(
    file_name: str, reason: str, username: str, session: ses.Session
) -> str:
    """
    Function to call Snowflake procedure to delete records from file upload
    :param file_name: name of original csv file that containes records for deletion
    :param reason: why data needs to be deleted
    :param username: username who initated deletion
    :param session: Active Snowflake session
    :return: deletion response text
    """
    response: str = session.call(
        f"FANSIFTER_APP_REPORTING.{CRM_FANS_SCHEMA}.DELETE_FILE_UPLOAD",
        file_name,
        reason,
        username,
    )
    return response


def get_upload_status(name: str, session: ses.Session) -> Tuple[int, pd.DataFrame]:
    """
    Function to retreive number of succesfull rows and failures if exists
    :param name: fileupload name
    :param session: Active Snowflake session
    :return: number of succesful rows and Pandas dataframe containg errors
    """
    try:
        row_count: int = (
            session.table(
                f"FANSIFTER_APP_REPORTING.{CRM_FANS_SCHEMA}.{VALIDATION_TABLE}"
            )
            .filter(col("file_source_description") == name)
            .select(col("EVENT_ROW_ID"))
            .distinct()
            .count()
        )
        failed_rows: pd.DataFrame = (
            session.table(f"FANSIFTER_APP_REPORTING.{CRM_FANS_SCHEMA}.{ERROR_TABLE}")
            .filter(col("RECORD_CONTENT")["file_source_description"] == name)
            .groupBy("NOT_VALID_REASON")
            .agg(count(col("EVENT_ROW_ID")).as_("FANS"))
            .to_pandas()
        )

    except (ProgrammingError, DatabaseError, ValueError) as e:
        st.write(f"An error occurred: {e}. Returning empty list of ids.")
        row_count = 0
        failed_rows = pd.DataFrame()

    return row_count, failed_rows