Create feedback.py

feedback.py

@@ -0,0 +1,70 @@
+# 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']}`."