""" src/api/search.py — PRIMO + PubMed search endpoint for LibBee v3.4. Search approach: send the full boolean query as a single q=any,contains, parameter — exactly how the PRIMO web UI works. This preserves phrase quotes, parentheses, AND/OR operators, and gives results identical to the web UI. Earlier versions split the boolean into multiple q= params (one per AND group), but PRIMO does not reliably handle OR within a single q= value in multi-param mode. The correct approach is a single q= with the full boolean, URL-encoded. """ import re import logging from datetime import datetime from typing import Optional from fastapi import APIRouter from pydantic import BaseModel from urllib.parse import quote logger = logging.getLogger(__name__) router = APIRouter() # ── Models ───────────────────────────────────────────────────────────────────── class SearchRequest(BaseModel): query: str boolean_query: Optional[str] = None # LLM-built boolean; used for PRIMO if present source: str = "primo" limit: int = 5 peer_reviewed: bool = False open_access: bool = False year_from: Optional[str] = None year_to: Optional[str] = None resource_type: str = "articles" # "articles" | "books" | "both" # ── PRIMO query builder ──────────────────────────────────────────────────────── def _boolean_to_primo_params(boolean_query: str) -> str: """ Convert a boolean query into a single PRIMO q= parameter. Sends the full boolean as: q=any,contains, This matches exactly how the PRIMO web UI constructs advanced search URLs. Examples: '("machine learning" OR "deep learning") AND ("healthcare")' → 'q=any,contains,%28%22machine+learning%22+OR+%22deep+learning%22%29...' '"climate change"' → 'q=any,contains,%22climate+change%22' 'petroleum engineering' → 'q=any,contains,petroleum+engineering' """ if not boolean_query or not boolean_query.strip(): return "q=any,contains,library" # Send the full boolean as a single q= parameter — PRIMO parses it correctly return f"q=any,contains,{quote(boolean_query.strip())}" # ── PRIMO helpers ────────────────────────────────────────────────────────────── def _build_primo_search_link(title: str, record_id: Optional[str] = None) -> Optional[str]: base = 'https://khalifa.primo.exlibrisgroup.com/discovery' if record_id: return f"{base}/fulldisplay?docid={quote(str(record_id))}&vid=971KUOSTAR_INST:KU" clean_title = (title or '').strip() if not clean_title: return None return ( f"{base}/search?query=any,contains,{quote(clean_title)}" "&tab=Everything&search_scope=MyInst_and_CI&vid=971KUOSTAR_INST:KU&lang=en" ) async def search_primo(req: SearchRequest, api_key: str) -> dict: """ Search the PRIMO API. Prefers req.boolean_query (LLM-built) over req.query for structured results. Converts boolean to multi-q-param format for accurate PRIMO results. """ import httpx if not api_key: return {"error": "PRIMO_API_KEY not configured", "results": [], "total": 0} raw_query = (req.boolean_query or req.query or "").strip() has_bool = bool(re.search(r'\b(AND|OR)\b', raw_query)) and '(' in raw_query if has_bool: query_params = _boolean_to_primo_params(raw_query) else: clean = re.sub(r'["\']', '', raw_query).strip() query_params = f"q=any,contains,{quote(clean)}" facets = "" if req.peer_reviewed: facets += "&qInclude=facet_tlevel,exact,peer_reviewed" if req.open_access: facets += "&qInclude=facet_tlevel,exact,open_access" if req.year_from or req.year_to: yf = req.year_from or "1900" yt = req.year_to or str(datetime.now().year) facets += f"&multiFacets=facet_searchcreationdate,include,{yf}%7C,%7C{yt}" if req.resource_type == "books": facets += "&qInclude=facet_rtype,exact,books" elif req.resource_type == "articles": facets += "&qInclude=facet_rtype,exact,articles" vid = "971KUOSTAR_INST:KU" qs = ( f"?vid={vid}&tab=Everything&scope=MyInst_and_CI" f"&{query_params}" f"&lang=en&sort=rank&limit={req.limit}&offset=0&mode=advanced" f"&apikey={api_key}{facets}" ) base = "https://api-eu.hosted.exlibrisgroup.com/primo/v1/search" async with httpx.AsyncClient(timeout=15) as client: for region in ["api-eu", "api-na", "api-ap"]: url = base.replace("api-eu", region) + qs try: r = await client.get(url, headers={"Accept": "application/json"}) if r.status_code == 200: data = r.json() total = data.get("info", {}).get("total", 0) results = [] for doc in data.get("docs", []): pnx = doc.get("pnx", {}) d = pnx.get("display", {}) a = pnx.get("addata", {}) s = pnx.get("search", {}) c = pnx.get("control", {}) l = pnx.get("links", {}) record_id = (c.get("recordid") or [None])[0] title = (d.get("title") or ["Untitled"])[0] primo_url = _build_primo_search_link(title, record_id) doi = (a.get("doi") or [None])[0] raw_link = (l.get("openurl") or l.get("linktorsrc") or [None])[0] results.append({ "record_id": record_id, "title": title, "creator": "; ".join( d.get("creator") or d.get("contributor") or [] ) or "Unknown", "date": ( s.get("creationdate") or a.get("risdate") or a.get("date") or [""] )[0], "type": (d.get("type") or [""])[0], "source": (d.get("source") or a.get("jtitle") or [""])[0], "description": ((d.get("description") or [""])[0] or "")[:400], "doi": doi, "primo_url": primo_url, "link": raw_link or primo_url, "open_access": (d.get("oa") or [""])[0] == "free_for_read", "_source": "PRIMO", }) return {"total": total, "results": results, "source": "PRIMO"} except Exception as e: logger.warning("PRIMO region %s failed: %s", region, e) continue return {"error": "PRIMO API unavailable", "results": [], "total": 0, "source": "PRIMO"} # ── PubMed search ────────────────────────────────────────────────────────────── async def search_pubmed(req: SearchRequest) -> dict: """Search PubMed via E-utilities.""" import httpx base = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils" query = (req.query or "").strip() if req.year_from or req.year_to: yf = req.year_from or "1900" yt = req.year_to or str(datetime.now().year) query = f"({query}) AND ({yf}:{yt}[dp])" try: async with httpx.AsyncClient(timeout=15) as client: r = await client.get( f"{base}/esearch.fcgi", params={ "db": "pubmed", "term": query, "retmax": req.limit, "retmode": "json", "sort": "relevance" } ) if r.status_code != 200: return {"error": "PubMed search failed", "results": [], "source": "PubMed"} esearch = r.json().get("esearchresult", {}) ids = esearch.get("idlist", []) # PubMed Automatic Term Mapping: how the free-text query was # expanded to MeSH terms - surfaced to teach search strategy. query_translation = esearch.get("querytranslation", "") count = int(esearch.get("count", 0)) if not ids: return {"total": 0, "results": [], "source": "PubMed", "query_translation": query_translation} r2 = await client.get( f"{base}/esummary.fcgi", params={"db": "pubmed", "id": ",".join(ids), "retmode": "json"} ) if r2.status_code != 200: return {"error": "PubMed fetch failed", "results": [], "source": "PubMed"} data = r2.json().get("result", {}) results = [] for pid in ids: rec = data.get(pid, {}) if not isinstance(rec, dict): continue authors_list = rec.get("authors", []) authors = ", ".join(a.get("name", "") for a in authors_list[:3]) if len(authors_list) > 3: authors += " et al." eloc = rec.get("elocationid", "") doi = eloc.replace("doi: ", "") if "doi:" in eloc else None results.append({ "title": rec.get("title", ""), "creator": authors, "date": rec.get("pubdate", ""), "source": rec.get("fulljournalname", rec.get("source", "")), "doi": doi, "pmid": pid, "link": f"https://pubmed.ncbi.nlm.nih.gov/{pid}/", "type": "Journal Article", "_source": "PubMed", }) return {"total": count, "results": results, "source": "PubMed", "query_translation": query_translation} except Exception as e: return {"error": f"PubMed: {str(e)}", "results": [], "source": "PubMed"} # ── FastAPI endpoint ─────────────────────────────────────────────────────────── @router.post("") async def search_endpoint(req: SearchRequest): """ Search PRIMO or PubMed. Accepts boolean_query (preferred, LLM-built) or falls back to query. """ from src.config import get_settings settings = get_settings() source = (req.source or "primo").lower() if source == "pubmed": return await search_pubmed(req) return await search_primo(req, settings.primo_api_key)