app.py

import os
import re
from pathlib import Path

import gradio as gr
import pandas as pd

try:
    from huggingface_hub import InferenceClient
except Exception:
    InferenceClient = None


APP_DIR = Path(__file__).resolve().parent
DATA_PATH = APP_DIR / "data" / "LampiranPublikasiIPLMFinal_Cimahi.xlsx"

DEFAULT_MODEL_ID = "deepseek-ai/DeepSeek-R1"
MODEL_ID = os.getenv("MODEL_ID", DEFAULT_MODEL_ID)
HF_PROVIDER = os.getenv("HF_PROVIDER", "auto").strip() or "auto"
HF_TOKEN = (
    os.getenv("HF_TOKENS")
    or os.getenv("HF_TOKEN")
    or os.getenv("HUGGINGFACEHUB_API_TOKEN")
)
MODEL_CANDIDATES = [
    MODEL_ID,
    "deepseek-ai/DeepSeek-R1",
    "deepseek-ai/DeepSeek-V3-0324",
    "deepseek-ai/DeepSeek-R1-Distill-Qwen-7B",
]

COLS = {
    "wilayah": "Wilayah",
    "jenis": "Jenis",
    "sdm": "VARIABEL SDM",
    "koleksi": "VARIABEL KOLEKSI",
    "pelayanan": "VARIABEL PELAYANAN",
    "pengelolaan": "VARIABEL PENGELOLAAN",
    "kepatuhan": "DIMENSI KEPATUHAN",
    "kinerja": "DIMENSI KINERJA",
    "iplm": "IPLM",
}

DIMENSION_LABELS = {
    "koleksi": "Koleksi",
    "sdm": "SDM",
    "pelayanan": "Pelayanan",
    "pengelolaan": "Pengelolaan",
    "kepatuhan": "Kepatuhan",
    "kinerja": "Kinerja",
    "iplm": "IPLM",
}


def normalize_name(value):
    text = str(value).strip()
    text = re.sub(r"\s+", " ", text)
    return text


def load_data():
    prov = pd.read_excel(DATA_PATH, sheet_name="IPLM_Prov_38")
    kab = pd.read_excel(DATA_PATH, sheet_name="IPLM_KabKota")

    prov = prov.rename(columns={"PROVINSI": COLS["wilayah"]})
    kab = kab.rename(columns={"KABUPATEN/KOTA": COLS["wilayah"]})

    prov[COLS["jenis"]] = "Provinsi"
    kab[COLS["jenis"]] = "Kabupaten/Kota"

    keep = [
        COLS["jenis"],
        COLS["wilayah"],
        COLS["sdm"],
        COLS["koleksi"],
        COLS["pelayanan"],
        COLS["pengelolaan"],
        COLS["kepatuhan"],
        COLS["kinerja"],
        COLS["iplm"],
    ]
    df = pd.concat([prov[keep], kab[keep]], ignore_index=True)
    df[COLS["wilayah"]] = df[COLS["wilayah"]].map(normalize_name)

    for col in keep[2:]:
        df[col] = pd.to_numeric(df[col], errors="coerce")

    df = df.dropna(subset=[COLS["wilayah"], COLS["iplm"]]).copy()
    df["rank_jenis"] = df.groupby(COLS["jenis"])[COLS["iplm"]].rank(
        method="min", ascending=False
    ).astype(int)
    return df.sort_values([COLS["jenis"], "rank_jenis", COLS["wilayah"]])


DATA = load_data()
WILAYAH_CHOICES = DATA[COLS["wilayah"]].tolist()


def find_row(wilayah):
    if not wilayah:
        return None
    mask = DATA[COLS["wilayah"]].str.casefold() == normalize_name(wilayah).casefold()
    if mask.any():
        return DATA.loc[mask].iloc[0]
    contains = DATA[COLS["wilayah"]].str.casefold().str.contains(
        re.escape(normalize_name(wilayah).casefold()), na=False
    )
    if contains.any():
        return DATA.loc[contains].iloc[0]
    return None


def fmt_value(value, scale="var"):
    if pd.isna(value):
        return "-"
    if scale == "iplm":
        return f"{float(value):.2f}"
    return f"{float(value):.3f}"


def classify_value(value, scale="var"):
    if pd.isna(value):
        return "belum tersedia"
    val = float(value)
    if scale == "iplm":
        val = val / 100
    if val >= 0.75:
        return "sangat kuat"
    if val >= 0.55:
        return "kuat"
    if val >= 0.35:
        return "perlu penguatan terarah"
    return "menjadi prioritas perbaikan"


def local_interpretation(row):
    weakest_key = min(
        ["koleksi", "sdm", "pelayanan", "pengelolaan"],
        key=lambda k: float(row[COLS[k]]),
    )
    strongest_key = max(
        ["koleksi", "sdm", "pelayanan", "pengelolaan"],
        key=lambda k: float(row[COLS[k]]),
    )
    wilayah = row[COLS["wilayah"]]
    jenis = row[COLS["jenis"]]

    return (
        f"**{wilayah}** ({jenis}) memiliki IPLM **{fmt_value(row[COLS['iplm']], 'iplm')}** "
        f"dan berada di peringkat **{int(row['rank_jenis'])}** pada kelompok {jenis.lower()}.\n\n"
        f"Dimensi terkuat adalah **{DIMENSION_LABELS[strongest_key]}** "
        f"({fmt_value(row[COLS[strongest_key]])}), sedangkan area yang paling perlu perhatian adalah "
        f"**{DIMENSION_LABELS[weakest_key]}** ({fmt_value(row[COLS[weakest_key]])}). "
        f"Kepatuhan berada pada {fmt_value(row[COLS['kepatuhan']])} dan kinerja berada pada "
        f"{fmt_value(row[COLS['kinerja']])}, sehingga rekomendasi awal perlu menyeimbangkan pemenuhan "
        f"komponen dasar dengan program layanan yang berdampak langsung ke pemustaka.\n\n"
        f"Pertama, prioritaskan intervensi pada {DIMENSION_LABELS[weakest_key].lower()} dengan target "
        f"indikator yang terukur. Kedua, gunakan kekuatan pada {DIMENSION_LABELS[strongest_key].lower()} "
        f"sebagai modal program lintas perangkat daerah. Ketiga, pantau perubahan IPLM secara berkala agar "
        f"perbaikan tidak berhenti di input administratif, tetapi terlihat pada kualitas layanan."
    )


def dataset_snapshot(row=None, limit=8):
    lines = [
        f"Dataset IPLM: {len(DATA)} wilayah total.",
        f"Provinsi: {(DATA[COLS['jenis']] == 'Provinsi').sum()} baris.",
        f"Kabupaten/Kota: {(DATA[COLS['jenis']] == 'Kabupaten/Kota').sum()} baris.",
    ]
    top = DATA.sort_values(COLS["iplm"], ascending=False).head(limit)
    bottom = DATA.sort_values(COLS["iplm"], ascending=True).head(limit)
    lines.append("\nTop IPLM:")
    lines.extend(
        f"- {r[COLS['wilayah']]} ({r[COLS['jenis']]}): {fmt_value(r[COLS['iplm']], 'iplm')}"
        for _, r in top.iterrows()
    )
    lines.append("\nBottom IPLM:")
    lines.extend(
        f"- {r[COLS['wilayah']]} ({r[COLS['jenis']]}): {fmt_value(r[COLS['iplm']], 'iplm')}"
        for _, r in bottom.iterrows()
    )

    if row is not None:
        lines.append("\nWilayah terpilih:")
        lines.extend(
            [
                f"- Nama: {row[COLS['wilayah']]}",
                f"- Jenis: {row[COLS['jenis']]}",
                f"- Rank kelompok: {int(row['rank_jenis'])}",
                f"- Koleksi: {fmt_value(row[COLS['koleksi']])} ({classify_value(row[COLS['koleksi']])})",
                f"- SDM: {fmt_value(row[COLS['sdm']])} ({classify_value(row[COLS['sdm']])})",
                f"- Pelayanan: {fmt_value(row[COLS['pelayanan']])} ({classify_value(row[COLS['pelayanan']])})",
                f"- Pengelolaan: {fmt_value(row[COLS['pengelolaan']])} ({classify_value(row[COLS['pengelolaan']])})",
                f"- Kepatuhan: {fmt_value(row[COLS['kepatuhan']])}",
                f"- Kinerja: {fmt_value(row[COLS['kinerja']])}",
                f"- IPLM: {fmt_value(row[COLS['iplm']], 'iplm')}",
            ]
        )

    return "\n".join(lines)


def build_prompt(question, row):
    context = dataset_snapshot(row)
    style_rules = """
Anda adalah analis kebijakan perpustakaan Indonesia. Tugas Anda adalah menafsirkan data IPLM dan memberi rekomendasi operasional.

Aturan jawaban:
- Jawab dalam Bahasa Indonesia.
- Fokus pada interpretasi data IPLM dan rekomendasi kebijakan perpustakaan.
- Gunakan angka eksplisit dari konteks.
- Jangan mengarang data di luar konteks.
- Bila membuat rekomendasi, tulis format: Pertama, Kedua, Ketiga.
- Untuk pertanyaan ringkas, jawab ringkas. Untuk analisis, beri struktur yang mudah dipakai.
"""
    return f"{style_rules}\n\nKonteks data:\n{context}\n\nPertanyaan pengguna:\n{question}"


def call_deepseek(question, row):
    if InferenceClient is None:
        raise RuntimeError("huggingface_hub belum tersedia.")
    if not HF_TOKEN:
        raise RuntimeError("HF_TOKENS/HF_TOKEN belum diset.")

    messages = [{"role": "user", "content": build_prompt(question, row)}]
    errors = []
    for model_id in dict.fromkeys(MODEL_CANDIDATES):
        try:
            client = InferenceClient(
                model=model_id,
                provider=HF_PROVIDER,
                token=HF_TOKEN,
                timeout=120,
            )
            response = client.chat_completion(
                messages=messages,
                max_tokens=900,
                temperature=0.35,
                top_p=0.9,
            )
            answer = response.choices[0].message.content.strip()
            return f"{answer}\n\n_Model: `{model_id}` via `{HF_PROVIDER}`_"
        except Exception as exc:
            errors.append(f"{model_id}: {exc}")

    raise RuntimeError("Semua kandidat DeepSeek gagal. " + " | ".join(errors))


def respond(message, history, wilayah):
    row = find_row(wilayah)
    if row is None and wilayah:
        return f"Wilayah **{wilayah}** belum ketemu di dataset. Coba pilih dari dropdown."

    question = message.strip() or "Buatkan interpretasi dan rekomendasi IPLM."

    try:
        return call_deepseek(question, row)
    except Exception as exc:
        fallback = local_interpretation(row) if row is not None else dataset_snapshot()
        return (
            f"{fallback}\n\n"
            f"_Mode lokal aktif karena DeepSeek belum bisa dipanggil: {exc}. "
            f"Di Hugging Face Space, set secret `HF_TOKENS` atau `HF_TOKEN`, dan opsional `MODEL_ID`._"
        )


def quick_question(kind, wilayah):
    base = {
        "interpret": "Buatkan interpretasi IPLM lengkap untuk wilayah ini.",
        "recommend": "Buatkan rekomendasi kebijakan prioritas untuk meningkatkan IPLM wilayah ini.",
        "compare": "Bandingkan posisi wilayah ini dengan konteks dataset nasional.",
        "risk": "Apa risiko atau kelemahan utama berdasarkan variabel IPLM wilayah ini?",
    }
    return base.get(kind, base["interpret"])


def run_chat(message, history, wilayah):
    history = history or []
    message = (message or "").strip()
    if not message:
        return history, ""

    history = history + [{"role": "user", "content": message}]
    answer = respond(message, history, wilayah)
    history = history + [{"role": "assistant", "content": answer}]
    return history, ""


def run_quick(kind, history, wilayah):
    return run_chat(quick_question(kind, wilayah), history, wilayah)


def build_preview(wilayah):
    row = find_row(wilayah)
    if row is None:
        return pd.DataFrame()
    preview_cols = [
        COLS["jenis"],
        COLS["wilayah"],
        "rank_jenis",
        COLS["koleksi"],
        COLS["sdm"],
        COLS["pelayanan"],
        COLS["pengelolaan"],
        COLS["kepatuhan"],
        COLS["kinerja"],
        COLS["iplm"],
    ]
    return pd.DataFrame([row[preview_cols]]).rename(columns={"rank_jenis": "Peringkat"})


CSS = """
.gradio-container { max-width: 1180px !important; }
#header { padding: 14px 0 6px; }
#header h1 { font-size: 28px; margin-bottom: 4px; }
#header p { margin: 0; color: #53606f; }
"""


with gr.Blocks(title="IPLM DeepSeek Chatbot") as demo:
    gr.Markdown(
        f"""
<div id="header">
<h1>IPLM DeepSeek Chatbot</h1>
<p>Chatbot interpretasi data IPLM berbasis data publikasi dan prompt benchmark. Model default: <code>{MODEL_ID}</code>.</p>
</div>
"""
    )

    with gr.Row():
        with gr.Column(scale=1, min_width=280):
            wilayah = gr.Dropdown(
                choices=WILAYAH_CHOICES,
                value="KOTA CIMAHI" if "KOTA CIMAHI" in WILAYAH_CHOICES else WILAYAH_CHOICES[0],
                label="Wilayah",
                filterable=True,
            )
            preview = gr.Dataframe(
                value=build_preview("KOTA CIMAHI" if "KOTA CIMAHI" in WILAYAH_CHOICES else WILAYAH_CHOICES[0]),
                label="Data Terpilih",
                interactive=False,
                wrap=True,
            )
            with gr.Row():
                btn_interpret = gr.Button("Interpretasi", size="sm")
                btn_recommend = gr.Button("Rekomendasi", size="sm")
            with gr.Row():
                btn_compare = gr.Button("Bandingkan", size="sm")
                btn_risk = gr.Button("Risiko", size="sm")

        with gr.Column(scale=2):
            chat = gr.Chatbot(height=520, label="Percakapan")
            msg = gr.Textbox(
                placeholder="Tanya: apa interpretasi IPLM Cimahi, prioritas intervensinya, atau bandingkan dengan wilayah lain...",
                lines=2,
                label="Pertanyaan",
            )
            with gr.Row():
                run_btn = gr.Button("Jalankan", variant="primary")
                clear_btn = gr.Button("Bersihkan")
            gr.Examples(
                examples=[
                    "Buatkan interpretasi IPLM wilayah ini dalam 3 paragraf.",
                    "Apa prioritas kebijakan paling penting berdasarkan variabel terlemah?",
                    "Buat ringkasan eksekutif untuk kepala daerah.",
                ],
                inputs=msg,
            )

    wilayah.change(fn=build_preview, inputs=wilayah, outputs=preview)
    msg.submit(fn=run_chat, inputs=[msg, chat, wilayah], outputs=[chat, msg])
    run_btn.click(fn=run_chat, inputs=[msg, chat, wilayah], outputs=[chat, msg])
    clear_btn.click(lambda: ([], ""), outputs=[chat, msg])
    btn_interpret.click(lambda h, w: run_quick("interpret", h, w), inputs=[chat, wilayah], outputs=[chat, msg])
    btn_recommend.click(lambda h, w: run_quick("recommend", h, w), inputs=[chat, wilayah], outputs=[chat, msg])
    btn_compare.click(lambda h, w: run_quick("compare", h, w), inputs=[chat, wilayah], outputs=[chat, msg])
    btn_risk.click(lambda h, w: run_quick("risk", h, w), inputs=[chat, wilayah], outputs=[chat, msg])


if __name__ == "__main__":
    demo.queue().launch(css=CSS)