Automating ODbL Attribution in Derived Products Jump to heading#

from __future__ import annotations

import json
import logging
import sqlite3
from dataclasses import dataclass
from pathlib import Path

import pyarrow as pa
import pyarrow.parquet as pq

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
logger = logging.getLogger("osm.licence.attribution")

CREDIT_URL = "https://www.openstreetmap.org/copyright"


class MissingAttribution(RuntimeError):
    """An artefact is about to ship without the credit it owes."""


@dataclass(frozen=True)
class Provenance:
    source_name: str        # e.g. "Geofabrik poland-latest"
    source_date: str        # ISO date the extract describes
    digest: str             # the published checksum actually consumed

    def credit(self) -> str:
        return (f"© OpenStreetMap contributors, {CREDIT_URL} — "
                f"derived from {self.source_name} ({self.source_date}), "
                f"ODbL 1.0")

    def as_metadata(self) -> dict[str, str]:
        return {
            "attribution": self.credit(),
            "licence": "ODbL-1.0",
            "source_name": self.source_name,
            "source_date": self.source_date,
            "source_digest": self.digest,
        }


def tag_mbtiles(path: Path, prov: Provenance) -> None:
    """Tile archives advertise attribution to clients through metadata."""
    conn = sqlite3.connect(path)
    try:
        for key, value in prov.as_metadata().items():
            conn.execute(
                "INSERT INTO metadata (name, value) VALUES (?, ?) "
                "ON CONFLICT(name) DO UPDATE SET value = excluded.value",
                (key, value))
        conn.commit()
    finally:
        conn.close()
    logger.info("tagged %s", path.name)


def tag_parquet(src: Path, dst: Path, prov: Provenance) -> None:
    """Rewrite with schema metadata: the credit travels inside the file."""
    table = pq.read_table(src)
    existing = table.schema.metadata or {}
    merged = dict(existing)
    merged.update({k.encode(): v.encode()
                   for k, v in prov.as_metadata().items()})
    pq.write_table(table.replace_schema_metadata(merged), dst)
    logger.info("wrote %s with schema metadata", dst.name)


def tag_database(conn: sqlite3.Connection, prov: Provenance) -> None:
    conn.execute("""
        CREATE TABLE IF NOT EXISTS data_provenance (
            key TEXT PRIMARY KEY, value TEXT NOT NULL,
            recorded_at TEXT NOT NULL DEFAULT (datetime('now'))
        )""")
    for key, value in prov.as_metadata().items():
        conn.execute("INSERT INTO data_provenance (key, value) VALUES (?, ?) "
                     "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
                     (key, value))
    conn.commit()


# --- the gate ---------------------------------------------------------------

def check_mbtiles(path: Path) -> bool:
    conn = sqlite3.connect(path)
    try:
        row = conn.execute(
            "SELECT value FROM metadata WHERE name = 'attribution'").fetchone()
    finally:
        conn.close()
    return bool(row and "OpenStreetMap" in row[0])


def check_parquet(path: Path) -> bool:
    meta = pq.read_schema(path).metadata or {}
    value = meta.get(b"attribution", b"").decode()
    return "OpenStreetMap" in value


CHECKS = {".mbtiles": check_mbtiles, ".pmtiles": None, ".parquet": check_parquet}


def gate(artefacts: list[Path]) -> None:
    """Fail the build on anything shipping without a credit."""
    missing: list[Path] = []
    for artefact in artefacts:
        check = CHECKS.get(artefact.suffix)
        if check is None:
            # An unknown format is NOT a pass: it is a gap in this gate.
            logger.error("no attribution check for %s — add one", artefact.name)
            missing.append(artefact)
            continue
        if not check(artefact):
            missing.append(artefact)
    if missing:
        raise MissingAttribution(
            "artefacts without attribution: "
            + ", ".join(p.name for p in missing))
    logger.info("attribution present on all %d artefact(s)", len(artefacts))


if __name__ == "__main__":
    prov = Provenance("Geofabrik poland-latest", "2026-09-16", "a1b2c3d4")
    logger.info(prov.credit())