Blame for src/agentcore/utils.py

"""Small shared helpers: HTML escaping, metrics, history windowing, LLM factory, sanitizers."""
import asyncio
import html
import json
import logging
import os
import re
import time
import uuid
from datetime import datetime
from typing import Any, Dict, List, Literal, Optional, Tuple
from urllib.parse import quote

import httpx
from pydantic import BaseModel, ConfigDict, Field

from src.config import get_settings, LIBBEE_VERSION
from src.services.staff_service import (
    STAFF_DIRECTORY,
    match_staff_name,
    match_staff_role,
    should_attempt_staff_lookup,
    staff_name_answer,
    staff_role_answer,
)

from src.agentcore.models import ChatMessage, SearchContextPayload
from src.agentcore.constants import (
    CURRENT_YEAR,
    HISTORY_WINDOW,
    _LEGACY_SYSTEM_PATTERNS,
    _RESOURCE_TYPE_NOISE,
)

logger = logging.getLogger(__name__)

def _get_runtime_config() -> dict:
    try:
        from src.services.runtime_store import JsonRuntimeStore
        settings = get_settings()
        store = JsonRuntimeStore(settings.config_path, default=settings.runtime_defaults)
        return store.load()
    except Exception:
        return {}


def _escape(text: str) -> str:
    return html.escape(text or "")


def _normalize_whitespace(text: str) -> str:
    return re.sub(r"\s+", " ", (text or "").strip())


def _title_case_topic(topic: str) -> str:
    return _normalize_whitespace(topic).strip().strip(".?")


def _quoted_search_parts(text: str) -> list[str]:
    """Separate quoted titles/phrases so request cleaning cannot alter them."""
    return re.split(r'("[^"\n]*"|“[^”\n]*”|(?<!\w)\'[^\'\n]+\'(?!\w))', text or "")


def _mask_quoted_search_text(text: str) -> str:
    return "".join(" " * len(part) if i % 2 else part
                   for i, part in enumerate(_quoted_search_parts(text)))


def _strip_resource_noise(text: str) -> str:
    parts = _quoted_search_parts(text)
    cleaned = "".join(part if i % 2 else _RESOURCE_TYPE_NOISE.sub(" ", part)
                      for i, part in enumerate(parts))
    cleaned = re.sub(r"\s+(AND|OR)\s+(AND|OR)\s+", " AND ", cleaned, flags=re.IGNORECASE)
    cleaned = re.sub(r"^\s*(AND|OR)\s+", "", cleaned, flags=re.IGNORECASE)
    cleaned = re.sub(r"\s+(AND|OR)\s*$", "", cleaned, flags=re.IGNORECASE)
    return _normalize_whitespace(cleaned)


def _isbn_search_value(text: str) -> str:
    """Recognise an ISBN request, not a number embedded in unrelated prose."""
    candidate = _light_strip_retrieval_boilerplate((text or "").strip()).strip(" .?")
    candidate = re.sub(
        r"^(?:(?:a|an|the)\s+)?(?:text\s?book|e[- ]?book|book)\s+(?:with\s+)?(?=ISBN)",
        "", candidate, flags=re.I,
    )
    match = re.fullmatch(r"(?:ISBN(?:-1[03])?\s*:?\s*)?([0-9][0-9Xx\s-]+)", candidate, flags=re.I)
    if not match:
        return ""
    value = re.sub(r"[\s-]", "", match.group(1)).upper()
    return value if re.fullmatch(r"(?:\d{9}[\dX]|97[89]\d{10})", value) else ""


def _safe_metrics_increment(key: str) -> None:
    try:
        from app import get_metrics_service
        get_metrics_service().incr(key)
    except Exception:
        return


def _safe_metrics_bucket(bucket: str, key: str) -> None:
    try:
        from app import get_metrics_service
        get_metrics_service().incr_bucket(bucket, key)
    except Exception:
        return


def _build_history_messages(history: List[ChatMessage]) -> List[dict]:
    msgs = []
    for m in history[-HISTORY_WINDOW:]:
        if m.role in ("user", "assistant") and m.content:
            msgs.append({"role": m.role, "content": m.content[:300]})
    return msgs


def _get_llm(model: str, temperature: float, max_tokens: int):
    settings = get_settings()
    if settings.anthropic_api_key and (model == "claude" or not settings.openai_api_key):
        from langchain_anthropic import ChatAnthropic
        return ChatAnthropic(
            model="claude-haiku-4-5-20251001",
            temperature=temperature,
            max_tokens=max_tokens,
            anthropic_api_key=settings.anthropic_api_key,
            timeout=15.0,
            max_retries=0,
        )
    from langchain_openai import ChatOpenAI
    return ChatOpenAI(
        model="gpt-4o-mini",
        temperature=temperature,
        max_tokens=max_tokens,
        openai_api_key=settings.openai_api_key,
        timeout=15.0,
        max_retries=0,
    )


def _shared_build_primo_boolean_query(topic: str) -> str:
    clean = _light_strip_retrieval_boilerplate(topic)
    clean = _normalize_whitespace(clean or topic)
    identifier = _isbn_search_value(clean)
    if identifier:
        return f'"{identifier}"'
    # Keep explicitly supplied Boolean operators and protected title phrases.
    if re.search(r"\b(?:AND|OR|NOT)\b", _mask_quoted_search_text(clean)):
        return _sanitize_boolean_for_primo(clean)
    stop = re.compile(
        r"\b(?:of|on|in|the|a|an|and|or|for|to|with|by|from|at|is|are|was|were|"
        r"be|been|have|has|had|do|does|did|will|would|could|should|may|its|"
        r"this|that|these|those|about|impact|role|effect|use|analysis|review|"
        r"what|how|why|when|where|which|between|within|across|among|using|"
        r"based|related|towards|toward|during|after|before|over|under)\b",
        re.IGNORECASE,
    )
    concepts = []
    for i, part in enumerate(_quoted_search_parts(clean)):
        if i % 2:
            concepts.append('"' + part[1:-1] + '"')
        else:
            for value in stop.split(part):
                value = _normalize_whitespace(value).strip(" ,;:.?")
                if len(value) > 1:
                    concepts.append(f'"{value}"' if " " in value else value)
    if not concepts:
        return f'"{clean}"' if clean else ""
    if len(concepts) == 1 and not concepts[0].startswith('"'):
        return f'"{concepts[0]}"'
    return " AND ".join(concepts)


def _make_primo_boolean_query(context: SearchContextPayload) -> str:
    topic = _title_case_topic(context.display_topic or context.topic)
    topic = _strip_resource_noise(topic) or (context.topic or "library search")
    return _shared_build_primo_boolean_query(topic)


def _shared_build_primo_discovery_url(
    boolean_query: str,
    resource_type: str = "articles",
    peer_reviewed: bool = False,
    open_access: bool = False,
    year_from: Optional[str] = None,
    year_to: Optional[str] = None,
    sort: str = "rank",
    query_field: str = "any",
) -> str:
    sort = sort if sort in {"date_d", "date_a"} else "rank"
    query_field = query_field if query_field in {"title", "creator", "sub"} else "any"
    base = (
        "https://khalifa.primo.exlibrisgroup.com/discovery/search"
        f"?vid=971KUOSTAR_INST:KU&tab=Everything&scope=MyInst_and_CI"
        f"&query={query_field},contains,{quote(boolean_query, safe='')}"
        f"&lang=en&search_scope=MyInst_and_CI&sortby={sort}&mode=advanced"
    )
    if query_field == "title" or _isbn_search_value(boolean_query.strip('"')):
        base += "&conVoc=false"
    facets = []
    if resource_type == "articles":
        facets.append("facet_rtype,include,articles")
    elif resource_type == "books":
        facets.append("facet_rtype,include,books")
    elif resource_type == "both":
        facets.append("facet_rtype,include,books|,|facet_rtype,include,articles")
    if peer_reviewed:
        facets.append("facet_tlevel,include,peer_reviewed")
    if open_access:
        facets.append("facet_tlevel,include,open_access")
    if year_from or year_to:
        yf = year_from or "0001"
        yt = year_to or "9999"
        facets.append(f"facet_searchcreationdate,include,{yf}|,|{yt}")
    for facet in facets:
        base += f"&multiFacets={quote(facet)}"
    return base


def _shared_build_pubmed_url(
    topic: str,
    year_from: Optional[str] = None,
    year_to: Optional[str] = None,
    peer_reviewed: bool = False,
    open_access: bool = False,
) -> str:
    clean = _strip_resource_noise(topic)
    term = clean or topic
    # PubMed has no peer-review filter. Journal Article is a publication type,
    # not a peer-review guarantee; keep the argument for API compatibility.
    if open_access:
        term = f"({term}) AND free full text[sb]"
    if year_from or year_to:
        yf = year_from or "0001"
        yt = year_to or "9999"
        term = f"({term}) AND {yf}:{yt}[dp]"
    return f"https://pubmed.ncbi.nlm.nih.gov/?term={quote(term, safe='')}"



def _primo_clean_url(context: SearchContextPayload) -> str:
    boolean_query = context.primo_boolean_query or _make_primo_boolean_query(context)
    return _shared_build_primo_discovery_url(
        boolean_query,
        resource_type=context.resource_type,
        peer_reviewed=context.peer_reviewed,
        open_access=context.open_access,
        year_from=context.year_from,
        year_to=context.year_to,
        sort=getattr(context, "sort", "rank"),
        query_field=getattr(context, "query_field", "any"),
    )


async def _grammar_refine_query(text: str, model: str) -> str:
    settings = get_settings()
    if not settings.openai_api_key and not settings.anthropic_api_key:
        return _normalize_whitespace(text)
    try:
        llm = _get_llm(model, temperature=0, max_tokens=60)
        response = await llm.ainvoke([
            {"role": "system", "content": "Rewrite the user's search question in clear grammatical English. Keep the meaning exactly the same. Return one sentence only."},
            {"role": "user", "content": text},
        ])
        refined = _normalize_whitespace(response.content)
        return refined or _normalize_whitespace(text)
    except Exception:
        return _normalize_whitespace(text)


def _light_strip_retrieval_boilerplate(text: str) -> str:
    """Remove request framing at the start; leave quoted titles untouched."""
    cleaned = (text or "").strip()
    framing = (
        r"^\s*(?:please\s+|(?:can|could|would) (?:you|u)\s+|help me\s+|"
        r"i(?:'|’)?m looking for\s+|i am looking for\s+|i would like\s+|"
        r"i(?:'|’)?d like\s+|i (?:need|want)\s+|"
        r"(?:find|fetch|retrieve)(?: me)?\s+|search(?: for)?\s+|look for\s+|"
        r"(?:get|show|give) me\s+|tell me about\s+)"
    )
    material = (
        r"^\s*(?:a |an |some |the )?(?:(?:peer[- ]reviewed|open[- ]access)\s+)?"
        r"(?:text\s?books?|e[- ]?books?|books?|articles?|papers?|journals?|"
        r"publications?|resources?|research|literature|studies|study)"
        r"(?:\s+(?:and|or)\s+(?:text\s?books?|e[- ]?books?|books?|articles?|papers?))*"
        r"\s+(?:(?:on|about|for|regarding|into|in|of|titled|called|entitled)\s+|(?=[\"“\'])|(?=ISBN\b))"
    )
    for _ in range(8):
        previous = cleaned
        cleaned = re.sub(framing, "", cleaned, count=1, flags=re.I)
        cleaned = re.sub(material, "", cleaned, count=1, flags=re.I)
        if cleaned == previous:
            break
    cleaned = re.sub(r"\s+(?:please|thanks|thank you|asap)[.!?]?$", "", cleaned, flags=re.I)
    return _normalize_whitespace(cleaned)


def _sanitize_llm_response(text: str) -> str:
    if not text:
        return text
    for pattern, replacement in _LEGACY_SYSTEM_PATTERNS:
        text = pattern.sub(replacement, text)
    return text


def _sanitize_boolean_for_primo(boolean: str) -> str:
    if not boolean:
        return boolean
    boolean = re.sub(r"'([^']+)'", r'"\1"', boolean)
    boolean = re.sub(r"^\s*(AND|OR)\s*", "", boolean, flags=re.IGNORECASE)
    boolean = re.sub(r"\s*(AND|OR)\s*$", "", boolean, flags=re.IGNORECASE)
    boolean = re.sub(r"\b(AND|OR)\s+(AND|OR)\b", r"\1", boolean, flags=re.IGNORECASE)
    return re.sub(r"\s+", " ", boolean).strip()


def _clean_database_keywords(boolean_query: str) -> str:
    return re.sub(r"\s+", " ",
                  re.sub(r"\b(AND|OR|NOT)\b|[()\"]", " ",
                         boolean_query, flags=re.IGNORECASE)).strip()


def _find_staff_by_token(token: str) -> Optional[dict]:
    token = (token or "").lower()
    for staff in STAFF_DIRECTORY:
        hay = (staff.get("full_name", "") + " " + staff.get("role", "")).lower()
        if token in hay:
            return staff
    return None