| """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={ |
| "max_results": 5, |
| "maintenance_mode": False, |
| "welcome_message": "", |
| "custom_instructions": "", |
| "announcement": "", |
| "maintenance_message": "", |
| }) |
| 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 _strip_resource_noise(text: str) -> str: |
| cleaned = _RESOURCE_TYPE_NOISE.sub(" ", text or "") |
| 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 _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 model == "claude" and settings.anthropic_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, |
| ) |
| from langchain_openai import ChatOpenAI |
| return ChatOpenAI( |
| model="gpt-4o-mini", |
| temperature=temperature, |
| max_tokens=max_tokens, |
| openai_api_key=settings.openai_api_key, |
| ) |
|
|
|
|
| def _shared_build_primo_boolean_query(topic: str) -> str: |
| clean = _strip_resource_noise(topic) |
| if not clean: |
| clean = topic |
| clean = _normalize_whitespace(clean) |
| words = clean.split() |
| if len(words) <= 4: |
| return f'"{clean}"' |
| _BOOL_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, |
| ) |
| parts = _BOOL_STOP.split(clean) |
| concepts = [_normalize_whitespace(p) for p in parts if _normalize_whitespace(p) and len(_normalize_whitespace(p)) > 2] |
| if not concepts: |
| return f'"{clean}"' |
| if len(concepts) == 1: |
| return f'"{concepts[0]}"' |
| quoted = [f'"{c}"' if ' ' in c else c for c in concepts[:4]] |
| return " AND ".join(quoted) |
|
|
|
|
| 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, |
| ) -> str: |
| base = ( |
| "https://khalifa.primo.exlibrisgroup.com/discovery/search" |
| f"?vid=971KUOSTAR_INST:KU&tab=Everything&scope=MyInst_and_CI" |
| f"&query=any,contains,{quote(boolean_query)}" |
| f"&lang=en&search_scope=MyInst_and_CI&sortby=rank&mode=advanced" |
| ) |
| facets = [] |
| if resource_type == "articles": |
| facets.append("facet_rtype,include,articles") |
| elif resource_type == "books": |
| facets.append("facet_rtype,include,books") |
| if peer_reviewed: |
| facets.append("facet_tlevel,include,peer_reviewed") |
| if open_access: |
| facets.append("facet_tlevel,include,online_resources") |
| 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, |
| ) -> str: |
| clean = _strip_resource_noise(topic) |
| term = clean or topic |
| if peer_reviewed: |
| term = f"({term}) AND Journal Article[pt]" |
| url = f"https://pubmed.ncbi.nlm.nih.gov/?term={quote(term)}" |
| if year_from or year_to: |
| yf = year_from or "1900" |
| yt = year_to or str(CURRENT_YEAR) |
| url += f"&filter=datesearch.y_{yf}-{yt}" |
| return url |
|
|
|
|
| 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, |
| ) |
|
|
|
|
| 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: |
| cleaned = re.sub(r"^\s*(please\s+)?(?:can you|could you|would you)\s+", "", (text or "").strip(), flags=re.IGNORECASE) |
| cleaned = re.sub(r"^\s*(please\s+)?help me\s+", "", cleaned, flags=re.IGNORECASE) |
| cleaned = re.sub(r"^\s*please\s+", "", cleaned, flags=re.IGNORECASE) |
| cleaned = re.sub(r"\s+(please|thanks|thank you|asap)$", "", cleaned, flags=re.IGNORECASE) |
| cleaned = re.sub( |
| r"^\s*(find|search for|search|look for|get me|show me|give me|fetch|retrieve|" |
| r"research on|articles on|papers on|literature on|studies on|" |
| r"tell me about|i need|i want|can you find|help me find|" |
| r"i am looking for|i'm looking for|i need articles on|" |
| r"i want articles on|i need papers on|i want papers on)\s+", |
| "", cleaned, flags=re.IGNORECASE |
| ) |
| cleaned = re.sub( |
| r"^\s*(research|articles?|papers?|books?|literature|studies|study|" |
| r"journals?|publications?|resources?)\s+(on|about|for|regarding|into)\s+", |
| "", cleaned, flags=re.IGNORECASE |
| ) |
| cleaned = re.sub( |
| r"^\s*(on|about|for|regarding|concerning|into|around|of|in|the)\s+", |
| "", cleaned, flags=re.IGNORECASE |
| ) |
| return re.sub(r"\s+", " ", cleaned).strip() |
|
|
|
|
| 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*(?:(?:19|20)\d{2}\s*(?:OR\s*(?:19|20)\d{2}\s*)*)\)", |
| "", boolean, flags=re.IGNORECASE, |
| ) |
| 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 |
|
|
|
|
|
|
