Spaces:
Running
Running
| """Conservative, deterministic applicability checks for linked policy options. | |
| Source verification and a permission for a particular manuscript/repository are | |
| different facts. Never combine the licence from one option with another option's | |
| location or embargo. Missing context remains a manual check. | |
| """ | |
| from __future__ import annotations | |
| from datetime import date, datetime, timezone | |
| import re | |
| from typing import Any | |
| SLOTS = ("postprint", "preprint", "published_version") | |
| _EMPTY = {"", "not stated", "not specified", "unknown", "unclear", "none stated"} | |
| def _text(value: Any) -> str: | |
| return " ".join(str(value or "").split()) | |
| def normalise_version(value: str) -> str: | |
| value = re.sub(r"[^a-z0-9]+", " ", _text(value).casefold()) | |
| if not value or value == "all versions": | |
| return "" | |
| if any(term in value for term in ("postprint", "post print", "accepted", "aam")): | |
| return "postprint" | |
| if any(term in value for term in ("preprint", "pre print", "submitted")): | |
| return "preprint" | |
| if any(term in value for term in ("published", "version of record", "publisher version", "vor")): | |
| return "published_version" | |
| return "unknown" | |
| def _values(value: Any) -> list[str]: | |
| if isinstance(value, list): | |
| return [text for item in value if (text := _text(item)) and text.casefold() not in _EMPTY] | |
| text = _text(value) | |
| return [part.strip() for part in re.split(r"[;,]", text) | |
| if part.strip().casefold() not in _EMPTY] | |
| def _location(value: str) -> str: | |
| text = re.sub(r"[-_]+", " ", _text(value).casefold()) | |
| text = " ".join(text.split()) | |
| if text.startswith("subject repository (e.g."): | |
| text = "subject repository" | |
| aliases = {"institutional repositories": "institutional repository", | |
| "khazna": "institutional repository", | |
| "ku repository": "institutional repository", | |
| "personal website": "author website", "author's website": "author website", | |
| "non commercial subject repositories": "non commercial subject repository", | |
| "subject repositories": "subject repository"} | |
| return aliases.get(text, text) | |
| def _location_match(target: str, locations: list[str]) -> str: | |
| desired = _location(target) | |
| known = [_location(value) for value in locations] | |
| if not desired or desired in {"unsure", "other", "any", "not sure"} or not known: | |
| return "unknown" | |
| if desired in known: | |
| return "yes" | |
| if desired.endswith("repository") and "any repository" in known: | |
| return "yes" | |
| if desired == "non commercial subject repository" and "subject repository" in known: | |
| return "yes" | |
| if desired == "subject repository" and "non commercial subject repository" in known: | |
| return "unknown" # The user has not established the required non-commercial status. | |
| return "no" | |
| def _licence(value: str) -> str: | |
| text = _text(value).casefold().replace("creative commons", "cc") | |
| match = re.search(r"creativecommons\.org/(?:licenses|publicdomain)/([^/]+)", text) | |
| if match: | |
| text = "cc " + ("0" if match.group(1) == "zero" else match.group(1)) | |
| text = re.sub(r"\b\d+\.\d+\b|\binternational\b|\bunported\b", "", text) | |
| text = text.replace("attribution", "by").replace("noncommercial", "nc") | |
| text = text.replace("non-commercial", "nc").replace("no derivatives", "nd") | |
| text = text.replace("noderivatives", "nd").replace("sharealike", "sa") | |
| return re.sub(r"[^a-z0-9]", "", text) | |
| def embargo_release(option: dict, today: date | None = None) -> dict: | |
| """Interpret an explicit article end date; never derive one from publication date.""" | |
| today = today or datetime.now(timezone.utc).date() | |
| end = _text(option.get("embargo_end")) | |
| months = option.get("embargo_months", option.get("_embargo_months")) | |
| status = _text(option.get("embargo_status") or option.get("_embargo_status")).lower() | |
| result = {"embargo_status": status if status in {"zero", "stated", "expired"} else "unknown", | |
| "embargo_months": months, "embargo_end": end, "embargo_date_valid": None} | |
| if end: | |
| try: | |
| # A date is an inclusive release date. Offset timestamps are first | |
| # converted to UTC; future time on today's date is not called expired. | |
| if re.fullmatch(r"\d{4}-\d{2}-\d{2}", end): | |
| release = date.fromisoformat(end) | |
| expired = release <= today | |
| else: | |
| stamp = datetime.fromisoformat(end.replace("Z", "+00:00")) | |
| if stamp.tzinfo is None: | |
| raise ValueError("article release timestamp requires a timezone") | |
| utc_stamp = stamp.astimezone(timezone.utc) | |
| # With a date-only evaluation clock, use the start of that UTC day. | |
| expired = utc_stamp <= datetime.combine(today, datetime.min.time(), timezone.utc) | |
| result.update(embargo_status="expired" if expired else "stated", embargo_date_valid=True) | |
| except (ValueError, TypeError, OverflowError): | |
| result.update(embargo_status="unknown", embargo_date_valid=False) | |
| return result | |
| def policy_options(green: dict) -> list[dict]: | |
| options = [] | |
| for slot in SLOTS: | |
| data = green.get(slot) | |
| if not isinstance(data, dict): | |
| continue | |
| entries = data.get("options") | |
| # Older snapshots have a single flat slot. An explicitly empty options | |
| # list means no policy was listed and must not become a permission. | |
| if not isinstance(entries, list): | |
| stated = _text(data.get("permission_status")).lower() in {"allowed", "conditional", "denied"} | |
| stated = stated or _text(data.get("allowed")).lower() in {"yes", "no", "conditional"} | |
| entries = [data] if stated else [] | |
| for entry in entries: | |
| if isinstance(entry, dict): | |
| options.append({**entry, "version": slot}) | |
| return options | |
| def _evaluate_single(green: dict, manuscript_version: str = "Accepted manuscript (AAM / postprint)", | |
| intended_repository: str = "Institutional repository", intended_licence: str = "", | |
| funder: str = "", today: date | None = None) -> dict: | |
| """Evaluate each complete option and return one applicable option, if any. | |
| ``meets`` establishes a sharing option for the supplied selection, not | |
| compliance with an unseen funder mandate. Free-text conditions always need | |
| review. Different options are never merged into an apparent permission. | |
| """ | |
| context = {"manuscript_version": manuscript_version, "intended_repository": intended_repository, | |
| "intended_licence": intended_licence, "funder": funder} | |
| base = {"state": "unknown", "permission_status": "unknown", "selected_option": None, | |
| "unresolved_conditions": [], "option_checks": [], "context": context, | |
| "embargo_status": "unknown", "embargo_months": None, "embargo_end": ""} | |
| if not isinstance(green, dict) or green.get("policy_status") != "Confirmed": | |
| base["unresolved_conditions"] = ["Verify the policy at Open Policy Finder before depositing."] | |
| return base | |
| version = normalise_version(manuscript_version) | |
| options = [o for o in policy_options(green) if not version or o["version"] == version] | |
| if not options: | |
| base["unresolved_conditions"] = ["No verified policy option was found for the selected manuscript version."] | |
| return base | |
| assessed = [] | |
| for option in options: | |
| reasons, mismatches = [], [] | |
| permission = _text(option.get("permission_status")).casefold() | |
| allowed = _text(option.get("allowed")).casefold() | |
| if not permission: | |
| permission = {"yes": "allowed", "no": "denied", "conditional": "conditional"}.get(allowed, "unknown") | |
| location = _location_match(intended_repository, _values(option.get("locations") or option.get("where"))) | |
| if location == "no": | |
| mismatches.append("This option does not list the selected deposit location.") | |
| elif location == "unknown": | |
| reasons.append("Confirm that this option permits the selected deposit location.") | |
| requested_licence = _licence(intended_licence) | |
| unspecified_licences = {"unsure", "notsure", "other", "notspecified", "unknown", "publishersbespokelicence"} | |
| if requested_licence in unspecified_licences: | |
| reasons.append("Confirm which licence you intend to apply before opening the file.") | |
| if requested_licence and requested_licence not in unspecified_licences: | |
| licences = _values(option.get("licences") or option.get("licence")) | |
| if not licences: | |
| reasons.append("The option does not establish permission to apply the selected licence.") | |
| elif requested_licence not in {_licence(value) for value in licences}: | |
| mismatches.append("This option does not list the selected licence.") | |
| conditions = _values(option.get("conditions")) | |
| reasons.extend(conditions) | |
| funders = _values(option.get("funders")) | |
| if funders: | |
| if not _text(funder): | |
| reasons.append("Confirm eligibility for the funder-specific policy option: " + "; ".join(funders)) | |
| elif _text(funder).casefold() not in {_text(value).casefold() for value in funders}: | |
| mismatches.append("The selected funder does not match this funder-specific option.") | |
| if _text(funder): | |
| reasons.append("Check the selected funder's current mandate; publisher permission alone does not establish funder compliance.") | |
| if permission == "conditional" and not conditions: | |
| reasons.append("The policy marks this permission as conditional; verify its eligibility conditions.") | |
| if permission not in {"allowed", "conditional", "denied"}: | |
| reasons.append("Permission is not established for this option.") | |
| timing = embargo_release(option, today) | |
| applicability_reasons = list(reasons) | |
| if timing["embargo_status"] == "unknown" and permission != "denied": | |
| reasons.append("Verify the release timing; this option does not state a usable embargo or article release date.") | |
| state = ("does_not_meet" if mismatches or permission == "denied" else | |
| "unknown" if permission not in {"allowed", "conditional"} else | |
| "conditional" if reasons else "meets") | |
| selected_permission = ("denied" if permission == "denied" else "unknown" if mismatches else | |
| "unknown" if permission not in {"allowed", "conditional"} else | |
| "conditional" if applicability_reasons else permission) | |
| assessed.append({"state": state, "permission_status": selected_permission, | |
| "option": option, "unresolved_conditions": list(dict.fromkeys(reasons)), | |
| "mismatches": mismatches, **timing}) | |
| base["option_checks"] = assessed | |
| viable = [item for item in assessed if item["state"] in {"meets", "conditional"}] | |
| if viable: | |
| order = {"zero": 0, "expired": 0, "stated": 1, "unknown": 2} | |
| winner = min(viable, key=lambda item: (item["state"] != "meets", len(item["unresolved_conditions"]), | |
| order[item["embargo_status"]], SLOTS.index(item["option"]["version"]))) | |
| # A denial for the same target is not silently overridden by an allowance. | |
| conflicts = [item for item in assessed if item["permission_status"] == "denied" | |
| and not item["mismatches"] and item["option"]["version"] == winner["option"]["version"]] | |
| if conflicts: | |
| winner = {**winner, "state": "conditional", "permission_status": "conditional", | |
| "unresolved_conditions": winner["unresolved_conditions"] + ["Conflicting permission options require source review."]} | |
| def route_key(item): | |
| record = item["option"] | |
| return (record["version"], tuple(sorted(_location(value) for value in _values(record.get("locations") or record.get("where")))), | |
| tuple(sorted(_licence(value) for value in _values(record.get("licences") or record.get("licence")))), | |
| tuple(sorted(_values(record.get("funders")))), tuple(sorted(_values(record.get("conditions"))))) | |
| def timing_key(item): | |
| if item["embargo_status"] in {"zero", "expired"}: | |
| return ("released",) | |
| return (item["embargo_status"], item["embargo_months"], item["embargo_end"]) | |
| # Two indistinguishable routes with contradictory timing cannot be | |
| # resolved by choosing whichever embargo happens to be shortest. | |
| timing_conflict = any(route_key(item) == route_key(winner) and timing_key(item) != timing_key(winner) | |
| for item in viable) | |
| if timing_conflict: | |
| winner = {**winner, "state": "conditional", "permission_status": "conditional", | |
| "unresolved_conditions": winner["unresolved_conditions"] + ["Otherwise matching policy options state different embargo timing; verify the applicable record."]} | |
| return {**base, **{key: value for key, value in winner.items() if key not in {"option", "mismatches"}}, | |
| "selected_option": winner["option"]} | |
| base["state"] = "unknown" if any(item["state"] == "unknown" for item in assessed) else "does_not_meet" | |
| # A location/licence mismatch is not a general prohibition on self-archiving. | |
| base["permission_status"] = "denied" if all(item["permission_status"] == "denied" for item in assessed) else "unknown" | |
| base["unresolved_conditions"] = list(dict.fromkeys(reason for item in assessed | |
| for reason in item["mismatches"] + item["unresolved_conditions"])) | |
| if not base["unresolved_conditions"]: | |
| base["unresolved_conditions"] = ["The verified record does not permit the selected version at the selected location."] | |
| return base | |
| def evaluate_policy(green: dict, manuscript_version: str = "Accepted manuscript (AAM / postprint)", | |
| intended_repository: str = "Institutional repository", intended_licence: str = "", | |
| funder: str = "", today: date | None = None) -> dict: | |
| """Evaluate one destination or require separate complete options for both.""" | |
| if _text(intended_repository).casefold() not in {"both institutional & subject", "both institutional and subject"}: | |
| return _evaluate_single(green, manuscript_version, intended_repository, intended_licence, funder, today) | |
| version = normalise_version(manuscript_version) | |
| versions = [manuscript_version] if version else list(SLOTS) | |
| groups = [] | |
| for selected_version in versions: | |
| results = [_evaluate_single(green, selected_version, destination, intended_licence, funder, today) | |
| for destination in ("Institutional repository", "Subject repository")] | |
| states = [item["state"] for item in results] | |
| state = ("does_not_meet" if "does_not_meet" in states else "unknown" if "unknown" in states | |
| else "conditional" if "conditional" in states else "meets") | |
| permissions = [item["permission_status"] for item in results] | |
| permission = ("allowed" if all(value == "allowed" for value in permissions) else | |
| "denied" if all(value == "denied" for value in permissions) else | |
| "conditional" if state == "conditional" else "unknown") | |
| reasons = list(dict.fromkeys(f"{item['context']['intended_repository']}: {reason}" | |
| for item in results for reason in item["unresolved_conditions"])) | |
| options = [item["selected_option"] for item in results if item["selected_option"]] | |
| statuses = [item["embargo_status"] for item in results] | |
| timing_same = len({(item["embargo_status"], item["embargo_months"], item["embargo_end"]) for item in results}) == 1 | |
| if all(status in {"zero", "expired"} for status in statuses): | |
| timing = {"embargo_status": "expired" if "expired" in statuses else "zero", | |
| "embargo_months": None, "embargo_end": next((item["embargo_end"] for item in results if item["embargo_end"]), "")} | |
| elif timing_same: | |
| timing = {key: results[0][key] for key in ("embargo_status", "embargo_months", "embargo_end")} | |
| else: | |
| timing = {"embargo_status": "unknown", "embargo_months": None, "embargo_end": ""} | |
| if state == "meets": | |
| state, permission = "conditional", "conditional" | |
| reasons.append("The destinations have different release conditions; follow each linked option separately before opening files.") | |
| groups.append({"state": state, "permission_status": permission, | |
| "selected_option": options[0] if len(options) == 2 else None, | |
| "selected_options": options if len(options) == 2 else [], | |
| "destination_checks": results, "option_checks": [check for item in results for check in item["option_checks"]], | |
| "unresolved_conditions": reasons, **timing, | |
| "context": {"manuscript_version": manuscript_version, "intended_repository": intended_repository, | |
| "intended_licence": intended_licence, "funder": funder}}) | |
| rank = {"meets": 0, "conditional": 1, "unknown": 2, "does_not_meet": 3} | |
| return min(groups, key=lambda item: rank[item["state"]]) | |