Spaces:
Running
Running
| """ | |
| OpenAlex enrichment — journal metadata + peer percentile calculation. | |
| Optional server-side API key raises the available request budget. | |
| The obsolete sample-derived percentile is deliberately not reported. | |
| Changes from v1.0: | |
| - Shared module-level httpx.AsyncClient (init_client / close_client for lifespan) | |
| - compute_percentile accepts an optional concept_cache dict to deduplicate | |
| identical concept peer-set fetches across journals in one request | |
| - _norm_issn removed — uses utils.norm_issn | |
| """ | |
| import httpx | |
| import asyncio | |
| import re | |
| import os | |
| import time | |
| import logging | |
| from collections import OrderedDict | |
| from datetime import datetime, timezone | |
| from typing import Optional | |
| from utils import norm_issn | |
| UA = "ResearchBee/1.8.2 (mailto:library@ku.ac.ae)" | |
| logger = logging.getLogger(__name__) | |
| _last_success = None | |
| _last_error = None | |
| _source_cache = OrderedDict() | |
| BASE = "https://api.openalex.org" | |
| _http_client: Optional[httpx.AsyncClient] = None | |
| async def _record_response(response) -> None: | |
| global _last_success, _last_error | |
| now = datetime.now(timezone.utc).isoformat(timespec="seconds") | |
| if response.is_success: | |
| _last_success = now | |
| _last_error = None | |
| else: | |
| _last_error = {"status": response.status_code, "at": now} | |
| def status() -> dict: | |
| return {"api_key_configured": bool((os.getenv("OPENALEX_API_KEY") or "").strip()), | |
| "last_success": _last_success, "last_error": _last_error, | |
| "availability": "not_checked" if _last_success is None and _last_error is None else ("degraded" if _last_error else "available")} | |
| async def _authorize_request(request) -> None: | |
| """Enforce the provider boundary before attaching its credential.""" | |
| url = request.url | |
| if url.scheme != "https" or url.host != "api.openalex.org" or url.port not in (None, 443): | |
| request.headers.pop("Authorization", None) | |
| raise ValueError("OpenAlex client refuses non-provider origins") | |
| key = (os.getenv("OPENALEX_API_KEY") or "").strip() | |
| if key: | |
| request.headers["Authorization"] = "Bearer " + key | |
| else: | |
| request.headers.pop("Authorization", None) | |
| def _record_failure(exc: Exception) -> None: | |
| global _last_error | |
| _last_error = {"type": type(exc).__name__, | |
| "at": datetime.now(timezone.utc).isoformat(timespec="seconds")} | |
| status_code = getattr(exc, "status_code", None) | |
| if status_code is not None: | |
| _last_error["status"] = status_code | |
| class OpenAlexUnavailable(RuntimeError): | |
| def __init__(self, status_code=None): | |
| self.status_code = status_code | |
| super().__init__("OpenAlex is temporarily unavailable") | |
| def init_client() -> None: | |
| global _http_client | |
| _http_client = httpx.AsyncClient( | |
| timeout=httpx.Timeout(10.0, connect=5.0), | |
| headers={"User-Agent": UA}, follow_redirects=False, | |
| event_hooks={"request": [_authorize_request], "response": [_record_response]}, | |
| ) | |
| async def close_client() -> None: | |
| global _http_client | |
| if _http_client and not _http_client.is_closed: | |
| await _http_client.aclose() | |
| _http_client = None | |
| def _get_client() -> httpx.AsyncClient: | |
| if _http_client is None or _http_client.is_closed: | |
| init_client() | |
| return _http_client | |
| async def enrich_journal(name: str, issn: Optional[str]) -> Optional[dict]: | |
| cache_key = (norm_issn(issn), name.strip().casefold()) | |
| cached = _source_cache.get(cache_key) | |
| if cached and time.monotonic() - cached[0] < 3600: | |
| _source_cache.move_to_end(cache_key) | |
| return dict(cached[1]) | |
| client = _get_client() | |
| try: | |
| if issn: | |
| clean = norm_issn(issn) or issn.strip() | |
| r = await client.get(f"{BASE}/sources/issn:{clean}") | |
| else: | |
| r = await client.get( | |
| f"{BASE}/sources", params={"search": name, "per_page": 1} | |
| ) | |
| if not r.is_success: | |
| return None | |
| d = r.json() | |
| s = d if issn else (d.get("results") or [None])[0] | |
| if not s: | |
| return None | |
| # An ISSN request must be confirmed by the response, not merely the URL. | |
| source_issns = {norm_issn(v) for v in [s.get("issn_l"), *(s.get("issn") or [])] if v} | |
| if issn and norm_issn(issn) not in source_issns: | |
| return None | |
| if not issn: | |
| from researchbee_journals import JournalIndex | |
| if JournalIndex.normalise_identity_title(name) != JournalIndex.normalise_identity_title(s.get("display_name", "")): | |
| return None | |
| result = { | |
| "source_updated_at": s.get("updated_date"), | |
| "retrieved_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), | |
| "openalex_id": s.get("id"), | |
| "display_name": s.get("display_name"), | |
| "issn_l": s.get("issn_l"), | |
| "issns": s.get("issn", []), | |
| "host_organization": s.get("host_organization_name"), | |
| "is_oa": s.get("is_oa"), | |
| "is_in_doaj": s.get("is_in_doaj"), | |
| "works_count": s.get("works_count"), | |
| "cited_by_count": s.get("cited_by_count"), | |
| "h_index": (s.get("summary_stats") or {}).get("h_index"), | |
| "two_yr_mean_citedness": (s.get("summary_stats") or {}).get("2yr_mean_citedness"), | |
| "homepage_url": s.get("homepage_url"), | |
| "top_concept": ((s.get("x_concepts") or [{}])[0]).get("display_name"), | |
| "top_concept_id": ((s.get("x_concepts") or [{}])[0]).get("id"), | |
| } | |
| _source_cache[cache_key] = (time.monotonic(), result) | |
| while len(_source_cache) > 512: | |
| _source_cache.popitem(last=False) | |
| return dict(result) | |
| except Exception as e: | |
| _record_failure(e) | |
| logger.warning("OpenAlex source lookup failed (%s)", type(e).__name__) | |
| return None | |
| async def compute_percentile(openalex: dict, concept_cache: Optional[dict] = None) -> Optional[dict]: | |
| """A truncated high-citation peer sample cannot establish a field quartile.""" | |
| return None | |
| async def enrich_all(journals: list) -> list: | |
| """Enrich a list of journal dicts in parallel, deduplicating concept peer-sets.""" | |
| concept_cache: dict = {} | |
| async def _enrich_one(j: dict) -> dict: | |
| oa = await enrich_journal(j.get("name", ""), j.get("issn")) | |
| pct = await compute_percentile(oa, concept_cache) if oa else None | |
| j["openalex"] = oa | |
| j["oa_percentile"] = pct | |
| if not j.get("issn") and oa: | |
| j["issn"] = oa.get("issn_l") or (oa.get("issns") or [None])[0] | |
| return j | |
| return list(await asyncio.gather(*[_enrich_one(j) for j in journals])) | |
| _EVIDENCE_STOP = { | |
| "this", "that", "with", "from", "into", "using", "used", "study", | |
| "research", "paper", "article", "results", "method", "methods", "based", | |
| "analysis", "approach", "between", "among", "their", "these", "those", | |
| } | |
| def _evidence_terms(text: str) -> set[str]: | |
| words = re.findall(r"[A-Za-z][A-Za-z\-]{3,}", (text or "").lower()) | |
| return {w.strip("-") for w in words if w not in _EVIDENCE_STOP} | |
| def _title_overlap(query: str, titles: list[str]) -> int: | |
| q = _evidence_terms(query) | |
| if not q or not titles: | |
| return 0 | |
| scores = [] | |
| for title in titles: | |
| t = _evidence_terms(title) | |
| if not t: | |
| continue | |
| overlap = len(q & t) | |
| # Dice coefficient, converted to 0..100. OpenAlex already orders the | |
| # search by relevance; this score is only a transparent local signal. | |
| scores.append((2 * overlap / max(len(q) + len(t), 1)) * 100) | |
| return round(max(scores, default=0.0)) | |
| async def similar_works_for_journal( | |
| name: str, | |
| issn: str = "", | |
| query: str = "", | |
| *, | |
| per_page: int = 3, | |
| ) -> dict: | |
| """Return representative recent works as evidence for journal fit. | |
| Failure is explicitly labelled unavailable, separately from no matches. Journal | |
| identity and ranking still come from the local verified database, so an | |
| OpenAlex outage never removes or invents a recommendation. | |
| """ | |
| source = await enrich_journal(name, issn or None) | |
| if not source or not source.get("openalex_id"): | |
| return {"source": source, "works": [], "similarity_score": 0, "status": "unavailable"} | |
| source_id = str(source["openalex_id"]).rsplit("/", 1)[-1] | |
| if not re.fullmatch(r"S\d+", source_id): | |
| return {"source": source, "works": [], "similarity_score": 0, "status": "unavailable"} | |
| current_year = datetime.now(timezone.utc).year | |
| query_text = " ".join((query or "").split())[:500] | |
| client = _get_client() | |
| async def _fetch(use_search: bool) -> list[dict]: | |
| params = { | |
| "filter": ( | |
| f"primary_location.source.id:{source_id},is_retracted:false," | |
| f"from_publication_date:{current_year - 5}-01-01" | |
| ), | |
| "per_page": min(max(per_page, 1), 5), | |
| "select": "id,title,publication_year,doi,cited_by_count,authorships", | |
| } | |
| if use_search and query_text: | |
| params["search"] = query_text | |
| params["sort"] = "relevance_score:desc" | |
| else: | |
| params["sort"] = "cited_by_count:desc" | |
| response = await client.get(f"{BASE}/works", params=params, timeout=6.0) | |
| if not response.is_success: | |
| raise OpenAlexUnavailable(response.status_code) | |
| works = [] | |
| for work in response.json().get("results", []): | |
| title = " ".join(str(work.get("title") or "").split()) | |
| if not title: | |
| continue | |
| authors = [ | |
| (a.get("author") or {}).get("display_name", "") | |
| for a in (work.get("authorships") or [])[:3] | |
| ] | |
| works.append({ | |
| "title": title, | |
| "year": work.get("publication_year"), | |
| "doi": work.get("doi", ""), | |
| "cited_by": work.get("cited_by_count", 0), | |
| "authors": [a for a in authors if a], | |
| "openalex_url": work.get("id", ""), | |
| }) | |
| return works | |
| try: | |
| works = await _fetch(bool(query_text)) | |
| if not works: | |
| works = await _fetch(False) | |
| score = _title_overlap(query_text, [w["title"] for w in works]) | |
| return { | |
| "source": source, | |
| "works": works, | |
| "similarity_score": score, | |
| "status": "available" if works else "no_recent_matches", | |
| } | |
| except Exception as exc: | |
| _record_failure(exc) | |
| logger.warning("OpenAlex evidence unavailable (%s)", type(exc).__name__) | |
| return {"source": source, "works": [], "similarity_score": 0, "status": "unavailable"} | |
| async def evidence_for_journals( | |
| journals: list[dict], | |
| query: str, | |
| *, | |
| limit: int = 12, | |
| per_page: int = 3, | |
| concurrency: int = 5, | |
| ) -> dict[str, dict]: | |
| """Fetch representative works for the strongest verified candidates.""" | |
| semaphore = asyncio.Semaphore(max(1, concurrency)) | |
| async def _one(journal: dict) -> tuple[str, dict]: | |
| key = str(journal.get("id") or journal.get("scopus_id") or journal.get("issn") or journal.get("name")) | |
| async with semaphore: | |
| evidence = await similar_works_for_journal( | |
| journal.get("name", ""), | |
| journal.get("issn", ""), | |
| query, | |
| per_page=per_page, | |
| ) | |
| return key, evidence | |
| selected = journals[: max(0, limit)] | |
| pairs = await asyncio.gather(*[_one(j) for j in selected]) if selected else [] | |
| return {key: evidence for key, evidence in pairs} | |