homelab-codex-ws/scripts/kb/check_okf.py

#!/usr/bin/env python3
"""Walidator konformancji OKF v0.1 dla bazy wiedzy homelab-codex.

Zaadaptowany z ~/narty-2027/saalbach-kb/check_okf.py. Wersja tamta sprawdzała
wyłącznie obecność frontmattera i niepuste `type` (OKF §9). Tutaj dochodzą
reguły tego repo:

  1. Frontmatter YAML obecny i parsowalny.
  2. `okf` przypięte do wersji PINNED_OKF ("0.1").
  3. `type` z zamkniętej listy TYPES.
  4. `visibility` ustawione i z listy {public, private}.
  5. `status` z listy {active, deprecated, planned}.
  6. `updated` w formacie YYYY-MM-DD.
  7. `as_of` (YYYY-MM-DD) wymagane wtedy i tylko wtedy, gdy `type: audit`.
  8. `superseded_by` niepuste wtedy i tylko wtedy, gdy `status: deprecated`.
  9. Każdy wpis `links` wskazuje na istniejący plik (ścieżka względna wobec
     katalogu dokumentu).
 10. Wpisy `contradicts` wyglądające jak ścieżka .md też muszą istnieć;
     pozostałe wpisy to wolny tekst.
 11. `stub` — o ile obecne — musi być boolem.

Zakres domyślny: kb/ oraz docs/sessions/, z wyłączeniem README-wskaźników
(POINTER_GLOBS) — te są nawigacją do kb-doca, nie dokumentami KB, i celowo nie
mają frontmattera OKF. Reszta repo (CLAUDE.md, README.md, .claude/skills/ itd.)
leży poza bazą wiedzy i nie podlega walidacji.

Tylko biblioteka standardowa: minimalny parser YAML wystarczający dla
frontmatterów w tym repo (klucze skalarne, listy inline, listy blokowe).
Uruchomienie: python3 scripts/kb/check_okf.py [katalog-repo]
"""

from __future__ import annotations

import re
import sys
from pathlib import Path, PurePosixPath

PINNED_OKF = "0.1"

TYPES = {
    "node",
    "service",
    "subsystem",
    "decision",
    "incident",
    "runbook",
    "phase",
    "audit",
    "session-log",
}

VISIBILITIES = {"public", "private"}
STATUSES = {"active", "deprecated", "planned"}

SCOPE = ("kb", "docs/sessions")

# README-wskazniki w services/<svc>/ i hosts/<node>/ to NAWIGACJA, nie dokumenty
# KB: nazwa + jedno zdanie + link do kb-doca, bez frontmattera OKF. Zrodlem prawdy
# jest kb-doc, do ktorego wskaznik odsyla. Wykluczenie jest tu zapisane wprost,
# zeby ewentualne poszerzenie SCOPE nie zaczelo ich nagle walidowac.
POINTER_GLOBS = (
    "services/*/README.md",
    "services/home-assistant/config/ken-legacy/README.md",
    "hosts/*/README.md",
)

DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")


def is_pointer(rel: str) -> bool:
    return any(PurePosixPath(rel).match(pat) for pat in POINTER_GLOBS)


def split_frontmatter(text: str) -> tuple[str | None, str]:
    """Zwraca (blok frontmattera lub None, reszta dokumentu)."""
    if not text.startswith("---\n"):
        return None, text
    end = text.find("\n---", 3)
    if end == -1:
        raise ValueError("blok frontmattera otwarty '---', ale nigdy nie zamknięty")
    return text[4:end], text[end + 4 :]


def parse_yaml(block: str) -> dict[str, object]:
    """Minimalny parser mapy YAML najwyższego poziomu."""
    out: dict[str, object] = {}
    key: str | None = None
    for lineno, raw in enumerate(block.splitlines(), start=2):
        line = raw.rstrip()
        if not line.strip() or line.lstrip().startswith("#"):
            continue
        if line.startswith((" ", "\t")):
            item = line.strip()
            if not item.startswith("- "):
                raise ValueError(f"linia {lineno}: nieoczekiwane wcięcie: {line!r}")
            if key is None:
                raise ValueError(f"linia {lineno}: element listy bez klucza")
            if not isinstance(out.get(key), list):
                out[key] = []
            out[key].append(scalar(item[2:]))  # type: ignore[union-attr]
            continue
        if ":" not in line:
            raise ValueError(f"linia {lineno}: brak ':' w {line!r}")
        key, _, value = line.partition(":")
        key = key.strip()
        if not key:
            raise ValueError(f"linia {lineno}: pusty klucz w {line!r}")
        out[key] = scalar(value.strip())
    return out


def scalar(value: str) -> object:
    value = value.strip()
    if not value:
        return ""
    if value.startswith("[") and value.endswith("]"):
        inner = value[1:-1].strip()
        return [scalar(v) for v in inner.split(",")] if inner else []
    if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'":
        return value[1:-1]
    if value in ("true", "false"):
        return value == "true"
    return value


def as_list(value: object) -> list[str]:
    if value in ("", None):
        return []
    if isinstance(value, list):
        return [str(v) for v in value if str(v).strip()]
    return [str(value)]


def scope_files(root: Path) -> list[Path]:
    files: list[Path] = []
    for rel in SCOPE:
        base = root / rel
        if base.is_dir():
            files.extend(
                p for p in base.rglob("*.md")
                if ".git" not in p.parts
                and not is_pointer(p.relative_to(root).as_posix())
            )
    return sorted(set(files))


def check_file(path: Path, root: Path) -> list[str]:
    rel = path.relative_to(root).as_posix()
    errors: list[str] = []
    text = path.read_text(encoding="utf-8")

    try:
        block, _ = split_frontmatter(text)
    except ValueError as exc:
        return [f"{rel}: {exc}"]

    if block is None:
        return [f"{rel}: brak bloku frontmattera YAML (§9.1)"]

    try:
        fm = parse_yaml(block)
    except ValueError as exc:
        return [f"{rel}: niepoprawny YAML we frontmatterze — {exc}"]

    okf = str(fm.get("okf", "")).strip()
    if okf != PINNED_OKF:
        errors.append(f"{rel}: `okf` musi być przypięte do {PINNED_OKF!r}, jest {okf!r}")

    doc_type = str(fm.get("type", "")).strip()
    if not doc_type:
        errors.append(f"{rel}: brak niepustego pola `type` (§9.2)")
    elif doc_type not in TYPES:
        errors.append(
            f"{rel}: `type` = {doc_type!r} spoza listy ({', '.join(sorted(TYPES))})"
        )

    visibility = str(fm.get("visibility", "")).strip()
    if not visibility:
        errors.append(f"{rel}: brak pola `visibility`")
    elif visibility not in VISIBILITIES:
        errors.append(f"{rel}: `visibility` = {visibility!r} spoza {sorted(VISIBILITIES)}")

    status = str(fm.get("status", "")).strip()
    if not status:
        errors.append(f"{rel}: brak pola `status`")
    elif status not in STATUSES:
        errors.append(f"{rel}: `status` = {status!r} spoza {sorted(STATUSES)}")

    updated = str(fm.get("updated", "")).strip()
    if not DATE_RE.match(updated):
        errors.append(f"{rel}: `updated` musi być YYYY-MM-DD, jest {updated!r}")

    as_of = str(fm.get("as_of", "")).strip()
    if doc_type == "audit":
        if not DATE_RE.match(as_of):
            errors.append(
                f"{rel}: `type: audit` wymaga `as_of` w formacie YYYY-MM-DD, jest {as_of!r}"
            )
    elif as_of:
        errors.append(f"{rel}: `as_of` dozwolone tylko dla `type: audit`")

    superseded = str(fm.get("superseded_by", "")).strip()
    if status == "deprecated" and not superseded:
        errors.append(f"{rel}: `status: deprecated` wymaga niepustego `superseded_by`")
    if status != "deprecated" and superseded:
        errors.append(f"{rel}: `superseded_by` dozwolone tylko dla `status: deprecated`")

    if "stub" in fm and not isinstance(fm["stub"], bool):
        errors.append(f"{rel}: `stub` musi być boolem, jest {fm['stub']!r}")

    for link in as_list(fm.get("links")):
        target = (path.parent / link).resolve()
        if not target.exists():
            errors.append(f"{rel}: `links` wskazuje na nieistniejący plik: {link}")

    for item in as_list(fm.get("contradicts")):
        if item.endswith(".md") and "/" in item:
            target = (path.parent / item).resolve()
            if not target.exists():
                errors.append(
                    f"{rel}: `contradicts` wskazuje na nieistniejący plik: {item}"
                )

    return errors


def main() -> int:
    root = Path(sys.argv[1] if len(sys.argv) > 1 else Path(__file__).resolve().parents[2])
    root = root.resolve()
    files = scope_files(root)

    print(f"Repo: {root}")
    print(f"Zakres: {', '.join(SCOPE)}")
    print(f"Sprawdzono plików .md: {len(files)}")

    by_type: dict[str, int] = {}
    errors: list[str] = []
    for path in files:
        errors.extend(check_file(path, root))
        try:
            block, _ = split_frontmatter(path.read_text(encoding="utf-8"))
            if block:
                t = str(parse_yaml(block).get("type", "?"))
                by_type[t] = by_type.get(t, 0) + 1
        except ValueError:
            pass

    if by_type:
        print("  per type: " + ", ".join(f"{t}={n}" for t, n in sorted(by_type.items())))
    print()

    if not files:
        print("BRAK plików w zakresie — nie ma czego walidować.")
        return 1

    if errors:
        print(f"NIEZGODNE z OKF v{PINNED_OKF} — {len(errors)} problem(ów):")
        for err in errors:
            print(f"  ✗ {err}")
        return 1

    print(f"ZGODNE z OKF v{PINNED_OKF}: frontmatter parsowalny, okf przypięte,")
    print("type/visibility/status z list, daty poprawne, links rozwiązywalne.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())