Configuring CC-BY 4.0 Licenses for Automated Dataset Publishing #

from __future__ import annotations

import logging
from typing import Final

from pydantic import BaseModel, Field, field_validator

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

# Closed vocabulary: every accepted label -> a single canonical record.
# Keys are lowercased at lookup time; SPDX identifiers themselves are NOT lowercased.
_LICENSE_TABLE: Final[dict[str, dict[str, str]]] = {
    "cc-by-4.0": {
        "spdx": "CC-BY-4.0",
        "name": "Creative Commons Attribution 4.0 International",
        "uri": "https://creativecommons.org/licenses/by/4.0/",
        "legal_code": "https://creativecommons.org/licenses/by/4.0/legalcode",
    },
    "cc-by-3.0": {
        "spdx": "CC-BY-3.0",
        "name": "Creative Commons Attribution 3.0 Unported",
        "uri": "https://creativecommons.org/licenses/by/3.0/",
        "legal_code": "https://creativecommons.org/licenses/by/3.0/legalcode",
    },
    "cc-by-sa-4.0": {
        "spdx": "CC-BY-SA-4.0",
        "name": "Creative Commons Attribution-ShareAlike 4.0 International",
        "uri": "https://creativecommons.org/licenses/by-sa/4.0/",
        "legal_code": "https://creativecommons.org/licenses/by-sa/4.0/legalcode",
    },
    "cc0-1.0": {
        "spdx": "CC0-1.0",
        "name": "Creative Commons Zero v1.0 Universal",
        "uri": "https://creativecommons.org/publicdomain/zero/1.0/",
        "legal_code": "https://creativecommons.org/publicdomain/zero/1.0/legalcode",
    },
}

# Common label aliases that must collapse onto a canonical table key.
_ALIASES: Final[dict[str, str]] = {
    "cc by 4.0": "cc-by-4.0",
    "attribution 4.0 international": "cc-by-4.0",
    "cc by 3.0": "cc-by-3.0",
    "attribution 3.0 unported": "cc-by-3.0",
    "cc by-sa 4.0": "cc-by-sa-4.0",
    "attribution-sharealike 4.0": "cc-by-sa-4.0",
    "cc0": "cc0-1.0",
    "public domain dedication": "cc0-1.0",
}

_CANONICAL_URIS: Final[frozenset[str]] = frozenset(r["uri"] for r in _LICENSE_TABLE.values())


class LicenseResolutionError(ValueError):
    """Raised when an input label cannot be resolved to a canonical CC-BY record."""


def resolve_license(label: str) -> dict[str, str]:
    """Resolve a free-text label to its canonical record, or reject it.

    Matching is case-insensitive on the label only; the returned SPDX
    identifier keeps its canonical case (e.g. "CC-BY-4.0").
    """
    key = " ".join(label.strip().lower().split())  # normalise whitespace + case
    key = _ALIASES.get(key, key)
    record = _LICENSE_TABLE.get(key)
    if record is None:
        raise LicenseResolutionError(
            f"Unresolvable license label {label!r}. "
            f"Supply an exact CC-BY version (e.g. 'CC-BY-4.0'); ambiguous strings are rejected."
        )
    return record


class RightsEntry(BaseModel):
    """A single DataCite 4.5+ rightsList entry, validated before serialization."""

    rights: str
    rightsUri: str
    rightsIdentifier: str
    rightsIdentifierScheme: str = Field(default="SPDX")
    schemeUri: str = Field(default="https://spdx.org/licenses/")

    @field_validator("rightsUri")
    @classmethod
    def uri_must_be_canonical(cls, v: str) -> str:
        # A legal-code URI is the canonical URI plus the "/legalcode" suffix.
        base = v.removesuffix("/legalcode")
        if not base.endswith("/"):
            base += "/"
        if base not in _CANONICAL_URIS:
            raise ValueError(f"Non-canonical license URI: {v!r}")
        return v


def build_rights_entry(label: str) -> RightsEntry:
    """Resolve a label and produce a validated DataCite rightsList entry."""
    record = resolve_license(label)
    entry = RightsEntry(
        rights=record["name"],
        rightsUri=record["legal_code"],
        rightsIdentifier=record["spdx"],
    )
    logging.info("Resolved %r -> %s", label, entry.rightsIdentifier)
    return entry


def inject_license(metadata: dict[str, object], label: str) -> dict[str, object]:
    """Attach a validated rightsList to a DataCite metadata payload."""
    entry = build_rights_entry(label)
    metadata["rightsList"] = [entry.model_dump()]
    return metadata

import time
from typing import Final

# Immutable, pinned fallback used only when the policy registry is unreachable.
_FALLBACK_LABEL: Final[str] = "CC-BY-4.0"
_CACHE_TTL_SECONDS: Final[int] = 900  # 15 minutes

_cache: dict[str, tuple[float, str]] = {}


def resolve_policy_label(dataset_id: str, registry_lookup) -> str:
    """Return the governing license label for a dataset, with cache-aside + fallback.

    `registry_lookup(dataset_id) -> str` calls the central policy registry.
    On any failure the pinned fallback is used and the event is logged for audit.
    """
    now = time.monotonic()
    cached = _cache.get(dataset_id)
    if cached and (now - cached[0]) < _CACHE_TTL_SECONDS:
        return cached[1]

    try:
        label = registry_lookup(dataset_id)
        resolve_license(label)  # validate before caching; reject drifted junk
        _cache[dataset_id] = (now, label)
        return label
    except Exception as exc:  # registry down OR returned an unresolvable label
        logging.warning(
            "Policy registry lookup failed for %s (%s); using pinned fallback %s",
            dataset_id, exc, _FALLBACK_LABEL,
        )
        return _FALLBACK_LABEL

import pytest


def test_canonical_label_injects_rightslist() -> None:
    meta = inject_license({"titles": [{"title": "Reef Temperature Series"}]}, "CC BY 4.0")
    entry = meta["rightsList"][0]
    assert entry["rightsIdentifier"] == "CC-BY-4.0"
    assert entry["rightsUri"].endswith("/by/4.0/legalcode")


def test_unversioned_label_is_rejected() -> None:
    with pytest.raises(LicenseResolutionError):
        resolve_license("CC-BY")


def test_forged_uri_fails_validation() -> None:
    with pytest.raises(Exception):
        RightsEntry(
            rights="Creative Commons Attribution 4.0 International",
            rightsUri="https://example.org/licenses/by/4.0/",
            rightsIdentifier="CC-BY-4.0",
        )