oaworks.py

"""
oaworks.py — ShareYourPaper / OA.Works Permissions client for ResearchBee.

CONFIRMED against a live response on 2026-07-27:
    GET https://api.oa.works/permissions/{doi}
    No authentication required.

Design rule: THIS MODULE NEVER GUESSES.
No LLM is involved anywhere in this file. Every value returned was read from a
live response, or it is None. Absent fields are reported as absent, never
interpreted into a convenient default.

Confirmed response shape
------------------------
{
  "best_permission": {...},
  "all_permissions": [ {...}, ... ]      # multiple versions, ranked by "score"
}

Each permission object:
  can_archive        bool
  version            "submittedVersion" | "acceptedVersion" | "publishedVersion"
  licence            "cc-by-nc" | ...            (absent on some records)
  licences           [{"type": "cc-by-nc"}]
  locations          ["Institutional Repository", "Non-commercial Subject Repository"]
  deposit_statement  str   <- exact wording the author must include; NEVER paraphrase
  copyright_owner    "journal" | "Publisher"
  copyright_name     str
  issuer             {id: [issn...], has_policy, parent_policy, type, journal_oa_type}
  meta               {monitoring, updated, added, ...}
  provenance         {archiving_policy: [...], author_rights, embargo, ...}
  score              int   <- higher wins; journal-level outranks publisher-level

NOTE ON EMBARGO: the observed response carried NO embargo field. We therefore
report "not stated" rather than "none". Do not change this without evidence —
telling an author to deposit immediately when an embargo applies is exactly the
failure this module exists to prevent.
"""

from __future__ import annotations

import asyncio
import json
import logging
import time
from typing import Optional
from urllib.parse import quote

import httpx

logger = logging.getLogger(__name__)

OA_WORKS_BASE = "https://api.oa.works"
PERMISSIONS_PATH = "/permissions/{doi}"          # CONFIRMED
CONTACT = "nikesh.narayanan@ku.ac.ae"
USER_AGENT = f"ResearchBee/1.0 (Khalifa University Library; mailto:{CONTACT})"

# Khalifa University of Science and Technology — https://ror.org/05hffr360
# Passed to OA.Works so institution-specific policy is applied. This matters:
# Elsevier records a 12-month embargo for UK institutions and 24 months for
# everyone else, and the ROR is how that gets disambiguated server-side.
KU_ROR = "05hffr360"

_TIMEOUT = httpx.Timeout(8.0, connect=4.0)
_CACHE: dict[str, tuple[float, Optional[dict]]] = {}
_CACHE_TTL = 60 * 60 * 24

VERSION_MAP = {
    "submittedversion": "preprint",
    "acceptedversion":  "postprint",
    "publishedversion": "published_version",
}

INSTITUTIONAL = "institutional repository"

# OA.Works data quirks seen in live responses:
#  * locations may carry stray whitespace, e.g. " Preprint Server"
#  * issuer.id mixes ISSNs with publisher names, and sometimes splits a name
#    on a comma into fragments like "Inc" — so match ISSNs properly rather
#    than testing for a hyphen (which "Wiley-Blackwell" would pass).
import re as _re
_ISSN_RE = _re.compile(r"^\d{4}-\d{3}[\dXx]$")


def _clean_locations(perm: dict) -> list[str]:
    seen, out = set(), []
    for loc in (perm.get("locations") or []):
        s = " ".join(str(loc).split())          # collapse and strip whitespace
        if s and s.lower() not in seen:
            seen.add(s.lower())
            out.append(s)
    return out


# ── DOI handling ───────────────────────────────────────────────────────────
def normalise_doi(raw: str) -> str:
    d = (raw or "").strip()
    for p in ("https://doi.org/", "http://doi.org/", "https://dx.doi.org/",
              "http://dx.doi.org/", "doi:"):
        if d.lower().startswith(p):
            d = d[len(p):]
            break
    d = d.strip().strip("/")
    return d if d.startswith("10.") and "/" in d else ""


def share_link(doi: str) -> str:
    d = normalise_doi(doi)
    return f"https://shareyourpaper.org/?doi={quote(d, safe='/')}" if d else ""


# ── Fetch ──────────────────────────────────────────────────────────────────
async def fetch_permissions(doi: str, ror: str = "") -> Optional[dict]:
    """GET permissions for a DOI. Returns raw JSON, or None on any failure."""
    doi = normalise_doi(doi)
    if not doi:
        return None

    key = f"{doi}|{ror}"
    hit = _CACHE.get(key)
    if hit and time.time() - hit[0] < _CACHE_TTL:
        return hit[1]

    url = OA_WORKS_BASE + PERMISSIONS_PATH.format(doi=quote(doi, safe="/"))
    params = {"ror": ror} if ror else {}
    payload: Optional[dict] = None

    try:
        async with httpx.AsyncClient(
            timeout=_TIMEOUT,
            headers={"User-Agent": USER_AGENT, "Accept": "application/json"},
            follow_redirects=True,
        ) as client:
            r = await client.get(url, params=params)
            if r.status_code == 200:
                payload = r.json()
            else:
                logger.info("oa.works %s -> HTTP %s", doi, r.status_code)
    except (httpx.HTTPError, json.JSONDecodeError) as e:
        logger.warning("oa.works fetch failed for %s: %s", doi, e)

    _CACHE[key] = (time.time(), payload)
    return payload


# ── Parsing ────────────────────────────────────────────────────────────────
def _licence_of(perm: dict) -> str:
    lic = perm.get("licence")
    if lic:
        return str(lic)
    types = [l.get("type") for l in (perm.get("licences") or [])
             if isinstance(l, dict) and l.get("type")]
    return ", ".join(types)


def _embargo_months(perm: dict) -> Optional[int]:
    """Raw embargo in months, or None when the record states none."""
    for k in ("embargo_months", "embargoMonths"):
        v = perm.get(k)
        if isinstance(v, (int, float)) and v > 0:
            return int(v)
        if isinstance(v, str) and v.strip().isdigit() and int(v) > 0:
            return int(v)
    return None


def _embargo_of(perm: dict) -> tuple[str, Optional[int]]:
    """
    Returns (display_text, months_or_None).

    Some publishers (notably T&F) record no embargo at all. We say
    "not stated" rather than "none" — see the module docstring.
    """
    m = _embargo_months(perm)
    if m is None:
        return "Not stated in the permission record", None
    end = perm.get("embargo_end")
    txt = f"{m} months"
    if end:
        txt += f" (ends {end} for this article)"
    return txt, m


def _permission_rank(perm: dict) -> tuple:
    """
    Sort key for choosing between competing permissions.

    CRITICAL: Elsevier publishes DIFFERENT embargoes for UK institutions
    (12 months) and everyone else (24 months), and both records carry the
    SAME score. Ranking by score alone can silently select the shorter UK
    embargo for a KU author. We therefore break score ties by the LONGER
    embargo — the conservative choice, and the one OA.Works' own
    best_permission makes.
    """
    return (perm.get("score", 0), _embargo_months(perm) or 0)


def _covered_versions(perm: dict) -> list[str]:
    """
    Slots this permission covers.

    The `versions` array is authoritative and may list several, e.g.
    ["acceptedVersion", "submittedVersion"]. Falls back to singular `version`.
    """
    raw = perm.get("versions")
    if not isinstance(raw, list) or not raw:
        raw = [perm.get("version", "")]
    out = []
    for v in raw:
        slot = VERSION_MAP.get(str(v).lower())
        if slot and slot not in out:
            out.append(slot)
    return out


def _blank() -> dict:
    return {"allowed": "Unclear", "where": "", "embargo": "",
            "licence": "", "conditions": ""}


def _slot_from(perm: dict) -> dict:
    locations = _clean_locations(perm)
    embargo_txt, _ = _embargo_of(perm)
    conditions = []
    if perm.get("deposit_statement"):
        conditions.append("A specific deposit statement must accompany the file.")
    issuer = perm.get("issuer") or {}
    if issuer.get("notes"):
        conditions.append(str(issuer["notes"]))
    return {
        "allowed": "Yes" if perm.get("can_archive") else "No",
        "where": ", ".join(locations),
        "embargo": embargo_txt,
        "licence": _licence_of(perm),
        "conditions": " ".join(conditions),
    }


def parse_permissions(payload: Optional[dict]) -> Optional[dict]:
    """
    Build ResearchBee's green_oa structure from a live OA.Works payload.

    Fills ALL versions present in all_permissions (highest score wins per
    version), not just best_permission. Returns None if nothing usable.
    """
    if not isinstance(payload, dict):
        return None

    perms = payload.get("all_permissions") or []
    best = payload.get("best_permission")
    if not perms and isinstance(best, dict):
        perms = [best]
    perms = [p for p in perms
             if isinstance(p, dict) and p.get("can_archive") is not None]
    if not perms:
        return None

    # Highest score wins; ties broken by the LONGER embargo (conservative).
    chosen: dict[str, dict] = {}
    for p in sorted(perms, key=_permission_rank, reverse=True):
        for slot in _covered_versions(p):
            if slot not in chosen:
                chosen[slot] = p
    if not chosen:
        return None

    slots = {"preprint": _blank(), "postprint": _blank(),
             "published_version": _blank()}
    for slot, perm in chosen.items():
        slots[slot] = _slot_from(perm)

    # Prefer OA.Works' own best_permission — it already applies their ranking,
    # including picking the non-UK Elsevier embargo. Fall back to our own
    # conservative ranking if it's missing.
    top = (best if isinstance(best, dict) and best.get("can_archive") is not None
           else max(perms, key=_permission_rank))
    issuer = top.get("issuer") or {}
    prov = top.get("provenance") or {}
    meta = top.get("meta") or {}

    evidence_urls: list[str] = []
    for v in prov.values():
        if isinstance(v, str):
            evidence_urls += [u.strip() for u in v.split(",")
                              if u.strip().startswith("http")]
        elif isinstance(v, list):
            evidence_urls += [str(u) for u in v if str(u).startswith("http")]

    ir_ok = any(INSTITUTIONAL in l.lower()
                for p in chosen.values() for l in _clean_locations(p))

    updated = meta.get("updated") or meta.get("added") or ""
    evidence = ("Source: ShareYourPaper Permissions (OA.Works) — the dataset behind "
                "cOAlition S's Plan S Journal Checker Tool. "
                f"Retrieved {time.strftime('%Y-%m-%d')}.")
    if updated:
        evidence += f" Policy record last updated {updated}."

    return {
        "policy_status": "Confirmed",
        **slots,
        "licence_notes": _licence_of(top),
        "repository_action_note": "",       # LLM fills this from these facts only
        "evidence_note": evidence,
        "risk_flag": None,

        # ---- surfaced to the UI — render directly, do NOT paraphrase ----
        "deposit_statement": top.get("deposit_statement", ""),
        "copyright_owner":   top.get("copyright_owner", ""),
        "copyright_name":    top.get("copyright_name", ""),
        "journal_oa_type":   issuer.get("journal_oa_type", ""),
        "policy_issuer":     issuer.get("parent_policy") or top.get("copyright_name", ""),
        "issuer_issns":      [str(i).strip() for i in (issuer.get("id") or [])
                              if _ISSN_RE.match(str(i).strip())],
        "embargo_end":       top.get("embargo_end", ""),
        "policy_updated":    updated,
        "evidence_urls":     sorted(set(evidence_urls))[:6],
        "monitoring":        meta.get("monitoring", ""),

        # ---- internal, for deterministic downstream logic ----
        "_can_archive":      bool(top.get("can_archive")),
        "_versions":         sorted(chosen.keys()),
        "_best_version":     VERSION_MAP.get(str(top.get("version", "")).lower(), ""),
        "_locations":        _clean_locations(top),
        "_institutional_ok": ir_ok,
        "_embargo_months":   _embargo_of(top)[1],
        "_licence":          _licence_of(top),
    }


def not_confirmed(journal: str = "", issn: str = "", reason: str = "") -> dict:
    verify = "https://openpolicyfinder.jisc.ac.uk/"
    if issn:
        verify += f"search?term={quote(issn)}"
    return {
        "policy_status": "Not confirmed",
        "preprint": _blank(), "postprint": _blank(),
        "published_version": _blank(),
        "licence_notes": "",
        "repository_action_note": ("ResearchBee could not verify this article's "
                                   "self-archiving rights against a trusted source."),
        "evidence_note": (f"No verified permission record found"
                          f"{(' — ' + reason) if reason else ''}. Verify at {verify}"),
        "risk_flag": "Policy not verified — confirm before depositing.",
        "deposit_statement": "",
        "evidence_urls": [],
        "_can_archive": False,
        "_versions": [],
        "_institutional_ok": False,
        "_embargo_months": None,
        "_verify_url": verify,
    }


# ── Deterministic recommendation — no LLM ──────────────────────────────────
def recommendation_from(green: dict) -> dict:
    ok = green.get("_can_archive", False)
    emb = green.get("_embargo_months")
    best = green.get("_best_version", "")
    checks = []
    if green.get("policy_status") != "Confirmed":
        checks.append("Verify the policy at Open Policy Finder before depositing.")
    if ok and emb is None:
        checks.append("No embargo is stated in the record — confirm against the "
                      "publisher's policy page before depositing immediately.")
    return {
        "best_version_to_deposit": best or "Not specified",
        "best_timing": (f"After a {emb}-month embargo" if emb
                        else ("Not stated — verify" if ok else "Not applicable")),
        "immediate_open_deposit_possible": "Yes" if (ok and emb is None) else "No",
        "embargoed_deposit_needed": "Yes" if emb else "No",
        "metadata_only_first": "Yes" if emb else "No",
        "manual_checks_required": checks,
    }


def next_actions_from(green: dict, doi: str,
                      khazna: dict | None = None) -> list[str]:
    acts: list[str] = []
    if green.get("_can_archive"):
        if green.get("_institutional_ok"):
            acts.append("Deposit the permitted version in the KU repository, Khazna.")
        if green.get("deposit_statement"):
            acts.append("Include the required deposit statement with the file.")
        link = share_link(doi)
        if link:
            acts.append(f"Or deposit via ShareYourPaper: {link}")
    else:
        acts.append("Self-archiving does not appear to be permitted — check "
                    "whether a KU APC agreement covers Gold OA for this journal.")
    if (khazna and khazna.get("checked") and khazna.get("in_khazna")
            and khazna.get("needs_deposit")):
        acts.insert(0, "This work is already recorded in the KU repository, Khazna, "
                       "but has no open full text attached — adding the file "
                       "would close the gap.")
    return acts


# ── Probe ──────────────────────────────────────────────────────────────────
async def probe(doi: str = "10.1080/19322909.2023.2221477") -> None:
    payload = await fetch_permissions(doi)
    print(json.dumps(parse_permissions(payload), indent=2, ensure_ascii=False))


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(probe())