Operational Context

A Python automation engineer is bulk-loading a rights export into a legacy Collection Management System and the ingest job dies with a ValueError on one chunk and silently truncates the URI on the next. The source rows carry RightsStatements.org v1.0 URIs — the twelve-term controlled vocabulary a digitization campaign stamps onto every asset — but the destination columns are a rights_statement free-text field, a copyright_status enum, and an access_restrictions flag whose lengths and allowed values were fixed years before the vocabulary existed. The task this page resolves is the deterministic transformation between the two: take a raw, possibly malformed URI string and produce a queryable enum plus the LIDO and IIIF fields that drive the public portal, without ever losing the original identifier. This is the URI-resolution step feeding the verdict engine in automating copyright status checks, and a sibling to the free-license path in routing Creative Commons licenses.

Root Cause Analysis

Three architectural mismatches produce every failure in this stage, and each has a distinct fix.

First, URI canonicalization drifts. Legacy ETL scripts strip trailing slashes, downgrade https:// to http://, and append tracking or language query parameters during string concatenation. RightsStatements.org URIs are opaque identifiers whose canonical form is https://rightsstatements.org/vocab/{TERM}/1.0/ — trailing slash included — so an exact-match validator rejects http://rightsstatements.org/vocab/InC/1.0 even though it names a valid statement. The fix is to normalize before you validate, never after.

Second, CMS schema rigidity truncates the value. Many collection tables enforce VARCHAR(50) on the rights column, and a full URI overruns it, so the database quietly stores a fragment that no longer resolves. Storing a short, stable enum in the queryable column and keeping the full URI in a wide audit column removes the length constraint from the hot path.

Third, and most dangerous, namespace collisions misclassify the record. A mixed batch interleaves RightsStatements.org URIs, Creative Commons licenses, and legacy in-house copyright codes. A parser that matches the first pattern it sees can read InC as “in collection” or fall through an unmatched URI to a “public domain” default. Publishing a still-protected work because the pipeline guessed public domain is the single most expensive failure mode in rights automation. The invariant that prevents it is simple: any URI the classifier does not positively recognize as In Copyright or No Copyright resolves to UNKNOWN and routes to human review — the same fail-safe posture used when handling orphan works in digital collections.

A fourth, non-semantic vector compounds the first three at scale: loading the entire export into RAM for row-wise transformation makes memory grow linearly with file size and triggers OOM kills on constrained runners — the same trap covered for large CSV batches. The solution below streams fixed-size chunks so peak memory is bounded by batch size, not row count.

Canonical Solution

The pipeline has three composable parts: a Polars expression that normalizes and classifies a whole chunk vectorized, a Pydantic v2 model that enforces the URI contract per record before any write, and a batched reader that keeps exactly one chunk reachable at a time. The diagram traces one chunk through the cycle.

RightsStatements.org URI to collection-field transformation pipeline A left-to-right flow: a CSV chunk feeds a Normalize URI step, which feeds a Classify family decision. The decision fans out to three stacked enum boxes — IN_COPYRIGHT for the InC family, NO_COPYRIGHT for the NoC family, and a dashed UNKNOWN box (CNE, UND, NKC, or unrecognized) marked as the fail-safe to human review. All three converge on a Validate against VALID_RS_URIS step, which emits the combined LIDO rightsType and IIIF rights output. CSV chunk read_csv_batched Normalize URI strip · https · single trailing slash Classify by prefix IN_COPYRIGHT InC family NO_COPYRIGHT NoC family UNKNOWN CNE · UND · NKC · unrecognized fail-safe → human review, never public domain Validate vs VALID_RS_URIS LIDO rightsType IIIF rights

The VALID_RS_URIS set is the authority the validator checks against — it enumerates the twelve v1.0 terms so a normalized URI that is not in the set is rejected as data corruption rather than silently mapped. The classifier keys on the term prefix (InC, NoC) so every member of a family routes identically, and the otherwise branch is the fail-safe catch-all.

python
# pipeline/rights_mapper.py
import polars as pl
from typing import Literal
from urllib.parse import urlparse
from pydantic import BaseModel, ConfigDict, field_validator

# The complete RightsStatements.org v1.0 controlled vocabulary.
VALID_RS_URIS: set[str] = {
    # In Copyright family
    "https://rightsstatements.org/vocab/InC/1.0/",
    "https://rightsstatements.org/vocab/InC-OW-EU/1.0/",
    "https://rightsstatements.org/vocab/InC-EDU/1.0/",
    "https://rightsstatements.org/vocab/InC-NC/1.0/",
    "https://rightsstatements.org/vocab/InC-RUU/1.0/",
    # No Copyright family (public-domain-equivalent, but the vocabulary
    # never asserts "public domain" itself — that comes from CC PDM/CC0).
    "https://rightsstatements.org/vocab/NoC-CR/1.0/",
    "https://rightsstatements.org/vocab/NoC-NC/1.0/",
    "https://rightsstatements.org/vocab/NoC-OKLR/1.0/",
    "https://rightsstatements.org/vocab/NoC-US/1.0/",
    # Other — every one of these means "status not determined".
    "https://rightsstatements.org/vocab/CNE/1.0/",
    "https://rightsstatements.org/vocab/UND/1.0/",
    "https://rightsstatements.org/vocab/NKC/1.0/",
}

CmsEnum = Literal["IN_COPYRIGHT", "PUBLIC_DOMAIN", "NO_COPYRIGHT", "UNKNOWN"]

class RightsMapping(BaseModel):
    model_config = ConfigDict(str_strip_whitespace=True)

    original_uri: str       # preserved verbatim for audit and manual review
    canonical_uri: str      # normalized form checked against the vocabulary
    cms_enum: CmsEnum
    lido_element: str
    iiif_rights_field: str

    @field_validator("canonical_uri", mode="before")
    @classmethod
    def normalize_uri(cls, v: str) -> str:
        # Force https, drop query/fragment, collapse to the single-slash
        # canonical form. Runs BEFORE validation so a well-formed URI with a
        # missing slash or http scheme is repaired, not rejected.
        p = urlparse(str(v).strip())
        path = p.path.rstrip("/")
        return f"https://{p.netloc}{path}/"

    @field_validator("canonical_uri")
    @classmethod
    def validate_rs_uri(cls, v: str) -> str:
        if v not in VALID_RS_URIS:
            raise ValueError(f"Unrecognized RightsStatements.org URI: {v}")
        return v


def classify(normalized: pl.Expr) -> pl.Expr:
    # Order matters only for readability here: the families are disjoint.
    return (
        pl.when(normalized.str.contains("/InC")).then(pl.lit("IN_COPYRIGHT"))
          .when(normalized.str.contains("/NoC")).then(pl.lit("NO_COPYRIGHT"))
          # CNE (Copyright Not Evaluated), UND (Undetermined), NKC (No Known
          # Copyright) all mean "unknown" — they must NEVER fall through to
          # PUBLIC_DOMAIN. The catch-all enforces that for any stray pattern.
          .otherwise(pl.lit("UNKNOWN"))
          .alias("cms_enum")
    )


def process_chunk(df: pl.DataFrame) -> pl.DataFrame:
    normalized = (
        df["raw_rights_uri"]
        .str.strip_chars()
        .str.replace(r"^http://", "https://")
        .str.replace_all(r"/+$", "")
        + "/"
    )
    return df.with_columns(
        normalized.alias("canonical_uri"),
        classify(normalized),
    )


def stream_pipeline(source_path: str, chunk_size: int = 10_000):
    # read_csv_batched yields true row chunks; LazyFrame.head only returns the
    # first n rows once and is NOT a chunk iterator — do not substitute it.
    reader = pl.read_csv_batched(source_path, batch_size=chunk_size)
    batches = reader.next_batches(1)
    while batches:
        yield process_chunk(batches[0])   # one chunk reachable at a time
        batches = reader.next_batches(1)

Note that cms_enum includes PUBLIC_DOMAIN in its Literal, but the RightsStatements.org classifier never emits it: the vocabulary has no public-domain statement, so that enum value is reserved for records arriving through the Creative Commons path — see automating CC BY-NC-ND tagging in Python. Keeping the value in the shared contract lets both pipelines write to one copyright_status column without a schema fork.

Edge Cases and Variants

  • XML / LIDO input instead of CSV. Replace read_csv_batched with xml.etree.ElementTree.iterparse, pulling the URI out of lido:rightsType/lido:conceptID, and call elem.clear() after each record. The normalize/classify/validate tail is identical.
  • API payloads. When records arrive from an OAI-PMH or REST feed, iterate a paged cursor and feed each page through process_chunk on a pl.DataFrame.from_dicts(page). Cap concurrency to the connection-pool ceiling.
  • Strict vs. lenient validation. In a backfill, catch ValidationError, keep original_uri, and route the row to a quarantine table for a registrar; in an interactive re-sync, surface the error to the operator instead of skipping silently.
  • Mixed-vocabulary batches. If a chunk interleaves Creative Commons URIs, dispatch on host first (rightsstatements.org vs creativecommons.org) so each vocabulary hits its own classifier — never let one parser’s default absorb the other’s URIs.
  • Version drift. A 2.0 term would normalize cleanly but fail validate_rs_uri because the set is pinned to 1.0; treat that rejection as a signal to extend VALID_RS_URIS deliberately, not as a bug.

Validation

Prove the two invariants that keep the mapping safe — malformed-but-valid URIs are repaired rather than rejected, and no ambiguous term ever reaches PUBLIC_DOMAIN — with an assert-based test that needs no database:

python
# test_rights_mapper.py  ->  run with:  pytest -q test_rights_mapper.py
import polars as pl
import pytest
from pydantic import ValidationError
from pipeline.rights_mapper import RightsMapping, process_chunk

def test_http_and_missing_slash_are_normalized():
    m = RightsMapping(
        original_uri="http://rightsstatements.org/vocab/InC/1.0",
        canonical_uri="http://rightsstatements.org/vocab/InC/1.0",
        cms_enum="IN_COPYRIGHT", lido_element="rightsType", iiif_rights_field="rights",
    )
    assert m.canonical_uri == "https://rightsstatements.org/vocab/InC/1.0/"

def test_unrecognized_uri_is_rejected():
    with pytest.raises(ValidationError):
        RightsMapping(
            original_uri="https://rightsstatements.org/vocab/InC/2.0/",
            canonical_uri="https://rightsstatements.org/vocab/InC/2.0/",
            cms_enum="IN_COPYRIGHT", lido_element="rightsType", iiif_rights_field="rights",
        )

def test_ambiguous_terms_never_map_to_public_domain():
    df = pl.DataFrame({"raw_rights_uri": [
        "https://rightsstatements.org/vocab/CNE/1.0/",
        "https://rightsstatements.org/vocab/UND/1.0/",
        "https://rightsstatements.org/vocab/NKC/1.0/",
        "https://rightsstatements.org/vocab/mystery/1.0/",
    ]})
    out = process_chunk(df)
    assert out["cms_enum"].to_list() == ["UNKNOWN"] * 4

A green run demonstrates that scheme and slash drift are repaired, that an out-of-vocabulary URI is caught at the row boundary, and that every “status not determined” term — including an unrecognized one — resolves to UNKNOWN rather than opening the asset.

Standards Alignment

The mapping conforms to three specifications at once. Under LIDO, the canonical URI populates lido:rightsType/lido:conceptID with lido:source set to RightsStatements.org, and the cms_enum drives the human-readable lido:term, so the record the LIDO-to-database mapping layer persists is round-trippable back to XML. For delivery, the IIIF Presentation API 3.0 rights property requires exactly one dereferenceable URI from an approved set that RightsStatements.org satisfies — the iiif_rights_field column carries the canonical form straight into the manifest, and an embargoed asset carries an InC URI until date-based embargo triggers flip it. Because the classifier keys on stable vocabulary prefixes rather than free text, it composes cleanly with authority reconciliation such as Getty AAT and TGN, and the strict Pydantic v2 contract is the same validation discipline applied in schema validation with Pydantic elsewhere in the pipeline. The urlparse normalization follows the standard-library urllib.parse semantics.

Frequently Asked Questions

Because the column is usually a short VARCHAR enum that other systems query and index. Store the stable cms_enum there for fast filtering and keep the full canonical_uri in a wide audit column. That separation removes the truncation risk from the hot path while preserving the exact identifier for export and review.

What is the difference between NoC and UNKNOWN, and why not call NoC public domain?

The NoC family asserts there is no copyright for a stated reason (expired, contractual, US federal work), so it maps to NO_COPYRIGHT. RightsStatements.org deliberately does not assert “public domain” — that positive claim belongs to Creative Commons PDM or CC0. UNKNOWN covers CNE, UND, and NKC, where the status was never determined. Collapsing any of these into PUBLIC_DOMAIN is the misclassification that publishes protected works.

How do I handle a URI that normalizes cleanly but is not in VALID_RS_URIS?

Let the Pydantic validator raise, catch the ValidationError, and route the record to a quarantine table with its original_uri intact. A clean-but-unrecognized URI is almost always a new vocabulary version or a typo — both warrant a human decision, not a silent default.

Does streaming with read_csv_batched change the classification results?

No. Chunking only bounds memory; each row is normalized, classified, and validated identically whether it arrives in a 1,000-row or 100,000-row batch. Keep exactly one chunk reachable at a time so peak memory tracks batch size rather than file size.