kb-wiki/check_okf.py

#!/usr/bin/env python3
"""Walidator konformancji OKF v0.1 dla kb-wiki.

Przeniesiony z `~/kb-wiki-etap1-sesja-2026-08-27/check_okf.py` przy
rekoncyliacji 2026-08-27 (patrz `_meta/conventions.md` §10 i
`homelab-codex-ws:kb/phases/kb-m5-faza5-wiki.md` §3) — rekomendacja tamtej
sesji: "Przenieść check_okf.py tej sesji do prawdziwego repo przy
rekoncyliacji", bo lipcowy proof-of-concept nie zostawił żadnego
skomitowanego walidatora. Rozszerzenia specyficzne dla kb-wiki
(`_meta/conventions.md` §10):

  1. `type` musi być z listy zamkniętej TYPES (podmiot|osoba|sprawa|umowa|temat|meta).
  2. `sources` niepusty na każdej stronie-koncepcie (inwariant 1, audytowalność).
  3. Każdy `envelope_id` z `sources` istnieje w bazie `kb` (SELECT-only,
     KB_DSN z env) -- pomijalne flagą --no-db.
  4. Każdy `chunks: [...]` id, jeśli podany, istnieje w `document_chunk` i
     należy do tego samego `envelope_id`.

Adaptacja przy przeniesieniu (rzeczywisty layout `kb-wiki`, którego lipcowy
proof-of-concept nie miał w zestawie testowym): `README.md` (root) i
`_meta/lint-reports/*.md` istniały w prawdziwym repo od 2026-07-21 bez
frontmattera stron-konceptów (to dokumentacja/raporty, nie skompilowane
encje) — wyjęte z wymogu frontmattera, tak samo jak pliki zarezerwowane, ale
bez ograniczenia "tylko root index.md może mieć pola".

To NIE jest scripts/kb/check_okf.py z homelab-codex-ws (ten ma SCOPE=("kb",
"docs/sessions") na sztywno i nie widzi layoutu kb-wiki -- patrz
_meta/conventions.md §10 dla rozjazdu). Ten plik jest osobnym walidatorem
tego repo, nie próbą dopasowania tamtego.

Tylko biblioteka standardowa poza opcjonalnym asyncpg (dociągane tylko gdy
weryfikacja bazy jest włączona). Uruchomienie:
  python3 check_okf.py [katalog-repo] [--no-db]
"""

from __future__ import annotations

import asyncio
import os
import sys
from pathlib import Path

RESERVED = {"index.md", "log.md"}
# README.md (root) i INDEX.md (root, nawigacja — OKF-owy index.md nie istnieje
# w tym repo, patrz kb-m5-faza5-wiki.md §3, "do rozstrzygnięcia przez
# operatora") istniały bez frontmattera stron-konceptów od 2026-07-21 —
# dokumentacja/nawigacja, nie skompilowana encja.
NO_FRONTMATTER_REQUIRED = {"README.md", "INDEX.md"}
TYPES = {"podmiot", "osoba", "sprawa", "umowa", "temat", "meta"}
STATUSES = {"active", "archived"}


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 (skalar/lista inline/lista
    blokowa; wystarczające dla frontmatterów tego repo, w tym listy map
    `sources:` z zagnieżdżonym `envelope_id`/`chunks`)."""
    out: dict[str, object] = {}
    key: str | None = None
    current_item: dict[str, object] | None = None
    for lineno, raw in enumerate(block.splitlines(), start=2):
        line = raw.rstrip()
        if not line.strip() or line.lstrip().startswith("#"):
            continue
        indent = len(line) - len(line.lstrip(" "))
        stripped = line.strip()
        if indent > 0:
            if stripped.startswith("- "):
                if key is None:
                    raise ValueError(f"linia {lineno}: element listy bez klucza")
                if not isinstance(out.get(key), list):
                    out[key] = []
                rest = stripped[2:]
                if ":" in rest and not (rest.startswith('"') or rest.startswith("'")):
                    # element listy = mapa (np. sources: - envelope_id: ... )
                    current_item = {}
                    out[key].append(current_item)  # type: ignore[union-attr]
                    k2, _, v2 = rest.partition(":")
                    current_item[k2.strip()] = scalar(v2.strip())
                else:
                    current_item = None
                    out[key].append(scalar(rest))  # type: ignore[union-attr]
                continue
            # wcięta linia bez "- " -- kolejne pole bieżącego elementu-mapy listy
            if current_item is not None and ":" in stripped:
                k2, _, v2 = stripped.partition(":")
                current_item[k2.strip()] = scalar(v2.strip())
                continue
            raise ValueError(f"linia {lineno}: nieoczekiwane wcięcie: {line!r}")
        current_item = None
        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]
    return value


def as_list(value: object) -> list:
    if value in ("", None):
        return []
    if isinstance(value, list):
        return value
    return [value]


def check_file(path: Path, root: Path) -> tuple[list[str], list[tuple[str, list]]]:
    """Zwraca (błędy, [(envelope_id, chunks), ...] do weryfikacji w bazie)."""
    rel = path.relative_to(root).as_posix()
    errors: list[str] = []
    to_verify: list[tuple[str, list]] = []
    text = path.read_text(encoding="utf-8")

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

    if (rel.count("/") == 0 and path.name in NO_FRONTMATTER_REQUIRED) or (
        rel.startswith("_meta/lint-reports/")
    ):
        return [], []

    if path.name in RESERVED:
        is_root_index = rel == "index.md"
        if block is None:
            return [], []
        if not is_root_index:
            return [f"{rel}: plik zarezerwowany nie może mieć frontmattera (OKF §6/§11)"], []
        try:
            fm = parse_yaml(block)
        except ValueError as exc:
            return [f"{rel}: niepoprawny YAML we frontmatterze — {exc}"], []
        extra = set(fm) - {"okf_version"}
        if extra:
            errors.append(
                f"{rel}: root index.md może mieć wyłącznie okf_version, "
                f"znaleziono też: {', '.join(sorted(extra))} (OKF §11)"
            )
        elif not str(fm.get("okf_version", "")).strip():
            errors.append(f"{rel}: puste okf_version (OKF §11)")
        return errors, []

    if block is None:
        return [f"{rel}: brak bloku frontmattera YAML (OKF §9.1)"], []
    try:
        fm = parse_yaml(block)
    except ValueError as exc:
        return [f"{rel}: niepoprawny YAML we frontmatterze — {exc}"], []

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

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

    # _meta/conventions.md jest samo type: temat, ale bez sources -- to jest
    # dokument o konwencjach, nie strona-koncept skompilowana z retrievalu.
    # Reguła "sources niepuste" dotyczy stron w podmioty/osoby/sprawy/umowy/tematy
    # poza _meta/.
    is_meta = rel.startswith("_meta/")

    sources = as_list(fm.get("sources"))
    if not is_meta:
        if not sources:
            errors.append(f"{rel}: `sources` puste — inwariant 1 (audytowalność) wymaga ≥1 wpisu")
        for item in sources:
            if not isinstance(item, dict) or not str(item.get("envelope_id", "")).strip():
                errors.append(f"{rel}: wpis `sources` bez `envelope_id`: {item!r}")
                continue
            env_id = str(item["envelope_id"])
            chunks = item.get("chunks", [])
            chunks = as_list(chunks) if not isinstance(chunks, list) else chunks
            to_verify.append((env_id, [str(c) for c in chunks if str(c).strip()]))

    return errors, to_verify


async def verify_sources_in_db(pairs: list[tuple[str, list]], dsn: str) -> list[str]:
    try:
        import asyncpg
    except ImportError:
        return ["[db] moduł `asyncpg` niedostępny — pomiń weryfikację bazy albo zainstaluj asyncpg"]

    errors: list[str] = []
    conn = await asyncpg.connect(dsn)
    try:
        env_ids = sorted({env for env, _ in pairs})
        rows = await conn.fetch(
            "SELECT id FROM envelope WHERE id = ANY($1::text[])", env_ids
        )
        existing_envs = {r["id"] for r in rows}
        for env in env_ids:
            if env not in existing_envs:
                errors.append(f"[db] envelope_id nie istnieje w bazie: {env!r}")

        chunk_ids = sorted({int(c) for _, chunks in pairs for c in chunks if c.isdigit()})
        non_numeric = sorted({c for _, chunks in pairs for c in chunks if not c.isdigit()})
        for c in non_numeric:
            errors.append(f"[db] chunk id nie jest liczbą całkowitą: {c!r}")
        if chunk_ids:
            crows = await conn.fetch(
                "SELECT id, envelope_id FROM document_chunk WHERE id = ANY($1::bigint[])",
                chunk_ids,
            )
            chunk_owner = {r["id"]: r["envelope_id"] for r in crows}
            for env, chunks in pairs:
                for c in chunks:
                    if not c.isdigit():
                        continue
                    cid = int(c)
                    if cid not in chunk_owner:
                        errors.append(f"[db] chunk id nie istnieje w document_chunk: {cid}")
                    elif chunk_owner[cid] != env:
                        errors.append(
                            f"[db] chunk id {cid} należy do envelope {chunk_owner[cid]!r}, "
                            f"nie do {env!r} podanego w sources"
                        )
    finally:
        await conn.close()
    return errors


def main() -> int:
    args = [a for a in sys.argv[1:] if not a.startswith("--")]
    no_db = "--no-db" in sys.argv[1:]
    root = Path(args[0] if args else Path(__file__).parent).resolve()

    files = sorted(p for p in root.rglob("*.md") if ".git" not in p.parts)
    print(f"Repo: {root}")
    print(f"Sprawdzono plików .md: {len(files)}")
    if not files:
        print("BRAK plików .md — nie ma czego walidować.")
        return 1

    concepts = [p for p in files if p.name not in RESERVED]
    reserved = [p for p in files if p.name in RESERVED]
    print(f"  koncepty: {len(concepts)}, zarezerwowane: {len(reserved)}")

    errors: list[str] = []
    all_pairs: list[tuple[str, list]] = []
    by_type: dict[str, int] = {}
    for path in files:
        errs, pairs = check_file(path, root)
        errors.extend(errs)
        all_pairs.extend(pairs)
        try:
            block, _ = split_frontmatter(path.read_text(encoding="utf-8"))
            if block and path.name not in RESERVED:
                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())))

    n_sources = len(all_pairs)
    n_degraded = sum(1 for _, chunks in all_pairs if not chunks)
    print(f"  przypisów sources: {n_sources} ({n_degraded} envelope-only / bez chunków)")
    print()

    db_errors: list[str] = []
    if all_pairs and not no_db:
        dsn = os.environ.get("KB_DSN", "")
        if not dsn:
            errors.append(
                "[db] KB_DSN nie ustawione w env — weryfikacja envelope_id/chunk_id pominięta "
                "(ustaw KB_DSN albo uruchom z --no-db żeby nie traktować tego jako błąd)"
            )
        else:
            db_errors = asyncio.run(verify_sources_in_db(all_pairs, dsn))
            errors.extend(db_errors)

    if errors:
        print(f"NIEZGODNE z OKF v0.1 (kb-wiki) — {len(errors)} problem(ów):")
        for err in errors:
            print(f"  ✗ {err}")
        return 1

    print("ZGODNE z OKF v0.1 (kb-wiki): frontmatter parsowalny, type/status poprawne,")
    print(f"sources niepuste, wszystkie {n_sources} przypisów zweryfikowane w bazie.")
    return 0


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