Spaces:
Build error
Build error
File size: 4,527 Bytes
0f026a1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 | """
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)
|