app.py

from __future__ import annotations

import json
import os
import subprocess
from pathlib import Path
from typing import Generator

import gradio as gr
import joblib
import matplotlib.pyplot as plt
import pandas as pd
import shap

APP_DIR = Path(__file__).parent.resolve()
STYLE_FILE = APP_DIR / "style.css"
ASSETS_DIR = APP_DIR / "assets"
DATA_DIR = APP_DIR / "data"
MODELS_DIR = APP_DIR / "models"
OUT_DIR = APP_DIR / "outputs"
FIG_DIR = OUT_DIR / "figures"
TAB_DIR = OUT_DIR / "tables"

MODEL_FILE = MODELS_DIR / "pipeline.joblib"
META_FILE = MODELS_DIR / "model_meta.json"
BG_FILE = MODELS_DIR / "background_sample.csv"
TEMPLATE_CSV = DATA_DIR / "batch_template.csv"

DEFAULTS = {
    "AGE": 42,
    "OPEN_ACC_DUR": 120,
    "GENDER_CD": "1",
    "HASNT_HOME_ADDRESS_INF": "N",
    "HASNT_MOBILE_TEL_NUM_INF": "N",
    "LOCAL_CUR_MON_AVG_BAL": 25000.0,
    "LOCAL_FIX_MON_AVG_BAL": 18000.0,
    "LOCAL_SAV_CUR_ALL_BAL": 28000.0,
    "POS_CONSUME_TX_AMT": 5000.0,
    "ATM_ALL_TX_NUM": 6,
    "COUNTER_ALL_TX_NUM": 2,
}
FEATURES = list(DEFAULTS.keys())

PIPE = None
META = None


def ensure_template_csv() -> None:
    if not TEMPLATE_CSV.exists():
        pd.DataFrame([DEFAULTS]).to_csv(TEMPLATE_CSV, index=False)


def load_assets() -> tuple[object | None, dict | None]:
    pipe = joblib.load(MODEL_FILE) if MODEL_FILE.exists() else None
    meta = json.loads(META_FILE.read_text(encoding="utf-8")) if META_FILE.exists() else None
    return pipe, meta


def refresh_model_state() -> str:
    global PIPE, META
    PIPE, META = load_assets()
    if PIPE is None:
        return "⚠️ 当前为演示状态：请先在 Pipeline 标签页点击 **Run Pipeline** 生成模型。"
    return "✅ 模型已加载，可以进行单条预测、批量预测和 SHAP 解释。"


def gauge_html(prob: float) -> str:
    pct = max(0.0, min(100.0, prob * 100.0))
    color = "#16a34a" if prob < 0.35 else ("#f59e0b" if prob < 0.65 else "#dc2626")
    return f"""
    <div style='background:rgba(255,255,255,0.88);padding:16px;border-radius:18px'>
      <div style='font-size:18px;font-weight:700;margin-bottom:8px'>Churn Probability Gauge</div>
      <div style='width:100%;height:20px;background:#e5e7eb;border-radius:999px;overflow:hidden'>
        <div style='width:{pct:.1f}%;height:20px;background:{color};border-radius:999px'></div>
      </div>
      <div style='margin-top:10px;font-size:28px;font-weight:800;color:{color}'>{pct:.1f}%</div>
    </div>
    """


def input_df(age, open_acc_dur, gender_cd, hasnt_home_address_inf, hasnt_mobile_tel_num_inf,
             local_cur_mon_avg_bal, local_fix_mon_avg_bal, local_sav_cur_all_bal,
             pos_consume_tx_amt, atm_all_tx_num, counter_all_tx_num) -> pd.DataFrame:
    return pd.DataFrame([{
        "AGE": int(age),
        "OPEN_ACC_DUR": int(open_acc_dur),
        "GENDER_CD": str(gender_cd),
        "HASNT_HOME_ADDRESS_INF": str(hasnt_home_address_inf),
        "HASNT_MOBILE_TEL_NUM_INF": str(hasnt_mobile_tel_num_inf),
        "LOCAL_CUR_MON_AVG_BAL": float(local_cur_mon_avg_bal),
        "LOCAL_FIX_MON_AVG_BAL": float(local_fix_mon_avg_bal),
        "LOCAL_SAV_CUR_ALL_BAL": float(local_sav_cur_all_bal),
        "POS_CONSUME_TX_AMT": float(pos_consume_tx_amt),
        "ATM_ALL_TX_NUM": int(atm_all_tx_num),
        "COUNTER_ALL_TX_NUM": int(counter_all_tx_num),
    }])


def predict_single(age, open_acc_dur, gender_cd, hasnt_home_address_inf, hasnt_mobile_tel_num_inf,
                   local_cur_mon_avg_bal, local_fix_mon_avg_bal, local_sav_cur_all_bal,
                   pos_consume_tx_amt, atm_all_tx_num, counter_all_tx_num):
    if PIPE is None:
        return {"error": "Run Pipeline first."}, "请先运行 Pipeline。", gauge_html(0.0), None
    df = input_df(age, open_acc_dur, gender_cd, hasnt_home_address_inf, hasnt_mobile_tel_num_inf,
                  local_cur_mon_avg_bal, local_fix_mon_avg_bal, local_sav_cur_all_bal,
                  pos_consume_tx_amt, atm_all_tx_num, counter_all_tx_num)
    prob = float(PIPE.predict_proba(df)[0, 1])
    pred = int(prob >= 0.5)
    risk = "低风险" if prob < 0.35 else ("中风险" if prob < 0.65 else "高风险")
    payload = {
        "churn_probability": round(prob, 6),
        "predicted_label": pred,
        "risk_level": risk,
    }
    summary = f"**预测结果**：{'流失' if pred == 1 else '留存'}  \n\n**概率**：{prob:.2%}  \n**风险等级**：{risk}"
    return payload, summary, gauge_html(prob), None


def predict_batch(file_obj):
    if PIPE is None:
        return None, None, "请先运行 Pipeline。"
    if file_obj is None:
        return None, None, "请先上传 CSV。"
    df = pd.read_csv(file_obj.name)
    missing = [c for c in FEATURES if c not in df.columns]
    if missing:
        return None, None, f"CSV 缺少列：{missing}"
    x = df[FEATURES].copy()
    proba = PIPE.predict_proba(x)[:, 1]
    pred = (proba >= 0.5).astype(int)
    out = df.copy()
    out["churn_proba"] = proba
    out["churn_pred"] = pred
    out_path = OUT_DIR / "batch_predictions.csv"
    out.to_csv(out_path, index=False)
    return out.head(50), str(out_path), "批量预测完成。"


def make_feature_importance_plot():
    fp = TAB_DIR / "feature_importance.csv"
    if not fp.exists():
        return None
    fi = pd.read_csv(fp)
    plt.figure(figsize=(8, 4.5))
    plt.barh(fi["feature"][::-1], fi["importance"][::-1])
    plt.title("Feature Importance")
    plt.xlabel("Importance")
    plt.tight_layout()
    fig_path = FIG_DIR / "feature_importance_runtime.png"
    plt.savefig(fig_path, dpi=160)
    plt.close()
    return str(fig_path)


def explain_single(age, open_acc_dur, gender_cd, hasnt_home_address_inf, hasnt_mobile_tel_num_inf,
                   local_cur_mon_avg_bal, local_fix_mon_avg_bal, local_sav_cur_all_bal,
                   pos_consume_tx_amt, atm_all_tx_num, counter_all_tx_num):
    if PIPE is None or not BG_FILE.exists():
        return None, "请先运行 Pipeline。"

    row = input_df(age, open_acc_dur, gender_cd, hasnt_home_address_inf, hasnt_mobile_tel_num_inf,
                   local_cur_mon_avg_bal, local_fix_mon_avg_bal, local_sav_cur_all_bal,
                   pos_consume_tx_amt, atm_all_tx_num, counter_all_tx_num)
    background = pd.read_csv(BG_FILE)
    background = background[FEATURES].head(40)

    def f(x):
        x_df = pd.DataFrame(x, columns=FEATURES)
        for c in ["GENDER_CD", "HASNT_HOME_ADDRESS_INF", "HASNT_MOBILE_TEL_NUM_INF"]:
            x_df[c] = x_df[c].astype(str)
        for c in [col for col in FEATURES if col not in ["GENDER_CD", "HASNT_HOME_ADDRESS_INF", "HASNT_MOBILE_TEL_NUM_INF"]]:
            x_df[c] = pd.to_numeric(x_df[c], errors="coerce")
        return PIPE.predict_proba(x_df)[:, 1]

    explainer = shap.Explainer(f, background, feature_names=FEATURES)
    sv = explainer(row)

    plt.figure(figsize=(9, 4.8))
    shap.plots.waterfall(sv[0], max_display=10, show=False)
    plt.tight_layout()
    out_path = FIG_DIR / "shap_waterfall.png"
    plt.savefig(out_path, dpi=160, bbox_inches="tight")
    plt.close()
    prob = float(PIPE.predict_proba(row)[0, 1])
    txt = f"SHAP 解释已生成。该客户流失概率约为 **{prob:.2%}**。"
    return str(out_path), txt


def run_pipeline_stream() -> Generator[tuple[str, str, str], None, None]:
    log_lines = []
    cmd = ["python", "-u", str(APP_DIR / "scripts" / "pipeline.py")]
    proc = subprocess.Popen(cmd, cwd=str(APP_DIR), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1)
    assert proc.stdout is not None
    yield "", "⏳ Pipeline 正在运行...", refresh_model_state()
    for line in proc.stdout:
        log_lines.append(line.rstrip("\n"))
        if len(log_lines) > 400:
            log_lines = log_lines[-400:]
        yield "\n".join(log_lines), "⏳ Pipeline 正在运行...", refresh_model_state()
    rc = proc.wait()
    status = "✅ Pipeline 运行完成。" if rc == 0 else f"❌ Pipeline 失败，退出码 {rc}。"
    model_status = refresh_model_state()
    yield "\n".join(log_lines), status, model_status


def build_ui():
    ensure_template_csv()
    gr.set_static_paths(paths=[str(ASSETS_DIR)])
    css = STYLE_FILE.read_text(encoding="utf-8") if STYLE_FILE.exists() else ""
    model_status = refresh_model_state()
    with gr.Blocks() as demo:
        gr.HTML(f"<style>{css}</style>")
        with gr.Column(elem_id="main_panel"):
            gr.Markdown("# 🏦 Bank Churn Pro Demo\n全屏背景 + Pipeline 日志 + 特征重要性 + 概率仪表盘 + CSV 批量预测 + SHAP 解释")
            model_state_md = gr.Markdown(model_status)
            pipeline_status_md = gr.Markdown("尚未运行 Pipeline。")

            with gr.Tabs():
                with gr.Tab("Pipeline"):
                    gr.Markdown("点击按钮执行 3 步流水线：数据准备 → 模型训练与特征重要性 → 验证与 SHAP 背景缓存")
                    run_btn = gr.Button("▶ Run Pipeline", variant="primary")
                    log_box = gr.Textbox(label="Pipeline Step 1/2/3 日志", lines=22, interactive=False)
                    fi_image = gr.Image(label="Feature Importance 图", type="filepath")
                    run_btn.click(fn=run_pipeline_stream, inputs=[], outputs=[log_box, pipeline_status_md, model_state_md]).then(fn=make_feature_importance_plot, inputs=[], outputs=fi_image)

                with gr.Tab("Single Prediction"):
                    with gr.Row():
                        with gr.Column():
                            age = gr.Slider(18, 100, value=DEFAULTS["AGE"], step=1, label="AGE")
                            open_acc_dur = gr.Slider(0, 400, value=DEFAULTS["OPEN_ACC_DUR"], step=1, label="OPEN_ACC_DUR")
                            gender_cd = gr.Dropdown(choices=["0", "1"], value=DEFAULTS["GENDER_CD"], label="GENDER_CD")
                            hasnt_home = gr.Dropdown(choices=["N", "Y"], value=DEFAULTS["HASNT_HOME_ADDRESS_INF"], label="HASNT_HOME_ADDRESS_INF")
                            hasnt_mobile = gr.Dropdown(choices=["N", "Y"], value=DEFAULTS["HASNT_MOBILE_TEL_NUM_INF"], label="HASNT_MOBILE_TEL_NUM_INF")
                            local_cur = gr.Number(value=DEFAULTS["LOCAL_CUR_MON_AVG_BAL"], label="LOCAL_CUR_MON_AVG_BAL")
                            local_fix = gr.Number(value=DEFAULTS["LOCAL_FIX_MON_AVG_BAL"], label="LOCAL_FIX_MON_AVG_BAL")
                            local_sav = gr.Number(value=DEFAULTS["LOCAL_SAV_CUR_ALL_BAL"], label="LOCAL_SAV_CUR_ALL_BAL")
                            pos_amt = gr.Number(value=DEFAULTS["POS_CONSUME_TX_AMT"], label="POS_CONSUME_TX_AMT")
                            atm_num = gr.Slider(0, 100, value=DEFAULTS["ATM_ALL_TX_NUM"], step=1, label="ATM_ALL_TX_NUM")
                            counter_num = gr.Slider(0, 100, value=DEFAULTS["COUNTER_ALL_TX_NUM"], step=1, label="COUNTER_ALL_TX_NUM")
                            pred_btn = gr.Button("Predict", variant="primary")
                        with gr.Column():
                            pred_json = gr.JSON(label="Prediction JSON")
                            pred_md = gr.Markdown()
                            gauge = gr.HTML(label="Gauge")
                    pred_btn.click(
                        fn=predict_single,
                        inputs=[age, open_acc_dur, gender_cd, hasnt_home, hasnt_mobile, local_cur, local_fix, local_sav, pos_amt, atm_num, counter_num],
                        outputs=[pred_json, pred_md, gauge, fi_image],
                    )

                with gr.Tab("CSV Batch"):
                    gr.Markdown("上传包含以下列的 CSV：" + ", ".join(FEATURES))
                    with gr.Row():
                        batch_file = gr.File(label="Upload CSV", file_types=[".csv"])
                        template_file = gr.File(value=str(TEMPLATE_CSV), label="Template CSV")
                    batch_btn = gr.Button("Run Batch Prediction")
                    batch_df = gr.Dataframe(label="Preview (Top 50)")
                    batch_out_file = gr.File(label="Download Result CSV")
                    batch_msg = gr.Markdown()
                    batch_btn.click(fn=predict_batch, inputs=[batch_file], outputs=[batch_df, batch_out_file, batch_msg])

                with gr.Tab("Explainability"):
                    gr.Markdown("使用当前表单中的同一组输入生成 SHAP waterfall 图。")
                    explain_btn = gr.Button("Generate SHAP Explainability")
                    shap_image = gr.Image(label="SHAP Explainability", type="filepath")
                    shap_md = gr.Markdown()
                    explain_btn.click(
                        fn=explain_single,
                        inputs=[age, open_acc_dur, gender_cd, hasnt_home, hasnt_mobile, local_cur, local_fix, local_sav, pos_amt, atm_num, counter_num],
                        outputs=[shap_image, shap_md],
                    )

            gr.Markdown("<div class='footer-note'>提示：首次进入请先运行 Pipeline，再使用预测、批量预测和解释功能。</div>")
    return demo


if __name__ == "__main__":
    demo = build_ui()
    demo.queue()
    port = int(os.environ.get("PORT", "7860"))
    demo.launch(server_name="0.0.0.0", server_port=port)