315 lines
12 KiB
Python
315 lines
12 KiB
Python
|
#!/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())
|