Create utils.py

utils.py

@@ -0,0 +1,135 @@
+"""
+Reusable helper utilities for the University Admissions RAG Chatbot.
+"""
+
+import hashlib
+import os
+import re
+import time
+from pathlib import Path
+from typing import Any
+
+from logging_config import get_logger
+
+logger = get_logger(__name__)
+
+
+# ── File helpers ─────────────────────────────────────────────────────────────
+
+def validate_file(
+    filepath: str,
+    allowed_extensions: tuple[str, ...] = (".pdf", ".docx", ".txt"),
+    max_size_mb: int = 20,
+) -> tuple[bool, str]:
+    """Return (is_valid, reason)."""
+    path = Path(filepath)
+
+    if not path.exists():
+        return False, f"File not found: {filepath}"
+
+    suffix = path.suffix.lower()
+    if suffix not in allowed_extensions:
+        return False, (
+            f"Extension '{suffix}' not allowed. "
+            f"Supported: {', '.join(allowed_extensions)}"
+        )
+
+    size_mb = path.stat().st_size / (1024 * 1024)
+    if size_mb > max_size_mb:
+        return False, f"File exceeds {max_size_mb} MB limit ({size_mb:.1f} MB)."
+
+    return True, "OK"
+
+
+def file_checksum(filepath: str) -> str:
+    """MD5 checksum of a file (for deduplication)."""
+    h = hashlib.md5()
+    with open(filepath, "rb") as f:
+        for chunk in iter(lambda: f.read(8192), b""):
+            h.update(chunk)
+    return h.hexdigest()
+
+
+def ensure_dir(path: str) -> str:
+    Path(path).mkdir(parents=True, exist_ok=True)
+    return path
+
+
+# ── Text helpers ─────────────────────────────────────────────────────────────
+
+def sanitize_input(text: str, max_length: int = 1000) -> str:
+    """Strip control characters and enforce length cap."""
+    text = re.sub(r"[\x00-\x08\x0b-\x0c\x0e-\x1f\x7f]", "", text)
+    return text.strip()[:max_length]
+
+
+def truncate_text(text: str, max_chars: int = 300) -> str:
+    if len(text) <= max_chars:
+        return text
+    return text[:max_chars].rstrip() + "…"
+
+
+# ── Timing helpers ────────────────────────────────────────────────────────────
+
+class Timer:
+    """Context-manager timer."""
+
+    def __enter__(self) -> "Timer":
+        self.start = time.perf_counter()
+        return self
+
+    def __exit__(self, *_: Any) -> None:
+        self.elapsed = time.perf_counter() - self.start
+
+    def __str__(self) -> str:
+        return f"{self.elapsed:.3f}s"
+
+
+# ── Chat history helpers ──────────────────────────────────────────────────────
+
+def format_history_for_context(
+    history: list[tuple[str, str]],
+    max_turns: int = 6,
+) -> str:
+    """Convert Gradio-style [(user, bot), ...] history to a plain string."""
+    recent = history[-max_turns:]
+    lines: list[str] = []
+    for user_msg, bot_msg in recent:
+        if user_msg:
+            lines.append(f"User: {user_msg}")
+        if bot_msg:
+            lines.append(f"Assistant: {bot_msg}")
+    return "\n".join(lines)
+
+
+def export_chat_history(history: list[tuple[str, str]]) -> str:
+    """Return a plain-text transcript."""
+    if not history:
+        return "No conversation history to export."
+    lines = ["University Admissions Assistant — Chat Transcript", "=" * 52, ""]
+    for i, (user_msg, bot_msg) in enumerate(history, 1):
+        lines.append(f"[Turn {i}]")
+        lines.append(f"You:       {user_msg}")
+        lines.append(f"Assistant: {bot_msg}")
+        lines.append("")
+    return "\n".join(lines)
+
+
+# ── Source formatting ─────────────────────────────────────────────────────────
+
+def format_sources(docs: list[Any]) -> str:
+    """Format LangChain Document objects into a readable source list."""
+    if not docs:
+        return ""
+    seen: set[str] = set()
+    parts: list[str] = []
+    for doc in docs:
+        src = doc.metadata.get("source", "Unknown source")
+        page = doc.metadata.get("page")
+        label = f"📄 {os.path.basename(src)}"
+        if page is not None:
+            label += f" (p. {page + 1})"
+        if label not in seen:
+            seen.add(label)
+            parts.append(label)
+    return "\n".join(parts)