"""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
import httpx
from pydantic import BaseModel, ConfigDict, Field
from src.config import get_settings, LIBBEE_VERSION
from src.agentcore.utils import _escape, _strip_resource_noise
logger = logging.getLogger(__name__)
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",
}
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(f"Semantic Scholar {r.status_code} for '{clean_topic}'")
return []
data = r.json()
results = []
for paper in data.get("data", []):
abstract = paper.get("abstract") or ""
if not abstract or len(abstract) < 50:
continue
author_list = paper.get("authors") or []
first_author_last = author_list[0].get("name", "Unknown").split()[-1] if author_list else "Unknown"
year = paper.get("year") or "n.d."
paper_id = paper.get("paperId", "")
doi = (paper.get("externalIds") or {}).get("DOI", "")
pdf_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(a.get("name", "") 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],
"link": link,
"_source": "Semantic Scholar",
})
return results
except Exception as e:
logger.error(f"Semantic Scholar search failed: {e}")
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})"}
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:
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 = work.get("authorships") or []
author_names = [
a.get("author", {}).get("display_name", "")
for a in authorships[:4]
if a.get("author", {}).get("display_name")
]
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 = (work.get("open_access") or {}).get("oa_url") or ""
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,
"link": link,
"_source": "OpenAlex",
})
logger.info(f"OpenAlex: {len(results)} papers for '{clean[:50]}'")
return results
except Exception as e:
logger.warning(f"OpenAlex search failed: {e}")
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:
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 = loc.get("url_for_pdf") or loc.get("url") or ""
except Exception as exc:
logger.debug(f"Unpaywall lookup failed for {doi}: {exc}")
_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(f"Crossref lookup failed for {doi}: {exc}")
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") and not p.get("oa_pdf_url")][:max_lookups]
if not targets:
return papers
oa_results = await asyncio.gather(
*(_unpaywall_oa_url(p["doi"]) for p in targets), return_exceptions=True
)
for p, oa in zip(targets, oa_results):
if isinstance(oa, str) and oa:
p["oa_pdf_url"] = oa
missing_meta = [p for p in targets if not p.get("year") or p.get("year") == "n.d."]
if missing_meta:
cf = await asyncio.gather(
*(_crossref_fill(p["doi"]) for p in missing_meta), return_exceptions=True
)
for p, meta in zip(missing_meta, cf):
if isinstance(meta, dict):
if meta.get("year") and (not p.get("year") or p.get("year") == "n.d."):
p["year"] = meta["year"]
if meta.get("venue") and not p.get("venue"):
p["venue"] = meta["venue"]
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):
papers: List[dict] = []
try:
papers = await _openalex_search(topic, limit=limit)
if not papers:
papers = await _semantic_scholar_search(topic, limit=limit)
if papers:
papers = await enrich_papers_with_oa(papers, max_lookups=max_lookups)
except Exception as exc:
logger.warning(f"Evidence panel failed for '{topic}': {exc}")
return [], ""
if not papers:
return [], ""
rows = ""
for paper in papers[:5]:
_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 = paper.get("link") or (f"https://doi.org/{paper['doi']}" if paper.get("doi") else "")
_oa = paper.get("oa_pdf_url") or ""
_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.