From fb46ac0feab3911a94fa890a5ea582842c9f7e6d Mon Sep 17 00:00:00 2001 From: Rodolpho Lopes Date: Tue, 26 May 2026 20:19:54 +0000 Subject: [PATCH] chore: move kb_check.py para dentro da knowledge-base MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Script e Makefile agora vivem junto à base que gerenciam. Caminho anterior: GLPI11/scripts/kb_check.py Co-Authored-By: Claude Sonnet 4.6 --- Makefile | 7 ++ kb_check.py | 288 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 295 insertions(+) create mode 100644 Makefile create mode 100644 kb_check.py diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..7be34b9 --- /dev/null +++ b/Makefile @@ -0,0 +1,7 @@ +.PHONY: kb-check kb-fix + +kb-check: + python3 kb_check.py + +kb-fix: + python3 kb_check.py --fix diff --git a/kb_check.py b/kb_check.py new file mode 100644 index 0000000..9a2957e --- /dev/null +++ b/kb_check.py @@ -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())