| """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__) |
|
|
| |
| _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_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")] |
| |
| 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 |
|
|
|
|
| |
| |
| |
| |
| |
|
|
| 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 |
| |
| |
| 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"))] |
| |
| 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 |
|
|
