"""Build or refresh the SQLite FTS fee index (no OpenSearch / no torch).""" from __future__ import annotations import argparse import logging import sys from .config import settings from .schedule_docs import build_documents from . import sqlite_store logger = logging.getLogger(__name__) def build_sqlite_index(*, force: bool = False) -> dict: path = sqlite_store.db_path() if path.exists() and sqlite_store.count_codes() > 0 and not force: n = sqlite_store.count_codes() logger.info("SQLite index already present (%d codes) at %s", n, path) return {"status": "exists", "codes": n, "path": str(path)} docs = build_documents() n = sqlite_store.replace_all(docs) return {"status": "built", "codes": n, "path": str(path)} def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--force", action="store_true", help="Rebuild even if an index already exists", ) args = parser.parse_args(argv) logging.basicConfig(level=logging.INFO) # Pilot / Spaces often use /tmp for writable caches. settings.ohip_data_dir = settings.ohip_data_dir or "/tmp/ohip" summary = build_sqlite_index(force=args.force) print(summary) return 0 if summary.get("codes", 0) > 0 else 1 if __name__ == "__main__": sys.exit(main())