""" 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())