""" 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 math import time from typing import Optional from datetime import date from policy_evaluation import embargo_release, evaluate_policy 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 _ERROR_TTL = 3 * 60 _CACHE_MAX = 512 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(), [] raw = perm.get("locations") or [] for loc in (raw if isinstance(raw, list) else [raw]): 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 if hit[1] is not None else _ERROR_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) now = time.time() for old_key, (stored_at, old_payload) in list(_CACHE.items()): if now - stored_at >= (_CACHE_TTL if old_payload is not None else _ERROR_TTL): _CACHE.pop(old_key, None) if key not in _CACHE and len(_CACHE) >= _CACHE_MAX: _CACHE.pop(min(_CACHE, key=lambda old_key: _CACHE[old_key][0]), None) _CACHE[key] = (now, payload) return payload # ── Parsing ──────────────────────────────────────────────────────────────── def _licence_of(perm: dict) -> str: lic = perm.get("licence") if lic: return str(lic) raw = perm.get("licences") or [] types = [l.get("type") for l in (raw if isinstance(raw, list) else [raw]) if isinstance(l, dict) and l.get("type")] return ", ".join(types) def _normalise_embargo_months(value) -> Optional[int]: """Return a non-negative month value only when the source states one.""" if isinstance(value, bool): return None if isinstance(value, (int, float)) and math.isfinite(value) and value >= 0 and value == int(value): return int(value) if isinstance(value, str): raw = value.strip().lower() if raw in {"none", "no embargo", "immediate", "0", "0.0"}: return 0 match = _re.fullmatch(r"(\d+)\s*(?:months?|mos?)?", raw) if match: return int(match.group(1)) return None def _embargo_details(perm: dict, today: date | None = None) -> tuple[str, Optional[int], str]: """ Return ``(display_text, months, status)`` without collapsing unknown to zero. ``status`` is one of: * ``zero`` — the permission record explicitly states zero/no embargo; * ``stated`` — a positive duration or article-specific end date is stated; * ``unknown`` — the record contains no usable embargo statement. """ # Article-specific release dates take precedence over the generic duration. if perm.get("embargo_end"): release = embargo_release(perm, today) end = release["embargo_end"] months = _normalise_embargo_months(perm.get("embargo_months", perm.get("embargoMonths"))) if release["embargo_date_valid"] is False: return "Embargo end date could not be validated — verify release timing", months, "unknown" if release["embargo_status"] == "expired": return f"Embargo expired (article release date {end})", months, "expired" return f"Embargo ends {end} for this article", months, "stated" for key in ("embargo_months", "embargoMonths"): if key not in perm: continue months = _normalise_embargo_months(perm.get(key)) if months is None: continue if months == 0: return "No embargo (0 months stated in the permission record)", 0, "zero" end = perm.get("embargo_end") text = f"{months} months" if end: text += f" (ends {end} for this article)" return text, months, "stated" # Some responses may use a generic embargo value rather than the confirmed # month keys. Accept only explicit, machine-readable values. if "embargo" in perm and not isinstance(perm.get("embargo"), dict): months = _normalise_embargo_months(perm.get("embargo")) if months is not None: if months == 0: return "No embargo (0 months stated in the permission record)", 0, "zero" end = perm.get("embargo_end") text = f"{months} months" if end: text += f" (ends {end} for this article)" return text, months, "stated" end = str(perm.get("embargo_end") or "").strip() if end: return f"Embargo ends {end} for this article", None, "stated" return "Not stated in the permission record", None, "unknown" def _embargo_months(perm: dict) -> Optional[int]: """Compatibility helper: 0, positive months, or None when not stated.""" return _embargo_details(perm)[1] def _embargo_of(perm: dict) -> tuple[str, Optional[int]]: text, months, _status = _embargo_details(perm) return text, months def _embargo_status(perm: dict) -> str: return _embargo_details(perm)[2] def _permission_rank(perm: dict) -> tuple: """Conservative sort key for competing permissions of equal source score.""" _text, months, status = _embargo_details(perm) # Prefer a known positive/end-dated embargo, then an unknown embargo, then # an explicit zero embargo. This avoids silently selecting "immediate" when # another equally authoritative record leaves timing unknown. status_rank = {"stated": 2, "unknown": 1, "zero": 0}.get(status, 1) try: score = float(perm.get("score", 0)) if not math.isfinite(score): score = 0 except (ValueError, TypeError): score = 0 return (score, status_rank, months 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, today: date | None = None) -> dict: locations = _clean_locations(perm) embargo_txt, _months, embargo_status = _embargo_details(perm, today) can_archive = perm.get("can_archive") is True conditions = [] if perm.get("deposit_statement"): conditions.append("A specific deposit statement must accompany the file.") for field in ("conditions", "requirements", "prerequisites"): value = perm.get(field) for item in (value if isinstance(value, list) else [value]): if isinstance(item, str) and item.strip(): conditions.append(item.strip()) elif isinstance(item, dict): text = item.get("text") or item.get("description") or item.get("name") if text: conditions.append(str(text)) issuer = perm.get("issuer") if isinstance(perm.get("issuer"), dict) else {} if issuer.get("notes"): conditions.append(str(issuer["notes"])) return { "allowed": "Yes" if can_archive else "No", "permission_status": "allowed" if can_archive else "denied", "embargo_status": embargo_status, "embargo_months": _months, "embargo_end": str(perm.get("embargo_end") or ""), "locations": locations, "licences": [item.strip() for item in _licence_of(perm).split(",") if item.strip()], "funders": _funders_of(perm), "deposit_statement": perm.get("deposit_statement", ""), "source": "ShareYourPaper Permissions (OA.Works)", "source_url": "https://shareyourpaper.org/", "where": ", ".join(locations), "embargo": embargo_txt, "licence": _licence_of(perm), "conditions": " ".join(conditions), } def _funders_of(perm: dict) -> list[str]: raw = perm.get("funders") or perm.get("funder") or [] if not isinstance(raw, list): raw = [raw] return [str(item.get("name") or item.get("id") or "").strip() if isinstance(item, dict) else str(item).strip() for item in raw if item] def parse_permissions(payload: Optional[dict], manuscript_version: str = "", intended_repository: str = "Institutional repository", intended_licence: str = "", funder: str = "", today: date | None = None) -> 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 isinstance(p.get("can_archive"), bool)] 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, today) # Preserve the fields of each complete permission route. Evaluation must # never combine one route's licence with another route's repository. slots[slot]["options"] = [_slot_from(item, today) for item in perms if slot in _covered_versions(item)] # 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 isinstance(best.get("can_archive"), bool) else max(perms, key=_permission_rank)) issuer = top.get("issuer") if isinstance(top.get("issuer"), dict) else {} prov = top.get("provenance") if isinstance(top.get("provenance"), dict) else {} meta = top.get("meta") if isinstance(top.get("meta"), dict) else {} 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( p.get("can_archive") is True and INSTITUTIONAL in location.lower() for p in chosen.values() for location 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}." can_archive = top.get("can_archive") is True embargo_text, embargo_months, embargo_status = _embargo_details(top, today) permission_status = "allowed" if can_archive else "denied" green = { "policy_status": "Confirmed", "permission_status": permission_status, "embargo_status": embargo_status, **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": 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_months, "_embargo_status": embargo_status, "_embargo_text": embargo_text, "_licence": _licence_of(top), } context = {"manuscript_version": manuscript_version, "intended_repository": intended_repository, "intended_licence": intended_licence, "funder": funder} green["_request_context"] = context evaluation = evaluate_policy(green, **context, today=today) green["policy_evaluation"] = evaluation selected = evaluation.get("selected_option") green["permission_status"] = evaluation["permission_status"] green["_can_archive"] = (True if evaluation["permission_status"] == "allowed" else False if evaluation["permission_status"] == "denied" else None) if selected: green.update({ "embargo_status": evaluation["embargo_status"], "_embargo_status": evaluation["embargo_status"], "_embargo_months": evaluation["embargo_months"], "_embargo_text": selected.get("embargo", ""), "embargo_end": evaluation["embargo_end"], "_best_version": selected["version"], "_locations": selected.get("locations", []), "_institutional_ok": any(INSTITUTIONAL == loc.lower() for loc in selected.get("locations", [])), "_licence": selected.get("licence", ""), "licence_notes": selected.get("licence", ""), "deposit_statement": selected.get("deposit_statement", ""), }) else: green.update({"_locations": [], "_institutional_ok": False, "_best_version": "", "_embargo_status": "unknown", "embargo_status": "unknown", "_embargo_months": None, "embargo_end": "", "deposit_statement": ""}) return green 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", "permission_status": "unknown", "embargo_status": "unknown", "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": "", "embargo_end": "", "evidence_urls": [], "_can_archive": None, "_versions": [], "_best_version": "", "_locations": [], "_institutional_ok": False, "_embargo_months": None, "_embargo_status": "unknown", "_embargo_text": "Not stated in a verified permission record", "_licence": "", "_verify_url": verify, } # ── Deterministic recommendation — no LLM ────────────────────────────────── def _evaluation_for(green: dict, manuscript_version: str | None = None, intended_repository: str | None = None, intended_licence: str | None = None, funder: str | None = None, today: date | None = None) -> dict: context = {"manuscript_version": "", "intended_repository": "Institutional repository", "intended_licence": "", "funder": ""} context.update(green.get("_request_context") or {}) supplied = {"manuscript_version": manuscript_version, "intended_repository": intended_repository, "intended_licence": intended_licence, "funder": funder} context.update({key: value for key, value in supplied.items() if value is not None}) return evaluate_policy(green, **context, today=today) def recommendation_from(green: dict, manuscript_version: str | None = None, intended_repository: str | None = None, intended_licence: str | None = None, funder: str | None = None, today: date | None = None) -> dict: evaluation = _evaluation_for(green, manuscript_version, intended_repository, intended_licence, funder, today) permission = evaluation["permission_status"] status = evaluation["embargo_status"] option = evaluation.get("selected_option") or {} checks = list(evaluation["unresolved_conditions"]) immediate, embargoed, metadata = "Unknown", "Unknown", "Unknown" timing = "Not determined — verify the applicable option" if permission == "denied": immediate, embargoed, metadata = "No", "No", "No" timing = "Not applicable for the selected version and location" elif evaluation["state"] == "does_not_meet": immediate = "No" timing = "No matching permission option established for the selected requirements" elif permission == "allowed": if status in {"zero", "expired"}: immediate, embargoed, metadata = "Yes", "No", "No" timing = (f"Embargo expired ({evaluation['embargo_end']}); open deposit is permitted by the matching record" if status == "expired" else "Immediate open deposit is permitted by the matching verified record") elif status == "stated": immediate, embargoed, metadata = "No", "Yes", "Yes" if evaluation["embargo_end"]: timing = f"After the stated embargo ends ({evaluation['embargo_end']})" elif isinstance(evaluation["embargo_months"], int) and evaluation["embargo_months"] > 0: timing = f"After a {evaluation['embargo_months']}-month embargo" checks.append("Confirm the policy's embargo starting event and exact release date before opening the file.") else: timing = "After the stated embargo; verify the release date" else: timing = "Embargo not stated — verify before opening the file" return { "permission_status": permission, "embargo_status": status, "best_version_to_deposit": option.get("version") or "Not determined", "best_timing": timing, "immediate_open_deposit_possible": immediate, "embargoed_deposit_needed": embargoed, "metadata_only_first": metadata, "manual_checks_required": list(dict.fromkeys(checks)), "requirement_state": evaluation["state"], "selected_option": evaluation.get("selected_option"), "policy_evaluation": evaluation, } def next_actions_from(green: dict, doi: str, khazna: dict | None = None, manuscript_version: str | None = None, intended_repository: str | None = None, intended_licence: str | None = None, funder: str | None = None, today: date | None = None) -> list[str]: evaluation = _evaluation_for(green, manuscript_version, intended_repository, intended_licence, funder, today) recommendation = recommendation_from(green, manuscript_version, intended_repository, intended_licence, funder, today) option = evaluation.get("selected_option") or {} acts = [] if evaluation["state"] == "meets": destination = evaluation["context"]["intended_repository"] if recommendation["immediate_open_deposit_possible"] == "Yes": acts.append(f"Prepare the {option.get('version', 'permitted version')} for {destination}, following the selected policy option.") else: acts.append(f"Confirm the embargo release date before opening the permitted file in {destination}.") elif evaluation["permission_status"] == "denied": acts.append("The verified record does not permit the selected manuscript version at the selected location; review other options with the Library.") else: acts.append("ResearchBee could not determine an unconditional permission for your selection. Verify the applicable option before depositing a file.") acts.extend(recommendation["manual_checks_required"]) if option.get("deposit_statement"): acts.append("Include the exact required deposit statement with the file if this option is confirmed applicable.") if evaluation["state"] == "meets" and share_link(doi): acts.append(f"Review the applicable deposit route at ShareYourPaper: {share_link(doi)}") 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 Khazna but has no open full text attached; verify the selected version, location and timing before adding a file.") return list(dict.fromkeys(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())