Nature Academic Search

{
  "mcpServers": {
    "academic-search": {
      "command": "uv",
      "args": [
        "run",
        "--no-project",
        "--directory",
        "<MCP_SERVER_DIR>",
        "--with",
        "mcp>=1.0.0,<2.0.0",
        "--with",
        "requests>=2.28.0,<3.0.0",
        "--with",
        "toml>=0.10.2,<2.0.0",
        "--with",
        "lxml>=4.9.0,<6.0.0",
        "--with",
        "pybliometrics>=4.4.1,<5.0.0",
        "python",
        "academic_search_server.py"
      ]
    }
  }
}

{
  "enabledMcpjsonServers": [
    "academic-search"
  ]
}

[academic-search]
trigger = ["查文献", "搜论文", "检索", "导入EndNote", "导出RIS", "检查参考文献", "验证DOI", "验证引用", "相关文献", "找类似论文", "构建检索式", "MeSH", "去重", "search papers", "find articles", "academic search", "literature search", "verify references", "check DOI", "verify citations", "download citation", "export .nbib", "export .ris", "BibTeX", "DOI", "PMID"]
priority = 5
suppress_active_skills = ["lit-process"]

#!/usr/bin/env bash
# Academic Search Skill + MCP Server Installer for Claude Code
# Usage: bash install.sh [PUBMED_EMAIL]
set -euo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
CLAUDE_DIR="${HOME}/.claude"
MCP_TARGET="${CLAUDE_DIR}/mcp_servers/academic-search"
SKILL_TARGET="${CLAUDE_DIR}/skills/academic-search"
MCP_JSON="${CLAUDE_DIR}/.mcp.json"

PUBMED_EMAIL="${1:-user@example.com}"

echo "=== Academic Search Installer ==="
echo "Target: ${CLAUDE_DIR}"
echo "PubMed email: ${PUBMED_EMAIL}"
echo

required_paths=(
    "README.md"
    "SKILL.md"
    "manifest.yaml"
    "static"
    "references"
    "scripts"
    "config"
    "mcp-server"
)

missing_paths=()
for relpath in "${required_paths[@]}"; do
    if [ ! -e "${SCRIPT_DIR}/${relpath}" ]; then
        missing_paths+=("${relpath}")
    fi
done

if [ "${#missing_paths[@]}" -gt 0 ]; then
    echo "ERROR: incomplete nature-academic-search skill directory." >&2
    echo "Missing required path(s):" >&2
    for relpath in "${missing_paths[@]}"; do
        echo "  - ${SCRIPT_DIR}/${relpath}" >&2
    done
    echo "Install the full skills/nature-academic-search folder and retry." >&2
    exit 1
fi

# 1. Install Python dependencies
echo "[1/5] Installing Python dependencies..."
pip install --quiet -r "${SCRIPT_DIR}/mcp-server/requirements.txt" 2>/dev/null || {
    echo "  pip failed, trying pip3..."
    pip3 install --quiet -r "${SCRIPT_DIR}/mcp-server/requirements.txt" 2>/dev/null || {
        echo "  WARNING: Could not install Python deps. Install manually:"
        echo "    pip install -r mcp-server/requirements.txt"
    }
}

# 2. Copy MCP server
echo "[2/5] Copying MCP server..."
mkdir -p "${MCP_TARGET}"
cp -r "${SCRIPT_DIR}/mcp-server/"* "${MCP_TARGET}/"

# 3. Copy Skill
echo "[3/5] Copying Skill..."
mkdir -p "${SKILL_TARGET}"
cp "${SCRIPT_DIR}/README.md" "${SKILL_TARGET}/"
cp "${SCRIPT_DIR}/SKILL.md" "${SKILL_TARGET}/"
cp "${SCRIPT_DIR}/manifest.yaml" "${SKILL_TARGET}/"
cp -r "${SCRIPT_DIR}/static" "${SKILL_TARGET}/"
cp -r "${SCRIPT_DIR}/references" "${SKILL_TARGET}/"
cp -r "${SCRIPT_DIR}/scripts" "${SKILL_TARGET}/"
cp -r "${SCRIPT_DIR}/config" "${SKILL_TARGET}/"

# 4. Merge .mcp.json
echo "[4/5] Configuring .mcp.json..."
if [ -f "${MCP_JSON}" ]; then
    # Check if academic-search already exists
    if grep -q '"academic-search"' "${MCP_JSON}" 2>/dev/null; then
        echo "  academic-search already in .mcp.json, skipping merge."
    else
        # Inject into existing mcpServers object
        python3 -c "
import json, sys
with open('${MCP_JSON}', 'r') as f:
    cfg = json.load(f)
cfg.setdefault('mcpServers', {})['academic-search'] = {
    'command': 'python3',
    'args': ['${MCP_TARGET}/academic_search_server.py'],
    'env': {'PUBMED_EMAIL': '${PUBMED_EMAIL}'}
}
with open('${MCP_JSON}', 'w') as f:
    json.dump(cfg, f, indent=2)
    f.write('\n')
print('  Merged academic-search into existing .mcp.json')
"
    fi
else
    cat > "${MCP_JSON}" <<MCPJSON
{
  "mcpServers": {
    "academic-search": {
      "command": "python3",
      "args": ["${MCP_TARGET}/academic_search_server.py"],
      "env": {
        "PUBMED_EMAIL": "${PUBMED_EMAIL}"
      }
    }
  }
}
MCPJSON
    echo "  Created new .mcp.json"
fi

# 5. Enable in settings.json
echo "[5/5] Enabling in settings.json..."
SETTINGS_JSON="${CLAUDE_DIR}/settings.json"
if [ -f "${SETTINGS_JSON}" ]; then
    python3 -c "
import json
with open('${SETTINGS_JSON}', 'r') as f:
    cfg = json.load(f)
enabled = cfg.setdefault('enabledMcpjsonServers', [])
if 'academic-search' not in enabled:
    enabled.append('academic-search')
    with open('${SETTINGS_JSON}', 'w') as f:
        json.dump(cfg, f, indent=2)
        f.write('\n')
    print('  Added academic-search to enabledMcpjsonServers')
else:
    print('  academic-search already enabled')
"
else
    echo '  WARNING: settings.json not found. Manually add "academic-search" to enabledMcpjsonServers.'
fi

echo
echo "=== Done ==="
echo
echo "Installed:"
echo "  MCP server : ${MCP_TARGET}/"
echo "  Skill      : ${SKILL_TARGET}/"
echo
echo "Next steps:"
echo "  1. Restart Claude Code (or /clear)"
echo "  2. Set your PubMed email in config.toml or PUBMED_EMAIL env var"
echo "  3. (Optional) Add NCBI_API_KEY for higher rate limits"
echo "  4. Test: ask Claude 'search papers about CRISPR'"
echo
echo "Optional: copy triggers to your data/triggers.toml"
echo "  See: config/triggers-academic-search.toml"

name: nature-academic-search
version: 2.0.0
description: >
  Declarative manifest for the static/dynamic split. SKILL.md uses this to
  decide which fragments to load for a literature-search request. The main axis
  is the workflow: the user's need maps to one of five coordinated workflows,
  each already a self-contained file.

# Note on axis paths: the five workflow files live in references/workflows/ and
# cross-reference the shared modules with ../ relative links. The workflow axis
# therefore points at them in place rather than moving them into
# static/fragments/, which would break those internal links. nature-academic-
# search does not use the prose-oriented _shared layer.

always_load:
  - static/core/tools.md
  - static/core/routing-and-ops.md

axes:
  workflow:
    detect: |
      Map the user's need to one workflow. Multiple may apply for a combined
      request (for example search then export); load each that applies.
        multi-source-search   — find literature across PubMed/CrossRef/arXiv and more
        citation-verification — verify or check citations extracted from a document
        mesh-strategy         — build a MeSH/PubMed search strategy
        citation-file-mgmt    — convert/manage .nbib/.ris/.bib citation files
        reference-mgmt        — BibTeX, related-article discovery, ID conversion
    values:
      multi-source-search:   references/workflows/wf1-multi-source-search.md
      citation-verification: references/workflows/wf2-citation-verification.md
      mesh-strategy:         references/workflows/wf3-mesh-strategy.md
      citation-file-mgmt:    references/workflows/wf4-citation-file-mgmt.md
      reference-mgmt:        references/workflows/wf5-reference-mgmt.md
    multi: true

references:
  on_demand:
    - condition: full source reliability tiers (T1/T2/T3) and fallback routing rules
      path: references/source-tiers.md
    - condition: deduplication across sources (used by WFs 1, 2, 5a)
      path: references/dedup-engine.md
    - condition: extracting citations from documents (WF 2)
      path: references/citation-parser.md
    - condition: query construction, source selection, and result ranking
      path: references/search-strategy.md
    - condition: RIS/BibTeX format specifications and field mappings
      path: references/ris-bibtex-format.md
    - condition: no-MCP fallback discovery search (OpenAlex, stdlib) when the MCP server is unavailable
      path: scripts/academic_search.py
    - condition: multi-source .nbib/.ris/.bib downloading and conversion
      path: scripts/format-converter.py
    - condition: pre-flight check that API endpoints are reachable before batch operations
      path: scripts/preflight.py

"""Academic search MCP server.

Unified entry point exposing four tools:
  - search_papers: multi-source concurrent search
  - get_paper_by_id: fetch details by DOI / PMID / arXiv ID
  - get_citation: formatted citation via CrossRef content negotiation
  - lookup_mesh: MeSH descriptor lookup
"""

from __future__ import annotations

import asyncio
import json
import re
from typing import Any

from mcp.server import FastMCP

from sources import ArxivSource, CrossRefSource, PubMedSource
from utils import AcademicSearchError, DataSourceError, setup_logging

mcp = FastMCP("academic-search")
logger = setup_logging()

# Warn if PubMed email is not configured
from utils.config import Config
_cfg = Config()
if not _cfg.pubmed_email:
    logger.warning(
        "PUBMED_EMAIL not set — PubMed requests will run without contact metadata. "
        "Set PUBMED_EMAIL or config.toml [pubmed] email for NCBI best practice."
    )

# Singleton source instances (shared across tool calls)
_crossref = CrossRefSource()
_pubmed = PubMedSource()
_arxiv = ArxivSource()


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _detect_id_type(id: str) -> str:
    """Auto-detect identifier type.

    Returns one of: "doi", "pmid", "arxiv".
    Raises ValueError when detection fails.
    """
    id = id.strip()
    if id.startswith("10.") and "/" in id:
        return "doi"
    if re.match(r"^\d{7,8}$", id):
        return "pmid"
    if re.match(r"^\d{4}\.\d{4,5}(v\d+)?$", id):
        return "arxiv"
    raise ValueError(f"Cannot detect ID type for: {id}")


def _resolve_id_type(id: str, id_type: str) -> str:
    """Resolve the effective ID type.

    If id_type is "auto", delegate to _detect_id_type.
    Otherwise normalise the explicit type string.
    """
    if id_type == "auto":
        return _detect_id_type(id)
    normalised = id_type.lower().strip()
    if normalised in ("doi", "pmid", "arxiv"):
        return normalised
    raise ValueError(f"Unsupported id_type: {id_type}")


def _json_ok(data: Any) -> str:
    """Serialize a successful result to JSON string."""
    return json.dumps(data, ensure_ascii=False, indent=2)


def _json_error(message: str, source: str | None = None) -> str:
    """Serialize an error result to JSON string."""
    payload: dict[str, Any] = {"error": message}
    if source:
        payload["source"] = source
    return json.dumps(payload, ensure_ascii=False, indent=2)


# ---------------------------------------------------------------------------
# Async wrappers for synchronous sources
# ---------------------------------------------------------------------------

async def _search_crossref(query: str, rows: int, filter_type: str | None) -> dict:
    return await asyncio.to_thread(_crossref.search, query, rows, filter_type)


async def _search_pubmed(query: str, rows: int) -> dict:
    return await asyncio.to_thread(_pubmed.search, query, rows)


async def _search_arxiv(query: str, rows: int) -> dict:
    return await asyncio.to_thread(_arxiv.search, query, rows)


async def _search_all(
    query: str,
    sources: list[str],
    rows: int,
    filter_type: str | None,
) -> dict:
    """Dispatch concurrent searches and merge results."""
    tasks: list[asyncio.Task] = []
    source_order: list[str] = []

    if "crossref" in sources:
        tasks.append(asyncio.create_task(_search_crossref(query, rows, filter_type)))
        source_order.append("crossref")
    if "pubmed" in sources:
        tasks.append(asyncio.create_task(_search_pubmed(query, rows)))
        source_order.append("pubmed")
    if "arxiv" in sources:
        tasks.append(asyncio.create_task(_search_arxiv(query, rows)))
        source_order.append("arxiv")

    if not tasks:
        return {"total": 0, "results": [], "errors": []}

    outcomes = await asyncio.gather(*tasks, return_exceptions=True)

    merged_results: list[dict] = []
    errors: list[dict] = []
    total = 0

    for src, outcome in zip(source_order, outcomes):
        if isinstance(outcome, BaseException):
            logger.error("Source %s failed: %s", src, outcome)
            errors.append({"source": src, "error": str(outcome)})
            continue
        total += outcome.get("total", 0)
        merged_results.extend(outcome.get("results", []))

    return {
        "total": total,
        "sources_queried": source_order,
        "result_count": len(merged_results),
        "results": merged_results,
        "errors": errors if errors else None,
    }


# ---------------------------------------------------------------------------
# Tools
# ---------------------------------------------------------------------------

@mcp.tool()
async def search_papers(
    query: str,
    sources: list[str] | None = None,
    rows: int = 5,
    type: str | None = None,
) -> str:
    """Search academic papers across multiple sources (CrossRef, PubMed, arXiv).

    Args:
        query: Search keywords or query string.
        sources: List of source names to query. Defaults to all three.
        rows: Number of results per source (max 50).
        type: Optional CrossRef work type filter (e.g. "journal-article").

    Returns:
        JSON string with total count, merged results, and any per-source errors.
    """
    if not query or not query.strip():
        return _json_error("Empty search query")

    if sources is None:
        sources = ["crossref", "pubmed", "arxiv"]

    # Validate source names
    valid_sources = {"crossref", "pubmed", "arxiv"}
    invalid = [s for s in sources if s not in valid_sources]
    if invalid:
        return _json_error(f"Invalid sources: {invalid}. Valid: {sorted(valid_sources)}")

    rows = max(1, min(rows, 50))

    logger.info("search_papers called", extra={
        "tool": "search_papers",
        "query": query,
        "sources": sources,
        "rows": rows,
    })

    try:
        result = await _search_all(query, sources, rows, type)
    except Exception as exc:
        logger.exception("search_papers failed")
        return _json_error(f"Search failed: {exc}")

    return _json_ok(result)


@mcp.tool()
def get_paper_by_id(id: str, id_type: str = "auto") -> str:
    """Get paper details by identifier (DOI, PMID, or arXiv ID).

    Args:
        id: Paper identifier. Auto-detected if id_type is "auto":
            - Starts with "10." -> DOI (CrossRef)
            - 7-8 digit number -> PMID (PubMed)
            - YYMM.NNNNN format -> arXiv ID (arXiv)
        id_type: Force identifier type ("doi", "pmid", "arxiv", or "auto").

    Returns:
        JSON string with detailed paper metadata.
    """
    if not id or not id.strip():
        return _json_error("Empty identifier")

    try:
        resolved_type = _resolve_id_type(id, id_type)
    except ValueError as exc:
        return _json_error(str(exc))

    logger.info("get_paper_by_id called", extra={
        "tool": "get_paper_by_id",
        "id": id,
        "id_type": resolved_type,
    })

    try:
        if resolved_type == "doi":
            result = _crossref.get_by_doi(id.strip())
        elif resolved_type == "pmid":
            result = _pubmed.get_by_pmid(id.strip())
        elif resolved_type == "arxiv":
            result = _arxiv.get_by_id(id.strip())
        else:
            return _json_error(f"Unsupported ID type: {resolved_type}")
    except DataSourceError as exc:
        logger.error("get_paper_by_id failed: %s", exc)
        return _json_error(str(exc), source=exc.source)
    except Exception as exc:
        logger.exception("get_paper_by_id failed unexpectedly")
        return _json_error(f"Unexpected error: {exc}")

    return _json_ok(result)


@mcp.tool()
def get_citation(id: str, id_type: str = "auto", style: str = "apa") -> str:
    """Get formatted citation for a paper.

    Uses CrossRef content negotiation for DOI-based citations.
    For PMID/arXiv IDs, fetches metadata first then generates a basic citation.

    Args:
        id: Paper identifier (DOI, PMID, or arXiv ID).
        id_type: Force identifier type ("doi", "pmid", "arxiv", or "auto").
        style: Citation style. Supported: apa, nature, ieee, harvard,
               vancouver, chicago, mla.

    Returns:
        JSON string with the formatted citation.
    """
    if not id or not id.strip():
        return _json_error("Empty identifier")

    try:
        resolved_type = _resolve_id_type(id, id_type)
    except ValueError as exc:
        return _json_error(str(exc))

    logger.info("get_citation called", extra={
        "tool": "get_citation",
        "id": id,
        "id_type": resolved_type,
        "style": style,
    })

    try:
        if resolved_type == "doi":
            citation = _crossref.get_citation(id.strip(), style=style)
            return _json_ok({"id": id, "style": style, "citation": citation})

        # For non-DOI IDs, fetch metadata and build a basic citation
        if resolved_type == "pmid":
            paper = _pubmed.get_by_pmid(id.strip())
        elif resolved_type == "arxiv":
            paper = _arxiv.get_by_id(id.strip())
        else:
            return _json_error(f"Unsupported ID type: {resolved_type}")

        citation = _format_basic_citation(paper, style)
        return _json_ok({"id": id, "style": style, "citation": citation})

    except DataSourceError as exc:
        logger.error("get_citation failed: %s", exc)
        return _json_error(str(exc), source=exc.source)
    except Exception as exc:
        logger.exception("get_citation failed unexpectedly")
        return _json_error(f"Unexpected error: {exc}")


def _format_basic_citation(paper: dict, style: str) -> str:
    """Generate a basic citation string from unified paper metadata.

    This is a fallback for non-DOI papers where CrossRef content
    negotiation is not available.
    """
    authors = paper.get("authors", [])
    title = paper.get("title", "Untitled")
    year = paper.get("year", "n.d.")
    journal = paper.get("journal", "")
    doi = paper.get("doi", "")
    arxiv_id = paper.get("arxiv_id", "")
    pmid = paper.get("pmid", "")

    # Author formatting
    if len(authors) > 3:
        author_str = f"{authors[0]} et al."
    elif authors:
        author_str = ", ".join(authors)
    else:
        author_str = "Unknown"

    if style == "nature":
        parts = [f"{author_str}. {title}."]
        if journal:
            parts.append(f" *{journal}*.")
        if year:
            parts.append(f" ({year}).")
        if doi:
            parts.append(f" https://doi.org/{doi}")
        return "".join(parts)

    if style == "ieee":
        ref = f"{author_str}, \"{title}\""
        if journal:
            ref += f", *{journal}*"
        if year:
            ref += f", {year}"
        ref += "."
        if doi:
            ref += f" doi: {doi}."
        return ref

    # Default APA-like
    parts = [f"{author_str} ({year}). {title}."]
    if journal:
        parts.append(f" *{journal}*.")
    if doi:
        parts.append(f" https://doi.org/{doi}")
    elif arxiv_id:
        parts.append(f" arXiv:{arxiv_id}")
    elif pmid:
        parts.append(f" PMID:{pmid}")
    return "".join(parts)


@mcp.tool()
def lookup_mesh(term: str) -> str:
    """Lookup MeSH (Medical Subject Headings) terms.

    Queries the MeSH database via NCBI E-utilities to find matching
    descriptor names and unique IDs.

    Args:
        term: Search term to look up in the MeSH vocabulary.

    Returns:
        JSON string with matching MeSH descriptors.
    """
    if not term or not term.strip():
        return _json_error("Empty MeSH lookup term")

    logger.info("lookup_mesh called", extra={
        "tool": "lookup_mesh",
        "term": term,
    })

    try:
        result = _pubmed.lookup_mesh(term.strip())
    except DataSourceError as exc:
        logger.error("lookup_mesh failed: %s", exc)
        return _json_error(str(exc), source=exc.source)
    except Exception as exc:
        logger.exception("lookup_mesh failed unexpectedly")
        return _json_error(f"Unexpected error: {exc}")

    return _json_ok(result)


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    mcp.run(transport="stdio")

[pubmed]
email = ""  # Set your PubMed email here or via PUBMED_EMAIL env var
api_key = ""
cache_dir = "~/.cache/academic-search"
cache_ttl = 86400

[crossref]
mailto = ""
timeout = 15

[arxiv]
timeout = 30

[general]
default_rows = 5
max_rows = 50

mcp>=1.0.0,<2.0.0
requests>=2.28.0,<3.0.0
toml>=0.10.2,<2.0.0
lxml>=4.9.0,<6.0.0
pybliometrics>=4.4.1,<5.0.0

"""Data source modules for academic search."""

from .crossref import CrossRefSource
from .pubmed import PubMedSource
from .arxiv import ArxivSource
from .scopus import ScopusSource
from .sciencedirect import ScienceDirectSource

__all__ = [
    "CrossRefSource",
    "PubMedSource",
    "ArxivSource",
    "ScopusSource",
    "ScienceDirectSource",
]

"""arXiv data source via REST API (Atom XML feed)."""

import re
import time
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET
from datetime import datetime

from utils.config import get_config
from utils.errors import DataSourceError
from utils.logging import setup_logging

ARXIV_API_URL = "https://export.arxiv.org/api/query"

ARXIV_NS = {
    "atom": "http://www.w3.org/2005/Atom",
    "arxiv": "http://arxiv.org/schemas/atom",
}

_MIN_REQUEST_INTERVAL = 3.0  # seconds between requests
_SOURCE_NAME = "arxiv"

logger = setup_logging("INFO")


class ArxivSource:
    """arXiv search and retrieval via the Atom API."""

    def __init__(self):
        self._last_request_time: float = 0.0
        self._timeout: int | None = None

    # ------------------------------------------------------------------
    # Public interface
    # ------------------------------------------------------------------

    def search(
        self,
        query: str,
        rows: int = 5,
        categories: list[str] | None = None,
        date_from: str | None = None,
        date_to: str | None = None,
    ) -> dict:
        """Search arXiv.

        Parameters
        ----------
        query : str
            Free-text search query.
        rows : int
            Max number of results to return.
        categories : list[str] | None
            arXiv categories to restrict to (e.g. ["cs.AI", "cs.LG"]).
        date_from : str | None
            Start date in YYYY-MM-DD format.
        date_to : str | None
            End date in YYYY-MM-DD format.

        Returns
        -------
        dict
            {"results": [...], "total": int, "source": "arxiv"}
        """
        search_query = self._build_query(query, categories)
        date_filter = self._build_date_filter(date_from, date_to)

        params = {
            "search_query": search_query,
            "start": 0,
            "max_results": rows,
            "sortBy": "relevance",
            "sortOrder": "descending",
        }
        if date_filter:
            params["search_query"] = f"{search_query}+AND+{date_filter}"

        raw = self._request(params)
        results = self._parse_feed(raw)
        return {
            "results": results,
            "total": len(results),
            "source": _SOURCE_NAME,
        }

    def get_by_id(self, arxiv_id: str) -> dict:
        """Retrieve a single paper by arXiv ID.

        Parameters
        ----------
        arxiv_id : str
            arXiv identifier, e.g. "2401.12345" or "2401.12345v1".

        Returns
        -------
        dict
            Paper record in unified format.
        """
        clean_id = self._normalize_id(arxiv_id)
        params = {
            "id_list": clean_id,
            "max_results": 1,
        }
        raw = self._request(params)
        results = self._parse_feed(raw)
        if not results:
            raise DataSourceError(
                _SOURCE_NAME,
                f"Paper not found: {arxiv_id}",
            )
        return results[0]

    # ------------------------------------------------------------------
    # Query construction
    # ------------------------------------------------------------------

    @staticmethod
    def _build_query(user_query: str, categories: list[str] | None = None) -> str:
        """Build the search_query parameter.

        Combines user query with optional category restrictions.
        Returns a string with +AND+/+OR+ connectors (URL-ready).
        """
        parts: list[str] = []
        parts.append(f"({user_query})")
        if categories:
            cat_expr = " OR ".join(f"cat:{cat}" for cat in categories)
            parts.append(f"({cat_expr})")
        combined = " AND ".join(parts)
        # Replace connectors and spaces for URL embedding
        combined = (
            combined.replace(" AND ", "+AND+")
            .replace(" OR ", "+OR+")
            .replace(" ", "+")
        )
        return combined

    @staticmethod
    def _build_date_filter(
        date_from: str | None, date_to: str | None
    ) -> str:
        """Build submittedDate filter expression.

        The arXiv API requires the literal +TO+ syntax and 14-digit
        timestamps in the format YYYYMMDDHHMM.

        Returns empty string when no date bounds are given.
        """
        if not date_from and not date_to:
            return ""

        # Default boundaries
        start_ts = "000000000000"
        end_ts = "999912312359"

        if date_from:
            start_ts = _date_to_ts(date_from)
        if date_to:
            end_ts = _date_to_ts(date_to, end_of_day=True)

        return f"submittedDate:[{start_ts}+TO+{end_ts}]"

    # ------------------------------------------------------------------
    # HTTP
    # ------------------------------------------------------------------

    def _request(self, params: dict) -> str:
        """Execute an HTTP GET to the arXiv API with rate limiting."""
        self._enforce_rate_limit()
        url = f"{ARXIV_API_URL}?{urllib.parse.urlencode(params)}"
        timeout = self._get_timeout()
        logger.debug("arXiv request: %s", url)

        try:
            req = urllib.request.Request(url)
            req.add_header("User-Agent", "academic-search/1.0")
            with urllib.request.urlopen(req, timeout=timeout) as resp:
                return resp.read().decode("utf-8")
        except urllib.error.HTTPError as exc:
            if exc.code in (429, 503):
                raise DataSourceError(
                    _SOURCE_NAME,
                    f"Rate limited or unavailable (HTTP {exc.code})",
                    original_error=exc,
                ) from exc
            raise DataSourceError(
                _SOURCE_NAME,
                f"HTTP error {exc.code}: {exc.reason}",
                original_error=exc,
            ) from exc
        except urllib.error.URLError as exc:
            raise DataSourceError(
                _SOURCE_NAME,
                f"Network error: {exc.reason}",
                original_error=exc,
            ) from exc
        except TimeoutError as exc:
            raise DataSourceError(
                _SOURCE_NAME,
                f"Request timed out after {timeout}s",
                original_error=exc,
            ) from exc

    def _enforce_rate_limit(self) -> None:
        elapsed = time.monotonic() - self._last_request_time
        if elapsed < _MIN_REQUEST_INTERVAL:
            time.sleep(_MIN_REQUEST_INTERVAL - elapsed)
        self._last_request_time = time.monotonic()

    def _get_timeout(self) -> int:
        if self._timeout is None:
            self._timeout = get_config().arxiv_timeout
        return self._timeout

    # ------------------------------------------------------------------
    # XML parsing
    # ------------------------------------------------------------------

    def _parse_feed(self, xml_text: str) -> list[dict]:
        """Parse arXiv Atom XML into a list of unified result dicts."""
        try:
            root = ET.fromstring(xml_text)
        except ET.ParseError as exc:
            raise DataSourceError(
                _SOURCE_NAME,
                f"Malformed XML response: {exc}",
                original_error=exc,
            ) from exc

        entries: list[dict] = []
        for entry in root.findall("atom:entry", ARXIV_NS):
            parsed = self._parse_entry(entry)
            if parsed:
                entries.append(parsed)
        return entries

    def _parse_entry(self, entry: ET.Element) -> dict | None:
        """Extract a single paper record from an Atom <entry>."""
        arxiv_id_raw = _text(entry, "atom:id", ARXIV_NS)
        if not arxiv_id_raw:
            return None

        title = _text(entry, "atom:title", ARXIV_NS) or ""
        title = re.sub(r"\s+", " ", title).strip()

        summary = _text(entry, "atom:summary", ARXIV_NS) or ""
        summary = re.sub(r"\s+", " ", summary).strip()

        authors = [
            name
            for name in (
                _text(author, "atom:name", ARXIV_NS)
                for author in entry.findall("atom:author", ARXIV_NS)
            )
            if name
        ]

        primary_cat = entry.find("arxiv:primary_category", ARXIV_NS)
        categories = []
        if primary_cat is not None:
            term = primary_cat.get("term")
            if term:
                categories.append(term)
        # Also collect secondary categories from atom:category elements
        for cat in entry.findall("atom:category", ARXIV_NS):
            scheme = cat.get("scheme", "")
            term = cat.get("term", "")
            if term and "arxiv" in scheme.lower() and term not in categories:
                categories.append(term)

        published = _text(entry, "atom:published", ARXIV_NS) or ""
        year = _extract_year(published)

        pdf_url = ""
        for link in entry.findall("atom:link", ARXIV_NS):
            if link.get("title") == "pdf":
                pdf_url = link.get("href", "")
                break

        arxiv_id = self._normalize_id(arxiv_id_raw)

        return {
            "title": title,
            "authors": authors,
            "year": year,
            "arxiv_id": arxiv_id,
            "categories": categories,
            "abstract": summary,
            "pdf_url": pdf_url,
            "source": _SOURCE_NAME,
        }

    # ------------------------------------------------------------------
    # Helpers
    # ------------------------------------------------------------------

    @staticmethod
    def _normalize_id(raw: str) -> str:
        """Strip URL prefix and version suffix from an arXiv ID.

        http://arxiv.org/abs/2401.12345v1 -> 2401.12345
        """
        arxiv_id = raw.strip()
        # Remove URL prefix
        match = re.search(r"(\d{4}\.\d{4,5})(v\d+)?$", arxiv_id)
        if match:
            return match.group(1)
        # Fallback: strip known prefixes
        for prefix in ("https://arxiv.org/abs/", "http://arxiv.org/abs/"):
            if arxiv_id.startswith(prefix):
                arxiv_id = arxiv_id[len(prefix):]
                break
        # Strip version
        arxiv_id = re.sub(r"v\d+$", "", arxiv_id)
        return arxiv_id


# ------------------------------------------------------------------
# Module-level helpers
# ------------------------------------------------------------------


def _text(parent: ET.Element, xpath: str, ns: dict) -> str | None:
    """Return stripped text of a sub-element, or None."""
    el = parent.find(xpath, ns)
    if el is not None and el.text:
        return el.text.strip()
    return None


def _date_to_ts(date_str: str, end_of_day: bool = False) -> str:
    """Convert YYYY-MM-DD to YYYYMMDDHHMM (14-digit timestamp).

    Parameters
    ----------
    date_str : str
        Date in YYYY-MM-DD format.
    end_of_day : bool
        If True, use 2359 as HHMM; otherwise 0000.
    """
    dt = datetime.strptime(date_str, "%Y-%m-%d")
    suffix = "2359" if end_of_day else "0000"
    return dt.strftime("%Y%m%d") + suffix


def _extract_year(published: str) -> int | None:
    """Extract year from ISO datetime string (e.g. 2024-01-15T...)."""
    try:
        return int(published[:4])
    except (ValueError, IndexError):
        return None

"""CrossRef data source for academic search."""

from urllib.parse import quote

import requests

from utils.config import get_config
from utils.errors import DataSourceError

CROSSREF_API = "https://api.crossref.org"


class CrossRefSource:
    """CrossRef API wrapper with unified result format."""

    SOURCE_NAME = "crossref"

    def __init__(self):
        config = get_config()
        mailto = config.crossref_mailto or "user@example.com"
        self._headers = {
            "User-Agent": f"ClaudeCode-MCP-Crossref/1.0 (mailto:{mailto})",
        }
        self._timeout = config.crossref_timeout

    # ------------------------------------------------------------------
    # Public API
    # ------------------------------------------------------------------

    def search(
        self, query: str, rows: int = 5, filter_type: str | None = None
    ) -> dict:
        """Search CrossRef works.

        Args:
            query: Keywords, author name, title, DOI prefix, etc.
            rows: Number of results (max 50).
            filter_type: Optional work type filter, e.g. "journal-article".

        Returns:
            {"total": int, "results": [unified_result, ...]}
        """
        params: dict = {"query": query, "rows": min(rows, 50)}
        if filter_type:
            params["filter"] = f"type:{filter_type}"

        data = self._request("/works", params=params)
        items = data.get("items", [])
        total = data.get("total-results", 0)

        results = [self._normalize_search_item(item) for item in items]
        return {"total": total, "results": results}

    def get_by_doi(self, doi: str) -> dict:
        """Get detailed metadata for a single work by DOI.

        Args:
            doi: Digital Object Identifier (e.g. "10.1038/nature12373").

        Returns:
            Unified result dict with extra fields (abstract, volume, etc.).
        """
        data = self._request(f"/works/{quote(doi, safe='/')}")
        return self._normalize_detail_item(data)

    def get_citation(self, doi: str, style: str = "apa") -> str:
        """Return a formatted citation string via CrossRef content negotiation.

        Args:
            doi: Digital Object Identifier.
            style: Citation style (apa, nature, vancouver, ieee, etc.).

        Returns:
            Formatted citation string.
        """
        url = f"{CROSSREF_API}/works/{quote(doi, safe='/')}/transform"
        headers = {
            **self._headers,
            "Accept": f"text/x-bibliography; style={style}",
        }
        try:
            resp = requests.get(url, headers=headers, timeout=self._timeout)
        except requests.RequestException as exc:
            raise DataSourceError(
                self.SOURCE_NAME,
                f"Network error fetching citation for {doi}: {exc}",
                original_error=exc,
            ) from exc

        if resp.status_code == 404:
            return f"Citation not available for DOI: {doi}"
        if resp.status_code == 406:
            raise DataSourceError(
                self.SOURCE_NAME,
                f"Unsupported citation style: {style}",
            )

        try:
            resp.raise_for_status()
        except requests.HTTPError as exc:
            raise DataSourceError(
                self.SOURCE_NAME,
                f"HTTP {resp.status_code} fetching citation for {doi}",
                original_error=exc,
            ) from exc

        return resp.text.strip()

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _request(self, path: str, params: dict | None = None) -> dict:
        """Issue GET to CrossRef API and return the ``message`` payload."""
        url = f"{CROSSREF_API}{path}"
        try:
            resp = requests.get(
                url, params=params, headers=self._headers, timeout=self._timeout
            )
            resp.raise_for_status()
        except requests.HTTPError as exc:
            status = exc.response.status_code if exc.response is not None else "?"
            raise DataSourceError(
                self.SOURCE_NAME,
                f"HTTP {status} from {url}",
                original_error=exc,
            ) from exc
        except requests.RequestException as exc:
            raise DataSourceError(
                self.SOURCE_NAME,
                f"Network error calling {url}: {exc}",
                original_error=exc,
            ) from exc

        return resp.json().get("message", {})

    # ------------------------------------------------------------------
    # Normalization
    # ------------------------------------------------------------------

    @staticmethod
    def _extract_authors(author_list: list[dict], limit: int = 0) -> list[str]:
        """Convert CrossRef author entries to ``["Given Family", ...]`` list.

        Args:
            author_list: Raw ``author`` array from CrossRef.
            limit: Max authors to include; 0 = all.
        """
        subset = author_list[:limit] if limit else author_list
        names = [
            f"{a.get('given', '')} {a.get('family', '')}".strip()
            for a in subset
        ]
        if limit and len(author_list) > limit:
            names.append("et al.")
        return names

    @staticmethod
    def _extract_year(item: dict) -> int | None:
        """Best-effort publication year extraction."""
        for key in ("published-print", "published-online", "created"):
            parts = item.get(key, {}).get("date-parts", [[None]])
            year = parts[0][0] if parts and parts[0] else None
            if year is not None:
                return year
        return None

    def _normalize_search_item(self, item: dict) -> dict:
        """Map a CrossRef work item to the unified search result format."""
        return {
            "title": (item.get("title") or [""])[0],
            "authors": self._extract_authors(item.get("author", []), limit=5),
            "year": self._extract_year(item),
            "doi": item.get("DOI"),
            "journal": (item.get("container-title") or [""])[0],
            "source": self.SOURCE_NAME,
            "citation_count": item.get("is-referenced-by-count", 0),
        }

    def _normalize_detail_item(self, item: dict) -> dict:
        """Map a CrossRef work item to the unified detail result format."""
        base = self._normalize_search_item(item)
        base.update({
            "authors": self._extract_authors(item.get("author", [])),
            "abstract": item.get("abstract", ""),
            "volume": item.get("volume", ""),
            "issue": item.get("issue", ""),
            "pages": item.get("page", ""),
            "publisher": item.get("publisher", ""),
            "type": item.get("type"),
            "references_count": item.get("references-count", 0),
            "url": item.get("URL"),
        })
        return base

"""Shared helpers for pybliometrics-backed Elsevier sources."""

from __future__ import annotations

from pathlib import Path
from threading import Lock
from typing import Any

from pybliometrics import init as pybliometrics_init
from pybliometrics.utils.constants import CONFIG_FILE

from utils.errors import DataSourceError


_init_lock = Lock()
_initialised = False


def ensure_pybliometrics_config(source: str) -> None:
    """Initialise pybliometrics from its configured file.

    pybliometrics creates a config interactively when the file is absent.  MCP
    servers cannot prompt, so missing or invalid config is reported explicitly.
    """
    config_path = Path(CONFIG_FILE)
    if not config_path.exists():
        raise DataSourceError(
            source,
            f"pybliometrics config not found at {config_path}",
        )

    global _initialised
    with _init_lock:
        if _initialised:
            return
        try:
            pybliometrics_init(config_path=config_path)
        except (FileNotFoundError, ValueError) as exc:
            raise DataSourceError(
                source,
                f"pybliometrics config is invalid at {config_path}: {exc}",
                original_error=exc,
            ) from exc
        _initialised = True


def record_to_dict(value: Any) -> Any:
    """Recursively convert pybliometrics records into JSON-safe structures."""
    if value is None or isinstance(value, (str, int, float, bool)):
        return value
    if hasattr(value, "_asdict"):
        return {k: record_to_dict(v) for k, v in value._asdict().items()}
    if isinstance(value, dict):
        return {str(k): record_to_dict(v) for k, v in value.items()}
    if isinstance(value, (list, tuple, set)):
        return [record_to_dict(v) for v in value]
    return value


def safe_attr(obj: Any, name: str) -> Any:
    """Read an optional pybliometrics property."""
    try:
        return getattr(obj, name)
    except (AttributeError, KeyError, TypeError):
        return None


def split_semicolon(value: str | None) -> list[str]:
    """Split pybliometrics semicolon-joined fields."""
    if not value:
        return []
    return [item.strip() for item in value.split(";") if item.strip()]


def year_from_date(value: str | None) -> int | None:
    """Extract a publication year from an ISO-like date string."""
    if not value or len(value) < 4:
        return None
    try:
        return int(value[:4])
    except ValueError:
        return None

"""PubMed data source via NCBI E-utilities API."""

from __future__ import annotations

import time
import xml.etree.ElementTree as ET
from typing import Any

import requests

from utils.config import get_config
from utils.errors import DataSourceError
from utils.logging import setup_logging

logger = setup_logging()

BASE_URL = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/"
SOURCE_NAME = "pubmed"

# Rate limit: 3 req/s without key, 10 req/s with key
_REQ_INTERVAL_WITH_KEY = 0.11
_REQ_INTERVAL_WITHOUT_KEY = 0.35

_last_request_ts: float = 0.0


def _throttle(api_key: str) -> None:
    """Enforce NCBI rate limits."""
    global _last_request_ts
    interval = _REQ_INTERVAL_WITH_KEY if api_key else _REQ_INTERVAL_WITHOUT_KEY
    elapsed = time.monotonic() - _last_request_ts
    if elapsed < interval:
        time.sleep(interval - elapsed)
    _last_request_ts = time.monotonic()


def _get(endpoint: str, params: dict[str, Any], timeout: int = 30) -> requests.Response:
    """Send GET request to NCBI E-utilities with throttling and error handling."""
    cfg = get_config()
    api_key = cfg.pubmed_api_key

    _throttle(api_key)

    merged = dict(params)
    if cfg.pubmed_email:
        merged["email"] = cfg.pubmed_email
    if api_key:
        merged["api_key"] = api_key

    url = BASE_URL + endpoint
    try:
        resp = requests.get(url, params=merged, timeout=timeout)
        resp.raise_for_status()
    except requests.Timeout as exc:
        raise DataSourceError(SOURCE_NAME, f"Request timed out: {url}", exc) from exc
    except requests.HTTPError as exc:
        status = exc.response.status_code if exc.response is not None else "?"
        raise DataSourceError(
            SOURCE_NAME, f"HTTP {status} from {url}", exc
        ) from exc
    except requests.RequestException as exc:
        raise DataSourceError(SOURCE_NAME, f"Request failed: {url}", exc) from exc

    return resp


def _parse_article(article: ET.Element) -> dict[str, Any]:
    """Parse a single PubmedArticle XML element into the unified result dict."""
    citation = article.find("MedlineCitation")
    if citation is None:
        raise DataSourceError(SOURCE_NAME, "Missing MedlineCitation in article XML")

    pmid_el = citation.find("PMID")
    pmid = pmid_el.text.strip() if pmid_el is not None and pmid_el.text else ""

    art = citation.find("Article")
    if art is None:
        raise DataSourceError(SOURCE_NAME, f"Missing Article for PMID {pmid}")

    # Title
    title_el = art.find("ArticleTitle")
    title = title_el.text.strip() if title_el is not None and title_el.text else ""

    # Authors
    authors: list[str] = []
    author_list = art.find("AuthorList")
    if author_list is not None:
        for author in author_list.findall("Author"):
            last = author.find("LastName")
            fore = author.find("ForeName")
            if last is not None and last.text:
                name = last.text.strip()
                if fore is not None and fore.text:
                    name = f"{name} {fore.text.strip()}"
                authors.append(name)
            elif (collective := author.find("CollectiveName")) is not None and collective.text:
                authors.append(collective.text.strip())

    # Abstract
    abstract_parts: list[str] = []
    abstract_el = art.find("Abstract")
    if abstract_el is not None:
        for text_el in abstract_el.findall("AbstractText"):
            label = text_el.get("Label", "")
            content = "".join(text_el.itertext()).strip()
            if label and content:
                abstract_parts.append(f"{label}: {content}")
            elif content:
                abstract_parts.append(content)
    abstract = " ".join(abstract_parts)

    # Journal
    journal_el = art.find("Journal")
    journal = ""
    year = None
    if journal_el is not None:
        title_el = journal_el.find("Title")
        if title_el is not None and title_el.text:
            journal = title_el.text.strip()
        # Year from JournalIssue/PubDate
        issue = journal_el.find("JournalIssue")
        if issue is not None:
            pub_date = issue.find("PubDate")
            if pub_date is not None:
                year_el = pub_date.find("Year")
                if year_el is not None and year_el.text:
                    try:
                        year = int(year_el.text.strip())
                    except ValueError:
                        pass
                if year is None:
                    medline_date = pub_date.find("MedlineDate")
                    if medline_date is not None and medline_date.text:
                        # Extract first 4-digit year from string like "2024 Jan-Feb"
                        import re

                        m = re.search(r"\d{4}", medline_date.text)
                        if m:
                            year = int(m.group())

    # DOI
    doi = ""
    for eloi in art.findall("ELocationID"):
        if eloi.get("EIdType") == "doi" and eloi.text:
            doi = eloi.text.strip()
            break

    return {
        "title": title,
        "authors": authors,
        "year": year,
        "pmid": pmid,
        "doi": doi,
        "journal": journal,
        "abstract": abstract,
        "source": SOURCE_NAME,
    }


class PubMedSource:
    """PubMed data source providing search, fetch, and MeSH lookup."""

    def search(
        self,
        query: str,
        rows: int = 5,
        sort: str = "relevance",
    ) -> dict[str, Any]:
        """Search PubMed and return structured results.

        Args:
            query: PubMed search query string.
            rows: Number of results to return.
            sort: Sort order -- "relevance" (Best Match) or "date".

        Returns:
            Dict with keys: total, query, results (list of unified result dicts).
        """
        if not query or not query.strip():
            raise DataSourceError(SOURCE_NAME, "Empty search query")

        cfg = get_config()
        rows = min(rows, cfg.max_rows)
        sort_param = "relevance" if sort == "relevance" else "pub_date"

        # Step 1: esearch to get WebEnv + query_key
        search_params: dict[str, Any] = {
            "db": "pubmed",
            "term": query.strip(),
            "retmax": rows,
            "usehistory": "y",
            "retmode": "xml",
            "sort": sort_param,
        }
        resp = _get("esearch.fcgi", search_params)
        root = ET.fromstring(resp.content)

        count_el = root.find("Count")
        total = int(count_el.text) if count_el is not None and count_el.text else 0

        web_env_el = root.find("WebEnv")
        query_key_el = root.find("QueryKey")

        if web_env_el is None or query_key_el is None or not web_env_el.text or not query_key_el.text:
            # No results
            return {"total": 0, "query": query, "results": []}

        web_env = web_env_el.text.strip()
        query_key = query_key_el.text.strip()

        # Step 2: efetch to get article details
        fetch_params: dict[str, Any] = {
            "db": "pubmed",
            "query_key": query_key,
            "WebEnv": web_env,
            "retmax": rows,
            "retmode": "xml",
            "rettype": "abstract",
        }
        resp = _get("efetch.fcgi", fetch_params)
        fetch_root = ET.fromstring(resp.content)

        results: list[dict[str, Any]] = []
        for article in fetch_root.findall("PubmedArticle"):
            try:
                results.append(_parse_article(article))
            except DataSourceError as exc:
                logger.warning("Failed to parse article: %s", exc)
                continue

        return {"total": total, "query": query, "results": results}

    def get_by_pmid(self, pmid: str) -> dict[str, Any]:
        """Fetch a single article by PMID.

        Args:
            pmid: PubMed ID (numeric string).

        Returns:
            Unified result dict for the article.

        Raises:
            DataSourceError: If PMID is invalid or article not found.
        """
        if not pmid or not pmid.strip().isdigit():
            raise DataSourceError(SOURCE_NAME, f"Invalid PMID: {pmid}")

        fetch_params: dict[str, Any] = {
            "db": "pubmed",
            "id": pmid.strip(),
            "retmode": "xml",
            "rettype": "abstract",
        }
        resp = _get("efetch.fcgi", fetch_params)
        root = ET.fromstring(resp.content)

        article = root.find("PubmedArticle")
        if article is None:
            raise DataSourceError(SOURCE_NAME, f"PMID {pmid} not found")

        return _parse_article(article)

    def lookup_mesh(self, term: str) -> dict[str, Any]:
        """Look up a MeSH descriptor by term.

        Queries the MeSH database via E-utilities to find matching
        descriptor names and unique IDs.

        Args:
            term: Search term to look up in MeSH.

        Returns:
            Dict with keys: term, results (list of {name, mesh_id, ui}).
        """
        if not term or not term.strip():
            raise DataSourceError(SOURCE_NAME, "Empty MeSH lookup term")

        # Use esearch on MeSH database
        search_params: dict[str, Any] = {
            "db": "mesh",
            "term": term.strip(),
            "retmax": 10,
            "retmode": "xml",
        }
        resp = _get("esearch.fcgi", search_params)
        root = ET.fromstring(resp.content)

        id_list = root.find("IdList")
        if id_list is None or len(id_list) == 0:
            return {"term": term, "results": []}

        ids = [id_el.text.strip() for id_el in id_list.findall("Id") if id_el.text]

        if not ids:
            return {"term": term, "results": []}

        # efetch from mesh db to get descriptor details
        fetch_params: dict[str, Any] = {
            "db": "mesh",
            "id": ",".join(ids),
            "retmode": "xml",
        }
        resp = _get("efetch.fcgi", fetch_params)
        fetch_root = ET.fromstring(resp.content)

        results: list[dict[str, str]] = []
        for descriptor in fetch_root.findall(".//DescriptorRecord"):
            name_el = descriptor.find("DescriptorName/String")
            ui_el = descriptor.find("DescriptorUI")
            name = name_el.text.strip() if name_el is not None and name_el.text else ""
            ui = ui_el.text.strip() if ui_el is not None and ui_el.text else ""
            if name:
                results.append({"name": name, "mesh_id": ui, "ui": ui})

        return {"term": term, "results": results}

"""ScienceDirect data source via pybliometrics."""

from __future__ import annotations

from typing import Any

from pybliometrics.utils import URLS, get_content

from utils.errors import DataSourceError

from .elsevier_common import ensure_pybliometrics_config, year_from_date


SOURCE_NAME = "sciencedirect"


class ScienceDirectSource:
    """pybliometrics-backed ScienceDirect metadata operations."""

    SOURCE_NAME = SOURCE_NAME

    def search(
        self,
        query: str,
        rows: int = 5,
        view: str | None = None,
        refresh: bool | int = False,
    ) -> dict[str, Any]:
        """Search ScienceDirect article metadata and return the requested page."""
        if not query or not query.strip():
            raise DataSourceError(SOURCE_NAME, "Empty search query")

        rows = max(1, min(rows, 50))

        def run() -> dict[str, Any]:
            data = self._search_api(
                "ScienceDirectSearch",
                {
                    "query": query.strip(),
                    "count": rows,
                    "start": 0,
                    "view": view or "STANDARD",
                },
            )
            records = _search_entries(data)
            return {
                "total": _total_results(data),
                "query": query,
                "source": SOURCE_NAME,
                "results": [self._normalize_search_entry(r) for r in records[:rows]],
            }

        return self._run("ScienceDirect search", run)

    def get_article_metadata(
        self,
        query: str,
        rows: int = 5,
        view: str | None = None,
        refresh: bool | int = False,
    ) -> dict[str, Any]:
        """Retrieve ScienceDirect article metadata using a metadata API query."""
        if not query or not query.strip():
            raise DataSourceError(SOURCE_NAME, "Empty article metadata query")

        rows = max(1, min(rows, 50))

        def run() -> dict[str, Any]:
            data = self._search_api(
                "ArticleMetadata",
                {
                    "query": query.strip(),
                    "count": rows,
                    "start": 0,
                    "view": view or "STANDARD",
                },
            )
            records = _search_entries(data)
            return {
                "total": _total_results(data),
                "query": query,
                "source": SOURCE_NAME,
                "results": [self._normalize_metadata_entry(r) for r in records[:rows]],
            }

        return self._run("ScienceDirect article metadata retrieval", run)

    def _run(self, operation: str, callback):
        ensure_pybliometrics_config(SOURCE_NAME)
        try:
            return callback()
        except DataSourceError:
            raise
        except Exception as exc:
            raise DataSourceError(
                SOURCE_NAME,
                f"{operation} failed: {exc}",
                original_error=exc,
            ) from exc

    @staticmethod
    def _search_api(api: str, params: dict[str, Any]) -> dict[str, Any]:
        response = get_content(URLS[api], api, params=params)
        return response.json()

    @staticmethod
    def _normalize_search_entry(data: dict[str, Any]) -> dict[str, Any]:
        links = _links(data.get("link"))
        return {
            "title": data.get("dc:title"),
            "authors": _authors(data),
            "year": year_from_date(data.get("prism:coverDate")),
            "doi": _doi(data),
            "pii": data.get("pii"),
            "journal": data.get("prism:publicationName"),
            "volume": data.get("prism:volume"),
            "pages": _join_pages(data.get("prism:startingPage"), data.get("prism:endingPage")),
            "openaccess_status": data.get("openaccess"),
            "link": links.get("scidir"),
            "api_link": links.get("self") or data.get("prism:url"),
            "source": SOURCE_NAME,
        }

    @staticmethod
    def _normalize_metadata_entry(data: dict[str, Any]) -> dict[str, Any]:
        links = _links(data.get("link"))
        return {
            "title": data.get("dc:title"),
            "authors": _authors(data),
            "year": year_from_date(data.get("prism:coverDate")),
            "doi": _doi(data),
            "eid": data.get("eid"),
            "pii": data.get("pii"),
            "abstract": data.get("dc:description"),
            "journal": data.get("prism:publicationName"),
            "pages": _join_pages(data.get("prism:startingPage"), data.get("prism:endingPage")),
            "aggregation_type": data.get("prism:aggregationType"),
            "publication_type": data.get("prism:publicationType"),
            "author_keywords": data.get("authkeywords"),
            "openaccess_status": data.get("openaccess"),
            "link": links.get("scidir") or links.get("self"),
            "api_link": links.get("self") or data.get("prism:url"),
            "source": SOURCE_NAME,
        }


def _search_entries(data: dict[str, Any]) -> list[dict[str, Any]]:
    entries = data.get("search-results", {}).get("entry", [])
    return [
        entry
        for entry in _as_list(entries)
        if isinstance(entry, dict)
        and (entry.get("dc:title") or entry.get("prism:doi") or entry.get("pii"))
    ]


def _total_results(data: dict[str, Any]) -> int:
    total = data.get("search-results", {}).get("opensearch:totalResults", 0)
    return _int_or_none(total) or 0


def _as_list(value: Any) -> list[Any]:
    if value is None:
        return []
    if isinstance(value, list):
        return value
    return [value]


def _authors(data: dict[str, Any]) -> list[str]:
    authors = []
    author_data = data.get("authors", {}).get("author", [])
    for item in _as_list(author_data):
        name = item.get("$") if isinstance(item, dict) else item
        if name:
            authors.append(name)
    creator = data.get("dc:creator")
    if isinstance(creator, list):
        authors.extend(item.get("$") for item in creator if isinstance(item, dict))
    elif isinstance(creator, str) and not authors:
        authors.append(creator)
    return [a for a in authors if a]


def _links(value: Any) -> dict[str, str]:
    out = {}
    for item in _as_list(value):
        if not isinstance(item, dict):
            continue
        ref = item.get("@ref") or item.get("rel")
        href = item.get("@href") or item.get("href")
        if ref and href:
            out[ref] = href
    return out


def _doi(data: dict[str, Any]) -> str | None:
    if data.get("prism:doi"):
        return data["prism:doi"]
    identifier = data.get("dc:identifier")
    if isinstance(identifier, str) and identifier.startswith("doi:"):
        return identifier[4:]
    return None


def _join_pages(start: str | None, end: str | None) -> str:
    if start and end:
        return f"{start}-{end}"
    return start or end or ""


def _int_or_none(value: Any) -> int | None:
    try:
        return int(value)
    except (TypeError, ValueError):
        return None