Projets/ObsiGate
1
1
Fork 0
Code Issues 16 Pull Requests Actions Packages Projects Releases 16 Wiki Activity

Add advanced search engine with inverted index, thread pool execution, configuration API, and comprehensive diagnostics

Browse Source
This commit is contained in:
bruno
2026-03-23 13:21:20 -04:00
parent af7d1c0d2e
commit b40fcae62f
8 changed files with 885 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/indexer.py
+6
View File
@@ -22,6 +22,10 @@ vault_config: Dict[str, Dict[str, Any]] = {}
# Thread-safe lock for index updates
_index_lock = threading.Lock()
# Generation counter — incremented on each index rebuild so consumers
# (e.g. the inverted index in search.py) can detect staleness.
_index_generation: int = 0
# O(1) lookup table for wikilink resolution: {filename_lower: [{vault, path}, ...]}
_file_lookup: Dict[str, List[Dict[str, str]]] = {}
@@ -318,6 +322,7 @@ async def build_index() -> None:
new_path_index[vname] = vdata.get("paths", [])
# Atomic swap under lock for thread safety during concurrent reads
global _index_generation
with _index_lock:
index.clear()
index.update(new_index)
@@ -325,6 +330,7 @@ async def build_index() -> None:
_file_lookup.update(new_lookup)
path_index.clear()
path_index.update(new_path_index)
_index_generation += 1
total_files = sum(len(v["files"]) for v in index.values())
logger.info(f"Index built: {len(index)} vaults, {total_files} total files")
backend/main.py
+178 -5
View File
@@ -1,8 +1,12 @@
import asyncio
import json as _json
import re
import html as html_mod
import logging
import mimetypes
from concurrent.futures import ThreadPoolExecutor
from contextlib import asynccontextmanager
from functools import partial
from pathlib import Path
from typing import Optional, List, Dict, Any
@@ -111,11 +115,14 @@ class SearchResultItem(BaseModel):
class SearchResponse(BaseModel):
"""Full-text search response."""
"""Full-text search response with optional pagination."""
query: str
vault_filter: str
tag_filter: Optional[str]
count: int
total: int = Field(0, description="Total results before pagination")
offset: int = Field(0, description="Current pagination offset")
limit: int = Field(200, description="Page size")
results: List[SearchResultItem]
@@ -165,6 +172,7 @@ class AdvancedSearchResponse(BaseModel):
offset: int
limit: int
facets: SearchFacets
query_time_ms: float = Field(0, description="Server-side query time in milliseconds")
class TitleSuggestion(BaseModel):
@@ -210,16 +218,25 @@ class HealthResponse(BaseModel):
# Application lifespan (replaces deprecated on_event)
# ---------------------------------------------------------------------------
# Thread pool for offloading CPU-bound search from the event loop.
# Sized to 2 workers so concurrent searches don't starve other requests.
_search_executor: Optional[ThreadPoolExecutor] = None
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Application lifespan: build index on startup, cleanup on shutdown."""
global _search_executor
_search_executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="search")
logger.info("ObsiGate starting \u2014 building index...")
await build_index()
logger.info("ObsiGate ready.")
yield
_search_executor.shutdown(wait=False)
_search_executor = None
app = FastAPI(title="ObsiGate", version="1.1.0", lifespan=lifespan)
app = FastAPI(title="ObsiGate", version="1.2.0", lifespan=lifespan)
# Resolve frontend path relative to this file
FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"
@@ -687,22 +704,38 @@ async def api_search(
q: str = Query("", description="Search query"),
vault: str = Query("all", description="Vault filter"),
tag: Optional[str] = Query(None, description="Tag filter"),
limit: int = Query(50, ge=1, le=200, description="Results per page"),
offset: int = Query(0, ge=0, description="Pagination offset"),
):
"""Full-text search across vaults with relevance scoring.
Supports combining free-text queries with tag filters.
Results are ranked by a multi-factor scoring algorithm.
Pagination via ``limit`` and ``offset`` (defaults preserve backward compat).
Args:
q: Free-text search string.
vault: Vault name or ``"all"`` to search everywhere.
tag: Comma-separated tag names to require.
limit: Max results per page (1–200).
offset: Pagination offset.
Returns:
``SearchResponse`` with ranked results and snippets.
"""
results = search(q, vault_filter=vault, tag_filter=tag)
return {"query": q, "vault_filter": vault, "tag_filter": tag, "count": len(results), "results": results}
loop = asyncio.get_event_loop()
# Fetch full result set (capped at DEFAULT_SEARCH_LIMIT internally)
all_results = await loop.run_in_executor(
_search_executor,
partial(search, q, vault_filter=vault, tag_filter=tag),
)
total = len(all_results)
page = all_results[offset: offset + limit]
return {
"query": q, "vault_filter": vault, "tag_filter": tag,
"count": len(page), "total": total, "offset": offset, "limit": limit,
"results": page,
}
@app.get("/api/tags", response_model=TagsResponse)
@@ -793,7 +826,12 @@ async def api_advanced_search(
Returns:
``AdvancedSearchResponse`` with scored results, facets, and pagination info.
"""
return advanced_search(q, vault_filter=vault, tag_filter=tag, limit=limit, offset=offset, sort_by=sort)
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
_search_executor,
partial(advanced_search, q, vault_filter=vault, tag_filter=tag,
limit=limit, offset=offset, sort_by=sort),
)
@app.get("/api/suggest", response_model=SuggestResponse)
@@ -924,6 +962,141 @@ async def api_attachment_stats(vault: Optional[str] = Query(None, description="V
return {"vaults": stats}
# ---------------------------------------------------------------------------
# Configuration API
# ---------------------------------------------------------------------------
_CONFIG_PATH = Path(__file__).resolve().parent.parent / "config.json"
_DEFAULT_CONFIG = {
"search_workers": 2,
"debounce_ms": 300,
"results_per_page": 50,
"min_query_length": 2,
"search_timeout_ms": 30000,
"max_content_size": 100000,
"snippet_context_chars": 120,
"max_snippet_highlights": 5,
"title_boost": 3.0,
"path_boost": 1.5,
"tag_boost": 2.0,
"prefix_max_expansions": 50,
}
def _load_config() -> dict:
"""Load config from disk, merging with defaults."""
config = dict(_DEFAULT_CONFIG)
if _CONFIG_PATH.exists():
try:
stored = _json.loads(_CONFIG_PATH.read_text(encoding="utf-8"))
config.update(stored)
except Exception as e:
logger.warning(f"Failed to read config.json: {e}")
return config
def _save_config(config: dict) -> None:
"""Persist config to disk."""
try:
_CONFIG_PATH.write_text(
_json.dumps(config, indent=2, ensure_ascii=False),
encoding="utf-8",
)
except Exception as e:
logger.error(f"Failed to write config.json: {e}")
raise HTTPException(status_code=500, detail=f"Failed to save config: {e}")
@app.get("/api/config")
async def api_get_config():
"""Return current configuration with defaults for missing keys."""
return _load_config()
@app.post("/api/config")
async def api_set_config(body: dict = Body(...)):
"""Update configuration. Only known keys are accepted.
Keys matching ``_DEFAULT_CONFIG`` are validated and persisted.
Unknown keys are silently ignored.
Returns the full merged config after update.
"""
current = _load_config()
updated_keys = []
for key, value in body.items():
if key in _DEFAULT_CONFIG:
expected_type = type(_DEFAULT_CONFIG[key])
if isinstance(value, expected_type) or (expected_type is float and isinstance(value, (int, float))):
current[key] = value
updated_keys.append(key)
else:
raise HTTPException(
status_code=400,
detail=f"Invalid type for '{key}': expected {expected_type.__name__}, got {type(value).__name__}",
)
_save_config(current)
logger.info(f"Config updated: {updated_keys}")
return current
# ---------------------------------------------------------------------------
# Diagnostics API
# ---------------------------------------------------------------------------
@app.get("/api/diagnostics")
async def api_diagnostics():
"""Return index statistics and system diagnostics.
Includes document counts, token counts, memory estimates,
and inverted index status.
"""
from backend.search import get_inverted_index
import sys
inv = get_inverted_index()
# Per-vault stats
vault_stats = {}
total_files = 0
total_tags = 0
for vname, vdata in index.items():
file_count = len(vdata.get("files", []))
tag_count = len(vdata.get("tags", {}))
vault_stats[vname] = {"file_count": file_count, "tag_count": tag_count}
total_files += file_count
total_tags += tag_count
# Memory estimate for inverted index
word_index_entries = sum(len(docs) for docs in inv.word_index.values())
mem_estimate_mb = round(
(sys.getsizeof(inv.word_index) + word_index_entries * 80
+ len(inv.doc_info) * 200
+ len(inv._sorted_tokens) * 60) / (1024 * 1024), 2
)
return {
"index": {
"total_files": total_files,
"total_tags": total_tags,
"vaults": vault_stats,
},
"inverted_index": {
"unique_tokens": len(inv.word_index),
"total_postings": word_index_entries,
"documents": inv.doc_count,
"sorted_tokens": len(inv._sorted_tokens),
"is_stale": inv.is_stale(),
"memory_estimate_mb": mem_estimate_mb,
},
"config": _load_config(),
"search_executor": {
"active": _search_executor is not None,
"max_workers": _search_executor._max_workers if _search_executor else 0,
},
}
# ---------------------------------------------------------------------------
# Static files & SPA fallback
# ---------------------------------------------------------------------------
backend/search.py
+180 -86
View File
@@ -1,10 +1,13 @@
import bisect
import logging
import math
import re
import time
import unicodedata
from collections import defaultdict
from typing import List, Dict, Any, Optional, Tuple
from backend import indexer as _indexer
from backend.indexer import index
logger = logging.getLogger("obsigate.search")
@@ -256,12 +259,21 @@ class InvertedIndex:
self.tag_prefix_index: Dict[str, List[str]] = defaultdict(list)
self.title_norm_map: Dict[str, List[Dict[str, str]]] = defaultdict(list)
self.doc_count: int = 0
self._source_id: Optional[int] = None
self.doc_info: Dict[str, Dict[str, Any]] = {}
self.doc_vault: Dict[str, str] = {}
self.vault_docs: Dict[str, set] = defaultdict(set)
self.tag_docs: Dict[str, set] = defaultdict(set)
self._sorted_tokens: List[str] = []
self._source_generation: int = -1
def is_stale(self) -> bool:
"""Check if the inverted index needs rebuilding."""
current_id = id(index)
return current_id != self._source_id
"""Check if the inverted index needs rebuilding.
Uses the indexer's generation counter which increments on every
rebuild, instead of ``id(index)`` which never changes since the
global dict is mutated in-place.
"""
return _indexer._index_generation != self._source_generation
def rebuild(self) -> None:
"""Rebuild inverted index from the global ``index`` dict.
@@ -276,12 +288,25 @@ class InvertedIndex:
self.tag_prefix_index = defaultdict(list)
self.title_norm_map = defaultdict(list)
self.doc_count = 0
self.doc_info = {}
self.doc_vault = {}
self.vault_docs = defaultdict(set)
self.tag_docs = defaultdict(set)
for vault_name, vault_data in index.items():
for file_info in vault_data.get("files", []):
doc_key = f"{vault_name}::{file_info['path']}"
self.doc_count += 1
# --- Document metadata for O(1) lookup ---
self.doc_info[doc_key] = file_info
self.doc_vault[doc_key] = vault_name
self.vault_docs[vault_name].add(doc_key)
# --- Per-document tag index ---
for tag in file_info.get("tags", []):
self.tag_docs[tag.lower()].add(doc_key)
# --- Title tokens ---
title_tokens = tokenize(file_info.get("title", ""))
for token in set(title_tokens):
@@ -316,7 +341,8 @@ class InvertedIndex:
if tag not in self.tag_prefix_index[prefix]:
self.tag_prefix_index[prefix].append(tag)
self._source_id = id(index)
self._sorted_tokens = sorted(self.word_index.keys())
self._source_generation = _indexer._index_generation
logger.info(
"Inverted index built: %d documents, %d unique tokens, %d tags",
self.doc_count,
@@ -358,6 +384,32 @@ class InvertedIndex:
return 0.0
return tf * self.idf(term)
def get_prefix_tokens(self, prefix: str, max_expansions: int = 50) -> List[str]:
"""Get all tokens starting with *prefix* using binary search.
Uses a pre-sorted token list for O(log V + k) lookup instead
of O(V) linear scan over the vocabulary.
Args:
prefix: Normalized prefix string.
max_expansions: Cap on returned tokens to bound work.
Returns:
List of matching tokens (including exact match if present).
"""
if not prefix or not self._sorted_tokens:
return []
lo = bisect.bisect_left(self._sorted_tokens, prefix)
results: List[str] = []
for i in range(lo, len(self._sorted_tokens)):
if self._sorted_tokens[i].startswith(prefix):
results.append(self._sorted_tokens[i])
if len(results) >= max_expansions:
break
else:
break
return results
# Singleton inverted index
_inverted_index = InvertedIndex()
@@ -582,6 +634,10 @@ def advanced_search(
) -> Dict[str, Any]:
"""Advanced full-text search with TF-IDF scoring, facets, and pagination.
Uses the inverted index for O(k × postings) candidate retrieval instead
of O(N) full document scan. Prefix matching uses binary search on a
sorted token list for O(log V + k) instead of O(V) linear scan.
Parses the query for operators (``tag:``, ``vault:``, ``title:``,
``path:``), falls back remaining tokens to TF-IDF scored free-text
search using the inverted index. Results include highlighted snippets
@@ -596,8 +652,10 @@ def advanced_search(
sort_by: ``"relevance"`` or ``"modified"``.
Returns:
Dict with ``results``, ``total``, ``offset``, ``limit``, ``facets``.
Dict with ``results``, ``total``, ``offset``, ``limit``, ``facets``,
``query_time_ms``.
"""
t0 = time.monotonic()
query = query.strip() if query else ""
parsed = _parse_advanced_query(query)
@@ -616,98 +674,132 @@ def advanced_search(
has_terms = len(query_terms) > 0
if not has_terms and not all_tags and not parsed["title"] and not parsed["path"]:
return {"results": [], "total": 0, "offset": offset, "limit": limit, "facets": {"tags": {}, "vaults": {}}}
return {"results": [], "total": 0, "offset": offset, "limit": limit,
"facets": {"tags": {}, "vaults": {}}, "query_time_ms": 0}
inv = get_inverted_index()
# ------------------------------------------------------------------
# Step 1: Candidate retrieval via inverted index (replaces O(N) scan)
# ------------------------------------------------------------------
if has_terms:
# Union of posting lists for all terms + prefix expansions
candidates: set = set()
for term in query_terms:
# Exact term matches
candidates.update(inv.word_index.get(term, {}).keys())
# Prefix matches — O(log V + k) via binary search
if len(term) >= MIN_PREFIX_LENGTH:
for expanded in inv.get_prefix_tokens(term):
if expanded != term:
candidates.update(inv.word_index.get(expanded, {}).keys())
else:
# Filter-only search: start with tag-filtered subset or all docs
if all_tags:
tag_sets = [inv.tag_docs.get(t.lower(), set()) for t in all_tags]
candidates = set.intersection(*tag_sets) if tag_sets else set()
else:
candidates = set(inv.doc_info.keys())
# ------------------------------------------------------------------
# Step 2: Apply filters on candidate set
# ------------------------------------------------------------------
if effective_vault != "all":
candidates &= inv.vault_docs.get(effective_vault, set())
if all_tags and has_terms:
for t in all_tags:
candidates &= inv.tag_docs.get(t.lower(), set())
if parsed["title"]:
norm_title_filter = normalize_text(parsed["title"])
candidates = {
dk for dk in candidates
if norm_title_filter in normalize_text(inv.doc_info[dk].get("title", ""))
}
if parsed["path"]:
norm_path_filter = normalize_text(parsed["path"])
candidates = {
dk for dk in candidates
if norm_path_filter in normalize_text(inv.doc_info[dk].get("path", ""))
}
# ------------------------------------------------------------------
# Step 3: Score only the candidates (not all N documents)
# ------------------------------------------------------------------
scored_results: List[Tuple[float, Dict[str, Any]]] = []
facet_tags: Dict[str, int] = defaultdict(int)
facet_vaults: Dict[str, int] = defaultdict(int)
for vault_name, vault_data in index.items():
if effective_vault != "all" and vault_name != effective_vault:
# Pre-compute prefix expansions once per term (avoid repeated binary search)
prefix_expansions: Dict[str, List[str]] = {}
if has_terms:
for term in query_terms:
if len(term) >= MIN_PREFIX_LENGTH:
prefix_expansions[term] = [
t for t in inv.get_prefix_tokens(term) if t != term
]
for doc_key in candidates:
file_info = inv.doc_info.get(doc_key)
if file_info is None:
continue
vault_name = inv.doc_vault[doc_key]
for file_info in vault_data.get("files", []):
doc_key = f"{vault_name}::{file_info['path']}"
score = 0.0
if has_terms:
for term in query_terms:
tfidf = inv.tf_idf(term, doc_key)
score += tfidf
# --- Tag filter ---
if all_tags:
file_tags_lower = [t.lower() for t in file_info.get("tags", [])]
if not all(t.lower() in file_tags_lower for t in all_tags):
continue
# Title boost — check if term appears in title tokens
norm_title = normalize_text(file_info.get("title", ""))
if term in norm_title:
score += tfidf * TITLE_BOOST
# --- Title filter ---
if parsed["title"]:
norm_title_filter = normalize_text(parsed["title"])
norm_file_title = normalize_text(file_info.get("title", ""))
if norm_title_filter not in norm_file_title:
continue
# Path boost
norm_path = normalize_text(file_info.get("path", ""))
if term in norm_path:
score += tfidf * PATH_BOOST
# --- Path filter ---
if parsed["path"]:
norm_path_filter = normalize_text(parsed["path"])
norm_file_path = normalize_text(file_info.get("path", ""))
if norm_path_filter not in norm_file_path:
continue
# --- Scoring ---
score = 0.0
if has_terms:
# TF-IDF scoring for each term
for term in query_terms:
tfidf = inv.tf_idf(term, doc_key)
score += tfidf
# Title boost — check if term appears in title tokens
norm_title = normalize_text(file_info.get("title", ""))
if term in norm_title:
score += tfidf * TITLE_BOOST
# Path boost
norm_path = normalize_text(file_info.get("path", ""))
if term in norm_path:
score += tfidf * PATH_BOOST
# Tag boost
for tag in file_info.get("tags", []):
if term in normalize_text(tag):
score += tfidf * TAG_BOOST
break
# Also add prefix matching bonus for partial words
for term in query_terms:
if len(term) >= MIN_PREFIX_LENGTH:
for indexed_term, docs in inv.word_index.items():
if indexed_term.startswith(term) and indexed_term != term:
if doc_key in docs:
score += inv.tf_idf(indexed_term, doc_key) * 0.5
else:
# Filter-only search (tag/title/path): score = 1
score = 1.0
if score > 0:
# Build highlighted snippet
content = file_info.get("content", "")
if has_terms:
snippet = _extract_highlighted_snippet(content, query_terms)
else:
snippet = _escape_html(content[:200].strip()) if content else ""
result = {
"vault": vault_name,
"path": file_info["path"],
"title": file_info["title"],
"tags": file_info.get("tags", []),
"score": round(score, 4),
"snippet": snippet,
"modified": file_info.get("modified", ""),
}
scored_results.append((score, result))
# Facets
facet_vaults[vault_name] = facet_vaults.get(vault_name, 0) + 1
# Tag boost
for tag in file_info.get("tags", []):
facet_tags[tag] = facet_tags.get(tag, 0) + 1
if term in normalize_text(tag):
score += tfidf * TAG_BOOST
break
# Prefix matching bonus (bounded by pre-computed expansions)
for term, expansions in prefix_expansions.items():
for expanded_term in expansions:
score += inv.tf_idf(expanded_term, doc_key) * 0.5
else:
# Filter-only search (tag/title/path): score = 1
score = 1.0
if score > 0:
# Build highlighted snippet
content = file_info.get("content", "")
if has_terms:
snippet = _extract_highlighted_snippet(content, query_terms)
else:
snippet = _escape_html(content[:200].strip()) if content else ""
result = {
"vault": vault_name,
"path": file_info["path"],
"title": file_info["title"],
"tags": file_info.get("tags", []),
"score": round(score, 4),
"snippet": snippet,
"modified": file_info.get("modified", ""),
}
scored_results.append((score, result))
# Facets
facet_vaults[vault_name] = facet_vaults.get(vault_name, 0) + 1
for tag in file_info.get("tags", []):
facet_tags[tag] = facet_tags.get(tag, 0) + 1
# Sort
if sort_by == "modified":
@@ -717,6 +809,7 @@ def advanced_search(
total = len(scored_results)
page = scored_results[offset: offset + limit]
elapsed_ms = round((time.monotonic() - t0) * 1000, 1)
return {
"results": [r for _, r in page],
@@ -727,6 +820,7 @@ def advanced_search(
"tags": dict(sorted(facet_tags.items(), key=lambda x: -x[1])[:20]),
"vaults": dict(sorted(facet_vaults.items(), key=lambda x: -x[1])),
},
"query_time_ms": elapsed_ms,
}