from datetime import datetime
from pathlib import Path

import pandas as pd
import streamlit as st
from sklearn.compose import ColumnTransformer
from sklearn.metrics import accuracy_score, f1_score
from sklearn.model_selection import train_test_split
from sklearn.naive_bayes import GaussianNB
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder
from sklearn.tree import DecisionTreeClassifier


APP_TITLE = "Klasifikasi Kelayakan Keringanan UKT"
DATA_PATH = Path(__file__).with_name("klasifikasi_mhs.csv")
HISTORY_PATH = Path(__file__).with_name("riwayat_prediksi_ukt.csv")

TARGET = "Kelayakan Keringanan UKT"
FEATURES = [
    "Tempat Tinggal",
    "Pekerjaan Orang Tua",
    "Penghasilan Orang Tua",
    "Jumlah Tanggungan Orang Tua",
    "Kendaraan",
]

LABEL_TEXT = {
    0: "Layak menerima keringanan UKT",
    1: "Tidak layak menerima keringanan UKT",
}

TEMPAT_TINGGAL_OPTIONS = {
    "Rumah sendiri": 0,
    "Bukan rumah sendiri": 1,
}

MODEL_OPTIONS = {
    "Decision Tree": "decision_tree",
    "Naive Bayes": "naive_bayes",
}


st.set_page_config(
    page_title=APP_TITLE,
    layout="wide",
)


def normalize_target(series: pd.Series) -> pd.Series:
    return (
        series.astype(str)
        .str.strip()
        .replace({"1 1": "1"})
        .astype(int)
    )


def load_data() -> pd.DataFrame:
    if not DATA_PATH.exists():
        st.error(f"File data tidak ditemukan: {DATA_PATH}")
        st.stop()

    data = pd.read_csv(DATA_PATH)
    data[TARGET] = normalize_target(data[TARGET])
    return data


def save_training_row(row: dict) -> None:
    data = load_data()
    updated = pd.concat([data, pd.DataFrame([row])], ignore_index=True)
    updated.to_csv(DATA_PATH, index=False)


def save_prediction_history(row: dict) -> None:
    row["Waktu Prediksi"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    history = pd.DataFrame([row])
    if HISTORY_PATH.exists():
        old_history = pd.read_csv(HISTORY_PATH)
        history = pd.concat([old_history, history], ignore_index=True)
    history.to_csv(HISTORY_PATH, index=False)


def make_one_hot_encoder() -> OneHotEncoder:
    try:
        return OneHotEncoder(handle_unknown="ignore", sparse_output=False)
    except TypeError:
        return OneHotEncoder(handle_unknown="ignore", sparse=False)


def make_preprocessor() -> ColumnTransformer:
    return ColumnTransformer(
        transformers=[
            ("pekerjaan", make_one_hot_encoder(), ["Pekerjaan Orang Tua"]),
            (
                "angka",
                "passthrough",
                [
                    "Tempat Tinggal",
                    "Penghasilan Orang Tua",
                    "Jumlah Tanggungan Orang Tua",
                    "Kendaraan",
                ],
            ),
        ],
        remainder="drop",
    )


def build_models() -> dict:
    return {
        "decision_tree": Pipeline(
            steps=[
                ("preprocess", make_preprocessor()),
                (
                    "model",
                    DecisionTreeClassifier(
                        max_depth=4,
                        random_state=42,
                        class_weight="balanced",
                    ),
                ),
            ]
        ),
        "naive_bayes": Pipeline(
            steps=[
                ("preprocess", make_preprocessor()),
                ("model", GaussianNB()),
            ]
        ),
    }


def train_models(data: pd.DataFrame) -> tuple[dict, pd.DataFrame]:
    x = data[FEATURES]
    y = data[TARGET]

    models = build_models()
    metrics = []

    can_split = y.nunique() == 2 and y.value_counts().min() >= 2
    if can_split:
        x_train, x_test, y_train, y_test = train_test_split(
            x,
            y,
            test_size=0.2,
            random_state=42,
            stratify=y,
        )
    else:
        x_train, x_test, y_train, y_test = x, x, y, y

    for display_name, model_key in MODEL_OPTIONS.items():
        model = models[model_key]
        model.fit(x_train, y_train)
        prediction = model.predict(x_test)
        metrics.append(
            {
                "Model": display_name,
                "Accuracy": accuracy_score(y_test, prediction),
                "F1-Score": f1_score(y_test, prediction, average="weighted", zero_division=0),
            }
        )

    return models, pd.DataFrame(metrics).round(4)


def make_input_dataframe(
    tempat_tinggal: int,
    pekerjaan: str,
    penghasilan: int,
    tanggungan: int,
    kendaraan: int,
) -> pd.DataFrame:
    return pd.DataFrame(
        [
            {
                "Tempat Tinggal": tempat_tinggal,
                "Pekerjaan Orang Tua": pekerjaan,
                "Penghasilan Orang Tua": penghasilan,
                "Jumlah Tanggungan Orang Tua": tanggungan,
                "Kendaraan": kendaraan,
            }
        ]
    )


def predict_with_models(models: dict, input_df: pd.DataFrame) -> pd.DataFrame:
    results = []
    for display_name, model_key in MODEL_OPTIONS.items():
        model = models[model_key]
        prediction = int(model.predict(input_df)[0])

        probability_text = "-"
        if hasattr(model, "predict_proba"):
            probabilities = model.predict_proba(input_df)[0]
            classes = list(model.classes_)
            probability = probabilities[classes.index(prediction)]
            probability_text = f"{probability:.2%}"

        results.append(
            {
                "Model": display_name,
                "Prediksi": LABEL_TEXT[prediction],
                "Kode Prediksi": prediction,
                "Keyakinan Model": probability_text,
            }
        )
    return pd.DataFrame(results)


def render_prediction_result(results: pd.DataFrame, selected_model_name: str) -> int:
    selected = results[results["Model"] == selected_model_name].iloc[0]
    prediction_code = int(selected["Kode Prediksi"])
    prediction_text = selected["Prediksi"]

    if prediction_code == 0:
        st.success(f"Hasil prediksi {selected_model_name}: {prediction_text}")
    else:
        st.warning(f"Hasil prediksi {selected_model_name}: {prediction_text}")

    st.dataframe(
        results[["Model", "Prediksi", "Keyakinan Model"]],
        use_container_width=True,
        hide_index=True,
    )
    return prediction_code


def render_form(prefix: str, data: pd.DataFrame) -> pd.DataFrame:
    pekerjaan_list = sorted(data["Pekerjaan Orang Tua"].dropna().astype(str).unique())
    pekerjaan_options = pekerjaan_list + ["Lainnya"]

    col_left, col_right = st.columns(2)
    with col_left:
        tempat_label = st.selectbox(
            "Tempat tinggal",
            options=list(TEMPAT_TINGGAL_OPTIONS.keys()),
            key=f"{prefix}_tempat",
        )
        pekerjaan_choice = st.selectbox(
            "Pekerjaan orang tua",
            options=pekerjaan_options,
            key=f"{prefix}_pekerjaan_choice",
        )
        if pekerjaan_choice == "Lainnya":
            pekerjaan = st.text_input(
                "Tulis pekerjaan orang tua",
                value="",
                key=f"{prefix}_pekerjaan_baru",
            ).strip()
        else:
            pekerjaan = pekerjaan_choice

    with col_right:
        penghasilan = st.number_input(
            "Penghasilan orang tua per bulan",
            min_value=0,
            value=3_000_000,
            step=100_000,
            key=f"{prefix}_penghasilan",
        )
        tanggungan = st.number_input(
            "Jumlah tanggungan orang tua",
            min_value=0,
            value=3,
            step=1,
            key=f"{prefix}_tanggungan",
        )
        kendaraan = st.number_input(
            "Jumlah kendaraan",
            min_value=0,
            value=1,
            step=1,
            key=f"{prefix}_kendaraan",
        )

    if not pekerjaan:
        st.warning("Isi pekerjaan orang tua terlebih dahulu.")
        return pd.DataFrame()

    return make_input_dataframe(
        tempat_tinggal=TEMPAT_TINGGAL_OPTIONS[tempat_label],
        pekerjaan=pekerjaan,
        penghasilan=int(penghasilan),
        tanggungan=int(tanggungan),
        kendaraan=int(kendaraan),
    )


data = load_data()
models, metrics_df = train_models(data)

st.title(APP_TITLE)
st.caption("Deployment interaktif menggunakan Decision Tree dan Naive Bayes")

model_name = st.sidebar.radio("Model utama", list(MODEL_OPTIONS.keys()))
st.sidebar.divider()
st.sidebar.write("Arti label")
st.sidebar.write("0 = Layak")
st.sidebar.write("1 = Tidak Layak")

tab_ringkasan, tab_pilih, tab_input, tab_data = st.tabs(
    [
        "Ringkasan",
        "Pilih Data Lama",
        "Input Data Baru",
        "Data & Riwayat",
    ]
)

with tab_ringkasan:
    total_data = len(data)
    layak_count = int((data[TARGET] == 0).sum())
    tidak_layak_count = int((data[TARGET] == 1).sum())

    col_a, col_b, col_c = st.columns(3)
    col_a.metric("Total data latih", total_data)
    col_b.metric("Layak", layak_count)
    col_c.metric("Tidak layak", tidak_layak_count)

    st.subheader("Performa model")
    st.dataframe(metrics_df, use_container_width=True, hide_index=True)

    st.subheader("Distribusi label")
    label_counts = (
        data[TARGET]
        .map(LABEL_TEXT)
        .value_counts()
        .rename_axis("Label")
        .reset_index(name="Jumlah")
    )
    st.bar_chart(label_counts, x="Label", y="Jumlah")

with tab_pilih:
    st.subheader("Cek prediksi dari data yang sudah ada")

    pilihan = [
        (
            idx,
            f"Data {idx + 1} | {row['Pekerjaan Orang Tua']} | "
            f"Rp{int(row['Penghasilan Orang Tua']):,} | "
            f"Tanggungan {int(row['Jumlah Tanggungan Orang Tua'])}",
        )
        for idx, row in data.iterrows()
    ]
    selected_index = st.selectbox(
        "Pilih data mahasiswa",
        options=[idx for idx, _ in pilihan],
        format_func=lambda value: dict(pilihan)[value],
    )

    selected_row = data.loc[[selected_index], FEATURES]
    st.dataframe(selected_row, use_container_width=True, hide_index=True)

    actual_label = int(data.loc[selected_index, TARGET])
    st.info(f"Label asli pada dataset: {LABEL_TEXT[actual_label]}")

    selected_results = predict_with_models(models, selected_row)
    render_prediction_result(selected_results, model_name)

with tab_input:
    st.subheader("Prediksi data mahasiswa baru")
    input_df = render_form("predict", data)

    if not input_df.empty:
        st.dataframe(input_df, use_container_width=True, hide_index=True)
        input_results = predict_with_models(models, input_df)
        predicted_label = render_prediction_result(input_results, model_name)

        with st.form("save_prediction_form"):
            st.write("Simpan data ini")
            save_mode = st.radio(
                "Jenis penyimpanan",
                [
                    "Simpan sebagai riwayat prediksi saja",
                    "Tambahkan ke dataset latih dengan label aktual",
                ],
            )

            actual_label_text = st.selectbox(
                "Label aktual",
                ["Layak menerima keringanan UKT", "Tidak layak menerima keringanan UKT"],
                disabled=save_mode == "Simpan sebagai riwayat prediksi saja",
            )
            submitted = st.form_submit_button("Simpan")

        if submitted:
            row = input_df.iloc[0].to_dict()
            row["Prediksi Model Utama"] = predicted_label
            row["Prediksi Teks"] = LABEL_TEXT[predicted_label]

            if save_mode == "Tambahkan ke dataset latih dengan label aktual":
                actual_label = 0 if actual_label_text.startswith("Layak") else 1
                training_row = input_df.iloc[0].to_dict()
                training_row[TARGET] = actual_label
                save_training_row(training_row)
                st.success("Data baru sudah ditambahkan ke dataset latih. Aplikasi akan memuat ulang model.")
                st.rerun()
            else:
                save_prediction_history(row)
                st.success("Riwayat prediksi sudah disimpan.")

with tab_data:
    st.subheader("Dataset saat ini")
    st.dataframe(data, use_container_width=True, hide_index=True)

    st.download_button(
        "Download dataset CSV",
        data=data.to_csv(index=False).encode("utf-8"),
        file_name="klasifikasi_mhs_terbaru.csv",
        mime="text/csv",
    )

    st.subheader("Riwayat prediksi")
    if HISTORY_PATH.exists():
        history_df = pd.read_csv(HISTORY_PATH)
        st.dataframe(history_df, use_container_width=True, hide_index=True)
        st.download_button(
            "Download riwayat prediksi",
            data=history_df.to_csv(index=False).encode("utf-8"),
            file_name="riwayat_prediksi_ukt.csv",
            mime="text/csv",
        )
    else:
        st.info("Belum ada riwayat prediksi yang disimpan.")