"""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 [], (
'Peer-reviewed filter: '
'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.
'
)
if context is not None and getattr(context, "resource_type", "articles") == "books":
return [], 'Use the prepared PRIMO search for books and e-books.
'
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 [], 'No matching open-index records were available for this response. Use the prepared database searches for more results.
'
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'{_t}'
if _link else f'{_t}'
)
_oa_html = (
f' \u00b7 \U0001f513 OA full text'
if _oa else ""
)
rows += (
f''
f'{_title_html}
'
f'{_meta}{_oa_html}
'
)
html = (
''
'
\U0001f50e Open scholarly evidence
'
'
'
'Top open-index results (OpenAlex / Semantic Scholar) with legal open-access '
'links via Unpaywall and metadata via Crossref where available.
'
f'{rows}