85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "scripts"))
|
|
|
|
from management_console_server import query_rag # noqa: E402
|
|
|
|
|
|
def run_checks() -> dict:
|
|
official = query_rag(
|
|
{
|
|
"question": "форма при открытии пример артикул",
|
|
"source_type": "official_1c_docs",
|
|
"limit": 8,
|
|
}
|
|
)
|
|
official_bad = [
|
|
row
|
|
for row in official.get("results") or []
|
|
if row.get("source_type") in {"metadata", "bsl_modules", "examples"}
|
|
or "example" in str(row.get("source_path") or "").casefold()
|
|
]
|
|
|
|
metadata = query_rag(
|
|
{
|
|
"question": "какие реквизиты у справочника Номенклатура",
|
|
"source_type": "metadata",
|
|
"limit": 5,
|
|
}
|
|
)
|
|
metadata_answer = str(metadata.get("answer") or "").casefold()
|
|
|
|
checks = [
|
|
{
|
|
"id": "official_docs_exclude_examples",
|
|
"status": "passed" if not official_bad else "failed",
|
|
"details": official_bad,
|
|
},
|
|
{
|
|
"id": "metadata_scope_warns",
|
|
"status": "passed" if "примеры" in metadata_answer and "не подтверждают текущую базу" in metadata_answer else "failed",
|
|
"answer": metadata.get("answer"),
|
|
},
|
|
]
|
|
return {
|
|
"schema": "onec_rag_source_governance_check.v1",
|
|
"status": "ok" if all(item["status"] == "passed" for item in checks) else "failed",
|
|
"checks": checks,
|
|
"samples": {
|
|
"official_1c_docs": {
|
|
"source_scope": official.get("source_scope"),
|
|
"result_count": official.get("result_count"),
|
|
"source_types": sorted({row.get("source_type") for row in official.get("results") or [] if row.get("source_type")}),
|
|
},
|
|
"metadata": {
|
|
"source_scope": metadata.get("source_scope"),
|
|
"result_count": metadata.get("result_count"),
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Check 1C RAG source governance.")
|
|
parser.add_argument("--output", type=Path)
|
|
parser.add_argument("--print", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
report = run_checks()
|
|
if args.output:
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
if args.print or not args.output:
|
|
print(json.dumps(report, ensure_ascii=False, indent=2))
|
|
return 0 if report["status"] == "ok" else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|