Blame for src/api/search.py

"""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()
# 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


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")
        # Primo reserves semicolons for additional field clauses, even when
        # percent-encoded; accepting one here changes the meaning of a query.
        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]:
    # PNX link fields can contain $$U<url>$$D<label>, rather than a bare URL.
    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"))
        # 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 _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}")

    # 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={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:
                # 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=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
                # 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 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)