-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathesir_cli.py
More file actions
115 lines (84 loc) · 3.68 KB
/
Copy pathesir_cli.py
File metadata and controls
115 lines (84 loc) · 3.68 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
from __future__ import annotations
"""Console CLI for indexing and searching without GUI.
Commands:
esir index -m MODEL -r ROOT [--chunk-chars N] [--overlap-chars N] [--batch-size N] [--ver V]
esir search -m MODEL -q QUERY [-k N]
esir clear [--all | --root DIR]
Comments are in English as requested.
"""
import argparse
import json
import sys
from typing import Any
from esir_lib import clear_all, clear_under_root, index_directory, search_query
def _add_common_model(p: argparse.ArgumentParser) -> None:
p.add_argument("-m", "--model", required=True, help="Path to GGUF embedding model")
def cmd_index(args: argparse.Namespace) -> int:
def log(msg: str) -> None:
print(msg, file=sys.stderr)
files, chunks = index_directory(
model_path=args.model,
root=args.root,
chunk_chars=args.chunk_chars,
overlap_chars=args.overlap_chars,
scanner_ver=args.ver,
batch_size=args.batch_size,
log=log,
)
print(json.dumps({"files_indexed": files, "chunks_added": chunks}))
return 0
def cmd_search(args: argparse.Namespace) -> int:
rows = search_query(args.model, args.query, k=args.k)
if args.json:
print(json.dumps(rows, ensure_ascii=False))
else:
for r in rows:
meta = r.get("meta", {})
print(f"{r.get('score', 0):.3f} {meta.get('path','')}#{r.get('chunk_index')}\n{r.get('text','')[:2000]}\n")
return 0
def cmd_clear(args: argparse.Namespace) -> int:
if args.all:
clear_all()
print("{\"cleared\": \"all\"}")
return 0
if args.root:
n = clear_under_root(args.root)
print(json.dumps({"cleared_under_root": n}))
return 0
print("Nothing to clear. Use --all or --root DIR", file=sys.stderr)
return 1
def build_parser() -> argparse.ArgumentParser:
ap = argparse.ArgumentParser(prog="esir", description="Local semantic indexer & search (GGUF + USearch)")
sub = ap.add_subparsers(dest="cmd", required=True)
p_index = sub.add_parser("index", help="Index a directory incrementally")
_add_common_model(p_index)
p_index.add_argument("-r", "--root", required=True, help="Directory to scan")
p_index.add_argument("--chunk-chars", type=int, default=2000, help="Chunk size in characters")
p_index.add_argument("--overlap-chars", type=int, default=200, help="Overlap between chunks")
p_index.add_argument("--batch-size", type=int, default=32, help="Embedding batch size")
p_index.add_argument("--ver", default="1", help="Scanner version string")
p_index.set_defaults(func=cmd_index)
p_search = sub.add_parser("search", help="Search the index with a query")
_add_common_model(p_search)
p_search.add_argument("-q", "--query", required=True, help="Search query text")
p_search.add_argument("-k", type=int, default=100, help="Top K results")
p_search.add_argument("--json", action="store_true", help="Output JSON")
p_search.set_defaults(func=cmd_search)
p_clear = sub.add_parser("clear", help="Clear index and metadata")
g = p_clear.add_mutually_exclusive_group(required=True)
g.add_argument("--all", action="store_true", help="Clear entire index and metadata")
g.add_argument("--root", help="Clear only entries under directory prefix")
p_clear.set_defaults(func=cmd_clear)
return ap
def main(argv: list[str] | None = None) -> int:
ap = build_parser()
args = ap.parse_args(argv)
return args.func(args)
def main_index() -> int:
return main(["index", *sys.argv[1:]])
def main_search() -> int:
return main(["search", *sys.argv[1:]])
def main_clear() -> int:
return main(["clear", *sys.argv[1:]])
if __name__ == "__main__":
raise SystemExit(main())