| """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}"' |
| |
| 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 |
| |
| |
| 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 |
|
|
|
|
|
|
