60 lines
1.8 KiB
Python
60 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
from search_1c_rag_vector import DEFAULT_CORPUS, DEFAULT_INDEX, freshness, load_meta
|
|
|
|
|
|
def check_vector_freshness(index_path: Path, corpus_path: Path) -> dict:
|
|
if not index_path.exists():
|
|
return {
|
|
"status": "missing",
|
|
"reason": "vector index is missing",
|
|
"index": str(index_path),
|
|
"corpus": str(corpus_path),
|
|
}
|
|
try:
|
|
conn = sqlite3.connect(index_path)
|
|
try:
|
|
meta = load_meta(conn)
|
|
finally:
|
|
conn.close()
|
|
except sqlite3.Error as exc:
|
|
return {
|
|
"status": "invalid",
|
|
"reason": str(exc),
|
|
"index": str(index_path),
|
|
"corpus": str(corpus_path),
|
|
}
|
|
report = freshness(meta, corpus_path)
|
|
return {
|
|
**report,
|
|
"index": str(index_path),
|
|
"embedding_model": meta.get("embedding_model"),
|
|
"embedding_dimensions": meta.get("embedding_dimensions"),
|
|
"doc_count": meta.get("doc_count"),
|
|
"built_at": meta.get("built_at"),
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Check whether the 1C RAG vector index is fresh.")
|
|
parser.add_argument("--index", type=Path, default=DEFAULT_INDEX)
|
|
parser.add_argument("--corpus", type=Path, default=DEFAULT_CORPUS)
|
|
parser.add_argument("--print", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
report = check_vector_freshness(args.index, args.corpus)
|
|
if args.print:
|
|
print(json.dumps(report, ensure_ascii=False, indent=2))
|
|
else:
|
|
print(f"1C RAG vector freshness: {report['status']}")
|
|
return 0 if report["status"] in {"fresh", "missing"} else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|