chore: move kb_check.py para dentro da knowledge-base
Script e Makefile agora vivem junto à base que gerenciam. Caminho anterior: GLPI11/scripts/kb_check.py Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Rodolpho Lopes
parent
2b10435643
commit
fb46ac0fea
2 changed files with 295 additions and 0 deletions
7
Makefile
Normal file
7
Makefile
Normal file
|
|
@ -0,0 +1,7 @@
|
|||
.PHONY: kb-check kb-fix
|
||||
|
||||
kb-check:
|
||||
python3 kb_check.py
|
||||
|
||||
kb-fix:
|
||||
python3 kb_check.py --fix
|
||||
288
kb_check.py
Normal file
288
kb_check.py
Normal file
|
|
@ -0,0 +1,288 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Knowledge base consistency checker/fixer for AI-oriented records."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
KB_ROOT = Path(__file__).resolve().parent
|
||||
INDEX_PATH = KB_ROOT / "index.json"
|
||||
SCHEMA_PATH = KB_ROOT / "schemas" / "knowledge-record.schema.json"
|
||||
RECORDS_ROOT = KB_ROOT / "records"
|
||||
|
||||
|
||||
def fail(errors: list[str]) -> int:
|
||||
print("KB CHECK: FAILED")
|
||||
for err in errors:
|
||||
print(f"- {err}")
|
||||
return 1
|
||||
|
||||
|
||||
def ok(prefix: str = "KB CHECK") -> int:
|
||||
print(f"{prefix}: OK")
|
||||
return 0
|
||||
|
||||
|
||||
def parse_scalar(value: str) -> Any:
|
||||
value = value.strip()
|
||||
if value == "[]":
|
||||
return []
|
||||
return value
|
||||
|
||||
|
||||
def parse_front_matter(markdown_text: str) -> dict[str, Any]:
|
||||
lines = markdown_text.splitlines()
|
||||
if len(lines) < 3 or lines[0].strip() != "---":
|
||||
raise ValueError("front matter ausente ou invalido (linha inicial '---').")
|
||||
|
||||
end_idx = None
|
||||
for i in range(1, len(lines)):
|
||||
if lines[i].strip() == "---":
|
||||
end_idx = i
|
||||
break
|
||||
if end_idx is None:
|
||||
raise ValueError("front matter sem delimitador final '---'.")
|
||||
|
||||
data: dict[str, Any] = {}
|
||||
current_key: str | None = None
|
||||
|
||||
for raw in lines[1:end_idx]:
|
||||
if not raw.strip():
|
||||
continue
|
||||
if raw.startswith(" - ") and current_key:
|
||||
if not isinstance(data.get(current_key), list):
|
||||
data[current_key] = []
|
||||
data[current_key].append(raw[4:].strip())
|
||||
continue
|
||||
|
||||
if ":" not in raw:
|
||||
raise ValueError(f"linha de metadado invalida: '{raw}'")
|
||||
|
||||
key, val = raw.split(":", 1)
|
||||
key = key.strip()
|
||||
val = val.strip()
|
||||
|
||||
if val == "":
|
||||
data[key] = []
|
||||
current_key = key
|
||||
else:
|
||||
data[key] = parse_scalar(val)
|
||||
current_key = key
|
||||
|
||||
return data
|
||||
|
||||
|
||||
def validate_metadata(
|
||||
metadata: dict[str, Any], schema: dict[str, Any], record_path: Path
|
||||
) -> list[str]:
|
||||
errors: list[str] = []
|
||||
required = schema.get("required", [])
|
||||
props = schema.get("properties", {})
|
||||
|
||||
for key in required:
|
||||
if key not in metadata:
|
||||
errors.append(f"{record_path}: campo obrigatorio ausente '{key}'")
|
||||
|
||||
for key, rule in props.items():
|
||||
if key not in metadata:
|
||||
continue
|
||||
value = metadata[key]
|
||||
expected_type = rule.get("type")
|
||||
|
||||
if expected_type == "array" and not isinstance(value, list):
|
||||
errors.append(f"{record_path}: '{key}' deve ser array")
|
||||
continue
|
||||
if expected_type == "string" and not isinstance(value, str):
|
||||
errors.append(f"{record_path}: '{key}' deve ser string")
|
||||
continue
|
||||
|
||||
enum = rule.get("enum")
|
||||
if enum and value not in enum:
|
||||
errors.append(f"{record_path}: '{key}' fora do enum permitido")
|
||||
|
||||
pattern = rule.get("pattern")
|
||||
if pattern and isinstance(value, str):
|
||||
if not re.match(pattern, value):
|
||||
errors.append(f"{record_path}: '{key}' nao corresponde ao padrao esperado")
|
||||
|
||||
if key in {"created_at", "updated_at"} and isinstance(value, str):
|
||||
if not re.match(r"^\d{4}-\d{2}-\d{2}$", value):
|
||||
errors.append(f"{record_path}: '{key}' deve estar em formato YYYY-MM-DD")
|
||||
|
||||
return errors
|
||||
|
||||
|
||||
def load_json(path: Path) -> dict[str, Any]:
|
||||
with path.open("r", encoding="utf-8") as f:
|
||||
return json.load(f)
|
||||
|
||||
|
||||
def build_summary(text: str, title: str) -> str:
|
||||
for raw in text.splitlines():
|
||||
line = raw.strip()
|
||||
if not line or line.startswith("#") or line.startswith("---"):
|
||||
continue
|
||||
return line[:180]
|
||||
return title[:180]
|
||||
|
||||
|
||||
def collect_records(schema: dict[str, Any]) -> tuple[list[dict[str, Any]], list[str]]:
|
||||
errors: list[str] = []
|
||||
records: list[dict[str, Any]] = []
|
||||
|
||||
record_files = sorted(RECORDS_ROOT.rglob("*.md"))
|
||||
if not record_files:
|
||||
errors.append("nenhum registro encontrado em knowledge-base/records")
|
||||
return records, errors
|
||||
|
||||
for file_path in record_files:
|
||||
rel_path = str(file_path.relative_to(KB_ROOT))
|
||||
text = file_path.read_text(encoding="utf-8")
|
||||
try:
|
||||
metadata = parse_front_matter(text)
|
||||
except ValueError as exc:
|
||||
errors.append(f"{file_path}: {exc}")
|
||||
continue
|
||||
|
||||
validation_errors = validate_metadata(metadata, schema, file_path)
|
||||
if validation_errors:
|
||||
errors.extend(validation_errors)
|
||||
continue
|
||||
|
||||
rec_id = metadata.get("id")
|
||||
if not isinstance(rec_id, str):
|
||||
errors.append(f"{file_path}: id invalido")
|
||||
continue
|
||||
|
||||
records.append(
|
||||
{
|
||||
"id": rec_id,
|
||||
"title": metadata["title"],
|
||||
"domain": metadata["domain"],
|
||||
"tags": metadata["tags"],
|
||||
"status": metadata["status"],
|
||||
"severity": metadata["severity"],
|
||||
"path": rel_path,
|
||||
"summary": build_summary(text, metadata["title"]),
|
||||
}
|
||||
)
|
||||
|
||||
return records, errors
|
||||
|
||||
|
||||
def check_index_against_records(index: dict[str, Any], records: list[dict[str, Any]]) -> list[str]:
|
||||
errors: list[str] = []
|
||||
indexed_records = index.get("records", [])
|
||||
|
||||
index_by_id: dict[str, dict[str, Any]] = {}
|
||||
index_by_path: dict[str, dict[str, Any]] = {}
|
||||
for rec in indexed_records:
|
||||
rec_id = rec.get("id")
|
||||
rec_path = rec.get("path")
|
||||
if not rec_id or not rec_path:
|
||||
errors.append("index.json: cada registro deve conter 'id' e 'path'")
|
||||
continue
|
||||
if rec_id in index_by_id:
|
||||
errors.append(f"index.json: id duplicado '{rec_id}'")
|
||||
if rec_path in index_by_path:
|
||||
errors.append(f"index.json: path duplicado '{rec_path}'")
|
||||
index_by_id[rec_id] = rec
|
||||
index_by_path[rec_path] = rec
|
||||
|
||||
records_by_id = {rec["id"]: rec for rec in records}
|
||||
|
||||
for rec in records:
|
||||
indexed = index_by_id.get(rec["id"])
|
||||
if not indexed:
|
||||
errors.append(f"{rec['path']}: id '{rec['id']}' nao encontrado em index.json")
|
||||
continue
|
||||
|
||||
if indexed.get("path") != rec["path"]:
|
||||
errors.append(
|
||||
f"{rec['path']}: path em index.json diverge "
|
||||
f"(index='{indexed.get('path')}', file='{rec['path']}')"
|
||||
)
|
||||
|
||||
for key in ("title", "domain", "status", "severity"):
|
||||
if indexed.get(key) != rec[key]:
|
||||
errors.append(
|
||||
f"{rec['path']}: campo '{key}' divergente entre registro e index.json"
|
||||
)
|
||||
|
||||
for rec in indexed_records:
|
||||
rec_id = rec.get("id")
|
||||
rec_path = rec.get("path")
|
||||
if not isinstance(rec_path, str):
|
||||
continue
|
||||
abs_path = KB_ROOT / rec_path
|
||||
if not abs_path.exists():
|
||||
errors.append(f"index.json: arquivo referenciado inexistente '{rec_path}'")
|
||||
if isinstance(rec_id, str) and rec_id not in records_by_id:
|
||||
errors.append(f"index.json: id '{rec_id}' sem arquivo correspondente em records/")
|
||||
|
||||
return errors
|
||||
|
||||
|
||||
def write_fixed_index(old_index: dict[str, Any], records: list[dict[str, Any]]) -> None:
|
||||
new_index = {
|
||||
"version": old_index.get("version", "1.0.0"),
|
||||
"last_updated": date.today().isoformat(),
|
||||
"records": sorted(records, key=lambda r: r["id"]),
|
||||
}
|
||||
with INDEX_PATH.open("w", encoding="utf-8") as f:
|
||||
json.dump(new_index, f, ensure_ascii=False, indent=2)
|
||||
f.write("\n")
|
||||
|
||||
|
||||
def run_check(fix: bool) -> int:
|
||||
if not INDEX_PATH.exists():
|
||||
return fail([f"index nao encontrado em {INDEX_PATH}"])
|
||||
if not SCHEMA_PATH.exists():
|
||||
return fail([f"schema nao encontrado em {SCHEMA_PATH}"])
|
||||
if not RECORDS_ROOT.exists():
|
||||
return fail([f"diretorio de registros nao encontrado em {RECORDS_ROOT}"])
|
||||
|
||||
schema = load_json(SCHEMA_PATH)
|
||||
index = load_json(INDEX_PATH)
|
||||
records, record_errors = collect_records(schema)
|
||||
|
||||
if record_errors and fix:
|
||||
return fail(
|
||||
["kb-fix nao pode continuar enquanto houver metadados invalidos em registros."]
|
||||
+ record_errors
|
||||
)
|
||||
|
||||
if fix:
|
||||
write_fixed_index(index, records)
|
||||
print("KB FIX: index.json sincronizado com registros validos.")
|
||||
check_errors = check_index_against_records(load_json(INDEX_PATH), records)
|
||||
if check_errors:
|
||||
return fail(check_errors)
|
||||
return ok("KB FIX")
|
||||
|
||||
check_errors = record_errors + check_index_against_records(index, records)
|
||||
if check_errors:
|
||||
return fail(check_errors)
|
||||
return ok()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument(
|
||||
"--fix",
|
||||
action="store_true",
|
||||
help="corrige automaticamente index.json com base nos registros validos",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
return run_check(fix=args.fix)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Loading…
Reference in a new issue