app.py

# app.py
# -*- coding: utf-8 -*-
"""
Top-level Gradio app for ESG ABSA (Rule-based | Classical ML | Deep | Hybrid++ | Explainability)

Expect directory layout:
  ./app.py
  ./core/
      __init__.py
      utils.py
      lexicons.py
      rule_based.py
      classical_ml.py
      deep_model.py
      hybrid_model.py
      explainability.py
      app_state.py    # small dict wrapper (see notes below)
"""

import os
import tempfile
import traceback

import gradio as gr
import pandas as pd

# Try to import core modules; provide helpful error if missing
try:
    from core.utils import parse_document, safe_plot
    from core.rule_based import run_rule_based, explain_rule_based_sentence
    from core.classical_ml import run_classical_ml, explain_classical_sentence
    from core.deep_model import run_deep_learning, explain_deep_sentence, plot_attention_plotly
    from core.hybrid_model import run_hierarchical_hybrid, explain_hybrid_sentence, plot_ontology_scatter
    from core.explainability import compare_explain, explain_sentence_across_models, plot_consistency_summary
    from core.app_state import app_state
except Exception as e:
    raise ImportError(f"Failed to import core modules. Make sure core/ package exists and contains the modules. Error: {e}")

EXAMPLE_TEXT = """## TANTANGAN DAN RESPONS TERHADAP ISU KEBERLANJUTAN
The ban on corn imports pushed us to become more self-sufficient by using locally sourced raw materials. Partnerships with local farmers became vital to secure supply and reduce reliance on international markets.
"""

def _safe_call(fn, *args, fallback=(None, pd.DataFrame(), None)):
    """
    Helper to call model functions safely and return consistent outputs.
    fallback default should match (csv_path, df, fig, ...) shape expected by callers.
    """
    try:
        return fn(*args)
    except Exception as e:
        traceback.print_exc()
        # return reasonable fallback shaped to caller expected outputs
        return fallback

# ---------------------------
# Gradio app UI
# ---------------------------
with gr.Blocks(title="ESG ABSA – Unified Explainability Dashboard (CPU-friendly demo)") as demo:
    gr.Markdown("# 🌱 ESG ABSA — Unified (Rule-based • Classical • Deep • Hybrid++)")
    gr.Markdown("Paste an ESG section or full report (headers `## ...` supported). Run any tab to produce CSV, preview and visualizations. Use Explainability tab to compare results across models.")

    # Tab 1: Rule-based
    with gr.Tab("1) Rule-based"):
        t1 = gr.Textbox(lines=14, value=EXAMPLE_TEXT, label="Input Text")
        r1_btn = gr.Button("Run Rule-based")
        r1_file = gr.File(label="Download CSV")
        r1_df = gr.DataFrame(label="Preview (Rule-based)", interactive=False)
        r1_plot = gr.Plot(label="Visualization")
        r1_expl = gr.DataFrame(label="Per-sentence Rule Explanations", interactive=False)

        def _run_rule(text):
            csv, df, fig = run_rule_based(text)
            # explanation column
            df2 = df.copy()
            df2["Rule_Explanation"] = df2["Sentence_Text"].apply(lambda t: "; ".join(explain_rule_based_sentence(t)))
            app_state["last_sentences"] = df2
            return csv, df2, fig, df2[["Sentence_ID", "Rule_Explanation"]]

        r1_btn.click(fn=_run_rule, inputs=t1, outputs=[r1_file, r1_df, r1_plot, r1_expl])

    # Tab 2: Classical ML
    with gr.Tab("2) Classical ML"):
        t2 = gr.Textbox(lines=14, value=EXAMPLE_TEXT, label="Input Text")
        r2_btn = gr.Button("Run Classical ML")
        r2_file = gr.File(label="Download CSV")
        r2_df = gr.DataFrame(label="Validation Predictions", interactive=False)
        r2_plot = gr.Plot(label="Visualization")
        r2_coef_sent = gr.DataFrame(label="Global Coefficients (Sentiment)", interactive=False)
        r2_coef_aspect = gr.DataFrame(label="Global Coefficients (Aspect)", interactive=False)

        def _run_classical(text):
            csv, out, fig, coef_s, coef_a = run_classical_ml(text)
            # update last_sentences for dashboard (prefer df_val)
            if app_state.get("classical") and app_state["classical"].get("df_val") is not None:
                app_state["last_sentences"] = app_state["classical"]["df_val"]
            return csv, out, fig, coef_s, coef_a

        r2_btn.click(fn=_run_classical, inputs=t2, outputs=[r2_file, r2_df, r2_plot, r2_coef_sent, r2_coef_aspect])
        # optional initial load for convenience
        demo.load(fn=_run_classical, inputs=[t2], outputs=[r2_file, r2_df, r2_plot, r2_coef_sent, r2_coef_aspect])

    # Tab 3: Deep Learning
    with gr.Tab("3) Deep Learning (mBERT demo)"):
        t3 = gr.Textbox(lines=14, value=EXAMPLE_TEXT, label="Input Text")
        e3 = gr.Slider(1, 2, value=1, step=1, label="Epochs (light demo)")
        r3_btn = gr.Button("Train & Predict (mBERT)")
        r3_file = gr.File(label="Download CSV")
        r3_df = gr.DataFrame(label="Predictions", interactive=False)
        r3_plot = gr.Plot(label="Visualization")
        r3_interp = gr.DataFrame(label="Interpretability (tokens)", interactive=False)

        def _run_deep(text, epochs):
            csv, df, fig, interp = run_deep_learning(text, epochs)
            if app_state.get("deep"):
                app_state["last_sentences"] = app_state["deep"]["df"]
            return csv, df, fig, interp

        r3_btn.click(fn=_run_deep, inputs=[t3, e3], outputs=[r3_file, r3_df, r3_plot, r3_interp])
        demo.load(fn=_run_deep, inputs=[t3, e3], outputs=[r3_file, r3_df, r3_plot, r3_interp])

    # Tab 4: Hybrid++ (Hierarchical + MTL + Ontology)
    with gr.Tab("4) Hybrid++ (Hierarchical + MTL + Ontology)"):
        t4 = gr.Textbox(lines=14, value=EXAMPLE_TEXT, label="Input Text")
        e4 = gr.Slider(minimum=1, maximum=50, value=3, label="Epochs")
        tw4 = gr.Slider(minimum=0.0, maximum=2.0, value=1.5, step=0.1, label="Tone weight")
        aw4 = gr.Slider(minimum=0.0, maximum=1.0, value=0.2, step=0.05, label="Alignment weight")
        r4_btn = gr.Button("Run Hybrid++")
        r4_file = gr.File(label="Download CSV")
        r4_df = gr.DataFrame(label="Predictions", interactive=False)
        r4_plot_t = gr.Plot(label="Tone × Sentiment")
        r4_plot_a = gr.Plot(label="Ontology Alignment (cosine)")
        r4_plot_s = gr.Plot(label="Tone Distribution by Section")
        r4_metrics = gr.DataFrame(label="Metrics", interactive=False)

        def _run_hybrid(text, epochs, tw, aw):
            csv, df, f1, f2, f3, metrics = run_hierarchical_hybrid(text, epochs, tone_weight=tw, align_weight=aw)
            if app_state.get("hybrid"):
                app_state["last_sentences"] = app_state["hybrid"]["df"]
            return csv, df, f1, f2, f3, metrics

        r4_btn.click(fn=_run_hybrid, inputs=[t4, e4, tw4, aw4], outputs=[r4_file, r4_df, r4_plot_t, r4_plot_a, r4_plot_s, r4_metrics])
        demo.load(fn=_run_hybrid, inputs=[t4, e4, tw4, aw4], outputs=[r4_file, r4_df, r4_plot_t, r4_plot_a, r4_plot_s, r4_metrics])

    # Tab 5: Explainability Dashboard
    with gr.Tab("5) 🧠 Explainability Dashboard"):
        gr.Markdown("Compare explanations across Rule-based, Classical ML, Deep, and Hybrid models.")
        # Input can be either a sentence typed directly or a dropdown from last run (we provide direct textbox)
        sent_input = gr.Textbox(label="Enter a sentence for cross-model explanation", lines=3, placeholder="Type or paste a sentence from your input...")
        compare_btn = gr.Button("Compare Explainability Across Models")
        summary_table = gr.DataFrame(label="Cross-model summary", interactive=False)
        deep_plot_out = gr.Plot(label="Deep: Token attention (Plotly)")
        cls_plot_out = gr.Plot(label="Classical: Top TF-IDF features")
        hybrid_plot_out = gr.Plot(label="Hybrid: Ontology embedding scatter")
        consistency_plot = gr.Plot(label="Model consistency summary")

        def _compare_sentence(sentence_text):
            if not sentence_text or str(sentence_text).strip() == "":
                return pd.DataFrame([["Error", "Please enter a sentence."]], columns=["Model", "Explanation"]), None, None, None, None
            summary_df, deep_fig, cls_fig, hy_fig = compare_explain_for_sentence(sentence_text)
            return summary_df, deep_fig, cls_fig, hy_fig, plot_consistency_summary_safe()

        # We use a wrapper to reuse logic from core.explainability but accept direct sentence text
        def compare_explain_for_sentence(sentence_text):
            # Build summary rows similar to previous standalone compare_explain function but for a single sentence
            records = []
            # Rule
            try:
                rule_expl = explain_rule_based_sentence(sentence_text)
                records.append(["Rule-based", ", ".join(rule_expl[:6])])
            except Exception as e:
                records.append(["Rule-based", f"Error: {e}"])
            # Classical
            try:
                cls_expl = explain_classical_sentence(sentence_text)
                if isinstance(cls_expl, dict) and "error" in cls_expl:
                    records.append(["Classical ML", cls_expl["error"]])
                    cls_fig = None
                else:
                    pred = cls_expl.get("prediction", "N/A")
                    local = cls_expl.get("local_features", [])
                    local_text = "; ".join([f"{f['feature']} ({f['contribution']:.3f})" for f in local]) if local else "No local features"
                    records.append(["Classical ML", f"Pred: {pred}; Local: {local_text}"])
                    # global plot if available
                    top = cls_expl.get("global_top", None)
                    if top is not None and isinstance(top, pd.DataFrame) and not top.empty:
                        import plotly.express as px
                        cls_fig = px.bar(top.nlargest(12, "Coefficient"), x="Feature", y="Coefficient", color="Direction", title="Classical: Top features (sentiment)")
                        cls_fig.update_layout(height=300, margin=dict(l=10, r=10, t=30, b=10))
                    else:
                        cls_fig = None
            except Exception as e:
                records.append(["Classical ML", f"Error: {e}"])
                cls_fig = None

            # Deep
            try:
                deep_expl = explain_deep_sentence(sentence_text)
                if isinstance(deep_expl, dict) and "error" in deep_expl:
                    records.append(["Deep (mBERT)", deep_expl["error"]])
                    deep_fig = None
                else:
                    toks = deep_expl.get("tokens", [])
                    w = deep_expl.get("weights", [])
                    pairs = []
                    for tok, wt in zip(toks, w):
                        if tok in ["[PAD]", "[CLS]", "[SEP]"]:
                            continue
                        pairs.append(f"{tok}:{wt:.3f}")
                    records.append(["Deep (mBERT)", ", ".join(pairs[:12]) or "No tokens"])
                    deep_fig = plot_attention_plotly(toks, w, title="mBERT: Token attention (first-layer avg)")
            except Exception as e:
                records.append(["Deep (mBERT)", f"Error: {e}"])
                deep_fig = None

            # Hybrid
            try:
                hy_expl = explain_hybrid_sentence(sentence_text)
                if isinstance(hy_expl, dict) and "error" in hy_expl:
                    records.append(["Hybrid++", hy_expl["error"]])
                    hy_fig = None
                else:
                    records.append(["Hybrid++", f"Ontology alignment: {hy_expl['ontology_alignment']:.3f}; Path: {hy_expl['ontology_path']}; Section: '{hy_expl.get('section_name', 'N/A')}' (influence: {hy_expl.get('section_influence', 0.0):.2f})"])
                    # build hybrid scatter if hybrid state available
                    hy_state = app_state.get("hybrid")
                    hy_fig = plot_ontology_scatter(hy_state) if hy_state else None
            except Exception as e:
                records.append(["Hybrid++", f"Error: {e}"])
                hy_fig = None

            summary_df = pd.DataFrame(records, columns=["Model", "Top Explanation"])
            return summary_df, deep_fig, cls_fig, hy_fig

        def plot_consistency_summary_safe():
            try:
                return plot_consistency_summary()
            except Exception:
                return None

        compare_btn.click(fn=_compare_sentence, inputs=[sent_input], outputs=[summary_table, deep_plot_out, cls_plot_out, hybrid_plot_out, consistency_plot])

    demo.launch()