""" 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)