Spaces:
Running
Running
| """ | |
| 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()) | |