Upload 5 files

src/api/search.py

@@ -3,14 +3,16 @@ import asyncio
 import logging
 import re
 from typing import Optional
-from urllib.parse import quote
+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 = 5.0
+# One deadline covers all gateways, rather than one wait per region. A region
+# selected in HF's PRIMO_API_REGION avoids even the fast discovery attempts.
+PRIMO_REQUEST_TIMEOUT = 6.0
 PUBMED_REQUEST_TIMEOUT = 8.0
 
 
@@ -19,7 +21,9 @@ class SearchRequest(BaseModel):
     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=5, ge=1, le=20)
+    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}$")
@@ -41,9 +45,11 @@ class SearchRequest(BaseModel):
         return self
 
 
-def _boolean_to_primo_params(boolean_query: str) -> str:
+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."""
-    return f"q=any,contains,{quote(boolean_query.strip(), safe='')}"
+    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]:
@@ -51,10 +57,15 @@ def _year_bounds(req: SearchRequest) -> tuple[str, str]:
     return req.year_from or "0001", req.year_to or "9999"
 
 
-def _build_primo_search_link(title: str, record_id: Optional[str] = None) -> Optional[str]:
+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:
-        return f"{base}/fulldisplay?docid={quote(str(record_id), safe='')}&vid=971KUOSTAR_INST:KU"
+        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
@@ -74,17 +85,105 @@ def _first(value, default=""):
 def _result_link(raw_link: str, fallback: Optional[str]) -> Optional[str]:
     # PNX link fields can contain $$U<url>$$D<label>, rather than a bare URL.
     match = re.match(r"(?:\$\$U)?(https?://[^\s]+?)(?:\$\$|$)", raw_link)
-    return match.group(1) if match else fallback
-
-
-async def search_primo(req: SearchRequest, api_key: str) -> dict:
+    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"))
+        # A corrupt/empty item is not a real result to pad the requested count.
+        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 {"error": "PRIMO_API_KEY not configured", "results": [], "total": 0, "source": "PRIMO"}
+        return _primo_failure("PRIMO results are not configured", "not_configured", "not_configured")
 
-    query_params = _boolean_to_primo_params(req.boolean_query or req.query)
+    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")
@@ -96,75 +195,77 @@ async def search_primo(req: SearchRequest, api_key: str) -> dict:
     if req.resource_type != "both":
         facets.append(f"facet_rtype,exact,{req.resource_type}")
 
+    # Multiple qInclude values in the same category are ANDed, so an explicit
+    # Books OR Articles request must use multiFacets for that category.
+    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"
+
     # qInclude accepts one value, joining ANDed facets with |,|. Repeated
     # qInclude keys are not its documented API contract.
     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=rank&limit={req.limit}&offset=0"
-        f"{facet_params}"
+        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 region in ("api-eu", "api-na", "api-ap"):
-            url = f"https://{region}.hosted.exlibrisgroup.com/primo/v1/search{qs}"
+        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:
                 # httpx read timeouts reset per chunk. wait_for additionally
                 # bounds the complete request, including a slowly streamed body.
                 response = await asyncio.wait_for(
                     client.get(url, headers={"Accept": "application/json", "Authorization": f"apikey {api_key}"}),
-                    timeout=PRIMO_REQUEST_TIMEOUT,
+                    timeout=remaining,
                 )
                 if response.status_code != 200:
-                    logger.warning("PRIMO region %s returned HTTP %s", region, response.status_code)
+                    last_status = response.status_code
+                    logger.warning("PRIMO gateway %s returned HTTP %s", gateway, response.status_code)
                     if response.status_code == 429:
-                        return {"error": "PRIMO request limit reached; try again later", "results": [], "total": 0, "source": "PRIMO"}
+                        return _primo_failure("PRIMO request limit reached; try again later", "rate_limited", "rate_limited")
                     if response.status_code == 400:
-                        return {"error": "PRIMO could not process this search; check the search terms", "results": [], "total": 0, "source": "PRIMO"}
-                    continue
+                        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 = []
-                for doc in data.get("docs", []):
-                    if not isinstance(doc, dict):
-                        continue
-                    pnx = doc.get("pnx") or {}
-                    if not isinstance(pnx, dict):
-                        continue
-                    display = pnx.get("display") or {}
-                    addata = pnx.get("addata") or {}
-                    search = pnx.get("search") or {}
-                    control = pnx.get("control") or {}
-                    links = pnx.get("links") or {}
-                    record_id = _first(control.get("recordid"), None)
-                    title = _first(display.get("title"), "Untitled")
-                    primo_url = _build_primo_search_link(title, record_id)
-                    creators = display.get("creator") or display.get("contributor") or []
-                    if isinstance(creators, str):
-                        creators = [creators]
-                    raw_link = _first(links.get("openurl") or links.get("linktorsrc"))
-                    results.append({
-                        "record_id": record_id,
-                        "title": title,
-                        "creator": "; ".join(str(creator) for creator in creators) or "Unknown",
-                        "date": _first(search.get("creationdate") or addata.get("risdate") or addata.get("date")),
-                        "type": _first(display.get("type")),
-                        "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": _first(display.get("oa")) == "free_for_read",
-                        "_source": "PRIMO",
-                    })
-                return {"total": (data.get("info") or {}).get("total", 0), "results": results, "source": "PRIMO"}
-            except (httpx.HTTPError, TimeoutError, ValueError, TypeError, AttributeError) as exc:
+                results = _primo_records(data, req.limit)
+                info = data.get("info") if isinstance(data.get("info"), dict) else {}
+                raw_total = info.get("total")
+                total = None
+                # Unknown totals must stay unknown: a short page is not evidence
+                # that those records exhaust all matches in the catalogue.
+                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:
                 # Exception strings from HTTP clients can include the full URL,
                 # including the API key. Never put them in logs or responses.
-                logger.warning("PRIMO region %s failed (%s)", region, type(exc).__name__)
+                logger.warning("PRIMO gateway %s failed (%s)", gateway, type(exc).__name__)
 
-    return {"error": "PRIMO API unavailable", "results": [], "total": 0, "source": "PRIMO"}
+    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:
@@ -268,4 +369,5 @@ async def search_endpoint(req: SearchRequest):
 
     if req.source == "pubmed":
         return await search_pubmed(req)
-    return await search_primo(req, get_settings().primo_api_key)
+    settings = get_settings()
+    return await search_primo(req, settings.primo_api_key, settings.primo_api_region)