Upload app.py

app.py

@@ -46,7 +46,11 @@ source_records_map: dict[str, list[int]] = {}
 extension_counts: dict[str, int] = {}
 source_extension_counts: dict[str, dict[str, int]] = {}
 word_index: dict[str, set[int]] = {}
+latin_all_index: dict[str, set[int]] = {}
+latin_file_index: dict[str, set[int]] = {}
+latin_vocabulary_all: set[str] = set()
 TOKEN_RE = re.compile(r"[a-z0-9]+|[\u4e00-\u9fff\u3400-\u4dbf]+")
+LATIN_RE = re.compile(r"[a-z0-9]+")
 TOKENIZER_VERSION = "cjk-bigram-boundary-fts5-v6-snippet-anchors"
 LITERAL_PREFIX = "\0literal:"
 API_CACHE_TTL_SECONDS = 120
@@ -81,6 +85,7 @@ def index_tokens(text: str) -> set[str]:
             tokens.add(part)
     return tokens
 
+@lru_cache(maxsize=4096)
 def query_tokens(text: str) -> list[str]:
     tokens = []
     for part in query_terms(text):
@@ -90,6 +95,7 @@ def query_tokens(text: str) -> list[str]:
             tokens.append(part)
     return list(dict.fromkeys(tokens))
 
+@lru_cache(maxsize=4096)
 def literal_query_tokens(text: str) -> list[str]:
     normalized = normalize_text(text)
     return list(dict.fromkeys(
@@ -98,6 +104,7 @@ def literal_query_tokens(text: str) -> list[str]:
         if not TOKEN_RE.fullmatch(normalized[index + 1])
     ))
 
+@lru_cache(maxsize=4096)
 def wildcard_required_tokens(text: str) -> list[str]:
     tokens = []
     for fixed_part in re.split(r"[*?]+", normalize_text(text)):
@@ -724,8 +731,11 @@ def load_json_gz(path: Path):
     return json.loads(gzip.decompress(path.read_bytes()).decode("utf-8"))
 
 def build_indexes() -> None:
-    global word_index, source_records_map, extension_counts, source_extension_counts, name_order, size_order, source_name_order, source_size_order
+    global word_index, latin_all_index, latin_file_index, latin_vocabulary_all, source_records_map, extension_counts, source_extension_counts, name_order, size_order, source_name_order, source_size_order
     word_index = {}
+    latin_all_index = {}
+    latin_file_index = {}
+    latin_vocabulary_all = set()
     source_records_map = {}
     extension_counts = {}
     source_extension_counts = {}
@@ -745,6 +755,12 @@ def build_indexes() -> None:
         rec["_rank_path_key"] = rec["display_rel_path"].lower()
         for token in tokens:
             word_index.setdefault(token, set()).add(idx)
+            if LATIN_RE.fullmatch(token):
+                latin_all_index.setdefault(token, set()).add(idx)
+                latin_vocabulary_all.add(token)
+        for token in TOKEN_RE.findall(rec["_file_search_text"]):
+            if LATIN_RE.fullmatch(token):
+                latin_file_index.setdefault(token, set()).add(idx)
     name_order = sorted(
         range(len(records)),
         key=lambda idx: (records[idx]["display_rel_path"].lower(), records[idx]["doc_id"]),
@@ -900,15 +916,45 @@ def add_summaries(items: list[dict], query: str = "", matched_snippets: bool = F
         item["snippet_complete"] = bool(payload) or (not needs_matched_snippet and item["doc_id"] in summaries)
     return items
 
+def latin_substring_candidates(term: str, search_paths: bool) -> set[int]:
+    index = latin_all_index if search_paths else latin_file_index
+    candidates: set[int] = set()
+    for token in latin_vocabulary_all:
+        if term in token:
+            candidates |= index.get(token, ())
+    return candidates
+
+def metadata_prefilter(terms: list[str], search_paths: bool) -> set[int] | None:
+    candidates: set[int] | None = None
+    for term in terms:
+        if LATIN_RE.fullmatch(term):
+            term_candidates = latin_substring_candidates(term, search_paths)
+        else:
+            term_candidates = None
+            for token in query_tokens(term):
+                indices = word_index.get(token)
+                if not indices:
+                    term_candidates = set()
+                    break
+                term_candidates = set(indices) if term_candidates is None else term_candidates & indices
+                if not term_candidates:
+                    break
+        if term_candidates is None:
+            continue
+        candidates = term_candidates if candidates is None else candidates & term_candidates
+        if not candidates:
+            return set()
+    return candidates
+
 def metadata_matches(q: str, exact: bool, search_paths: bool) -> set[int]:
     field = "_search_text" if search_paths else "_file_search_text"
     if exact:
         terms = query_terms(q)
         normalized_query, pattern = compile_exact_query(q)
-        # Whole-token postings are not a lossless prefilter for Latin
-        # substrings or punctuation-only queries. CJK literals necessarily
-        # contain all of their indexed unigram/bigram tokens, so retain the
-        # fast candidate intersection for that common path.
+        # CJK literals necessarily contain all of their indexed unigram/bigram
+        # tokens, so candidate intersection is lossless there. Latin substrings
+        # are resolved through the per-field Latin token index, and
+        # punctuation-only queries retain the full-range scan.
         if has_wildcard_query(q):
             candidate_set = None
             for token in wildcard_required_tokens(normalized_query):
@@ -917,16 +963,10 @@ def metadata_matches(q: str, exact: bool, search_paths: bool) -> set[int]:
                 if not candidate_set:
                     return set()
             candidates = candidate_set if candidate_set is not None else range(len(records))
-        elif not terms or any(re.fullmatch(r"[a-z0-9]+", term) for term in terms):
+        elif not terms:
             candidates = range(len(records))
         else:
-            candidate_set = None
-            for token in query_tokens(q):
-                token_indices = word_index.get(token, set())
-                candidate_set = set(token_indices) if candidate_set is None else candidate_set & token_indices
-                if not candidate_set:
-                    return set()
-            candidates = candidate_set or set()
+            candidates = metadata_prefilter(terms, search_paths) or set()
         return {
             idx for idx in candidates
             if matches_normalized_exact_query(records[idx][field], normalized_query, pattern)
@@ -934,19 +974,11 @@ def metadata_matches(q: str, exact: bool, search_paths: bool) -> set[int]:
     terms = query_terms(q)
     if not terms:
         return set()
-    if any(re.fullmatch(r"[a-z0-9]+", term) for term in terms):
-        return {
-            idx for idx, rec in enumerate(records)
-            if all(term in rec[field] for term in terms)
-        }
-    candidates = None
-    for token in query_tokens(q):
-        token_indices = word_index.get(token, set())
-        candidates = set(token_indices) if candidates is None else candidates & token_indices
-        if not candidates:
-            return set()
+    candidates = metadata_prefilter(terms, search_paths)
+    if candidates is None:
+        candidates = range(len(records))
     return {
-        idx for idx in (candidates or set())
+        idx for idx in candidates
         if all(term in records[idx][field] for term in terms)
     }
 
@@ -1276,24 +1308,27 @@ def run_search(body: SearchRequest, sources_filter=None, timings=None):
         body.search_paths,
         body.fulltext,
     )
-    result = cached_search_payload(
-        key,
-        lambda: fulltext_search(
-            body.q, selected_sources, body.folders, body.min_size, body.max_size,
-            body.page, body.page_size, body.sort, body.exact, body.search_paths,
-            timings,
-        ) if body.fulltext else search(
-            body.q, selected_sources, body.folders, body.min_size, body.max_size,
-            body.page, body.page_size, body.sort, body.exact, body.search_paths,
-            timings,
-        ),
-        timings,
-    )
-    payload = dict(result)
-    requested_sources = set(selected_sources or source_counts)
-    payload["indexing"] = bool(body.fulltext and not requested_sources.issubset(ready_fulltext_sources()))
-    payload["index_generation"] = fulltext_generation
-    return payload
+    def build_payload():
+        result = (
+            fulltext_search(
+                body.q, selected_sources, body.folders, body.min_size, body.max_size,
+                body.page, body.page_size, body.sort, body.exact, body.search_paths,
+                timings,
+            )
+            if body.fulltext
+            else search(
+                body.q, selected_sources, body.folders, body.min_size, body.max_size,
+                body.page, body.page_size, body.sort, body.exact, body.search_paths,
+                timings,
+            )
+        )
+        payload = dict(result)
+        requested_sources = set(selected_sources or source_counts)
+        payload["indexing"] = bool(body.fulltext and not requested_sources.issubset(ready_fulltext_sources()))
+        payload["index_generation"] = fulltext_generation
+        return payload
+
+    return cached_search_payload(key, build_payload, timings)
 
 @app.post("/api/search")