| """PRIMO and PubMed search, with explicit filters and bounded upstream waits.""" |
| import asyncio |
| import logging |
| import re |
| from typing import Optional |
| from urllib.parse import quote, urlsplit |
|
|
| from fastapi import APIRouter |
| from pydantic import BaseModel, ConfigDict, Field, model_validator |
|
|
| logger = logging.getLogger(__name__) |
| router = APIRouter() |
| |
| |
| PRIMO_REQUEST_TIMEOUT = 6.0 |
| PUBMED_REQUEST_TIMEOUT = 8.0 |
|
|
|
|
| class SearchRequest(BaseModel): |
| model_config = ConfigDict(extra="forbid", str_strip_whitespace=True) |
| query: str = Field(min_length=1, max_length=1000) |
| boolean_query: Optional[str] = Field(default=None, max_length=1500) |
| source: str = Field(default="primo", pattern=r"^(primo|pubmed)$") |
| limit: int = Field(default=10, ge=1, le=20) |
| sort: str = Field(default="rank", pattern=r"^(rank|date_d|date_a)$") |
| query_field: str = Field(default="any", pattern=r"^(any|title|creator|sub)$") |
| peer_reviewed: bool = False |
| open_access: bool = False |
| year_from: Optional[str] = Field(default=None, pattern=r"^[0-9]{4}$") |
| year_to: Optional[str] = Field(default=None, pattern=r"^[0-9]{4}$") |
| resource_type: str = Field(default="articles", pattern=r"^(articles|books|both)$") |
|
|
| @model_validator(mode="after") |
| def validate_search(self): |
| self.boolean_query = self.boolean_query or None |
| for year in (self.year_from, self.year_to): |
| if year == "0000": |
| raise ValueError("Publication years must be between 0001 and 9999") |
| if self.year_from and self.year_to and self.year_from > self.year_to: |
| raise ValueError("year_from must not be later than year_to") |
| |
| |
| if self.source == "primo" and ";" in (self.boolean_query or self.query): |
| raise ValueError("PRIMO search terms cannot contain semicolons") |
| return self |
|
|
|
|
| def _boolean_to_primo_params(boolean_query: str, query_field: str = "any") -> str: |
| """Preserve exact phrases and all Boolean operators in one Primo q value.""" |
| if query_field not in {"any", "title", "creator", "sub"}: |
| raise ValueError("Unsupported PRIMO search field") |
| return f"q={query_field},contains,{quote(boolean_query.strip(), safe='')}" |
|
|
|
|
| def _year_bounds(req: SearchRequest) -> tuple[str, str]: |
| """Represent an omitted bound without silently excluding older/future works.""" |
| return req.year_from or "0001", req.year_to or "9999" |
|
|
|
|
| def _build_primo_search_link(title: str, record_id: Optional[str] = None, |
| context: Optional[str] = None) -> Optional[str]: |
| base = "https://khalifa.primo.exlibrisgroup.com/discovery" |
| if record_id: |
| context_param = f"&context={context}" if isinstance(context, str) and context in {"L", "PC"} else "" |
| return ( |
| f"{base}/fulldisplay?docid={quote(str(record_id), safe='')}&vid=971KUOSTAR_INST:KU" |
| f"{context_param}&lang=en&tab=Everything&search_scope=MyInst_and_CI" |
| ) |
| clean_title = (title or "").strip() |
| if not clean_title: |
| return None |
| return ( |
| f"{base}/search?query=any,contains,{quote(clean_title, safe='')}" |
| "&tab=Everything&search_scope=MyInst_and_CI&vid=971KUOSTAR_INST:KU&lang=en" |
| ) |
|
|
|
|
| def _first(value, default=""): |
| """PNX values are normally arrays; tolerate scalar/missing metadata.""" |
| if isinstance(value, list): |
| value = value[0] if value else default |
| return value if isinstance(value, str) else default |
|
|
|
|
| def _result_link(raw_link: str, fallback: Optional[str]) -> Optional[str]: |
| |
| match = re.match(r"(?:\$\$U)?(https?://[^\s]+?)(?:\$\$|$)", raw_link) |
| if not match: |
| return fallback |
| candidate = match.group(1) |
| try: |
| parsed = urlsplit(candidate) |
| if parsed.hostname and not parsed.username and not parsed.password: |
| return candidate |
| except ValueError: |
| pass |
| return fallback |
|
|
|
|
| def _pnx_section(pnx: dict, section: str) -> dict: |
| value = pnx.get(section) |
| return value if isinstance(value, dict) else {} |
|
|
|
|
| def _primo_records(data: dict, limit: int) -> list[dict]: |
| """Normalize returned PNX metadata; never infer circulation availability.""" |
| results = [] |
| seen = set() |
| for doc in data.get("docs", []): |
| if not isinstance(doc, dict) or not isinstance(doc.get("pnx"), dict): |
| continue |
| pnx = doc["pnx"] |
| display = _pnx_section(pnx, "display") |
| addata = _pnx_section(pnx, "addata") |
| search = _pnx_section(pnx, "search") |
| sort = _pnx_section(pnx, "sort") |
| control = _pnx_section(pnx, "control") |
| links = _pnx_section(pnx, "links") |
| record_id = _first(control.get("recordid"), None) or _first(doc.get("id"), None) |
| title = _first(display.get("title")) or _first(addata.get("btitle") or addata.get("atitle")) |
| |
| if not title: |
| continue |
| if record_id and record_id in seen: |
| continue |
| if record_id: |
| seen.add(record_id) |
| primo_url = _build_primo_search_link(title, record_id, _first(doc.get("context"))) |
| creators = display.get("creator") or display.get("contributor") or addata.get("au") or [] |
| if isinstance(creators, str): |
| creators = [creators] |
| if not isinstance(creators, list): |
| creators = [] |
| raw_link = _first(links.get("openurl") or links.get("linktorsrc")) |
| results.append({ |
| "record_id": record_id, |
| "title": title, |
| "creator": "; ".join(creator for creator in creators if isinstance(creator, str)), |
| "date": (_first(search.get("creationdate")) or _first(addata.get("risdate")) |
| or _first(addata.get("date")) or _first(sort.get("creationdate"))), |
| "type": _first(display.get("type") or addata.get("ristype")), |
| "source": _first(display.get("source") or addata.get("jtitle")), |
| "description": _first(display.get("description"))[:400], |
| "doi": _first(addata.get("doi"), None), |
| "primo_url": primo_url, |
| "link": _result_link(raw_link, primo_url), |
| "open_access": True if _first(display.get("oa")) == "free_for_read" else None, |
| "_source": "PRIMO", |
| }) |
| if len(results) >= limit: |
| break |
| return results |
|
|
|
|
| def _primo_failure(message: str, code: str, status: str = "unavailable") -> dict: |
| return {"error": message, "error_code": code, "status": status, |
| "results": [], "total": 0, "source": "PRIMO"} |
|
|
|
|
| def _primo_gateways(region: str) -> tuple[str, ...]: |
| """Allow only Ex Libris's documented gateway hosts; no custom secret target.""" |
| hosts = { |
| "eu": "api-eu.hosted.exlibrisgroup.com", |
| "na": "api-na.hosted.exlibrisgroup.com", |
| "ap": "api-ap.hosted.exlibrisgroup.com", |
| "ca": "api-ca.hosted.exlibrisgroup.com", |
| "aps": "api-aps.hosted.exlibrisgroup.com", |
| "cn": "api-cn.hosted.exlibrisgroup.com.cn", |
| } |
| if region == "auto": |
| return tuple(hosts[name] for name in ("eu", "na", "ap")) |
| return (hosts[region],) if region in hosts else () |
|
|
|
|
| async def search_primo(req: SearchRequest, api_key: str, region: str = "auto") -> dict: |
| """Search Primo using the full preferred query and documented facet syntax.""" |
| import httpx |
|
|
| if not api_key: |
| return _primo_failure("PRIMO results are not configured", "not_configured", "not_configured") |
|
|
| gateways = _primo_gateways(region) |
| if not gateways: |
| return _primo_failure("PRIMO API region is not configured correctly", "invalid_region", "not_configured") |
|
|
| query_params = _boolean_to_primo_params(req.boolean_query or req.query, req.query_field) |
| facets = [] |
| if req.peer_reviewed: |
| facets.append("facet_tlevel,exact,peer_reviewed") |
| if req.open_access: |
| facets.append("facet_tlevel,exact,open_access") |
| if req.year_from or req.year_to: |
| year_from, year_to = _year_bounds(req) |
| facets.append(f"facet_searchcreationdate,exact,[{year_from} TO {year_to}]") |
| if req.resource_type != "both": |
| facets.append(f"facet_rtype,exact,{req.resource_type}") |
|
|
| |
| |
| mixed_types = "facet_rtype,include,books|,|facet_rtype,include,articles" |
| mixed_params = f"&multiFacets={quote(mixed_types, safe='')}" if req.resource_type == "both" else "" |
| compact_query = re.sub(r'["\s-]', "", (req.boolean_query or req.query)).upper() |
| exact_identifier = bool(re.fullmatch(r"(?:[0-9]{9}[0-9X]|[0-9]{13})", compact_query)) |
| controlled_vocabulary = "false" if req.query_field == "title" or exact_identifier else "true" |
|
|
| |
| |
| facet_params = f"&qInclude={quote('|,|'.join(facets), safe='')}" if facets else "" |
| qs = ( |
| "?vid=971KUOSTAR_INST:KU&tab=Everything&scope=MyInst_and_CI" |
| f"&{query_params}&lang=en&sort={req.sort}&limit={req.limit}&offset=0" |
| f"&conVoc={controlled_vocabulary}&skipDelivery=true" |
| f"{facet_params}{mixed_params}" |
| ) |
|
|
| deadline = asyncio.get_running_loop().time() + PRIMO_REQUEST_TIMEOUT |
| last_status = None |
| async with httpx.AsyncClient(timeout=PRIMO_REQUEST_TIMEOUT) as client: |
| for gateway in gateways: |
| remaining = deadline - asyncio.get_running_loop().time() |
| if remaining <= 0: |
| return _primo_failure("PRIMO took too long to respond; use the search link", "timeout") |
| url = f"https://{gateway}/primo/v1/search{qs}" |
| try: |
| |
| |
| response = await asyncio.wait_for( |
| client.get(url, headers={"Accept": "application/json", "Authorization": f"apikey {api_key}"}), |
| timeout=remaining, |
| ) |
| if response.status_code != 200: |
| last_status = response.status_code |
| logger.warning("PRIMO gateway %s returned HTTP %s", gateway, response.status_code) |
| if response.status_code == 429: |
| return _primo_failure("PRIMO request limit reached; try again later", "rate_limited", "rate_limited") |
| if response.status_code == 400: |
| return _primo_failure("PRIMO could not process this search; check the search terms", "invalid_query", "error") |
| if response.status_code in (401, 403, 404): |
| continue |
| return _primo_failure("PRIMO API unavailable", "upstream_error") |
| data = response.json() |
| if not isinstance(data, dict) or not isinstance(data.get("docs"), list): |
| raise ValueError("Invalid PRIMO response") |
| results = _primo_records(data, req.limit) |
| info = data.get("info") if isinstance(data.get("info"), dict) else {} |
| raw_total = info.get("total") |
| total = None |
| |
| |
| if ((type(raw_total) is int) or |
| (isinstance(raw_total, str) and re.fullmatch(r"[0-9]+", raw_total.strip()))): |
| parsed_total = int(raw_total) |
| if parsed_total >= len(results): |
| total = parsed_total |
| if data["docs"] and not results: |
| return _primo_failure("PRIMO returned unreadable record details; use the search link", "invalid_response") |
| return {"total": total, "results": results, "source": "PRIMO", "status": "ok"} |
| except (TimeoutError, httpx.TimeoutException): |
| logger.warning("PRIMO search exceeded the total request deadline") |
| return _primo_failure("PRIMO took too long to respond; use the search link", "timeout") |
| except (httpx.HTTPError, ValueError, TypeError, AttributeError) as exc: |
| |
| |
| logger.warning("PRIMO gateway %s failed (%s)", gateway, type(exc).__name__) |
|
|
| if last_status in (401, 403): |
| return _primo_failure("PRIMO access could not be verified; use the search link", "unauthorized") |
| return _primo_failure("PRIMO API unavailable", "upstream_error") |
|
|
|
|
| def _pubmed_query(req: SearchRequest) -> str: |
| query = req.boolean_query or req.query |
| if req.year_from or req.year_to: |
| year_from, year_to = _year_bounds(req) |
| query = f"({query}) AND ({year_from}:{year_to}[dp])" |
| if req.open_access: |
| query = f"({query}) AND free full text[sb]" |
| if req.resource_type == "books": |
| query = f'({query}) AND "pubmed books"[sb]' |
| elif req.resource_type == "articles": |
| query = f'({query}) NOT "pubmed books"[sb]' |
| return query |
|
|
|
|
| async def search_pubmed(req: SearchRequest) -> dict: |
| """Search PubMed; free availability does not establish an open-access licence.""" |
| import httpx |
|
|
| base = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils" |
| warnings = [] |
| if req.peer_reviewed: |
| warnings.append("PubMed has no general peer-reviewed-only filter; peer review must be checked for each record.") |
| if req.open_access: |
| warnings.append("The PubMed access filter finds free full text; it does not verify an open-access licence.") |
| common = {"source": "PubMed", "warnings": warnings} |
|
|
| def failed(message): |
| return {**common, "error": message, "results": [], "total": 0} |
|
|
| try: |
| async with httpx.AsyncClient(timeout=PUBMED_REQUEST_TIMEOUT) as client: |
| response = await asyncio.wait_for(client.get( |
| f"{base}/esearch.fcgi", |
| params={"db": "pubmed", "term": _pubmed_query(req), "retmax": req.limit, |
| "retmode": "json", "sort": "relevance"}, |
| ), timeout=PUBMED_REQUEST_TIMEOUT) |
| if response.status_code != 200: |
| return failed("PubMed search failed") |
| payload = response.json() |
| esearch = payload.get("esearchresult", {}) |
| if payload.get("error") or esearch.get("ERROR") or esearch.get("errorlist"): |
| return failed("PubMed could not process this search; check the search terms") |
| ids = esearch.get("idlist") |
| if not isinstance(ids, list): |
| return failed("PubMed returned an unreadable search response; try again") |
| query_translation = esearch.get("querytranslation", "") |
| count = int(esearch.get("count", 0)) |
| if not ids: |
| return {**common, "total": count, "results": [], "query_translation": query_translation} |
|
|
| summary = await asyncio.wait_for(client.get( |
| f"{base}/esummary.fcgi", |
| params={"db": "pubmed", "id": ",".join(ids), "retmode": "json"}, |
| ), timeout=PUBMED_REQUEST_TIMEOUT) |
| if summary.status_code != 200: |
| return failed("PubMed fetch failed") |
| summary_payload = summary.json() |
| if summary_payload.get("error"): |
| return failed("PubMed fetch failed") |
| data = summary_payload.get("result", {}) |
| results = [] |
| for pid in ids: |
| record = data.get(pid) |
| if not isinstance(record, dict) or record.get("error"): |
| continue |
| authors_list = record.get("authors") or [] |
| authors = ", ".join(author.get("name", "") for author in authors_list[:3]) |
| if len(authors_list) > 3: |
| authors += " et al." |
| doi = next((article.get("value") for article in record.get("articleids", []) |
| if article.get("idtype") == "doi"), None) |
| if not doi: |
| match = re.search(r"\bdoi:\s*(\S+)", record.get("elocationid", ""), re.IGNORECASE) |
| doi = match.group(1) if match else None |
| results.append({ |
| "title": record.get("title", ""), |
| "creator": authors, |
| "date": record.get("pubdate", ""), |
| "source": record.get("fulljournalname", record.get("source", "")), |
| "doi": doi, |
| "pmid": pid, |
| "link": f"https://pubmed.ncbi.nlm.nih.gov/{pid}/", |
| "type": "; ".join(record.get("pubtype") or []) or "PubMed record", |
| "free_full_text": True if req.open_access else None, |
| "_source": "PubMed", |
| }) |
| if ids and not results: |
| return failed("PubMed returned no readable record details; try again") |
| return {**common, "total": count, "results": results, "query_translation": query_translation} |
| except (httpx.HTTPError, TimeoutError, ValueError, TypeError, AttributeError) as exc: |
| logger.warning("PubMed search failed (%s)", type(exc).__name__) |
| return failed("PubMed is temporarily unavailable; try again") |
|
|
|
|
| @router.post("") |
| async def search_endpoint(req: SearchRequest): |
| """Search PRIMO or PubMed, preferring boolean_query when supplied.""" |
| from src.config import get_settings |
|
|
| if req.source == "pubmed": |
| return await search_pubmed(req) |
| settings = get_settings() |
| return await search_primo(req, settings.primo_api_key, settings.primo_api_region) |
|
|
