feat: add KB schemas and validator

Pydantic validation for entry frontmatter, series YAML, source
registry IDs, and definition references. Tests cover accept/reject
paths including missing sources and unknown source_id.

schemas/definition.schema.yaml

@@ -0,0 +1,9 @@
+type: object
+required: [id, term, source_id, source_definition]
+properties:
+  id: {type: string}
+  term: {type: string}
+  source_id: {type: string}
+  source_definition: {type: string}
+  applies_to: {type: array, items: {type: string}}
+  do_not_confuse_with: {type: array, items: {type: string}}

schemas/entry.schema.yaml

@@ -0,0 +1,20 @@
+# KB-Eintrag (YAML-Frontmatter). Spiegel der Pydantic-Modelle in scripts/validate_kb.py.
+type: object
+required: [id, path, title, status, updated, summary, sources, series, definitions, coverage]
+properties:
+  id: {type: string}
+  path: {type: string}
+  title: {type: string}
+  status: {type: string, enum: [stub, partial, complete]}
+  updated: {type: string}
+  summary: {type: string}
+  sources: {type: array, items: {type: string}}
+  series: {type: array, items: {type: string}}
+  definitions: {type: array, items: {type: string}}
+  coverage:
+    type: object
+    required: [geo]
+    properties:
+      geo: {type: string}
+      time: {type: ["string", "null"]}
+  notes_methodik: {type: string}

schemas/series.schema.yaml

@@ -0,0 +1,38 @@
+type: object
+required:
+  - id
+  - title
+  - unit
+  - value_type
+  - geo
+  - definition_ids
+  - denominator
+  - source_id
+  - source_table
+  - source_date
+  - retrieved
+  - points
+properties:
+  id: {type: string}
+  title: {type: string}
+  unit: {type: string}
+  value_type: {type: string, enum: [count, share, quote_per_100k, euro, percent]}
+  geo: {type: string}
+  definition_ids: {type: array, items: {type: string}}
+  denominator: {type: string}
+  source_id: {type: string}
+  source_table: {type: string}
+  source_date: {type: string}
+  retrieved: {type: string}
+  points:
+    type: array
+    items:
+      type: object
+      required: [year, value]
+      properties:
+        year: {type: integer}
+        value: {type: number}
+        share: {type: ["number", "null"]}
+        note: {type: ["string", "null"]}
+        month: {type: ["integer", "null"]}
+  caveats: {type: array, items: {type: string}}

schemas/source.schema.yaml

@@ -0,0 +1,26 @@
+type: object
+required:
+  - id
+  - name
+  - url
+  - type
+  - update_frequency
+  - access
+  - reliability
+  - reliability_reason
+  - feeds
+  - status
+properties:
+  id: {type: string}
+  name: {type: string}
+  url: {type: string}
+  type: {type: string, enum: [amtlich, wissenschaftlich, gerichtlich, parlamentarisch, ngo, journalistisch, thinktank]}
+  update_frequency: {type: string}
+  access: {type: string}
+  reliability: {type: string, enum: [hoch, mittel, niedrig, ungeprueft]}
+  reliability_reason: {type: string}
+  feeds: {type: array, items: {type: string}}
+  status: {type: string, enum: [active, inactive]}
+  license: {type: string}
+  latest_edition: {type: string}
+  priority: {type: integer}

scripts/validate_kb.py

@@ -0,0 +1,359 @@
+#!/usr/bin/env python3
+"""Validate KB markdown frontmatter, series YAML, source registry, and ID refs."""
+
+from __future__ import annotations
+
+import sys
+from datetime import date, datetime
+from pathlib import Path
+from typing import Any, Iterable
+
+import frontmatter
+import yaml
+from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator
+
+ROOT = Path(__file__).resolve().parents[1]
+KB_DIR = ROOT / "kb"
+
+ENTRY_STATUSES = {"stub", "partial", "complete"}
+SOURCE_TYPES = {
+    "amtlich",
+    "wissenschaftlich",
+    "gerichtlich",
+    "parlamentarisch",
+    "ngo",
+    "journalistisch",
+    "thinktank",
+}
+SOURCE_STATUSES = {"active", "inactive"}
+VALUE_TYPES = {"count", "share", "quote_per_100k", "euro", "percent"}
+RELIABILITY = {"hoch", "mittel", "niedrig", "ungeprueft"}
+
+META_SKIP_PREFIXES = (
+    "kb/_meta/",
+    "kb/sources/SOURCES.md",
+)
+
+
+class Coverage(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    geo: str
+    time: str | None = None
+
+
+def _as_str(v: Any) -> Any:
+    if isinstance(v, datetime):
+        return v.date().isoformat()
+    if isinstance(v, date):
+        return v.isoformat()
+    return v
+
+
+class EntryFrontmatter(BaseModel):
+    model_config = ConfigDict(extra="allow")
+    id: str
+    path: str
+    title: str
+    status: str
+    updated: str
+    summary: str
+    sources: list[str]
+    series: list[str] = Field(default_factory=list)
+    definitions: list[str] = Field(default_factory=list)
+    coverage: Coverage
+    notes_methodik: str = ""
+
+    @field_validator("updated", mode="before")
+    @classmethod
+    def updated_str(cls, v: Any) -> Any:
+        return _as_str(v)
+
+    @field_validator("status")
+    @classmethod
+    def status_ok(cls, v: str) -> str:
+        if v not in ENTRY_STATUSES:
+            raise ValueError(f"status must be one of {sorted(ENTRY_STATUSES)}")
+        return v
+
+    @field_validator("sources", mode="before")
+    @classmethod
+    def sources_required(cls, v: Any) -> list[str]:
+        if v is None:
+            raise ValueError("sources is required")
+        if not isinstance(v, list):
+            raise ValueError("sources must be a list")
+        return v
+
+
+class SeriesPoint(BaseModel):
+    model_config = ConfigDict(extra="allow")
+    year: int
+    value: float | int
+    share: float | None = None
+    note: str | None = None
+    month: int | None = None
+
+
+class SeriesDoc(BaseModel):
+    model_config = ConfigDict(extra="allow")
+    id: str
+    title: str
+    unit: str
+    value_type: str
+    geo: str
+    definition_ids: list[str] = Field(default_factory=list)
+    denominator: str
+    source_id: str
+    source_table: str
+    source_date: str
+    retrieved: str
+    points: list[SeriesPoint]
+    caveats: list[str] = Field(default_factory=list)
+
+    @field_validator("source_date", "retrieved", mode="before")
+    @classmethod
+    def dates_str(cls, v: Any) -> Any:
+        return _as_str(v)
+
+    @field_validator("value_type")
+    @classmethod
+    def value_type_ok(cls, v: str) -> str:
+        if v not in VALUE_TYPES:
+            raise ValueError(f"value_type must be one of {sorted(VALUE_TYPES)}")
+        return v
+
+
+class SourceDoc(BaseModel):
+    model_config = ConfigDict(extra="allow")
+    id: str
+    name: str
+    url: str
+    type: str
+    update_frequency: str
+    access: str
+    reliability: str
+    reliability_reason: str
+    feeds: list[str]
+    status: str
+    license: str | None = None
+    latest_edition: str | None = None
+    priority: int | None = None
+
+    @field_validator("type")
+    @classmethod
+    def type_ok(cls, v: str) -> str:
+        if v not in SOURCE_TYPES:
+            raise ValueError(f"type must be one of {sorted(SOURCE_TYPES)}")
+        return v
+
+    @field_validator("status")
+    @classmethod
+    def status_ok(cls, v: str) -> str:
+        if v not in SOURCE_STATUSES:
+            raise ValueError(f"status must be one of {sorted(SOURCE_STATUSES)}")
+        return v
+
+    @field_validator("reliability")
+    @classmethod
+    def reliability_ok(cls, v: str) -> str:
+        if v not in RELIABILITY:
+            raise ValueError(f"reliability must be one of {sorted(RELIABILITY)}")
+        return v
+
+
+class DefinitionDoc(BaseModel):
+    model_config = ConfigDict(extra="allow")
+    id: str
+    term: str
+    source_id: str
+    source_definition: str
+    applies_to: list[str] = Field(default_factory=list)
+    do_not_confuse_with: list[str] = Field(default_factory=list)
+
+
+def _fmt_pydantic(err: ValidationError, prefix: str) -> list[str]:
+    out = []
+    for e in err.errors():
+        loc = ".".join(str(x) for x in e["loc"])
+        out.append(f"{prefix}: {loc}: {e['msg']}")
+    return out
+
+
+def _load_yaml(path: Path) -> Any:
+    with path.open(encoding="utf-8") as f:
+        return yaml.safe_load(f)
+
+
+def validate_entry_file(path: Path) -> list[str]:
+    errors: list[str] = []
+    try:
+        post = frontmatter.load(path)
+    except Exception as exc:  # noqa: BLE001
+        return [f"{path}: cannot parse frontmatter: {exc}"]
+    meta = dict(post.metadata)
+    if "sources" not in meta:
+        errors.append(f"{path}: missing field sources")
+        return errors
+    try:
+        entry = EntryFrontmatter.model_validate(meta)
+    except ValidationError as err:
+        return _fmt_pydantic(err, str(path))
+    if entry.status == "complete":
+        if not entry.summary.strip() or entry.summary.strip().startswith("Strukturknoten"):
+            errors.append(f"{path}: complete entries need a factual summary")
+        if not entry.series and "|" not in post.content:
+            errors.append(f"{path}: complete entries need series or a markdown table")
+    if entry.status != "stub" and not entry.sources:
+        errors.append(f"{path}: non-stub entries need at least one source")
+    return errors
+
+
+def validate_series_file(path: Path) -> list[str]:
+    try:
+        data = _load_yaml(path)
+    except Exception as exc:  # noqa: BLE001
+        return [f"{path}: cannot parse yaml: {exc}"]
+    if not isinstance(data, dict):
+        return [f"{path}: series yaml must be a mapping"]
+    try:
+        SeriesDoc.model_validate(data)
+    except ValidationError as err:
+        return _fmt_pydantic(err, str(path))
+    return []
+
+
+def load_source_ids(path: Path) -> set[str]:
+    data = _load_yaml(path) or []
+    if isinstance(data, dict) and "sources" in data:
+        data = data["sources"]
+    if not isinstance(data, list):
+        raise ValueError(f"{path}: expected a list of sources")
+    ids: set[str] = set()
+    for item in data:
+        if not isinstance(item, dict) or "id" not in item:
+            raise ValueError(f"{path}: source item missing id")
+        ids.add(item["id"])
+    return ids
+
+
+def validate_sources_file(path: Path) -> tuple[set[str], list[str]]:
+    errors: list[str] = []
+    try:
+        data = _load_yaml(path) or []
+    except Exception as exc:  # noqa: BLE001
+        return set(), [f"{path}: cannot parse yaml: {exc}"]
+    if isinstance(data, dict) and "sources" in data:
+        data = data["sources"]
+    if not isinstance(data, list):
+        return set(), [f"{path}: expected a list of sources"]
+    ids: set[str] = set()
+    for i, item in enumerate(data):
+        try:
+            src = SourceDoc.model_validate(item)
+        except ValidationError as err:
+            errors.extend(_fmt_pydantic(err, f"{path}[{i}]"))
+            continue
+        if src.id in ids:
+            errors.append(f"{path}: duplicate source id {src.id}")
+        ids.add(src.id)
+    return ids, errors
+
+
+def validate_series_source_refs(path: Path, known_ids: set[str]) -> list[str]:
+    try:
+        data = _load_yaml(path)
+    except Exception as exc:  # noqa: BLE001
+        return [f"{path}: cannot parse yaml: {exc}"]
+    if not isinstance(data, dict):
+        return [f"{path}: series yaml must be a mapping"]
+    sid = data.get("source_id")
+    if not sid:
+        return [f"{path}: missing source_id"]
+    if sid not in known_ids:
+        return [f"{path}: unknown source_id {sid}"]
+    return []
+
+
+def _iter_kb_markdown(kb_dir: Path) -> Iterable[Path]:
+    for path in sorted(kb_dir.rglob("*.md")):
+        rel = path.relative_to(kb_dir.parent).as_posix()
+        if rel.startswith("kb/sources/items/"):
+            continue
+        if rel.startswith("kb/_meta/"):
+            continue
+        if rel == "kb/sources/SOURCES.md":
+            continue
+        yield path
+
+
+def _load_definition_ids(path: Path) -> tuple[set[str], list[str]]:
+    if not path.exists():
+        return set(), [f"{path}: missing definitions.yaml"]
+    data = _load_yaml(path) or []
+    if isinstance(data, dict) and "definitions" in data:
+        data = data["definitions"]
+    if not isinstance(data, list):
+        return set(), [f"{path}: expected a list of definitions"]
+    ids: set[str] = set()
+    errors: list[str] = []
+    for i, item in enumerate(data):
+        try:
+            d = DefinitionDoc.model_validate(item)
+        except ValidationError as err:
+            errors.extend(_fmt_pydantic(err, f"{path}[{i}]"))
+            continue
+        ids.add(d.id)
+    return ids, errors
+
+
+def validate_kb(kb_dir: Path = KB_DIR) -> list[str]:
+    errors: list[str] = []
+    sources_path = kb_dir / "sources" / "sources.yaml"
+    defs_path = kb_dir / "definitions.yaml"
+    known_sources, src_errors = validate_sources_file(sources_path)
+    errors.extend(src_errors)
+    def_ids, def_errors = _load_definition_ids(defs_path)
+    errors.extend(def_errors)
+
+    if not kb_dir.exists():
+        return [f"{kb_dir}: missing kb directory"]
+
+    for md in _iter_kb_markdown(kb_dir):
+        errors.extend(validate_entry_file(md))
+        try:
+            post = frontmatter.load(md)
+            meta = dict(post.metadata)
+        except Exception:
+            continue
+        for sid in meta.get("sources") or []:
+            if sid not in known_sources:
+                errors.append(f"{md}: unknown source_id {sid}")
+        for did in meta.get("definitions") or []:
+            if did not in def_ids:
+                errors.append(f"{md}: unknown definition_id {did}")
+
+    for series in sorted(kb_dir.rglob("series/*.yaml")):
+        errors.extend(validate_series_file(series))
+        errors.extend(validate_series_source_refs(series, known_sources))
+        data = _load_yaml(series) or {}
+        for did in data.get("definition_ids") or []:
+            if did not in def_ids:
+                errors.append(f"{series}: unknown definition_id {did}")
+
+    return errors
+
+
+def main(argv: list[str] | None = None) -> int:
+    del argv
+    errors = validate_kb()
+    if errors:
+        for e in errors:
+            print(e, file=sys.stderr)
+        print(f"{len(errors)} error(s)", file=sys.stderr)
+        return 1
+    print("KB validation ok")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/conftest.py

@@ -0,0 +1,6 @@
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))

tests/fixtures/invalid_entry.md

@@ -0,0 +1,16 @@
+---
+id: kb.fixture.invalid
+path: fixtures/invalid
+title: Ungültiger Fixture-Eintrag
+status: stub
+updated: 2026-08-21
+summary: "Fehlt das Feld sources."
+series: []
+definitions: []
+coverage:
+  geo: DE
+  time: null
+notes_methodik: ""
+---
+
+# Ungültig

tests/fixtures/invalid_series.yaml

@@ -0,0 +1,15 @@
+id: series.fixture.bad
+title: Unvollständige Zeitreihe
+unit: personen
+value_type: count
+geo: DE-Bund
+definition_ids: []
+denominator: "x"
+source_id: src-fixture
+source_table: "T-FIXTURE"
+source_date: "2026-04"
+retrieved: "2026-08-21"
+points:
+  - year: 2024
+    share: 0.4
+caveats: []

tests/fixtures/sources.yaml

@@ -0,0 +1,12 @@
+- id: src-fixture
+  name: Fixture-Quelle
+  url: https://example.invalid/fixture
+  type: amtlich
+  update_frequency: jaehrlich
+  access: oeffentlich
+  reliability: hoch
+  reliability_reason: "Testdaten."
+  feeds: [fixtures]
+  status: active
+  license: "test"
+  latest_edition: "2026"

tests/fixtures/valid_entry.md

@@ -0,0 +1,19 @@
+---
+id: kb.fixture.valid
+path: fixtures/valid
+title: Gültiger Fixture-Eintrag
+status: stub
+updated: 2026-08-21
+summary: "Strukturknoten für Schema-Tests."
+sources: []
+series: []
+definitions: []
+coverage:
+  geo: DE
+  time: null
+notes_methodik: ""
+---
+
+# Gültiger Fixture-Eintrag
+
+Nur für Tests.

tests/fixtures/valid_series.yaml

@@ -0,0 +1,18 @@
+id: series.fixture.count
+title: Fixture-Zeitreihe
+unit: personen
+value_type: count
+geo: DE-Bund
+definition_ids: []
+denominator: "Tatverdächtige insgesamt"
+source_id: src-fixture
+source_table: "T-FIXTURE"
+source_date: "2026-04"
+retrieved: "2026-08-21"
+points:
+  - year: 2024
+    value: 100
+    share: 0.4
+    note: null
+caveats:
+  - "Nur Testdaten."

tests/test_schema.py

@@ -0,0 +1,35 @@
+"""Schema-Tests für KB-Einträge und Zeitreihen. Produktionsmodul: scripts.validate_kb."""
+
+from pathlib import Path
+
+import pytest
+
+FIXTURES = Path(__file__).parent / "fixtures"
+
+
+def test_valid_entry_is_accepted():
+    from scripts.validate_kb import validate_entry_file
+
+    errors = validate_entry_file(FIXTURES / "valid_entry.md")
+    assert errors == []
+
+
+def test_entry_without_sources_is_rejected():
+    from scripts.validate_kb import validate_entry_file
+
+    errors = validate_entry_file(FIXTURES / "invalid_entry.md")
+    assert any("sources" in e.lower() for e in errors)
+
+
+def test_valid_series_is_accepted():
+    from scripts.validate_kb import validate_series_file
+
+    errors = validate_series_file(FIXTURES / "valid_series.yaml")
+    assert errors == []
+
+
+def test_series_point_without_value_is_rejected():
+    from scripts.validate_kb import validate_series_file
+
+    errors = validate_series_file(FIXTURES / "invalid_series.yaml")
+    assert any("value" in e.lower() for e in errors)

tests/test_source_refs.py

@@ -0,0 +1,27 @@
+"""Referenzen auf source_id müssen in der Registry existieren."""
+
+from pathlib import Path
+
+FIXTURES = Path(__file__).parent / "fixtures"
+
+
+def test_unknown_source_id_is_rejected():
+    from scripts.validate_kb import load_source_ids, validate_series_source_refs
+
+    known = load_source_ids(FIXTURES / "sources.yaml")
+    errors = validate_series_source_refs(
+        FIXTURES / "valid_series.yaml",
+        known_ids=known - {"src-fixture"},
+    )
+    assert any("src-fixture" in e for e in errors)
+
+
+def test_known_source_id_is_accepted():
+    from scripts.validate_kb import load_source_ids, validate_series_source_refs
+
+    known = load_source_ids(FIXTURES / "sources.yaml")
+    errors = validate_series_source_refs(
+        FIXTURES / "valid_series.yaml",
+        known_ids=known,
+    )
+    assert errors == []