Blame for src/agentcore/scholarly.py

"""Open scholarly-index clients: Semantic Scholar, OpenAlex, Crossref and Unpaywall (OA links)."""
import asyncio
import html
import json
import logging
import os
import re
import time
import uuid
from datetime import datetime
from typing import Any, Dict, List, Literal, Optional, Tuple
from urllib.parse import quote, urlsplit

import httpx
from pydantic import BaseModel, ConfigDict, Field

from src.config import get_settings, LIBBEE_VERSION

from src.agentcore.utils import _escape, _get_runtime_config, _strip_resource_noise

logger = logging.getLogger(__name__)

# Optional enrichment must not hold the entire chat response open indefinitely.
_EVIDENCE_BUDGET_SECONDS = 8.0
_INDEX_TIMEOUT_SECONDS = 3.0
_ENRICHMENT_TIMEOUT_SECONDS = 2.0


def _safe_http_url(value: str) -> str:
    value = str(value or "").strip()
    try:
        parsed = urlsplit(value)
        return value if parsed.scheme in {"http", "https"} and parsed.netloc and not parsed.username else ""
    except ValueError:
        return ""


async def _semantic_scholar_search(topic: str, limit: int = 8) -> List[dict]:
    clean_topic = _strip_resource_noise(topic)
    if not clean_topic:
        clean_topic = topic
    try:
        params = {
            "query": clean_topic,
            "limit": limit,
            "fields": "title,authors,year,venue,abstract,externalIds,openAccessPdf,citationCount",
        }
        async with httpx.AsyncClient(timeout=15) as client:
            r = await client.get("https://api.semanticscholar.org/graph/v1/paper/search", params=params)
            if r.status_code != 200:
                logger.warning("Semantic Scholar status: %s", r.status_code)
                return []
            data = r.json()
            results = []
            for paper in data.get("data", []):
                if not isinstance(paper, dict):
                    continue
                abstract = paper.get("abstract") or ""
                if not abstract or len(abstract) < 50:
                    continue
                author_list = [a for a in (paper.get("authors") or []) if isinstance(a, dict)]
                first_name = str(author_list[0].get("name") or "Unknown").split() if author_list else []
                first_author_last = first_name[-1] if first_name else "Unknown"
                year = paper.get("year") or "n.d."
                paper_id = paper.get("paperId", "")
                doi = (paper.get("externalIds") or {}).get("DOI", "")
                pdf_url = _safe_http_url((paper.get("openAccessPdf") or {}).get("url", ""))
                link = (
                    pdf_url
                    or (f"https://doi.org/{doi}" if doi else "")
                    or (f"https://www.semanticscholar.org/paper/{paper_id}" if paper_id else "#")
                )
                results.append({
                    "title": paper.get("title", "Untitled"),
                    "authors": (
                        ", ".join(str(a.get("name") or "") for a in author_list[:3])
                        + (" et al." if len(author_list) > 3 else "")
                    ),
                    "first_author": first_author_last,
                    "year": year,
                    "citation_key": f"{first_author_last}, {year}",
                    "venue": paper.get("venue", ""),
                    "abstract": abstract[:700],
                    "doi": doi,
                    "oa_pdf_url": pdf_url,
                    "cited_by_count": paper.get("citationCount") or 0,
                    "link": link,
                    "_source": "Semantic Scholar",
                })
            return results
    except Exception as e:
        logger.error(f"Semantic Scholar search failed: {type(e).__name__}")
        return []


async def _openalex_search(topic: str, limit: int = 8) -> List[dict]:
    try:
        from urllib.parse import quote as _oa_quote
        clean = re.sub(r'\b(AND|OR|NOT)\b|[()"\'\\]', ' ', topic, flags=re.IGNORECASE)
        clean = re.sub(r'\s+', ' ', clean).strip()[:200]
        encoded = _oa_quote(clean, safe='')
        select = "id,title,authorships,publication_year,primary_location,abstract_inverted_index,doi,cited_by_count,open_access"
        url = (
            f"https://api.openalex.org/works"
            f"?search={encoded}"
            f"&filter=has_abstract:true,type:article,is_paratext:false"
            f"&sort=relevance_score:desc"
            f"&per_page={limit}"
            f"&select={select}"
            f"&mailto={quote(get_settings().contact_email, safe='')}"
        )
        _contact = get_settings().contact_email
        headers = {"User-Agent": f"LibBee/{LIBBEE_VERSION} (mailto:{_contact})"}
        api_key = getattr(get_settings(), "openalex_api_key", "")
        if api_key:
            headers["Authorization"] = f"Bearer {api_key}"
        async with httpx.AsyncClient(timeout=10) as client:
            r = await client.get(url, headers=headers)
            if r.status_code != 200:
                logger.warning(f"OpenAlex search failed: {r.status_code}")
                return []
            works = r.json().get("results", [])
            results = []
            for work in works:
                if not isinstance(work, dict):
                    continue
                title = work.get("title") or "Untitled"
                inv = work.get("abstract_inverted_index") or {}
                abstract = ""
                if inv:
                    word_positions = []
                    for word, positions in inv.items():
                        for pos in positions:
                            word_positions.append((pos, word))
                    word_positions.sort(key=lambda x: x[0])
                    abstract = " ".join(w for _, w in word_positions)[:700]
                if not abstract:
                    continue
                authorships = [a for a in (work.get("authorships") or []) if isinstance(a, dict)]
                author_names = [
                    str((a.get("author") or {}).get("display_name") or "").strip()
                    for a in authorships[:4]
                    if str((a.get("author") or {}).get("display_name") or "").strip()
                ]
                authors_str = ", ".join(author_names[:3])
                if len(author_names) > 3:
                    authors_str += " et al."
                first_last = author_names[0].split()[-1] if author_names else "Unknown"
                year = str(work.get("publication_year") or "n.d.")
                doi = (work.get("doi") or "").replace("https://doi.org/", "")
                oa_url = _safe_http_url((work.get("open_access") or {}).get("oa_url"))
                link = (
                    oa_url if oa_url
                    else (f"https://doi.org/{doi}" if doi else "")
                    or work.get("id", "https://openalex.org")
                )
                venue = (
                    ((work.get("primary_location") or {}).get("source") or {})
                    .get("display_name") or ""
                )
                results.append({
                    "title": title,
                    "authors": authors_str or "Unknown",
                    "first_author": first_last,
                    "year": year,
                    "citation_key": f"{first_last}, {year}",
                    "venue": venue,
                    "abstract": abstract,
                    "doi": doi,
                    "oa_pdf_url": oa_url,
                    "cited_by_count": int(work.get("cited_by_count") or 0),
                    "link": link,
                    "_source": "OpenAlex",
                })
            logger.debug("OpenAlex returned %s records", len(results))
            return results
    except Exception as e:
        logger.warning(f"OpenAlex search failed: {type(e).__name__}")
        return []




# ── Unpaywall + Crossref enrichment ──────────────────────────────────────────
# Unpaywall (https://unpaywall.org/products/api) resolves a DOI to its best
# legal open-access location. Crossref fills publication-year / venue gaps.
# Both are free, keyless "polite pool" APIs identified by contact e-mail.
# Every call is best-effort: failures never break the response.

_UNPAYWALL_TTL = 24 * 60 * 60
_unpaywall_cache: Dict[str, Any] = {}


async def _unpaywall_oa_url(doi: str) -> str:
    """Return the best legal OA PDF/landing URL for a DOI, or '' if none/unknown."""
    doi = (doi or "").strip().lower().removeprefix("https://doi.org/")
    if not doi:
        return ""
    now = time.time()
    hit = _unpaywall_cache.get(doi)
    if hit and now - hit[0] < (_UNPAYWALL_TTL if hit[1] else 60):
        return hit[1]
    url = ""
    try:
        email = get_settings().contact_email
        async with httpx.AsyncClient(timeout=4) as client:
            r = await client.get(
                f"https://api.unpaywall.org/v2/{quote(doi, safe='')}",
                params={"email": email},
            )
            if r.status_code == 200:
                loc = (r.json() or {}).get("best_oa_location") or {}
                url = _safe_http_url(loc.get("url_for_pdf") or loc.get("url"))
    except Exception as exc:
        logger.debug("Unpaywall lookup failed: %s", type(exc).__name__)
    if len(_unpaywall_cache) >= 2048:
        oldest = min(_unpaywall_cache, key=lambda key: _unpaywall_cache[key][0])
        _unpaywall_cache.pop(oldest, None)
    _unpaywall_cache[doi] = (now, url)
    return url


async def _crossref_fill(doi: str) -> dict:
    """Return {'year': int|None, 'venue': str} from Crossref for a DOI (best effort)."""
    doi = (doi or "").strip().lower().removeprefix("https://doi.org/")
    if not doi:
        return {}
    try:
        _contact = get_settings().contact_email
        headers = {"User-Agent": f"LibBee/{LIBBEE_VERSION} (mailto:{_contact})"}
        async with httpx.AsyncClient(timeout=4) as client:
            r = await client.get(f"https://api.crossref.org/works/{quote(doi, safe='')}",
                                 headers=headers)
            if r.status_code != 200:
                return {}
            msg = (r.json() or {}).get("message") or {}
            year = None
            issued = ((msg.get("issued") or {}).get("date-parts") or [[None]])[0]
            if issued and issued[0]:
                year = issued[0]
            venue = (msg.get("container-title") or [""])[0]
            return {"year": year, "venue": venue}
    except Exception as exc:
        logger.debug("Crossref lookup failed: %s", type(exc).__name__)
        return {}


async def enrich_papers_with_oa(papers: List[dict], max_lookups: int = 5) -> List[dict]:
    """Concurrently attach 'oa_pdf_url' (Unpaywall) and fill missing year/venue
    (Crossref) for papers that carry a DOI. Mutates and returns the list."""
    targets = [p for p in papers if p.get("doi")][:max_lookups]
    oa_targets = [p for p in targets if not p.get("oa_pdf_url")]
    metadata_targets = [p for p in targets if not p.get("year") or p.get("year") == "n.d." or not p.get("venue")]
    # Metadata repair is independent of whether an OA link already exists.
    results = await asyncio.gather(
        *(_unpaywall_oa_url(p["doi"]) for p in oa_targets),
        *(_crossref_fill(p["doi"]) for p in metadata_targets),
        return_exceptions=True,
    )
    for paper, url in zip(oa_targets, results[:len(oa_targets)]):
        if isinstance(url, str) and _safe_http_url(url):
            paper["oa_pdf_url"] = url
    for paper, meta in zip(metadata_targets, results[len(oa_targets):]):
        if not isinstance(meta, dict):
            continue
        if meta.get("year") and (not paper.get("year") or paper.get("year") == "n.d."):
            paper["year"] = meta["year"]
        if meta.get("venue") and not paper.get("venue"):
            paper["venue"] = meta["venue"]
        if paper.get("first_author"):
            paper["citation_key"] = f"{paper['first_author']}, {paper.get('year') or 'n.d.'}"
    return papers


# ── Shared evidence panel ────────────────────────────────────────────────────
# Used by BOTH search handlers (_run_search_mode and _research_snapshot) so the
# panel appears on ordinary searches, not only on "deep research" phrasing.
# Returns (papers, html). Never raises: on any failure returns ([], "") and the
# caller renders exactly its pre-3.8 output.

async def fetch_evidence_panel(topic: str, limit: int = 6, max_lookups: int = 5, context=None):
    try:
        configured_limit = int(_get_runtime_config().get("max_results", get_settings().max_results))
    except (TypeError, ValueError):
        configured_limit = get_settings().max_results
    limit = max(1, min(limit, configured_limit, 20))
    if context is not None and getattr(context, "peer_reviewed", False):
        return [], (
            '<div class="evidence-note" role="note"><strong>Peer-reviewed filter:</strong> '
            'Open-index evidence is not displayed because those records do not provide a '
            'reliable peer-review guarantee. Use the prepared PRIMO search, where the '
            'Peer Reviewed filter is applied.</div>'
        )
    if context is not None and getattr(context, "resource_type", "articles") == "books":
        return [], '<div class="evidence-note">Use the prepared PRIMO search for books and e-books.</div>'
    papers: List[dict] = []
    deadline = time.monotonic() + _EVIDENCE_BUDGET_SECONDS
    # Retry the other provider within a fixed overall budget; the prepared
    # catalogue links remain useful even when both public indexes are slow.
    for search in (_openalex_search, _semantic_scholar_search):
        try:
            remaining = max(0.01, deadline - time.monotonic())
            papers = await asyncio.wait_for(search(topic, limit=min(20, limit * 3)), timeout=min(_INDEX_TIMEOUT_SECONDS, remaining))
        except Exception as exc:
            logger.warning("Open-index lookup unavailable: %s", type(exc).__name__)
            papers = []
        if papers:
            break
    if papers:
        try:
            remaining = max(0.01, deadline - time.monotonic())
            await asyncio.wait_for(enrich_papers_with_oa(papers, max_lookups=max_lookups), timeout=min(_ENRICHMENT_TIMEOUT_SECONDS, remaining))
        except Exception as exc:
            logger.debug("Optional evidence enrichment unavailable: %s", type(exc).__name__)
        if context is not None:
            year_from = getattr(context, "year_from", None)
            year_to = getattr(context, "year_to", None)
            if year_from and str(year_from).isdigit():
                papers = [p for p in papers if str(p.get("year", "")).isdigit() and int(p["year"]) >= int(year_from)]
            if year_to and str(year_to).isdigit():
                papers = [p for p in papers if str(p.get("year", "")).isdigit() and int(p["year"]) <= int(year_to)]
            if getattr(context, "open_access", False):
                papers = [p for p in papers if _safe_http_url(p.get("oa_pdf_url"))]
        # Duplicates should not be presented as independent supporting sources.
        seen = set()
        unique = []
        for paper in papers:
            key = str(paper.get("doi") or paper.get("title") or "").strip().lower()
            if key and key not in seen:
                seen.add(key)
                paper["link"] = _safe_http_url(paper.get("link")) or (f"https://doi.org/{quote(str(paper['doi']), safe='/')}" if paper.get("doi") else "")
                paper["oa_pdf_url"] = _safe_http_url(paper.get("oa_pdf_url"))
                unique.append(paper)
        papers = unique[:limit]
    if not papers:
        return [], '<div class="evidence-note">No matching open-index records were available for this response. Use the prepared database searches for more results.</div>'

    rows = ""
    for paper in papers:
        _t = _escape(str(paper.get("title") or "Untitled"))
        _y = _escape(str(paper.get("year") or "n.d."))
        _v = _escape(str(paper.get("venue") or ""))
        _c = paper.get("cited_by_count")
        _link = _safe_http_url(paper.get("link") or (f"https://doi.org/{paper['doi']}" if paper.get("doi") else ""))
        _oa = _safe_http_url(paper.get("oa_pdf_url"))
        _bits = [b for b in (_y, _v) if b]
        if isinstance(_c, int) and _c > 0:
            _bits.append(f"{_c:,} citations")
        _meta = " \u00b7 ".join(_bits)
        _title_html = (
            f'<a href="{_escape(_link)}" target="_blank" rel="noopener noreferrer" style="color:#003366;font-weight:600">{_t}</a>'
            if _link else f'<span style="font-weight:600">{_t}</span>'
        )
        _oa_html = (
            f' \u00b7 <a href="{_escape(_oa)}" target="_blank" rel="noopener noreferrer" '
            f'style="color:#047857;font-weight:700">\U0001f513 OA full text</a>'
            if _oa else ""
        )
        rows += (
            f'<div style="margin:7px 0;padding-left:2px;font-size:.84rem;line-height:1.55">'
            f'{_title_html}<br>'
            f'<span style="color:#6b7280;font-size:.78rem">{_meta}{_oa_html}</span></div>'
        )
    html = (
        '<div class="evidence-panel">'
        '<div class="evidence-title">\U0001f50e Open scholarly evidence</div>'
        '<div class="evidence-description">'
        'Top open-index results (OpenAlex / Semantic Scholar) with legal open-access '
        'links via Unpaywall and metadata via Crossref where available.</div>'
        f'{rows}</div>'
    )
    return papers, html