Add governed Statbel edition probe

backend/README.md

@@ -1444,8 +1444,8 @@ docker exec geointel python /app/scripts/audit_source_freshness.py \
 Use `--fail-on due` to make a planned review date fail automation, or
 `--fail-on never` for reporting only. The command never starts a refresh.
 
-An operator can explicitly add the official GRB, orthophoto and ALZ edition
-check:
+An operator can explicitly add the official GRB, orthophoto, Statbel and ALZ
+edition check:
 
 ```bash
 docker exec geointel python /app/scripts/audit_source_freshness.py \
@@ -1459,14 +1459,18 @@ Use `--refresh-catalogs` to bypass the 15-minute in-memory cache and
 `--fail-on-catalog` only when temporary official-provider unavailability must
 fail an operator job. This path reads bounded WFS/WMS capabilities and their
 fixed ISO 19139 metadata records. It confirms GRB `GBG`, `WBN`, `WGO`, `ADP`
-and orthophoto `Ortho`, `Vliegdagcontour`. It also reads the exact official ALZ
+and orthophoto `Ortho`, `Vliegdagcontour`. It parses the exact official Statbel
+DCAT Turtle catalog to identify the latest population-by-statistical-sector
+year, landing page, license and allowed distribution identities. It never
+follows those ZIP/XLSX links. It also reads the exact official ALZ
 publication page and validates only allowlisted archive-link identities. It
 never requests feature, raster or ALZ ZIP content. ALZ v1/v2 campaign snapshots
 remain provisional; only a v3 publication is compared with a local definitive
 historical edition.
 
 Runtime controls are `SOURCE_CATALOG_PROBE_ENABLED`,
-`SOURCE_CATALOG_GRB_WFS_URL`, `SOURCE_CATALOG_ALZ_RELEASE_URL`,
+`SOURCE_CATALOG_GRB_WFS_URL`, `SOURCE_CATALOG_STATBEL_DCAT_URL`,
+`SOURCE_CATALOG_STATBEL_MAX_RESPONSE_MB`, `SOURCE_CATALOG_ALZ_RELEASE_URL`,
 `SOURCE_CATALOG_PROBE_TIMEOUT_SECONDS`, `SOURCE_CATALOG_PROBE_MAX_RESPONSE_MB` and
 `SOURCE_CATALOG_PROBE_CACHE_TTL_SECONDS`. Metadata links remain restricted to
 the official HTTPS CSW path on `metadata.vlaanderen.be` even when an operator
@@ -1474,6 +1478,12 @@ overrides the capabilities endpoint. The ALZ release URL is fail-closed to the
 exact HTTPS host/path and cannot be redirected to another page or download
 host.
 
+The Statbel catalog has a separate 5 MiB default response bound. Population
+year, sector-geometry year and REDEGEO layout remain separate concepts: the
+2025 population release uses the new layout, while the concurrently published
+old layout is transition evidence only. The existence of 2026 sector geometry
+does not imply a 2026 population-by-sector release.
+
 ## Governed regional GRB refresh
 
 `GET /api/v1/projects/{project_id}/datasets/grb-refresh-plan` combines the

backend/app/core/config.py

@@ -40,6 +40,16 @@ class Settings(BaseSettings):
         default="https://landbouwcijfers.vlaanderen.be/open-geodata-landbouwgebruikspercelen",
         validation_alias="SOURCE_CATALOG_ALZ_RELEASE_URL",
     )
+    source_catalog_statbel_dcat_url: str = Field(
+        default="https://doc.statbel.be/publications/DCAT/DCAT_opendata_datasets.ttl",
+        validation_alias="SOURCE_CATALOG_STATBEL_DCAT_URL",
+    )
+    source_catalog_statbel_max_response_mb: int = Field(
+        default=5,
+        ge=1,
+        le=10,
+        validation_alias="SOURCE_CATALOG_STATBEL_MAX_RESPONSE_MB",
+    )
     source_catalog_probe_timeout_seconds: int = Field(
         default=10,
         ge=1,

backend/app/schemas/source_catalog.py

@@ -14,7 +14,7 @@ SourceCatalogComparisonStatus = Literal["same", "different", "not_comparable", "
 class SourceCatalogProbeItem(BaseModel):
     source_name: str
     display_name: str
-    service_type: Literal["WFS", "WMS", "HTML"]
+    service_type: Literal["WFS", "WMS", "HTML", "DCAT"]
     endpoint_url: str
     status: SourceCatalogProbeStatus
     reachable: bool

backend/app/services/source_catalog_probe_service.py

@@ -24,6 +24,11 @@ from app.schemas.source_catalog import (
     SourceCatalogProbeReport,
     SourceCatalogProbeSummary,
 )
+from app.services.statbel_catalog_probe import (
+    StatbelCatalogError,
+    parse_statbel_population_catalog,
+    validate_statbel_catalog_url,
+)
 
 
 _GMD = "http://www.isotc211.org/2005/gmd"
@@ -44,6 +49,7 @@ _ALZ_SNAPSHOT = re.compile(
     re.IGNORECASE,
 )
 _ALZ_EDITION = re.compile(r"^(20\d{2})-(?:definitive|v3)$", re.IGNORECASE)
+_YEAR_EDITION = re.compile(r"^20\d{2}$")
 
 
 class CatalogProbeFailure(RuntimeError):
@@ -159,12 +165,19 @@ def _with_capabilities_query(base_url: str, service_type: str) -> str:
     return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, query, ""))
 
 
-def _bounded_fetch(url: str, settings: Settings, opener: Callable[..., Any] | None = None) -> FetchResult:
+def _bounded_fetch(
+    url: str,
+    settings: Settings,
+    opener: Callable[..., Any] | None = None,
+    *,
+    max_response_mb: int | None = None,
+    accept: str = "application/xml,text/xml,text/html,application/json",
+) -> FetchResult:
     request = Request(
         url,
-        headers={"Accept": "application/xml,text/xml,text/html,application/json", "User-Agent": "GeoIntel/0.1 source-catalog-probe"},
+        headers={"Accept": accept, "User-Agent": "GeoIntel/0.1 source-catalog-probe"},
     )
-    max_bytes = settings.source_catalog_probe_max_response_mb * 1024 * 1024
+    max_bytes = (max_response_mb or settings.source_catalog_probe_max_response_mb) * 1024 * 1024
     try:
         with (opener or urlopen)(request, timeout=settings.source_catalog_probe_timeout_seconds) as response:
             content_length = _header(response.headers, "Content-Length")
@@ -509,6 +522,56 @@ def _probe_alz_remote(
     )
 
 
+def _probe_statbel_remote(
+    contract: ProbeContract,
+    settings: Settings,
+    *,
+    opener: Callable[..., Any] | None,
+    now: datetime,
+) -> RemoteProbe:
+    try:
+        catalog_url = validate_statbel_catalog_url(contract.endpoint_url)
+        response = _bounded_fetch(
+            catalog_url,
+            settings,
+            opener,
+            max_response_mb=settings.source_catalog_statbel_max_response_mb,
+            accept="text/turtle,application/x-turtle,application/octet-stream,text/plain",
+        )
+        validate_statbel_catalog_url(response.final_url)
+        content_type = response.content_type.lower()
+        if content_type and not any(token in content_type for token in ("turtle", "octet-stream", "text/plain")):
+            raise StatbelCatalogError(
+                "CATALOG_STATBEL_INVALID_CONTENT_TYPE",
+                "De officiële Statbel DCAT-catalogus heeft geen ondersteund Turtle-contenttype.",
+            )
+        release = parse_statbel_population_catalog(response.content)
+    except StatbelCatalogError as exc:
+        raise CatalogProbeFailure(exc.code, exc.message) from exc
+
+    layout = "nieuwe REDEGEO-sectorindeling" if release.current_distribution_variant == "new" else "actuele sectorindeling"
+    message = f"De officiële Statbel DCAT-catalogus bevestigt bevolkingseditie {release.version} met de {layout}."
+    if release.legacy_distribution_available:
+        message += " De oude 2025-indeling is alleen overgangsevidentie en wordt niet als actuele GeoIntel-editie gebruikt."
+    return RemoteProbe(
+        status="available",
+        reachable=True,
+        checked_at=now,
+        expected_layers=contract.expected_layers,
+        matched_layers=contract.expected_layers,
+        advertised_layer_count=release.distribution_count,
+        metadata_url=release.landing_page,
+        metadata_identifier=release.identifier,
+        remote_title=f"Bevolking per statistische sector {release.version} ({layout})",
+        remote_version=release.version,
+        remote_modified_at=release.catalog_modified_at,
+        capabilities_sha256=sha256(response.content).hexdigest(),
+        capabilities_etag=response.etag,
+        capabilities_last_modified_at=response.last_modified_at,
+        message=message,
+    )
+
+
 def _probe_remote(
     contract: ProbeContract,
     settings: Settings,
@@ -519,6 +582,8 @@ def _probe_remote(
     try:
         if contract.service_type == "HTML":
             return _probe_alz_remote(contract, settings, opener=opener, now=now)
+        if contract.service_type == "DCAT":
+            return _probe_statbel_remote(contract, settings, opener=opener, now=now)
         capabilities = _bounded_fetch(_validate_capabilities_url(contract.endpoint_url), settings, opener)
         _validate_capabilities_url(capabilities.final_url)
         content_type = capabilities.content_type.lower()
@@ -635,6 +700,10 @@ def _latest_local_version(source_name: str, datasets: list[Dataset]) -> str | No
                 ),
             )
             return latest.source_version
+    elif source_name == "statbel":
+        annual = [item for item in candidates if _YEAR_EDITION.fullmatch((item.source_version or "").strip())]
+        if annual:
+            return max(annual, key=lambda item: int((item.source_version or "0").strip())).source_version
     if not candidates:
         return None
     latest = max(
@@ -660,6 +729,8 @@ def _normalized_version(source_name: str, version: str | None) -> str | None:
     if source_name == _ALZ_SOURCE_NAME:
         match = _ALZ_EDITION.fullmatch(value)
         return f"{match.group(1)}-v3" if match else None
+    if source_name == "statbel" and _YEAR_EDITION.fullmatch(value):
+        return value
     return None
 
 
@@ -711,6 +782,13 @@ class SourceCatalogProbeService:
                 endpoint_url=_with_capabilities_query(active_settings.orthophoto_wms_url, "WMS"),
                 expected_layers=(active_settings.orthophoto_wms_layer, "Vliegdagcontour"),
             ),
+            ProbeContract(
+                source_name="statbel",
+                display_name="Bevolking per statistische sector (Statbel)",
+                service_type="DCAT",
+                endpoint_url=active_settings.source_catalog_statbel_dcat_url,
+                expected_layers=("population_txt_current", "landing_page", "cc_by_4_0"),
+            ),
             ProbeContract(
                 source_name=_ALZ_SOURCE_NAME,
                 display_name="Landbouwgebruikspercelen (ALZ)",
@@ -789,8 +867,9 @@ class SourceCatalogProbeService:
             summary=summary,
             items=items,
             limitations=[
-                "Deze expliciete controle leest alleen allowlisted WFS/WMS-capabilities, gekoppelde ISO 19139 metadata en de officiële ALZ-publicatiepagina.",
+                "Deze expliciete controle leest alleen allowlisted WFS/WMS-capabilities, gekoppelde ISO 19139 metadata, de officiële Statbel DCAT-catalogus en de officiële ALZ-publicatiepagina.",
                 "Er worden geen features, rasters of modelbestanden opgehaald en geen datasets aangemaakt of overschreven.",
+                "Statbel distributielinks worden alleen als release-evidentie gevalideerd; de oude en nieuwe 2025-sectorindeling blijven semantisch gescheiden.",
                 "Voor ALZ is alleen de nieuwste definitieve v3-editie vergelijkbaar; voorlopige v1/v2-snapshots zijn uitsluitend informatief.",
                 "Een versieverschil is controlesignaal, geen bewijs dat een lokale dataset onbruikbaar is en geen automatische importopdracht.",
             ],

backend/app/services/statbel_catalog_probe.py

@@ -0,0 +1,256 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+import re
+from urllib.parse import urlsplit
+
+from rdflib import Graph, Literal, URIRef
+from rdflib.namespace import RDF
+
+
+_DCAT = "http://www.w3.org/ns/dcat#"
+_DCT = "http://purl.org/dc/terms/"
+_EXPECTED_TITLE = "Bevolking per statistische sector"
+_CATALOG_HOST = "doc.statbel.be"
+_CATALOG_PATH = "/publications/DCAT/DCAT_opendata_datasets.ttl"
+_STATBEL_HOST = "statbel.fgov.be"
+_LANDING_PATH = re.compile(r"^/nl/open-data/bevolking-statistische-sector(?:-\d+)?$")
+_DISTRIBUTION_PATH = re.compile(
+    r"^/sites/default/files/files/opendata/bevolking/sectoren/"
+    r"OPENDATA_SECTOREN_(20\d{2})(?:_(NEW|OLD))?\.(zip|xlsx)$",
+    re.IGNORECASE,
+)
+_DISTRIBUTION_FRAGMENT = re.compile(r"^distribution\d+$")
+_ALTERNATIVE_YEAR = re.compile(r"\[Periode:\s*(20\d{2})\]", re.IGNORECASE)
+_CC_BY_4 = "https://creativecommons.org/licenses/by/4.0/"
+
+DCT_TITLE = URIRef(f"{_DCT}title")
+DCT_ALTERNATIVE = URIRef(f"{_DCT}alternative")
+DCT_IDENTIFIER = URIRef(f"{_DCT}identifier")
+DCT_LICENSE = URIRef(f"{_DCT}license")
+DCT_MODIFIED = URIRef(f"{_DCT}modified")
+DCT_TEMPORAL = URIRef(f"{_DCT}temporal")
+DCAT_CATALOG = URIRef(f"{_DCAT}Catalog")
+DCAT_DATASET = URIRef(f"{_DCAT}Dataset")
+DCAT_DISTRIBUTION = URIRef(f"{_DCAT}distribution")
+DCAT_LANDING_PAGE = URIRef(f"{_DCAT}landingPage")
+DCAT_START_DATE = URIRef(f"{_DCAT}startDate")
+
+
+class StatbelCatalogError(RuntimeError):
+    def __init__(self, code: str, message: str) -> None:
+        super().__init__(message)
+        self.code = code
+        self.message = message
+
+
+@dataclass(frozen=True)
+class StatbelPopulationRelease:
+    identifier: str
+    version: str
+    landing_page: str
+    catalog_modified_at: datetime | None
+    distribution_count: int
+    current_distribution_variant: str
+    legacy_distribution_available: bool
+
+
+def validate_statbel_catalog_url(url: str) -> str:
+    parsed = urlsplit(url)
+    if (
+        parsed.scheme != "https"
+        or parsed.hostname != _CATALOG_HOST
+        or parsed.port not in {None, 443}
+        or parsed.username
+        or parsed.password
+        or parsed.path != _CATALOG_PATH
+        or parsed.query
+        or parsed.fragment
+    ):
+        raise StatbelCatalogError(
+            "CATALOG_STATBEL_URL_REJECTED",
+            "De ingestelde Statbel DCAT-catalogus valt buiten de toegestane officiële URL.",
+        )
+    return url
+
+
+def _dutch_literal(values: list[object], expected: str | None = None) -> Literal | None:
+    for value in values:
+        if not isinstance(value, Literal) or value.language != "nl":
+            continue
+        if expected is None or str(value).strip() == expected:
+            return value
+    return None
+
+
+def _dataset_year(graph: Graph, subject: object) -> int | None:
+    years: set[int] = set()
+    for alternative in graph.objects(subject, DCT_ALTERNATIVE):
+        if isinstance(alternative, Literal) and alternative.language == "nl":
+            match = _ALTERNATIVE_YEAR.search(str(alternative))
+            if match:
+                years.add(int(match.group(1)))
+    for period in graph.objects(subject, DCT_TEMPORAL):
+        for value in graph.objects(period, DCAT_START_DATE):
+            match = re.match(r"^(20\d{2})-\d{2}-\d{2}$", str(value))
+            if match:
+                years.add(int(match.group(1)))
+    if len(years) > 1:
+        raise StatbelCatalogError(
+            "CATALOG_STATBEL_PERIOD_AMBIGUOUS",
+            "De officiële Statbel dataset bevat tegenstrijdige referentiejaren.",
+        )
+    return next(iter(years), None)
+
+
+def _validate_landing_page(url: str) -> str:
+    parsed = urlsplit(url)
+    if (
+        parsed.scheme != "https"
+        or parsed.hostname != _STATBEL_HOST
+        or parsed.port not in {None, 443}
+        or parsed.username
+        or parsed.password
+        or not _LANDING_PATH.fullmatch(parsed.path)
+        or parsed.query
+        or parsed.fragment
+    ):
+        raise StatbelCatalogError(
+            "CATALOG_STATBEL_LANDING_PAGE_REJECTED",
+            "De Statbel DCAT-dataset verwijst niet naar de toegestane Nederlandstalige landingspagina.",
+        )
+    return url
+
+
+def _validate_distribution(url: str, expected_year: int) -> tuple[str | None, str]:
+    parsed = urlsplit(url)
+    match = _DISTRIBUTION_PATH.fullmatch(parsed.path)
+    if (
+        parsed.scheme != "https"
+        or parsed.hostname != _STATBEL_HOST
+        or parsed.port not in {None, 443}
+        or parsed.username
+        or parsed.password
+        or parsed.query
+        or not match
+        or (parsed.fragment and not _DISTRIBUTION_FRAGMENT.fullmatch(parsed.fragment))
+    ):
+        raise StatbelCatalogError(
+            "CATALOG_STATBEL_DISTRIBUTION_REJECTED",
+            "De Statbel DCAT-dataset bevat een distributie buiten de toegestane officiële URL-structuur.",
+        )
+    if int(match.group(1)) != expected_year:
+        raise StatbelCatalogError(
+            "CATALOG_STATBEL_DISTRIBUTION_YEAR_MISMATCH",
+            "De Statbel distributie hoort niet bij het gepubliceerde referentiejaar.",
+        )
+    return match.group(2).lower() if match.group(2) else None, match.group(3).lower()
+
+
+def _catalog_modified_at(graph: Graph) -> datetime | None:
+    dates: list[datetime] = []
+    for catalog in graph.subjects(RDF.type, DCAT_CATALOG):
+        for value in graph.objects(catalog, DCT_MODIFIED):
+            try:
+                dates.append(datetime.fromisoformat(str(value)).replace(tzinfo=timezone.utc))
+            except ValueError:
+                continue
+    return max(dates) if dates else None
+
+
+def parse_statbel_population_catalog(content: bytes) -> StatbelPopulationRelease:
+    try:
+        text = content.decode("utf-8")
+    except UnicodeDecodeError as exc:
+        raise StatbelCatalogError(
+            "CATALOG_STATBEL_INVALID_TURTLE",
+            "De officiële Statbel DCAT-catalogus is niet geldige UTF-8 Turtle.",
+        ) from exc
+    graph = Graph()
+    try:
+        graph.parse(data=text, format="turtle")
+    except Exception as exc:
+        raise StatbelCatalogError(
+            "CATALOG_STATBEL_INVALID_TURTLE",
+            "De officiële Statbel DCAT-catalogus kon niet als RDF/Turtle worden gelezen.",
+        ) from exc
+
+    candidates: list[tuple[int, object]] = []
+    for subject in graph.subjects(RDF.type, DCAT_DATASET):
+        title = _dutch_literal(list(graph.objects(subject, DCT_TITLE)), _EXPECTED_TITLE)
+        if title is None:
+            continue
+        year = _dataset_year(graph, subject)
+        if year is not None:
+            candidates.append((year, subject))
+    if not candidates:
+        raise StatbelCatalogError(
+            "CATALOG_STATBEL_POPULATION_MISSING",
+            "De officiële Statbel DCAT-catalogus bevat geen herkenbare bevolking-per-sectorpublicatie.",
+        )
+    latest_year = max(year for year, _subject in candidates)
+    latest = [subject for year, subject in candidates if year == latest_year]
+    if len(latest) != 1:
+        raise StatbelCatalogError(
+            "CATALOG_STATBEL_POPULATION_AMBIGUOUS",
+            "De officiële Statbel DCAT-catalogus bevat meerdere bevolking-per-sectorpublicaties voor hetzelfde nieuwste jaar.",
+        )
+    subject = latest[0]
+
+    identifiers = [str(value).strip() for value in graph.objects(subject, DCT_IDENTIFIER) if str(value).strip()]
+    if len(set(identifiers)) != 1:
+        raise StatbelCatalogError(
+            "CATALOG_STATBEL_IDENTIFIER_MISSING",
+            "De nieuwste Statbel bevolking-per-sectorpublicatie heeft geen eenduidige datasetidentiteit.",
+        )
+    landing_pages = [
+        _validate_landing_page(str(value))
+        for value in graph.objects(subject, DCAT_LANDING_PAGE)
+        if urlsplit(str(value)).path.startswith("/nl/")
+    ]
+    if len(set(landing_pages)) != 1:
+        raise StatbelCatalogError(
+            "CATALOG_STATBEL_LANDING_PAGE_MISSING",
+            "De nieuwste Statbel bevolking-per-sectorpublicatie heeft geen eenduidige Nederlandstalige landingspagina.",
+        )
+    licenses = {str(value) for value in graph.objects(subject, DCT_LICENSE)}
+    if _CC_BY_4 not in licenses:
+        raise StatbelCatalogError(
+            "CATALOG_STATBEL_LICENSE_MISSING",
+            "De nieuwste Statbel bevolking-per-sectorpublicatie bevestigt de vereiste CC BY 4.0-licentie niet.",
+        )
+
+    distributions = [str(value) for value in graph.objects(subject, DCAT_DISTRIBUTION)]
+    if not distributions:
+        raise StatbelCatalogError(
+            "CATALOG_STATBEL_DISTRIBUTION_MISSING",
+            "De nieuwste Statbel bevolking-per-sectorpublicatie bevat geen distributies.",
+        )
+    variants: list[tuple[str | None, str]] = [
+        _validate_distribution(value, latest_year) for value in distributions
+    ]
+    zip_variants = {variant for variant, file_type in variants if file_type == "zip"}
+    if latest_year == 2025:
+        current_variant = "new" if "new" in zip_variants else ""
+    elif "new" in zip_variants:
+        current_variant = "new"
+    elif None in zip_variants:
+        current_variant = "standard"
+    else:
+        current_variant = ""
+    if not current_variant:
+        raise StatbelCatalogError(
+            "CATALOG_STATBEL_CURRENT_DISTRIBUTION_MISSING",
+            "De nieuwste Statbel bevolking-per-sectorpublicatie bevat geen herkenbare actuele TXT/ZIP-distributie.",
+        )
+
+    return StatbelPopulationRelease(
+        identifier=identifiers[0],
+        version=str(latest_year),
+        landing_page=landing_pages[0],
+        catalog_modified_at=_catalog_modified_at(graph),
+        distribution_count=len(distributions),
+        current_distribution_variant=current_variant,
+        legacy_distribution_available="old" in zip_variants,
+    )

backend/pyproject.toml

@@ -16,6 +16,7 @@ dependencies = [
     "shapely>=2.0.4",
     "pyproj>=3.6.1",
     "python-multipart>=0.0.9",
+    "rdflib>=7.1,<8",
     "alembic>=1.13.2",
 ]
 

backend/tests/test_sprint222_source_catalog_probes.py

@@ -22,6 +22,7 @@ ORTHO_METADATA_URL = (
     "id=f5304d6d-0dd4-43fd-a726-427af31e8d61&OUTPUTSCHEMA=http://www.isotc211.org/2005/gmd"
 )
 ALZ_RELEASE_URL = "https://landbouwcijfers.vlaanderen.be/open-geodata-landbouwgebruikspercelen"
+STATBEL_DCAT_URL = "https://doc.statbel.be/publications/DCAT/DCAT_opendata_datasets.ttl"
 
 
 def _wfs_capabilities(metadata_url: str = GRB_METADATA_URL) -> bytes:
@@ -95,6 +96,30 @@ def _alz_release_page(*, include_snapshot: bool = True, download_host: str = "ww
     """.encode()
 
 
+def _statbel_dcat() -> bytes:
+    return b"""
+    @prefix dcat: <http://www.w3.org/ns/dcat#> .
+    @prefix dct: <http://purl.org/dc/terms/> .
+    @prefix xsd: <http://www.w3.org/2001/XMLSchema#> .
+
+    <http://data.gov.be/catalog/statbelopen> a dcat:Catalog ;
+      dct:modified "2026-07-07"^^xsd:date .
+
+    <https://statbel.fgov.be/node/6475#id> a dcat:Dataset ;
+      dct:title "Bevolking per statistische sector"@nl ;
+      dct:alternative "Bevolking per statistische sector [Periode: 2025]"@nl ;
+      dct:identifier "NodeID6475" ;
+      dct:license <https://creativecommons.org/licenses/by/4.0/> ;
+      dct:temporal [ dcat:startDate "2025-01-01"^^xsd:date ] ;
+      dcat:landingPage <https://statbel.fgov.be/nl/open-data/bevolking-statistische-sector-12> ;
+      dcat:distribution
+        <https://statbel.fgov.be/sites/default/files/files/opendata/bevolking/sectoren/OPENDATA_SECTOREN_2025_NEW.zip#distribution6475>,
+        <https://statbel.fgov.be/sites/default/files/files/opendata/bevolking/sectoren/OPENDATA_SECTOREN_2025_NEW.xlsx#distribution6475>,
+        <https://statbel.fgov.be/sites/default/files/files/opendata/bevolking/sectoren/OPENDATA_SECTOREN_2025_OLD.zip#distribution6475>,
+        <https://statbel.fgov.be/sites/default/files/files/opendata/bevolking/sectoren/OPENDATA_SECTOREN_2025_OLD.xlsx#distribution6475> .
+    """
+
+
 class _Response:
     def __init__(self, content: bytes, content_type: str = "text/xml", *, content_length: int | None = None) -> None:
         self.content = content
@@ -149,7 +174,7 @@ def _dataset(source_name: str, version: str) -> Dataset:
         id=uuid.uuid4(),
         project_id=uuid.uuid4(),
         name=f"{source_name} source",
-        dataset_type="vector" if source_name == "grb" else "raster",
+        dataset_type="vector" if source_name in {"grb", "statbel", "agentschap_landbouw_zeevisserij_agricultural_parcels"} else "raster",
         source=source_name,
         source_name=source_name,
         source_version=version,
@@ -163,6 +188,8 @@ def _settings(**overrides) -> Settings:
         "SOURCE_CATALOG_PROBE_ENABLED": True,
         "SOURCE_CATALOG_GRB_WFS_URL": "https://geo.api.vlaanderen.be/GRB/wfs",
         "SOURCE_CATALOG_ALZ_RELEASE_URL": ALZ_RELEASE_URL,
+        "SOURCE_CATALOG_STATBEL_DCAT_URL": STATBEL_DCAT_URL,
+        "SOURCE_CATALOG_STATBEL_MAX_RESPONSE_MB": 5,
         "SOURCE_CATALOG_PROBE_TIMEOUT_SECONDS": 3,
         "SOURCE_CATALOG_PROBE_MAX_RESPONSE_MB": 1,
         "SOURCE_CATALOG_PROBE_CACHE_TTL_SECONDS": 0,
@@ -176,6 +203,8 @@ def _settings(**overrides) -> Settings:
 def _opener(request, timeout):
     assert timeout == 3
     url = request.full_url
+    if url == STATBEL_DCAT_URL:
+        return _Response(_statbel_dcat(), "application/octet-stream")
     if url == ALZ_RELEASE_URL:
         return _Response(_alz_release_page(), "text/html; charset=utf-8")
     if "metadata.vlaanderen.be" in url:
@@ -195,6 +224,7 @@ def test_catalog_probe_reads_real_editions_and_compares_only_compatible_versions
             [
                 _dataset("grb", "2026-07-14"),
                 _dataset("digitaal_vlaanderen_orthophoto", "most_recent_at_2026-07-14"),
+                _dataset("statbel", "2025"),
                 _dataset("agentschap_landbouw_zeevisserij_agricultural_parcels", "2025-definitive"),
             ]
         ),
@@ -205,7 +235,7 @@ def test_catalog_probe_reads_real_editions_and_compares_only_compatible_versions
     )
     by_source = {item.source_name: item for item in report.items}
 
-    assert report.summary.available_count == 3
+    assert report.summary.available_count == 4
     assert report.summary.different_version_count == 1
     assert by_source["grb"].remote_version == "Toestand 2026-07-15"
     assert by_source["grb"].comparison_status == "different"
@@ -214,6 +244,14 @@ def test_catalog_probe_reads_real_editions_and_compares_only_compatible_versions
     assert by_source["digitaal_vlaanderen_orthophoto"].remote_version == "2025.04"
     assert by_source["digitaal_vlaanderen_orthophoto"].comparison_status == "not_comparable"
     assert by_source["digitaal_vlaanderen_orthophoto"].remote_published_at.year == 2025
+    statbel = by_source["statbel"]
+    assert statbel.service_type == "DCAT"
+    assert statbel.remote_version == "2025"
+    assert statbel.local_source_version == "2025"
+    assert statbel.comparison_status == "same"
+    assert statbel.metadata_identifier == "NodeID6475"
+    assert statbel.matched_layers == ["population_txt_current", "landing_page", "cc_by_4_0"]
+    assert "REDEGEO" in statbel.message
     alz = by_source["agentschap_landbouw_zeevisserij_agricultural_parcels"]
     assert alz.service_type == "HTML"
     assert alz.remote_version == "2025-v3"
@@ -309,10 +347,10 @@ def test_catalog_probe_cache_is_explicitly_bypassable() -> None:
     second = SourceCatalogProbeService.audit_project(db, project_id, settings=settings, opener=counting_opener, now=NOW)
     SourceCatalogProbeService.audit_project(db, project_id, settings=settings, opener=counting_opener, now=NOW, force=True)
 
-    assert first_call_count == 5
+    assert first_call_count == 6
     assert all(item.cached is False for item in first.items)
     assert all(item.cached is True for item in second.items)
-    assert len(calls) == 10
+    assert len(calls) == 12
 
 
 def test_catalog_probe_can_be_disabled_without_network_access() -> None:
@@ -327,7 +365,7 @@ def test_catalog_probe_can_be_disabled_without_network_access() -> None:
         now=NOW,
     )
 
-    assert report.summary.disabled_count == 3
+    assert report.summary.disabled_count == 4
     assert all(item.status == "disabled" for item in report.items)
 
 
@@ -350,7 +388,7 @@ def test_catalog_probe_route_returns_canonical_envelope(monkeypatch) -> None:
 
     assert list(response) == ["data"]
     assert response["data"]["project_id"] == project_id
-    assert response["data"]["summary"]["provider_count"] == 3
+    assert response["data"]["summary"]["provider_count"] == 4
 
 
 def test_catalog_probe_remains_explicit_and_never_imports_provider_data() -> None:

backend/tests/test_sprint226_statbel_catalog_probe.py

@@ -0,0 +1,287 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from email.message import Message
+from types import SimpleNamespace
+from urllib.error import URLError
+import uuid
+
+import pytest
+
+from app.core.config import Settings
+from app.models import Dataset
+from app.services.source_catalog_probe_service import SourceCatalogProbeService
+from app.services.statbel_catalog_probe import StatbelCatalogError, parse_statbel_population_catalog
+
+
+NOW = datetime(2026, 7, 16, 18, 0, tzinfo=timezone.utc)
+STATBEL_DCAT_URL = "https://doc.statbel.be/publications/DCAT/DCAT_opendata_datasets.ttl"
+ALZ_RELEASE_URL = "https://landbouwcijfers.vlaanderen.be/open-geodata-landbouwgebruikspercelen"
+
+
+def _dataset_block(
+    year: int,
+    *,
+    node_id: int,
+    host: str = "statbel.fgov.be",
+    include_new_zip: bool = True,
+    include_standard_zip: bool = False,
+) -> str:
+    distributions = [
+        f"https://{host}/sites/default/files/files/opendata/bevolking/sectoren/OPENDATA_SECTOREN_{year}_NEW.xlsx#distribution{node_id}",
+        f"https://{host}/sites/default/files/files/opendata/bevolking/sectoren/OPENDATA_SECTOREN_{year}_OLD.zip#distribution{node_id}",
+    ]
+    if include_new_zip:
+        distributions.append(
+            f"https://{host}/sites/default/files/files/opendata/bevolking/sectoren/OPENDATA_SECTOREN_{year}_NEW.zip#distribution{node_id}"
+        )
+    if include_standard_zip:
+        distributions.append(
+            f"https://{host}/sites/default/files/files/opendata/bevolking/sectoren/OPENDATA_SECTOREN_{year}.zip#distribution{node_id}"
+        )
+    distribution_values = ",\n        ".join(f"<{value}>" for value in distributions)
+    return f"""
+    <https://statbel.fgov.be/node/{node_id}#id> a dcat:Dataset ;
+      dct:title "Bevolking per statistische sector"@nl ;
+      dct:alternative "Bevolking per statistische sector [Periode: {year}]"@nl ;
+      dct:identifier "NodeID{node_id}" ;
+      dct:license <https://creativecommons.org/licenses/by/4.0/> ;
+      dct:temporal [ dcat:startDate "{year}-01-01"^^xsd:date ] ;
+      dcat:landingPage <https://statbel.fgov.be/nl/open-data/bevolking-statistische-sector-{node_id}> ;
+      dcat:distribution {distribution_values} .
+    """
+
+
+def _catalog(*blocks: str) -> bytes:
+    return f"""
+    @prefix dcat: <http://www.w3.org/ns/dcat#> .
+    @prefix dct: <http://purl.org/dc/terms/> .
+    @prefix xsd: <http://www.w3.org/2001/XMLSchema#> .
+    <http://data.gov.be/catalog/statbelopen> a dcat:Catalog ;
+      dct:modified "2026-07-07"^^xsd:date .
+    {''.join(blocks)}
+    """.encode()
+
+
+class _Response:
+    def __init__(
+        self,
+        content: bytes,
+        *,
+        content_type: str = "application/octet-stream",
+        final_url: str | None = None,
+        content_length: int | None = None,
+    ) -> None:
+        self.content = content
+        self.final_url = final_url
+        self.headers = Message()
+        self.headers["Content-Type"] = content_type
+        self.headers["Content-Length"] = str(content_length if content_length is not None else len(content))
+        self.headers["ETag"] = '"statbel-test"'
+        self.headers["Last-Modified"] = "Mon, 13 Jul 2026 08:22:54 GMT"
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_args):
+        return False
+
+    def read(self, size: int = -1) -> bytes:
+        return self.content if size < 0 else self.content[:size]
+
+    def geturl(self) -> str:
+        return self.final_url or STATBEL_DCAT_URL
+
+
+class _Query:
+    def __init__(self, datasets: list[Dataset]) -> None:
+        self.datasets = datasets
+
+    def filter(self, *_args):
+        return self
+
+    def all(self) -> list[Dataset]:
+        return self.datasets
+
+
+class _Db:
+    def __init__(self, datasets: list[Dataset]) -> None:
+        self.datasets = datasets
+
+    def get(self, _model, identifier):
+        return SimpleNamespace(id=identifier)
+
+    def query(self, _model):
+        return _Query(self.datasets)
+
+
+def _settings(**overrides) -> Settings:
+    values = {
+        "SOURCE_CATALOG_PROBE_ENABLED": True,
+        "SOURCE_CATALOG_GRB_WFS_URL": "https://geo.api.vlaanderen.be/GRB/wfs",
+        "SOURCE_CATALOG_ALZ_RELEASE_URL": ALZ_RELEASE_URL,
+        "SOURCE_CATALOG_STATBEL_DCAT_URL": STATBEL_DCAT_URL,
+        "SOURCE_CATALOG_STATBEL_MAX_RESPONSE_MB": 5,
+        "SOURCE_CATALOG_PROBE_TIMEOUT_SECONDS": 3,
+        "SOURCE_CATALOG_PROBE_MAX_RESPONSE_MB": 1,
+        "SOURCE_CATALOG_PROBE_CACHE_TTL_SECONDS": 0,
+        "ORTHOPHOTO_WMS_URL": "https://geo.api.vlaanderen.be/OMWRGBMRVL/wms",
+        "ORTHOPHOTO_WMS_LAYER": "Ortho",
+    }
+    values.update(overrides)
+    return Settings(**values)
+
+
+def _statbel_item(content: bytes, *, final_url: str | None = None, content_length: int | None = None, datasets=None):
+    def opener(request, timeout):
+        if request.full_url == STATBEL_DCAT_URL:
+            return _Response(content, final_url=final_url, content_length=content_length)
+        raise URLError("not needed for selected assertion")
+
+    report = SourceCatalogProbeService.audit_project(
+        _Db(datasets or []), uuid.uuid4(), settings=_settings(), opener=opener, now=NOW
+    )
+    return next(item for item in report.items if item.source_name == "statbel")
+
+
+def test_statbel_parser_selects_latest_population_release_and_redegeo_variant() -> None:
+    release = parse_statbel_population_catalog(
+        _catalog(_dataset_block(2024, node_id=5510), _dataset_block(2025, node_id=6475))
+    )
+
+    assert release.version == "2025"
+    assert release.identifier == "NodeID6475"
+    assert release.current_distribution_variant == "new"
+    assert release.legacy_distribution_available is True
+    assert release.distribution_count == 3
+    assert release.catalog_modified_at == datetime(2026, 7, 7, tzinfo=timezone.utc)
+
+
+def test_statbel_probe_compares_latest_local_year_not_latest_import_order() -> None:
+    older = Dataset(
+        id=uuid.uuid4(), project_id=uuid.uuid4(), name="Statbel 2024", dataset_type="vector",
+        source="statbel", source_name="statbel", source_version="2024", imported_at=NOW, status="ready",
+    )
+    newer = Dataset(
+        id=uuid.uuid4(), project_id=uuid.uuid4(), name="Statbel 2025", dataset_type="vector",
+        source="statbel", source_name="statbel", source_version="2025", imported_at=NOW.replace(year=2025), status="ready",
+    )
+    item = _statbel_item(_catalog(_dataset_block(2025, node_id=6475)), datasets=[older, newer])
+
+    assert item.status == "available"
+    assert item.local_source_version == "2025"
+    assert item.remote_version == "2025"
+    assert item.comparison_status == "same"
+    assert item.metadata_identifier == "NodeID6475"
+    assert item.matched_layers == ["population_txt_current", "landing_page", "cc_by_4_0"]
+
+
+def test_statbel_parser_rejects_untrusted_distribution_host() -> None:
+    with pytest.raises(StatbelCatalogError) as exc_info:
+        parse_statbel_population_catalog(_catalog(_dataset_block(2025, node_id=6475, host="example.com")))
+
+    assert exc_info.value.code == "CATALOG_STATBEL_DISTRIBUTION_REJECTED"
+
+
+def test_statbel_parser_requires_new_2025_txt_distribution() -> None:
+    with pytest.raises(StatbelCatalogError) as exc_info:
+        parse_statbel_population_catalog(
+            _catalog(_dataset_block(2025, node_id=6475, include_new_zip=False))
+        )
+
+    assert exc_info.value.code == "CATALOG_STATBEL_CURRENT_DISTRIBUTION_MISSING"
+
+
+def test_statbel_parser_rejects_conflicting_period_evidence() -> None:
+    content = _catalog(_dataset_block(2025, node_id=6475)).replace(
+        b'dcat:startDate "2025-01-01"',
+        b'dcat:startDate "2024-01-01"',
+    )
+
+    with pytest.raises(StatbelCatalogError) as exc_info:
+        parse_statbel_population_catalog(content)
+
+    assert exc_info.value.code == "CATALOG_STATBEL_PERIOD_AMBIGUOUS"
+
+
+def test_statbel_parser_rejects_duplicate_latest_release() -> None:
+    with pytest.raises(StatbelCatalogError) as exc_info:
+        parse_statbel_population_catalog(
+            _catalog(_dataset_block(2025, node_id=6475), _dataset_block(2025, node_id=7000))
+        )
+
+    assert exc_info.value.code == "CATALOG_STATBEL_POPULATION_AMBIGUOUS"
+
+
+def test_statbel_parser_requires_cc_by_4_license() -> None:
+    content = _catalog(_dataset_block(2025, node_id=6475)).replace(
+        b"https://creativecommons.org/licenses/by/4.0/",
+        b"https://example.com/unknown-license",
+    )
+
+    with pytest.raises(StatbelCatalogError) as exc_info:
+        parse_statbel_population_catalog(content)
+
+    assert exc_info.value.code == "CATALOG_STATBEL_LICENSE_MISSING"
+
+
+def test_statbel_probe_revalidates_catalog_url_after_redirect() -> None:
+    item = _statbel_item(
+        _catalog(_dataset_block(2025, node_id=6475)),
+        final_url="https://example.com/DCAT_opendata_datasets.ttl",
+    )
+
+    assert item.status == "unavailable"
+    assert item.error_code == "CATALOG_STATBEL_URL_REJECTED"
+
+
+def test_statbel_probe_rejects_noncanonical_configured_catalog_before_network() -> None:
+    calls: list[str] = []
+
+    def opener(request, timeout):
+        calls.append(request.full_url)
+        raise URLError("network must not be reached for the rejected Statbel URL")
+
+    report = SourceCatalogProbeService.audit_project(
+        _Db([]),
+        uuid.uuid4(),
+        settings=_settings(SOURCE_CATALOG_STATBEL_DCAT_URL="https://example.com/catalog.ttl"),
+        opener=opener,
+        now=NOW,
+    )
+    item = next(value for value in report.items if value.source_name == "statbel")
+
+    assert item.status == "unavailable"
+    assert item.error_code == "CATALOG_STATBEL_URL_REJECTED"
+    assert "https://example.com/catalog.ttl" not in calls
+
+
+def test_statbel_probe_uses_separate_bounded_catalog_limit() -> None:
+    item = _statbel_item(
+        _catalog(_dataset_block(2025, node_id=6475)),
+        content_length=6 * 1024 * 1024,
+    )
+
+    assert item.status == "unavailable"
+    assert item.error_code == "CATALOG_RESPONSE_TOO_LARGE"
+
+
+def test_statbel_probe_does_not_fetch_catalog_distributions() -> None:
+    calls: list[str] = []
+
+    def opener(request, timeout):
+        calls.append(request.full_url)
+        if request.full_url == STATBEL_DCAT_URL:
+            return _Response(_catalog(_dataset_block(2025, node_id=6475)))
+        if "OPENDATA_SECTOREN" in request.full_url:
+            raise AssertionError("Statbel distribution fetch attempted")
+        raise URLError("unrelated provider offline")
+
+    report = SourceCatalogProbeService.audit_project(
+        _Db([]), uuid.uuid4(), settings=_settings(), opener=opener, now=NOW
+    )
+    item = next(value for value in report.items if value.source_name == "statbel")
+
+    assert item.status == "available"
+    assert calls.count(STATBEL_DCAT_URL) == 1
+    assert all("OPENDATA_SECTOREN" not in value for value in calls)