nordisk/korp-collectors
src/dcat.rs
//! **#1 DCAT-AP-SE crawler** — the highest-leverage collector. It enumerates
//! *every* dataset + its distributions across the whole Swedish public sector by
//! parsing the dataportal.se (DIGG / EntryScape) catalogue, and emits a
//! normalised `(dataset, distribution{download_url, access_url, format, license,
//! publisher})` **registry** table into Iceberg. That registry then drives the
//! bulk/REST fetchers (#5/#7) their URLs.
//!
//! Two live inputs (config `mode`):
//! * `"rdf"` — GET the nightly RDF/XML dump (`https://admin.dataportal.se/all.rdf`)
//! and parse it once. (default)
//! * `"sparql"`— POST a SPARQL query to a SPARQL endpoint
//! (`https://data.europa.eu/data/sparql`) for **EU-wide** discovery, parsing
//! the SPARQL-results JSON into the same registry schema.
//!
//! RDF parsing is pure-Rust via `oxrdfxml` (oxigraph's RDF/XML reader) → `oxrdf`
//! triples. The offline contract is [`parse_dcat_rdfxml`]: a sample DCAT-AP-SE
//! RDF/XML document → registry rows, asserted as DATA.
use std::collections::HashMap;
use anyhow::{Result, anyhow};
use arrow_array::RecordBatch;
use arrow_schema::{DataType, Schema};
use oxrdf::{NamedOrBlankNode, Term};
use oxrdfxml::RdfXmlParser;
use super::arrow_util::{batch, schema, str_col};
use super::{
AuthKind, BatchSink, Bounds, CollectStats, Collector, CollectorCfg, CollectorMeta,
CollectorPath, Direction, GdprClass, SourceEndpoint, SourceKind, SourceTarget,
};
// ── DCAT / DCT / FOAF vocabulary IRIs ──────────────────────────────────────
const RDF_TYPE: &str = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type";
const DCAT_DATASET: &str = "http://www.w3.org/ns/dcat#Dataset";
const DCAT_DISTRIBUTION: &str = "http://www.w3.org/ns/dcat#distribution";
const DCAT_DOWNLOAD_URL: &str = "http://www.w3.org/ns/dcat#downloadURL";
const DCAT_ACCESS_URL: &str = "http://www.w3.org/ns/dcat#accessURL";
const DCAT_MEDIA_TYPE: &str = "http://www.w3.org/ns/dcat#mediaType";
const DCT_TITLE: &str = "http://purl.org/dc/terms/title";
const DCT_PUBLISHER: &str = "http://purl.org/dc/terms/publisher";
const DCT_LICENSE: &str = "http://purl.org/dc/terms/license";
const DCT_FORMAT: &str = "http://purl.org/dc/terms/format";
const FOAF_NAME: &str = "http://xmlns.com/foaf/0.1/name";
/// One row of the discovery registry — a (dataset, distribution) pair.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct RegistryRow {
pub dataset_uri: String,
pub dataset_title: Option<String>,
pub publisher: Option<String>,
pub dataset_license: Option<String>,
pub distribution_uri: Option<String>,
pub download_url: Option<String>,
pub access_url: Option<String>,
pub format: Option<String>,
pub media_type: Option<String>,
pub dist_license: Option<String>,
}
/// The registry's Arrow schema (all Utf8, nullable except `dataset_uri`).
pub fn registry_schema() -> Schema {
schema(&[
("dataset_uri", DataType::Utf8, false),
("dataset_title", DataType::Utf8, true),
("publisher", DataType::Utf8, true),
("dataset_license", DataType::Utf8, true),
("distribution_uri", DataType::Utf8, true),
("download_url", DataType::Utf8, true),
("access_url", DataType::Utf8, true),
("format", DataType::Utf8, true),
("media_type", DataType::Utf8, true),
("dist_license", DataType::Utf8, true),
])
}
/// Registry rows → one `RecordBatch`.
pub fn registry_batch(rows: &[RegistryRow]) -> Result<RecordBatch> {
let req = |f: fn(&RegistryRow) -> Option<String>| rows.iter().map(f).collect::<Vec<_>>();
let cols = vec![
str_col(
&rows
.iter()
.map(|r| Some(r.dataset_uri.clone()))
.collect::<Vec<_>>(),
),
str_col(&req(|r| r.dataset_title.clone())),
str_col(&req(|r| r.publisher.clone())),
str_col(&req(|r| r.dataset_license.clone())),
str_col(&req(|r| r.distribution_uri.clone())),
str_col(&req(|r| r.download_url.clone())),
str_col(&req(|r| r.access_url.clone())),
str_col(&req(|r| r.format.clone())),
str_col(&req(|r| r.media_type.clone())),
str_col(&req(|r| r.dist_license.clone())),
];
batch(®istry_schema(), cols)
}
// ── RDF triple store (subject → predicate → objects) ───────────────────────
type Graph = HashMap<String, HashMap<String, Vec<Term>>>;
fn subj_key(s: &NamedOrBlankNode) -> String {
match s {
NamedOrBlankNode::NamedNode(n) => n.as_str().to_string(),
NamedOrBlankNode::BlankNode(b) => format!("_:{}", b.as_str()),
}
}
/// A term's node identity (for linking dataset→distribution nodes).
fn node_key(t: &Term) -> Option<String> {
match t {
Term::NamedNode(n) => Some(n.as_str().to_string()),
Term::BlankNode(b) => Some(format!("_:{}", b.as_str())),
_ => None,
}
}
/// A term's lexical value: IRI for a node, the literal value for a literal.
fn lexical(t: &Term) -> String {
match t {
Term::NamedNode(n) => n.as_str().to_string(),
Term::BlankNode(b) => format!("_:{}", b.as_str()),
Term::Literal(l) => l.value().to_string(),
}
}
fn first<'a>(g: &'a Graph, subj: &str, pred: &str) -> Option<&'a Term> {
g.get(subj)
.and_then(|po| po.get(pred))
.and_then(|ts| ts.first())
}
fn first_lexical(g: &Graph, subj: &str, pred: &str) -> Option<String> {
first(g, subj, pred).map(lexical)
}
/// Resolve a publisher node to a human name: prefer `foaf:name`, else the node IRI.
fn publisher_of(g: &Graph, ds: &str) -> Option<String> {
let node = first(g, ds, DCT_PUBLISHER)?;
if let Some(key) = node_key(node) {
if let Some(name) = first_lexical(g, &key, FOAF_NAME) {
return Some(name);
}
}
Some(lexical(node))
}
/// Parse a DCAT-AP-SE **RDF/XML** document into registry rows — the offline
/// contract. One row per (dataset, distribution); a dataset with no
/// distribution yields one row with empty distribution fields.
pub fn parse_dcat_rdfxml(bytes: &[u8]) -> Result<Vec<RegistryRow>> {
let mut g: Graph = HashMap::new();
let mut types: HashMap<String, Vec<String>> = HashMap::new();
for triple in RdfXmlParser::new().for_slice(bytes) {
let t = triple.map_err(|e| anyhow!("RDF/XML parse: {e}"))?;
let s = subj_key(&t.subject);
let p = t.predicate.as_str().to_string();
if p == RDF_TYPE {
if let Some(k) = node_key(&t.object) {
types.entry(s.clone()).or_default().push(k);
}
}
g.entry(s).or_default().entry(p).or_default().push(t.object);
}
// Every subject typed dcat:Dataset, in a stable order.
let mut datasets: Vec<String> = types
.iter()
.filter(|(_, ts)| ts.iter().any(|t| t == DCAT_DATASET))
.map(|(s, _)| s.clone())
.collect();
datasets.sort();
let mut rows = Vec::new();
for ds in datasets {
let title = first_lexical(&g, &ds, DCT_TITLE);
let publisher = publisher_of(&g, &ds);
let ds_license = first_lexical(&g, &ds, DCT_LICENSE);
let dists: Vec<String> = g
.get(&ds)
.and_then(|po| po.get(DCAT_DISTRIBUTION))
.map(|ts| ts.iter().filter_map(node_key).collect())
.unwrap_or_default();
if dists.is_empty() {
rows.push(RegistryRow {
dataset_uri: ds.clone(),
dataset_title: title.clone(),
publisher: publisher.clone(),
dataset_license: ds_license.clone(),
..Default::default()
});
continue;
}
let mut dists = dists;
dists.sort();
for d in dists {
rows.push(RegistryRow {
dataset_uri: ds.clone(),
dataset_title: title.clone(),
publisher: publisher.clone(),
dataset_license: ds_license.clone(),
distribution_uri: Some(d.clone()),
download_url: first_lexical(&g, &d, DCAT_DOWNLOAD_URL),
access_url: first_lexical(&g, &d, DCAT_ACCESS_URL),
format: first_lexical(&g, &d, DCT_FORMAT),
media_type: first_lexical(&g, &d, DCAT_MEDIA_TYPE),
dist_license: first_lexical(&g, &d, DCT_LICENSE),
});
}
}
Ok(rows)
}
/// Parse a SPARQL-results-JSON response (EU `data.europa.eu/data/sparql`) into
/// registry rows. Expects bindings named `dataset`,`title`,`publisher`,
/// `license`,`distribution`,`downloadURL`,`accessURL`,`format`,`mediaType`.
pub fn parse_sparql_results(json: &serde_json::Value) -> Result<Vec<RegistryRow>> {
let bindings = json
.get("results")
.and_then(|r| r.get("bindings"))
.and_then(|b| b.as_array())
.ok_or_else(|| anyhow!("SPARQL JSON: no results.bindings array"))?;
let val = |b: &serde_json::Value, k: &str| {
b.get(k)
.and_then(|v| v.get("value"))
.and_then(|v| v.as_str())
.map(str::to_string)
};
let mut rows = Vec::with_capacity(bindings.len());
for b in bindings {
let dataset_uri = match val(b, "dataset") {
Some(d) => d,
None => continue,
};
rows.push(RegistryRow {
dataset_uri,
dataset_title: val(b, "title"),
publisher: val(b, "publisher"),
dataset_license: val(b, "license"),
distribution_uri: val(b, "distribution"),
download_url: val(b, "downloadURL"),
access_url: val(b, "accessURL"),
format: val(b, "format"),
media_type: val(b, "mediaType"),
dist_license: val(b, "license"),
});
}
Ok(rows)
}
/// A canned SPARQL CONSTRUCT/SELECT for EU dataset discovery (publisher country
/// SE by default). Kept here so the live `sparql` mode and any docs share it.
pub fn eu_discovery_query(limit: usize) -> String {
format!(
"PREFIX dcat: <http://www.w3.org/ns/dcat#> \
PREFIX dct: <http://purl.org/dc/terms/> \
SELECT ?dataset ?title ?publisher ?license ?distribution ?downloadURL ?accessURL ?format ?mediaType WHERE {{ \
?dataset a dcat:Dataset . \
OPTIONAL {{ ?dataset dct:title ?title }} \
OPTIONAL {{ ?dataset dct:publisher ?publisher }} \
OPTIONAL {{ ?dataset dct:license ?license }} \
OPTIONAL {{ ?dataset dcat:distribution ?distribution . \
OPTIONAL {{ ?distribution dcat:downloadURL ?downloadURL }} \
OPTIONAL {{ ?distribution dcat:accessURL ?accessURL }} \
OPTIONAL {{ ?distribution dct:format ?format }} \
OPTIONAL {{ ?distribution dcat:mediaType ?mediaType }} }} \
}} LIMIT {limit}"
)
}
/// The DCAT-AP-SE crawler collector.
#[derive(Default)]
pub struct DcatCrawler;
#[async_trait::async_trait]
impl Collector for DcatCrawler {
fn meta(&self) -> CollectorMeta {
CollectorMeta {
name: "dcat-ap-se".into(),
source_kind: SourceKind::Dcat,
path: CollectorPath::RustNative,
direction: Direction::Source,
formats: vec!["rdf/xml".into(), "turtle".into(), "sparql-json".into()],
license: "open (catalogue metadata, CC0/CC-BY)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
join_keys: vec!["dataset_uri".into(), "download_url".into()],
description: "dataportal.se DCAT-AP-SE registry crawl (+ EU data.europa.eu SPARQL); \
discovers every SE dataset + its distribution URLs."
.into(),
}
}
fn source_target(&self) -> SourceTarget {
SourceTarget::warehouse(
SourceEndpoint::DcatCatalog,
"https://admin.dataportal.se/all.rdf",
"dcat_registry",
)
}
/// **The one collector whose data owner is not its authority, and says so.**
///
/// DIGG runs Sveriges dataportal, but DIGG owns almost none of what it serves: the
/// portal carries **harvested metadata** describing datasets that belong to a few
/// hundred other Swedish public bodies. So `authority` and `owner` genuinely
/// differ here, and the licence an operator actually needs is not the portal's —
/// it is the per-dataset one this crawler already emits as the `dataset_license`
/// and `dist_license` COLUMNS. That is the most important thing on this card, and
/// it is the first quirk.
///
/// Verified live 2026-09-04: `docs.dataportal.se` carries "Copyright © 2025 DIGG
/// – Myndigheten för digital förvaltning", and `info@digg.se` is published on
/// DIGG's own pages for the portal. No rate limit was found published for either
/// the nightly dump or the EU SPARQL endpoint, so [`RateLimit::Unknown`] says so
/// rather than implying permission.
///
/// [`RateLimit::Unknown`]: crate::human_doc::RateLimit::Unknown
fn human_doc(&self) -> Option<crate::human_doc::HumanDoc> {
use crate::human_doc::{Contact, HumanDoc};
Some(
HumanDoc::new(
"DIGG — Myndigheten för digital förvaltning (the Swedish Agency for Digital \
Government), which runs Sveriges dataportal; in `sparql` mode the query goes \
instead to the EU Open Data Portal (data.europa.eu)",
"each publishing Swedish public-sector body — the portal holds only harvested \
DCAT-AP-SE metadata, and every emitted row carries the owner's own terms in \
`dataset_license` / `dist_license`",
"https://www.dataportal.se/anvandarvillkor",
"https://docs.dataportal.se/",
)
.with_contact(Contact::Email("info@digg.se".into()))
.with_quirk(
"The licence that matters is PER DATASET and is already a column. The portal's \
own terms cover the catalogue metadata; whether a given distribution may be \
downloaded and republished is what `dataset_license` / `dist_license` say. \
`meta().license` describes the crawl, not the harvest.",
)
.with_quirk(
"`params.mode` picks between two completely different live inputs and `base_url` \
must match the one chosen: `\"rdf\"` (the default) GETs a nightly RDF/XML dump \
— `https://admin.dataportal.se/all.rdf` — while `\"sparql\"` sends a \
percent-encoded SELECT to a SPARQL endpoint such as \
`https://data.europa.eu/data/sparql`. Pointing one mode at the other's URL is \
the usual first failure.",
)
.with_quirk(
"The RDF mode is a WHOLE-CATALOGUE download with no date filter, so every run \
re-reads the entire Swedish catalogue. There is no incremental key here; treat \
it as a scheduled full snapshot, not a poll.",
)
.with_quirk(
"`rows` counts (dataset, distribution) PAIRS, not datasets: a dataset with no \
`dcat:distribution` still emits exactly one row, with every distribution column \
null. Counting datasets means counting distinct `dataset_uri`.",
)
.with_quirk(
"Distributions are frequently blank nodes, and `distribution_uri` then carries a \
`_:`-prefixed local id that is NOT stable between dumps. Join on `download_url` \
instead — which is why it, and not the distribution URI, is a declared join \
key.",
)
.with_quirk(
"`publisher` resolves `foaf:name` when the publisher node has one and falls back \
to that node's raw IRI when it does not, so the column legitimately mixes human \
names with URIs. Normalise before grouping by it.",
)
.with_quirk(
"In `sparql` mode `dist_license` is filled from the same `license` binding as \
`dataset_license` (the canned query in [`eu_discovery_query`] binds one \
`?license`), so the two columns agree by construction there and only differ on \
the RDF path.",
),
)
}
/// **One row per DISTRIBUTION**, which is what a DCAT registry row is: a dataset
/// fanned out over its download formats.
///
/// The derived key was `["dataset_uri", "download_url"]` and `download_url` is
/// NULLABLE — a distribution that offers only an `access_url` (a landing page, a
/// query service) has none, and every such distribution of one dataset would have
/// keyed identically. `distribution_uri` is the RDF subject and is always present.
fn business_key(&self) -> crate::history::BusinessKey {
crate::history::BusinessKey::of(
&["dataset_uri", "distribution_uri"],
"one row per distribution — `distribution_uri` is the RDF subject and is \
always present, while `download_url` is null for an access-URL-only \
distribution",
)
}
/// **The DCAT catalogue as a graph** (lane EC O7): a dataset, who published it,
/// what it is distributed as, and under which licence and format.
fn graph_mapping(&self) -> Option<crate::graph_mapping::GraphMapping> {
use crate::graph_mapping::GraphMapping;
Some(
GraphMapping::new(
"dcat_catalog",
"dataportal.se — DCAT-registret",
"dcat_registry",
"a catalogue row carries no geography, so nothing here joins to the \
kommun spine; publisher is a NAME, not an orgnummer, so two spellings \
of one agency are two nodes.",
)
.vertex("Dataset", "dataset_uri", &["dataset_title", "dataset_license"])
// The distribution is a SECOND entity carried by the same row — it gets
// its own vertex (with its own properties) and its own outgoing edge.
.vertex(
"Distribution",
"distribution_uri",
&["download_url", "access_url", "media_type", "dist_license"],
)
.edge(
"PUBLISHED_BY",
("Dataset", "dataset_uri", "dataset_uri"),
("Organisation", "publisher", "publisher"),
)
.edge(
"HAS_DISTRIBUTION",
("Dataset", "dataset_uri", "dataset_uri"),
("Distribution", "distribution_uri", "distribution_uri"),
)
.edge(
"UNDER_LICENSE",
("Dataset", "dataset_uri", "dataset_uri"),
("License", "dataset_license", "dataset_license"),
)
.edge(
"IN_FORMAT",
("Distribution", "distribution_uri", "distribution_uri"),
("Format", "format", "format"),
),
)
}
fn schema(&self, _cfg: &CollectorCfg) -> Result<Schema> {
Ok(registry_schema())
}
async fn collect(
&self,
cfg: &CollectorCfg,
_bounds: Bounds,
sink: &BatchSink,
) -> Result<CollectStats> {
let http = cfg.http(10);
let auth = cfg.auth.resolve_async().await?;
let mode = cfg
.params
.get("mode")
.and_then(|v| v.as_str())
.unwrap_or("rdf");
let rows = match mode {
"sparql" => {
let limit = cfg
.params
.get("limit")
.and_then(|v| v.as_u64())
.unwrap_or(1000) as usize;
let query = eu_discovery_query(limit);
let url = format!(
"{}?query={}",
cfg.base_url,
super::http::percent_encode(&query)
);
let body = http.get_text(&url, auth.as_ref()).await?;
let json: serde_json::Value = serde_json::from_str(&body)?;
parse_sparql_results(&json)?
}
_ => {
// Nightly RDF/XML dump (default).
let bytes = http.get_bytes(&cfg.base_url, auth.as_ref()).await?;
parse_dcat_rdfxml(&bytes)?
}
};
if rows.is_empty() {
return Ok(CollectStats {
reason: Some("catalogue returned no datasets".into()),
..Default::default()
});
}
let n = rows.len() as u64;
sink.push(registry_batch(&rows)?)?;
Ok(CollectStats {
rows: n,
batches: 1,
table: cfg.table.clone(),
reason: None,
})
}
}