feedback.py

# feedback.py — Step 9
# Minimal CSV logger for Helpful / Not helpful signals.
# Writes to data/feedback.csv and auto-creates header on first write.

from pathlib import Path
import csv
import time

FEEDBACK_CSV = Path("data/feedback.csv")
FEEDBACK_CSV.parent.mkdir(parents=True, exist_ok=True)

_FIELDS = [
    "ts_question",
    "ts_feedback",
    "session",
    "feedback",        # "helpful" | "not_helpful"
    "note",            # optional free text
    "lang",
    "mode",
    "temperature",
    "max_tokens",
    "top_k",
    "question",
    "answer",
    "sources",
    "links",
    "forms",
]

def _ensure_header(path: Path):
    if not path.exists() or path.stat().st_size == 0:
        with path.open("w", newline="", encoding="utf-8") as f:
            csv.DictWriter(f, fieldnames=_FIELDS).writeheader()

def _flatten(s: str, lim: int = 2000) -> str:
    """Make single-line, truncate to keep rows compact."""
    if s is None:
        return ""
    s = " ".join(str(s).split())
    return s[:lim]

def log_feedback(entry: dict) -> str:
    """
    entry expects at least:
      ts_question, session, feedback, note, lang, mode, temperature, max_tokens, top_k,
      question, answer, sources, links, forms
    """
    _ensure_header(FEEDBACK_CSV)
    row = {k: "" for k in _FIELDS}
    row.update({
        "ts_question": entry.get("ts_question", int(time.time())),
        "ts_feedback": int(time.time()),
        "session": entry.get("session", ""),
        "feedback": entry.get("feedback", ""),
        "note": _flatten(entry.get("note", "")),
        "lang": entry.get("lang", ""),
        "mode": entry.get("mode", ""),
        "temperature": entry.get("temperature", ""),
        "max_tokens": entry.get("max_tokens", ""),
        "top_k": entry.get("top_k", ""),
        "question": _flatten(entry.get("question", ""), 1000),
        "answer": _flatten(entry.get("answer", ""), 2000),
        "sources": _flatten(entry.get("sources", ""), 1000),
        "links": _flatten(entry.get("links", ""), 1000),
        "forms": _flatten(entry.get("forms", ""), 1000),
    })
    with FEEDBACK_CSV.open("a", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=_FIELDS)
        w.writerow(row)
    return f"Thanks — logged as **{row['feedback']}** for session `{row['session']}`."