app.py

# app.py
# NeMo Guardrails Demo (starter UI) — CLI-style spelling quiz from words.csv
# All offline; no LLMs. Ready for Hugging Face Spaces.

from pathlib import Path
import os
import csv
import pandas as pd
import gradio as gr

# ---------- Data loading (robust to messy CSVs) ----------

WORDS_PATH = Path(__file__).parent / "words.csv"
EXPECTED_COLS = ["word", "difficulty", "definition", "origin", "sentence"]

def _load_words(path: Path = WORDS_PATH) -> pd.DataFrame:
    """Load a possibly-messy CSV:
       - tolerate commas inside sentence/origin
       - accept 5+ columns (extras glued into sentence)
       - create missing optional columns
    """
    if not path.exists():
        return pd.DataFrame(columns=["word", "definition", "origin", "sentence", "difficulty_score"])

    rows = []
    with open(path, "r", encoding="utf-8", newline="") as f:
        # try a forgiving CSV read first (honors quotes like "…")
        try:
            df_try = pd.read_csv(
                f,
                engine="python",
                quotechar='"',
                escapechar='\\',
                dtype=str,
                keep_default_na=False
            )
            df = df_try
        except Exception:
            # fall back to manual glue if pandas fails
            f.seek(0)
            reader = csv.reader(f)
            header = next(reader, None)
            header_ok = header and [h.strip().lower() for h in header[:5]] == EXPECTED_COLS
            if not header_ok:
                # treat first line as data
                f.seek(0)
                reader = csv.reader(f)

            for row in reader:
                if not row or all((x is None or str(x).strip() == "") for x in row):
                    continue
                if len(row) < 5:
                    row = row + [""] * (5 - len(row))
                base = row[:4]
                sentence = ",".join(row[4:])  # glue any extras back
                rows.append([*(str(x).strip() for x in base), sentence.strip()])
            df = pd.DataFrame(rows, columns=EXPECTED_COLS)

    # normalize header names
    mapper = {c: c.strip().lower() for c in df.columns}
    df = df.rename(columns=mapper)

    # rebuild to our schema when there are >5 columns
    if df.shape[1] >= 5:
        def pick(colname, default=""):
            return df[colname] if colname in df.columns else default
        base_df = pd.DataFrame({
            "word": pick("word", ""),
            "difficulty": pick("difficulty", ""),
            "definition": pick("definition", ""),
            "origin": pick("origin", ""),
        })
        # sentence = existing sentence + any extra cols joined by commas
        extras = []
        if "sentence" in df.columns:
            extras.append(df["sentence"].astype(str))
        extra_cols = [c for c in df.columns if c not in {"word","difficulty","definition","origin","sentence"}]
        for c in extra_cols:
            extras.append(df[c].astype(str))
        if extras:
            sentence_series = extras[0]
            for s in extras[1:]:
                sentence_series = sentence_series.str.cat(s, sep=",", na_rep="")
        else:
            sentence_series = pd.Series([""] * len(base_df))
        base_df["sentence"] = sentence_series
        df = base_df
    else:
        for c in ["definition", "origin", "sentence", "difficulty"]:
            if c not in df.columns:
                df[c] = ""

    # clean + types
    df["word"] = df["word"].astype(str).fillna("").str.strip()
    df["definition"] = df["definition"].astype(str).fillna("").str.strip()
    df["origin"] = df["origin"].astype(str).fillna("").str.strip()
    df["sentence"] = df["sentence"].astype(str).fillna("").str.strip()

    # difficulty score (numeric difficulty if present; else word length)
    if "difficulty" in df.columns:
        ds = pd.to_numeric(df["difficulty"], errors="coerce").fillna(0)
        df["difficulty_score"] = ds
    else:
        df["difficulty_score"] = df["word"].astype(str).str.len()

    # hardest → easiest
    df = df.sort_values("difficulty_score", ascending=False).reset_index(drop=True)
    return df

DF = _load_words()

# ---------- Quiz logic ----------

def _summary(state):
    if not state:
        return "No round active."
    n = state["n"]
    score = state["score"]
    return f"Round complete. Score this round: {score}/{n}."

def start_quiz(n_words, state):
    df = DF
    if df.empty:
        return (state,
                "words.csv not found or empty.",
                gr.update(value="", interactive=False),
                "Please add a words.csv with header: word,difficulty,definition,origin,sentence",
                "Score: 0/0",
                gr.update(value="", interactive=False))

    try:
        n = int(n_words or 5)
    except Exception:
        n = 5
    n = max(1, min(n, len(df)))  # clip

    # take top-n hardest words, then randomize order for the round
    block = df.head(n).sample(frac=1, random_state=None).reset_index(drop=True)

    s = {
        "i": 0,
        "n": n,
        "score": 0,
        "words": block["word"].astype(str).tolist(),
        "defs": block["definition"].astype(str).tolist(),
        "orig": block["origin"].astype(str).tolist(),
        "sent": block["sentence"].astype(str).tolist(),
    }

    current = s["words"][0]
    hist = f"Okay! We'll do {n} words this round, hardest → easiest (shuffled).\nSpell this word: {current}"
    status = "Type your spelling attempt, or click definition/origin/sentence."
    return s, hist, gr.update(value=current, interactive=False), status, f"Score: 0/{n}", gr.update(value="", interactive=True)

def _check_state(state):
    return bool(state and "words" in state and state["i"] < state["n"])

def check_attempt(state, attempt, history, current_word, score_md):
    if not _check_state(state):
        return state, history, current_word, "No round active. Click Start.", score_md, gr.update(value="")
    attempt = (attempt or "").strip()
    if not attempt:
        return state, history, current_word, "Type your attempt first.", score_md, gr.update(value="")

    target = state["words"][state["i"]]
    if attempt.lower() == target.lower():
        state["score"] += 1
        msg = "✅ Correct!"
    else:
        msg = "❌ Not quite. Try again or ask for definition/origin/sentence."

    history = f"{history}\nYou: {attempt}\nTutor: {msg}"
    score_md = f"Score: {state['score']}/{state['n']}"
    return state, history, current_word, msg, score_md, gr.update(value="")

def _safe_text(val, fallback="Not available."):
    v = (val or "").strip()
    return v if v else fallback

def show_def(state, history, current_word, score_md):
    if not _check_state(state):
        return state, history, current_word, "No round active. Click Start.", score_md
    i = state["i"]
    word = state["words"][i]
    text = _safe_text(state["defs"][i])
    history = f"{history}\nTutor (definition of {word}): {text}"
    return state, history, current_word, f"Definition of {word}: {text}", score_md

def show_origin(state, history, current_word, score_md):
    if not _check_state(state):
        return state, history, current_word, "No round active. Click Start.", score_md
    i = state["i"]
    word = state["words"][i]
    text = _safe_text(state["orig"][i])
    history = f"{history}\nTutor (origin of {word}): {text}"
    return state, history, current_word, f"Origin of {word}: {text}", score_md

def show_sentence(state, history, current_word, score_md):
    if not _check_state(state):
        return state, history, current_word, "No round active. Click Start.", score_md
    i = state["i"]
    word = state["words"][i]
    text = _safe_text(state["sent"][i])
    history = f"{history}\nTutor (sentence with {word}): {text}"
    return state, history, current_word, f"Sentence: {text}", score_md

def show_answer(state, history, current_word, score_md):
    if not _check_state(state):
        return state, history, current_word, "No round active. Click Start.", score_md
    i = state["i"]
    word = state["words"][i]
    history = f"{history}\nTutor: The correct spelling is **{word}**."
    return state, history, current_word, f"Answer: {word}", score_md

def next_word(state, history, current_word, score_md):
    if not _check_state(state):
        return state, history, current_word, "No round active. Click Start.", score_md
    state["i"] += 1
    if state["i"] >= state["n"]:
        summary = _summary(state)
        history = f"{history}\n{summary}"
        return {}, history, gr.update(value="", interactive=False), summary, f"Score: {state['score']}/{state['n']}"
    w = state["words"][state["i"]]
    history = f"{history}\nNext word: {w}"
    status = "Type your spelling attempt, or click definition/origin/sentence."
    return state, history, gr.update(value=w, interactive=False), status, f"Score: {state['score']}/{state['n']}"

def stop_round(state, history, current_word, score_md):
    if not state:
        return {}, history, current_word, "Stopped.", score_md
    summary = _summary(state)
    history = f"{history}\n{summary}"
    return {}, history, gr.update(value="", interactive=False), "Stopped.", f"Score: {state['score']}/{state['n']}"

# ---------- Upload handling ----------

def load_uploaded(file):
    """Replace words.csv with the uploaded file and refresh DF."""
    global DF
    try:
        # Copy uploaded file bytes into WORDS_PATH
        with open(file.name, "rb") as src, open(WORDS_PATH, "wb") as dst:
            dst.write(src.read())
        DF = _load_words()  # refresh global dataset
        return "Loaded new words.csv!", f"Loaded {len(DF)} rows."
    except Exception as e:
        return "", f"Upload failed: {e}"

# ---------- UI ----------

with gr.Blocks() as demo:
    gr.Markdown("# NeMo Guardrails Demo (starter UI)\n**CLI-style spelling quiz** — works offline from `words.csv`.")

    with gr.Row():
        n_words = gr.Number(value=5, precision=0, label="Words this round")
        start = gr.Button("Start quiz")

    # Optional: allow swapping in a new CSV live
    upload = gr.File(label="Upload words.csv", file_types=[".csv"], interactive=True)

    current_word = gr.Textbox(label="Spell this word", interactive=False)
    attempt = gr.Textbox(label="Your attempt")

    with gr.Row():
        check = gr.Button("check")
        bdef = gr.Button("definition")
        borg = gr.Button("origin")
        bsent = gr.Button("sentence")
        bshow = gr.Button("show answer")
        bnext = gr.Button("next")
        bstop = gr.Button("stop")

    status = gr.Markdown("")
    upload_status = gr.Markdown("")  # feedback for uploads
    score_md = gr.Markdown("Score: 0/0")
    history = gr.Textbox(label="History", lines=14)

    state = gr.State({})

    # wire events
    start.click(start_quiz, [n_words, state],
                [state, history, current_word, status, score_md, attempt])

    check.click(check_attempt, [state, attempt, history, current_word, score_md],
                [state, history, current_word, status, score_md, attempt])

    bdef.click(show_def, [state, history, current_word, score_md],
               [state, history, current_word, status, score_md], queue=False)
    borg.click(show_origin, [state, history, current_word, score_md],
               [state, history, current_word, status, score_md], queue=False)
    bsent.click(show_sentence, [state, history, current_word, score_md],
                [state, history, current_word, status, score_md], queue=False)
    bshow.click(show_answer, [state, history, current_word, score_md],
                [state, history, current_word, status, score_md], queue=False)
    bnext.click(next_word, [state, history, current_word, score_md],
                [state, history, current_word, status, score_md], queue=False)
    bstop.click(stop_round, [state, history, current_word, score_md],
                [state, history, current_word, status, score_md], queue=False)

    upload.upload(load_uploaded, [upload], [status, upload_status])

# ---------- Launch (blocking on Spaces) ----------

if __name__ == "__main__":
    try:
        demo.queue(concurrency_count=8, max_size=32)
    except Exception:
        pass

    demo.launch(
        server_name="0.0.0.0",
        server_port=int(os.getenv("PORT", "7860")),
        show_error=True,
    )