utils.py

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