import pandas as pd
import streamlit as st
from snowflake.snowpark import Session

from backend.actions.file_upload import (
    ACQUISITION_CHANNELS,
    DSPS,
    IS_CRM_GENERATED,
    check_data_quality,
    check_upload_status,
    delete_file_upload,
    filter_options_df,
    list_campaign_artists,
    list_campaign_stats,
    prepare_upload_dataframe,
    submit_upload,
    validate_fan_data,
)
from backend.adapters.snowflake import get_session, resolve_env
from backend.dtos import CampaignStats
from backend.permissions import require_access
from backend.queries import convert_country_to_iso2, query_file_upload_filter_options

session = get_session()
require_access(session, "FILE_UPLOAD")
env = resolve_env(session)

if env == "PROD":
    st.error("The File Upload page is disabled in PROD.")
    st.stop()

st.title("File Upload")

tab_upload, tab_delete, tab_status = st.tabs(["Upload", "Delete", "Status"])


@st.cache_data(show_spinner="Loading filter options…")
def _load_filter_options(_session: Session, _env: str) -> pd.DataFrame:
    return filter_options_df(query_file_upload_filter_options(_session, _env))


@st.cache_data(show_spinner="Loading artists…")
def _load_campaign_artists(_session: Session, _env: str) -> list[str]:
    return list_campaign_artists(_session, _env)


@st.cache_data(show_spinner="Loading campaign stats…")
def _load_campaign_stats(
    _session: Session, _env: str, artist_name: str
) -> list[CampaignStats]:
    return list_campaign_stats(_session, _env, artist_name)


with tab_upload:
    st.write(
        "Contact crm.support@sonymusic.com if the mailing list you are looking for "
        "doesn't appear."
    )

    filter_df = _load_filter_options(session, env)

    file_source_description = st.text_input(
        label="File Source Description", key="fu_file_source_description"
    )

    label_options = (
        filter_df[["label_name", "vendor_id"]]
        .drop_duplicates()
        .sort_values("label_name")
    )
    label_choices: dict[str, tuple[str, str]] = {
        f"{row.label_name} ({row.vendor_id})": (str(row.label_name), str(row.vendor_id))
        for row in label_options.itertuples()
    }
    select_label_display = st.selectbox(
        label="Label (name or vendor ID)",
        options=sorted(label_choices.keys()),
        index=None,
        key="fu_label",
    )
    select_label: str | None
    select_vendor_id: str | None
    if select_label_display is None:
        select_label, select_vendor_id = None, None
    else:
        select_label, select_vendor_id = label_choices[select_label_display]

    artist_options = filter_df[filter_df["vendor_id"] == select_vendor_id][
        ["artist_name", "type", "global_participant_id", "virtual_participant_id"]
    ].drop_duplicates()
    artist_choices: dict[str, tuple[str, str, str | None, str | None]] = {
        (
            f"Custom List: {row.artist_name}"
            if row.type == "custom list"
            else f"Artist: {row.artist_name}"
        ): (
            str(row.artist_name),
            str(row.type),
            str(row.global_participant_id)
            if pd.notna(row.global_participant_id)
            else None,
            str(row.virtual_participant_id)
            if pd.notna(row.virtual_participant_id)
            else None,
        )
        for row in artist_options.itertuples()
    }
    select_artist_display = st.selectbox(
        label="Artist / Custom List",
        options=sorted(artist_choices.keys()),
        index=None,
        key="fu_artist",
    )
    select_artist: str | None
    select_artist_type: str | None
    select_global_participant_id: str | None
    select_virtual_participant_id: str | None
    if select_artist_display is None:
        select_artist, select_artist_type = None, None
        select_global_participant_id, select_virtual_participant_id = None, None
    else:
        (
            select_artist,
            select_artist_type,
            select_global_participant_id,
            select_virtual_participant_id,
        ) = artist_choices[select_artist_display]

    mailing_list_values = sorted(
        filter_df[
            (filter_df["vendor_id"] == select_vendor_id)
            & (filter_df["artist_name"] == select_artist)
            & (filter_df["type"] == select_artist_type)
        ]["mailing_list_name"].unique()
    )
    select_mailing_list = None
    if select_artist:
        if len(mailing_list_values) == 1:
            select_mailing_list = mailing_list_values[0]
            st.write(f"Mailing list: **{select_mailing_list}**")
        elif len(mailing_list_values) == 0:
            st.warning("No mailing list found for this artist.")
        else:
            st.warning(
                "Multiple mailing lists found for this artist; contact "
                "crm.support@sonymusic.com to resolve."
            )

    select_dsp = st.selectbox(label="DSP", options=DSPS, index=None, key="fu_dsp")

    select_is_crm_generated = st.selectbox(
        label="Is CRM generated",
        options=IS_CRM_GENERATED,
        index=None,
        key="fu_crm_generated",
    )

    select_acquisition_channel = st.selectbox(
        label="Acquisition Channel",
        options=list(ACQUISITION_CHANNELS.keys()),
        index=None,
        key="fu_acquisition_channel",
    )

    overwrite_fan_data = st.checkbox("Overwrite Fan Data", key="fu_overwrite")

    st.divider()

    required_fields = [
        file_source_description,
        select_label,
        select_artist,
        select_mailing_list,
        select_dsp,
        select_is_crm_generated,
        select_acquisition_channel,
    ]

    if all(required_fields):
        assert select_label is not None
        assert select_artist is not None
        assert select_mailing_list is not None
        assert select_dsp is not None
        assert select_is_crm_generated is not None
        assert select_acquisition_channel is not None

        uploaded_file = st.file_uploader("Choose a file", type="csv", key="fu_uploader")

        if uploaded_file is not None:
            session = get_session()
            user = (session.get_current_user() or "UNKNOWN").strip('"')

            dataframe, errors = prepare_upload_dataframe(
                uploaded_file,
                uploaded_file.name,
                select_label,
                select_artist,
                select_mailing_list,
                file_source_description,
                select_dsp,
                select_is_crm_generated,
                select_acquisition_channel,
                overwrite_fan_data,
                user,
                select_vendor_id,
                select_global_participant_id,
                select_virtual_participant_id,
            )
            uploaded_file.seek(0)

            if errors:
                for error in errors:
                    st.error(error)
            else:
                dataframe = convert_country_to_iso2(session, dataframe, env)
                dataframe["TERRITORY"] = dataframe["COUNTRY_REGION"]

                upload_focus = ACQUISITION_CHANNELS[select_acquisition_channel]
                for warning in check_data_quality(dataframe, upload_focus):
                    if warning.startswith("Error!"):
                        st.error(warning)
                    else:
                        st.warning(warning)

                st.write(f"Total number of rows: {len(dataframe)}.")

                valid_dataframe, validation_failures = validate_fan_data(dataframe)
                invalid_row_count = len(dataframe) - len(valid_dataframe)

                if validation_failures:
                    st.write(
                        f"{invalid_row_count} row(s) excluded because they failed "
                        "validation:"
                    )
                    st.dataframe(
                        pd.DataFrame([f.model_dump() for f in validation_failures]),
                        width="content",
                    )

                st.write(f"Number of rows passing validation: {len(valid_dataframe)}.")
                st.write("Preview of first 10 valid rows.")
                st.dataframe(valid_dataframe.head(10), width="content")

                st.divider()

                if valid_dataframe.empty:
                    st.warning("No rows passed validation; nothing to upload.")
                elif st.button("Upload data", key="fu_upload_button"):
                    session = get_session()
                    result = submit_upload(
                        session, env, valid_dataframe, uploaded_file.name
                    )
                    if result.errors:
                        st.error(f"Upload failed: {'; '.join(result.errors)}")
                    else:
                        st.success(
                            f"File uploaded successfully — {result.rows_uploaded} "
                            f"row(s), campaign ID {result.campaign_id}."
                        )
    else:
        st.info("Make sure all fields above have been filled to enable file upload.")

with tab_delete:
    st.subheader(
        "Here you can delete a file upload if there is a need to fix source data "
        "and try again."
    )
    st.write("Please note that every deletion will be recorded in an audit table.")

    delete_file_name = st.text_input(
        label="Name of the csv file that was used for upload", key="fu_delete_file_name"
    )
    delete_reason = st.text_input(
        label="Brief reason why you need to delete data", key="fu_delete_reason"
    )
    confirm_deletion = st.checkbox(
        "Confirm you want to delete related records.", key="fu_delete_confirm"
    )

    delete_disabled = not (delete_file_name and delete_reason and confirm_deletion)

    if st.button(
        "Delete related rows", disabled=delete_disabled, key="fu_delete_button"
    ):
        session = get_session()
        user = (session.get_current_user() or "UNKNOWN").strip('"')
        response = delete_file_upload(
            session, env, delete_file_name, delete_reason, user
        )
        st.write(response)

with tab_status:
    st.subheader("Here you can check the status of an uploaded file.")
    st.write(
        "Background processes usually take 2-3 minutes. Select an artist below to "
        "see their related campaigns."
    )

    session = get_session()
    status_artist = st.selectbox(
        label="Artist",
        options=_load_campaign_artists(session, env),
        index=None,
        key="fu_status_artist",
    )

    status_upload_name = None

    if status_artist:
        campaign_stats = _load_campaign_stats(session, env, status_artist)

        if campaign_stats:
            st.dataframe(
                pd.DataFrame([c.model_dump() for c in campaign_stats]),
                width="content",
            )

            file_source_options = sorted(
                {c.file_source_description for c in campaign_stats}
            )
            status_upload_name = st.selectbox(
                label="File source description",
                options=file_source_options,
                index=None,
                key="fu_status_upload_name",
            )
        else:
            st.info("No campaigns found for this artist.")

    if st.button(
        "Check status", disabled=not status_upload_name, key="fu_status_button"
    ):
        assert status_upload_name is not None
        session = get_session()
        status_result = check_upload_status(session, env, status_upload_name)
        st.write(
            f"Number of rows successfully processed: {status_result.processed_count}"
        )
        if status_result.failures:
            st.write("Failed rows by reason:")
            st.dataframe(
                pd.DataFrame([f.model_dump() for f in status_result.failures]),
                width="content",
            )