openalex.py

"""
OpenAlex enrichment — journal metadata + peer percentile calculation.
Free API, no key required. Uses polite pool with mailto header.

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
from typing import Optional
from utils import norm_issn

UA = "ResearchNavigator/1.0 (mailto:library@ku.ac.ae)"
BASE = "https://api.openalex.org"

_http_client: Optional[httpx.AsyncClient] = None


def init_client() -> None:
    global _http_client
    _http_client = httpx.AsyncClient(
        timeout=httpx.Timeout(10.0, connect=5.0),
        headers={"User-Agent": UA},
    )


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]:
    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
        return {
            "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"),
        }
    except Exception as e:
        print(f"[OpenAlex] enrich error for {name}: {e}")
        return None


async def compute_percentile(
    openalex: dict,
    concept_cache: Optional[dict] = None,
) -> Optional[dict]:
    try:
        if not openalex or not openalex.get("top_concept_id"):
            return None
        val = openalex.get("two_yr_mean_citedness")
        if val is None:
            return None

        concept_id = str(openalex["top_concept_id"]).split("/")[-1]

        if concept_cache is not None and concept_id in concept_cache:
            peers = concept_cache[concept_id]
        else:
            client = _get_client()
            r = await client.get(
                f"{BASE}/sources",
                params={
                    "filter": f"concepts.id:{concept_id},type:journal",
                    "per_page": 200,
                    "sort": "summary_stats.2yr_mean_citedness:desc",
                },
            )
            if not r.is_success:
                return None
            peers = [
                s.get("summary_stats", {}).get("2yr_mean_citedness")
                for s in r.json().get("results", [])
            ]
            peers = [v for v in peers if isinstance(v, (int, float))]
            if concept_cache is not None:
                concept_cache[concept_id] = peers

        if len(peers) < 10:
            return None

        below = sum(1 for v in peers if v < val)
        pct = round(below / len(peers) * 100)
        q = "Q1" if pct >= 75 else "Q2" if pct >= 50 else "Q3" if pct >= 25 else "Q4"
        return {
            "percentile": pct,
            "quartile": q,
            "peer_count": len(peers),
            "concept": openalex.get("top_concept"),
        }
    except Exception as e:
        print(f"[OpenAlex] percentile error: {e}")
        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]))