from sortedcontainers import SortedList import logging import math import re import time import unicodedata from collections import defaultdict from typing import List, Dict, Any, Optional, Tuple from snowballstemmer import stemmer as _snowball_stemmer from backend import indexer as _indexer from backend.indexer import index logger = logging.getLogger("obsigate.search") # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- DEFAULT_SEARCH_LIMIT = 200 ADVANCED_SEARCH_DEFAULT_LIMIT = 50 SNIPPET_CONTEXT_CHARS = 120 MAX_SNIPPET_HIGHLIGHTS = 5 TITLE_BOOST = 3.0 # TF-IDF multiplier for title matches PATH_BOOST = 1.5 # TF-IDF multiplier for path matches TAG_BOOST = 2.0 # TF-IDF multiplier for tag matches MIN_PREFIX_LENGTH = 2 # Minimum chars for prefix matching SUGGEST_LIMIT = 10 # Default max suggestions returned # Regex to tokenize text into alphanumeric words (Unicode-aware) _WORD_RE = re.compile(r"[\w]+", re.UNICODE) # French stemmer (lazy-init singleton) _FR_STEMMER: object | None = None def _get_stemmer(): """Return a cached French snowball stemmer instance.""" global _FR_STEMMER if _FR_STEMMER is None: _FR_STEMMER = _snowball_stemmer("french") return _FR_STEMMER def stem_token(token: str) -> str: """Reduce a word to its French stem (e.g. 'mangeons' → 'mang').""" try: return _get_stemmer().stemWord(token) except Exception: return token # fallback: keep original token if stemmer crashes def stem_tokens(tokens: list[str] | set[str]) -> set[str]: """Return the set of unique stems for a list of tokens.""" result = set() for t in tokens: try: stemmed = stem_token(t) except Exception: stemmed = t result.add(stemmed) return result # --------------------------------------------------------------------------- # Accent / Unicode normalization helpers # --------------------------------------------------------------------------- def normalize_text(text: str) -> str: """Normalize text for accent-insensitive comparison. Decomposes Unicode characters (NFD), strips combining diacritical marks, then lowercases the result. For example ``"Éléphant"`` → ``"elephant"``. Args: text: Raw input string. Returns: Lowercased, accent-stripped string. """ if not text: return "" # NFD decomposition splits base char + combining mark nfkd = unicodedata.normalize("NFKD", text) # Strip combining marks (category "Mn" = Mark, Nonspacing) stripped = "".join(ch for ch in nfkd if unicodedata.category(ch) != "Mn") return stripped.lower() def tokenize(text: str) -> List[str]: """Split text into normalized tokens (accent-stripped, lowercased words). Args: text: Raw text to tokenize. Returns: List of normalized word tokens. """ return _WORD_RE.findall(normalize_text(text)) # --------------------------------------------------------------------------- # Tag filter helper (unchanged for backward compat) # --------------------------------------------------------------------------- def _normalize_tag_filter(tag_filter: Optional[str]) -> List[str]: """Parse a comma-separated tag filter string into a clean list. Strips whitespace and leading ``#`` from each tag. Args: tag_filter: Raw tag filter string (e.g. ``"docker,linux"``). Returns: List of normalised tag strings, empty list if input is falsy. """ if not tag_filter: return [] return [tag.strip().lstrip("#") for tag in tag_filter.split(",") if tag.strip()] # --------------------------------------------------------------------------- # Snippet extraction helpers # --------------------------------------------------------------------------- def _extract_snippet(content: str, query: str, context_chars: int = SNIPPET_CONTEXT_CHARS) -> str: """Extract a text snippet around the first occurrence of *query*. Returns up to ``context_chars`` characters before and after the match. Falls back to the first 200 characters when the query is not found. Args: content: Full text to search within. query: The search term. context_chars: Number of context characters on each side. Returns: Snippet string, optionally prefixed/suffixed with ``...``. """ lower_content = content.lower() lower_query = query.lower() pos = lower_content.find(lower_query) if pos == -1: return content[:200].strip() start = max(0, pos - context_chars) end = min(len(content), pos + len(query) + context_chars) snippet = content[start:end].strip() if start > 0: snippet = "..." + snippet if end < len(content): snippet = snippet + "..." return snippet def _extract_highlighted_snippet( content: str, query_terms: List[str], context_chars: int = SNIPPET_CONTEXT_CHARS, max_highlights: int = MAX_SNIPPET_HIGHLIGHTS, ) -> str: """Extract a snippet and wrap matching terms in ```` tags. Performs accent-normalized matching so ``"resume"`` highlights ``"résumé"``. Returns at most *max_highlights* highlighted regions to keep snippets concise. Args: content: Full text to search within. query_terms: Normalized search terms. context_chars: Number of context characters on each side. max_highlights: Maximum highlighted regions. Returns: HTML snippet string with ```` highlights. """ if not content or not query_terms: return content[:200].strip() if content else "" norm_content = normalize_text(content) # Find best position — first occurrence of any query term best_pos = len(content) for term in query_terms: pos = norm_content.find(term) if pos != -1 and pos < best_pos: best_pos = pos if best_pos == len(content): # No match found — return beginning of content return _escape_html(content[:200].strip()) start = max(0, best_pos - context_chars) end = min(len(content), best_pos + context_chars + 40) raw_snippet = content[start:end].strip() prefix = "..." if start > 0 else "" suffix = "..." if end < len(content) else "" # Highlight all term occurrences in the snippet highlighted = _highlight_terms(raw_snippet, query_terms, max_highlights) return prefix + highlighted + suffix def _extract_regex_snippet( content: str, pattern_text: str, context_chars: int = SNIPPET_CONTEXT_CHARS, max_highlights: int = MAX_SNIPPET_HIGHLIGHTS, ) -> str: """Extract a snippet and highlight actual regex matches. Unlike ``_extract_highlighted_snippet`` which works with tokenized terms, this function compiles the raw regex pattern and wraps each match in ```` tags. Falls back to the beginning of content if no match. Args: content: Full text to search within. pattern_text: Raw regex pattern string. context_chars: Number of context characters on each side. max_highlights: Maximum highlighted regions. Returns: HTML snippet string with ```` highlights. """ if not content or not pattern_text: return content[:200].strip() if content else "" try: pattern = re.compile(pattern_text, re.IGNORECASE) except re.error: return _escape_html(content[:200].strip()) matches = list(pattern.finditer(content)) if not matches: return _escape_html(content[:200].strip()) # Find the first match position for centering the snippet best_pos = matches[0].start() start = max(0, best_pos - context_chars) end = min(len(content), best_pos + context_chars + 40) snippet = content[start:end].strip() prefix = "..." if start > 0 else "" suffix = "..." if end < len(content) else "" # Highlight regex matches in the snippet (re-compile on snippet for correct positions) snippet_matches = list(pattern.finditer(snippet)) if not snippet_matches: return prefix + _escape_html(snippet) + suffix parts = [] prev = 0 count = 0 for m in snippet_matches: if count >= max_highlights: break if m.start() > prev: parts.append(_escape_html(snippet[prev:m.start()])) parts.append(f"{_escape_html(snippet[m.start():m.end()])}") prev = m.end() count += 1 if prev < len(snippet): parts.append(_escape_html(snippet[prev:])) return prefix + "".join(parts) + suffix def _highlight_terms(text: str, terms: List[str], max_highlights: int) -> str: """Wrap occurrences of *terms* in *text* with ```` tags. Uses accent-normalized comparison so diacritical variants are matched. Escapes HTML in non-highlighted portions to prevent XSS. Args: text: Raw text snippet. terms: Normalized search terms. max_highlights: Cap on highlighted regions. Returns: HTML-safe string with ```` wrapped matches. """ if not terms or not text: return _escape_html(text) norm = normalize_text(text) # Collect (start, end) spans for all term matches spans: List[Tuple[int, int]] = [] for term in terms: idx = 0 while idx < len(norm): pos = norm.find(term, idx) if pos == -1: break spans.append((pos, pos + len(term))) idx = pos + 1 if not spans: return _escape_html(text) # Merge overlapping spans and limit count spans.sort() merged: List[Tuple[int, int]] = [spans[0]] for s, e in spans[1:]: if s <= merged[-1][1]: merged[-1] = (merged[-1][0], max(merged[-1][1], e)) else: merged.append((s, e)) merged = merged[:max_highlights] # Build result with highlights parts: List[str] = [] prev = 0 for s, e in merged: if s > prev: parts.append(_escape_html(text[prev:s])) parts.append(f"{_escape_html(text[s:e])}") prev = e if prev < len(text): parts.append(_escape_html(text[prev:])) return "".join(parts) def _escape_html(text: str) -> str: """Escape HTML special characters.""" return ( text.replace("&", "&") .replace("<", "<") .replace(">", ">") .replace('"', """) ) # --------------------------------------------------------------------------- # Inverted Index for TF-IDF # --------------------------------------------------------------------------- class InvertedIndex: """In-memory inverted index supporting TF-IDF scoring. Built initially via ``rebuild()`` from the global index, then maintained incrementally via ``add_document()`` / ``remove_document()`` hooks from the file watcher and API mutations. Attributes: word_index: ``{token: {doc_key: term_frequency}}`` title_index: ``{token: [doc_key, ...]}`` tag_norm_map: ``{normalized_tag: original_tag}`` tag_prefix_index: ``{prefix: [original_tag, ...]}`` doc_count: Total number of indexed documents. """ def __init__(self) -> None: self.word_index: Dict[str, Dict[str, int]] = defaultdict(dict) self.title_index: Dict[str, List[str]] = defaultdict(list) self.tag_norm_map: Dict[str, str] = {} 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.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: "SortedList" = SortedList() self._ready: bool = False # True after initial build def is_stale(self) -> bool: """Return True if the index has not been built yet.""" return not self._ready def rebuild(self) -> None: """Rebuild inverted index from the global ``index`` dict. Tokenizes titles and content of every file, computes term frequencies, and builds auxiliary indexes for tag and title prefix suggestions. """ logger.info("Building inverted index...") self.word_index = defaultdict(dict) self.title_index = defaultdict(list) self.tag_norm_map = {} 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): self.title_index[token].append(doc_key) # --- Normalized title for prefix suggestions --- norm_title = normalize_text(file_info.get("title", "")) if norm_title: self.title_norm_map[norm_title].append({ "vault": vault_name, "path": file_info["path"], "title": file_info["title"], }) # --- Content tokens (including title for combined scoring) --- content = file_info.get("content", "") full_text = (file_info.get("title", "") + " " + content) tokens = tokenize(full_text) tf: Dict[str, int] = defaultdict(int) for token in tokens: tf[token] += 1 # Also index stemmed forms (French stemming) # Compute stem frequencies in one pass instead of O(S×T) stem_freqs: Dict[str, int] = defaultdict(int) for token, freq in tf.items(): stemmed = stem_token(token) if stemmed != token: # only index stem if different stem_freqs[stemmed] += freq for token, freq in tf.items(): self.word_index[token][doc_key] = freq for stem, freq in stem_freqs.items(): existing = self.word_index[stem].get(doc_key, 0) self.word_index[stem][doc_key] = existing + freq # --- Tag indexes --- for tag in vault_data.get("tags", {}): norm_tag = normalize_text(tag) self.tag_norm_map[norm_tag] = tag # Build prefix entries for each prefix length ≥ MIN_PREFIX_LENGTH for plen in range(MIN_PREFIX_LENGTH, len(norm_tag) + 1): prefix = norm_tag[:plen] if tag not in self.tag_prefix_index[prefix]: self.tag_prefix_index[prefix].append(tag) self._sorted_tokens = SortedList(self.word_index.keys()) self._ready = True logger.info( "Inverted index built: %d documents, %d unique tokens, %d tags", self.doc_count, len(self.word_index), len(self.tag_norm_map), ) def add_document(self, vault_name: str, path: str, file_info: dict): """Add or update a single document incrementally.""" if not self._ready: return doc_key = f"{vault_name}::{path}" old_file_info = self.doc_info.get(doc_key) if old_file_info is not None: self._remove_doc_internals(doc_key, vault_name, old_file_info, skip_sorted_cleanup=True) else: self.doc_count += 1 # Metadata self.doc_info[doc_key] = file_info self.doc_vault[doc_key] = vault_name self.vault_docs[vault_name].add(doc_key) # Tags tags = file_info.get("tags", []) for tag in tags: self.tag_docs[tag.lower()].add(doc_key) norm_tag = normalize_text(tag) if norm_tag not in self.tag_norm_map: self.tag_norm_map[norm_tag] = tag for plen in range(MIN_PREFIX_LENGTH, len(norm_tag) + 1): prefix = norm_tag[:plen] if tag not in self.tag_prefix_index[prefix]: self.tag_prefix_index[prefix].append(tag) # Title tokens title = file_info.get("title", "") title_tokens = tokenize(title) for token in set(title_tokens): if token: self.title_index[token].append(doc_key) # Title norm map norm_title = normalize_text(title) if norm_title: self.title_norm_map[norm_title].append({"vault": vault_name, "path": path, "title": title}) # Word index (content + title TF) content = file_info.get("content", "") full_text = title + " " + content tokens = tokenize(full_text) tf: Dict[str, int] = defaultdict(int) for token in tokens: if token: tf[token] += 1 # Also compute stems in one pass stem_freqs: Dict[str, int] = defaultdict(int) for token, freq in tf.items(): stemmed = stem_token(token) if stemmed != token: stem_freqs[stemmed] += freq for token, freq in tf.items(): if not self.word_index.get(token): self._sorted_tokens.add(token) self.word_index[token][doc_key] = freq for stem, freq in stem_freqs.items(): if not self.word_index.get(stem): self._sorted_tokens.add(stem) existing = self.word_index[stem].get(doc_key, 0) self.word_index[stem][doc_key] = existing + freq def remove_document(self, vault_name: str, path: str): """Remove a single document incrementally.""" if not self._ready: return doc_key = f"{vault_name}::{path}" file_info = self.doc_info.get(doc_key) if file_info is None: return self._remove_doc_internals(doc_key, vault_name, file_info, skip_sorted_cleanup=False) self.doc_count -= 1 def _remove_doc_internals(self, doc_key: str, vault_name: str, file_info: dict, skip_sorted_cleanup: bool = False): """Remove one doc_key from all indexes without adjusting doc_count.""" # Metadata self.doc_info.pop(doc_key, None) self.doc_vault.pop(doc_key, None) if vault_name in self.vault_docs: self.vault_docs[vault_name].discard(doc_key) # Tags (per-document, NOT the global tag_norm_map) for tag in file_info.get("tags", []): td = self.tag_docs.get(tag.lower()) if td: td.discard(doc_key) if not td: del self.tag_docs[tag.lower()] # Title tokens title = file_info.get("title", "") for token in set(tokenize(title)): if not token: continue ti = self.title_index.get(token) if ti: try: ti.remove(doc_key) except ValueError: pass if not ti: del self.title_index[token] # Title norm map norm_title = normalize_text(title) if norm_title and norm_title in self.title_norm_map: self.title_norm_map[norm_title] = [ e for e in self.title_norm_map[norm_title] if not (e["vault"] == vault_name and e["path"] == file_info.get("path")) ] if not self.title_norm_map[norm_title]: del self.title_norm_map[norm_title] # Word index content = file_info.get("content", "") full_text = title + " " + content tokens_to_remove = set(tokenize(full_text)) # Also compute stems to clean up stems_to_clean = stem_tokens(tokens_to_remove) for token in tokens_to_remove: if not token: continue wi = self.word_index.get(token) if wi: wi.pop(doc_key, None) if not wi: del self.word_index[token] if not skip_sorted_cleanup: self._sorted_tokens.discard(token) for stem in stems_to_clean: wi = self.word_index.get(stem) if wi: wi.pop(doc_key, None) if not wi: del self.word_index[stem] if not skip_sorted_cleanup: self._sorted_tokens.discard(stem) def idf(self, term: str) -> float: """Inverse Document Frequency for a term. ``idf(t) = log(N / (1 + df(t)))`` where *df(t)* is the number of documents containing term *t*. Args: term: Normalized term. Returns: IDF score (≥ 0). """ df = len(self.word_index.get(term, {})) if df == 0: return 0.0 return math.log((self.doc_count + 1) / (1 + df)) def tf_idf(self, term: str, doc_key: str) -> float: """TF-IDF score for a term in a document. Uses raw term frequency (no log normalization) × IDF. Args: term: Normalized term. doc_key: ``"vault::path"`` document key. Returns: TF-IDF score. """ tf = self.word_index.get(term, {}).get(doc_key, 0) if tf == 0: 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 = self._sorted_tokens.bisect_left(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() def _on_index_change_hook(action: str, vault_name: str, path: str, file_info: dict): """Callback registered with indexer for incremental inverted index updates.""" inv = _inverted_index try: if action == 'add' and file_info: inv.add_document(vault_name, path, file_info) elif action == 'remove': inv.remove_document(vault_name, path) except Exception as e: logger.warning(f"Inverted index incremental update failed ({action} {vault_name}/{path}): {e}") # Register the hook with indexer (indexer is already imported at top of file) _indexer.set_index_change_hook(_on_index_change_hook) def init_inverted_index(): """Force initial inverted index build. Called after build_index completes on startup.""" if any(vdata.get("files") for vdata in index.values()): _inverted_index.rebuild() logger.info("Inverted index initialized.") def get_inverted_index() -> InvertedIndex: """Return the singleton inverted index. The inverted index is built once at startup in a background thread. If called before the build completes, returns an empty index (searches will return no results until the build finishes). No automatic rebuild — the startup sequence handles the initial build. Incremental updates are applied via ``update_inverted_index()``. """ return _inverted_index # --------------------------------------------------------------------------- # Backward-compatible search (unchanged API) # --------------------------------------------------------------------------- def search( query: str, vault_filter: str = "all", tag_filter: Optional[str] = None, limit: int = DEFAULT_SEARCH_LIMIT, ) -> List[Dict[str, Any]]: """Full-text search across indexed vaults with relevance scoring. Scoring heuristics (when a text query is provided): - **+20** exact title match (case-insensitive) - **+10** partial title match - **+5** query found in file path - **+3** query matches a tag name - **+1 per occurrence** in content (capped at 10) When only tag filters are active, all matching files receive score 1. Results are sorted descending by score and capped at *limit*. Uses the in-memory cached content from the index — **no disk I/O**. Args: query: Free-text search string. vault_filter: Vault name or ``"all"``. tag_filter: Comma-separated tag names to require. limit: Maximum number of results to return. Returns: List of result dicts sorted by descending relevance score. """ query = query.strip() if query else "" has_query = len(query) > 0 selected_tags = _normalize_tag_filter(tag_filter) if not has_query and not selected_tags: return [] query_lower = query.lower() results: List[Dict[str, Any]] = [] for vault_name, vault_data in index.items(): if vault_filter != "all" and vault_name != vault_filter: continue for file_info in vault_data["files"]: # Tag filter: all selected tags must be present if selected_tags and not all(tag in file_info["tags"] for tag in selected_tags): continue score = 0 snippet = file_info.get("content_preview", "") if has_query: title_lower = file_info["title"].lower() # Exact title match (highest weight) if query_lower == title_lower: score += 20 # Partial title match elif query_lower in title_lower: score += 10 # Path match (folder/filename relevance) if query_lower in file_info["path"].lower(): score += 5 # Tag name match for tag in file_info.get("tags", []): if query_lower in tag.lower(): score += 3 break # count once per file # Content match — use cached content (no disk I/O) content = file_info.get("content", "") content_lower = content.lower() if query_lower in content_lower: # Frequency-based scoring, capped to avoid over-weighting occurrences = content_lower.count(query_lower) score += min(occurrences, 10) snippet = _extract_snippet(content, query) else: # Tag-only filter: all matching files get score 1 score = 1 if score > 0: results.append({ "vault": vault_name, "path": file_info["path"], "title": file_info["title"], "tags": file_info["tags"], "score": score, "snippet": snippet, "modified": file_info["modified"], }) results.sort(key=lambda x: -x["score"]) return results[:limit] # --------------------------------------------------------------------------- # Advanced search with TF-IDF scoring # --------------------------------------------------------------------------- def _parse_advanced_query(raw_query: str) -> Dict[str, Any]: """Parse an advanced query string into structured filters and free text. Supported operators: - ``tag:`` or ``#`` — tag filter - ``vault:`` — vault filter - ``title:`` — title filter - ``path:`` — path filter - ``ext:`` — file extension filter - Remaining tokens are treated as free-text search terms. Args: raw_query: Raw query string from the user. Returns: Dict with keys ``tags``, ``vault``, ``title``, ``path``, ``ext``, ``terms``. """ parsed: Dict[str, Any] = { "tags": [], "vault": None, "title": None, "path": None, "ext": None, "terms": [], } if not raw_query: return parsed # Use shlex-like tokenizing but handle quotes manually tokens = _split_query_tokens(raw_query) for token in tokens: lower = token.lower() if lower.startswith("tag:"): tag_val = token[4:].strip().lstrip("#") if tag_val: parsed["tags"].append(tag_val) elif lower.startswith("#") and len(token) > 1: parsed["tags"].append(token[1:]) elif lower.startswith("vault:"): parsed["vault"] = token[6:].strip() elif lower.startswith("title:"): parsed["title"] = token[6:].strip() elif lower.startswith("path:"): parsed["path"] = token[5:].strip() elif lower.startswith("ext:"): parsed["ext"] = token[4:].strip().lstrip(".").lower() else: parsed["terms"].append(token) return parsed def _split_query_tokens(raw: str) -> List[str]: """Split a query string respecting quoted phrases. ``tag:"my tag" hello world`` → ``['tag:my tag', 'hello', 'world']`` Args: raw: Raw query string. Returns: List of token strings. """ tokens: List[str] = [] i = 0 n = len(raw) while i < n: # Skip whitespace while i < n and raw[i] == " ": i += 1 if i >= n: break # Check for operator with quoted value, e.g., tag:"foo bar" if i < n and raw[i] != '"': # Read until space or quote j = i while j < n and raw[j] != " ": if raw[j] == '"': # Read quoted portion j += 1 while j < n and raw[j] != '"': j += 1 if j < n: j += 1 # skip closing quote else: j += 1 token = raw[i:j].replace('"', "") tokens.append(token) i = j else: # Quoted token i += 1 # skip opening quote j = i while j < n and raw[j] != '"': j += 1 tokens.append(raw[i:j]) i = j + 1 # skip closing quote return tokens def _passes_search_filters( file_info: dict, query_terms: List[str], query_terms_raw: List[str], raw_query: str, case_sensitive: bool, whole_word: bool, regex: bool, include_paths: Optional[str], exclude_paths: Optional[str], ) -> bool: """Post-filter a candidate by case-sensitive, whole-word, regex, and path filters.""" title = file_info.get("title", "") content = file_info.get("content", "") path = file_info.get("path", "") search_text = f"{title} {content}" search_text_norm = normalize_text(search_text) # --- Regex mode --- if regex and raw_query: try: flags = 0 if case_sensitive else re.IGNORECASE if whole_word: pattern = re.compile(rf"\b{raw_query}\b", flags) else: pattern = re.compile(raw_query, flags) if not pattern.search(search_text): return False except re.error: return False return _passes_path_filters(path, include_paths, exclude_paths) # --- Case-sensitive (use raw, non-normalized terms) --- if case_sensitive and query_terms_raw: for term in query_terms_raw: if term not in search_text: return False # --- Whole-word (use normalized text + normalized terms) --- if whole_word and query_terms: for term in query_terms: pattern = re.compile(rf"\b{re.escape(term)}\b", re.IGNORECASE) if not pattern.search(search_text_norm): return False # --- Path filters (glob-like) --- return _passes_path_filters(path, include_paths, exclude_paths) def _passes_path_filters(path: str, include: Optional[str], exclude: Optional[str]) -> bool: """Check if a file path passes include/exclude glob patterns.""" import fnmatch if include: patterns = [p.strip() for p in include.split(",") if p.strip()] if patterns and not any(fnmatch.fnmatch(path, p) for p in patterns): return False if exclude: patterns = [p.strip() for p in exclude.split(",") if p.strip()] if patterns and any(fnmatch.fnmatch(path, p) for p in patterns): return False return True def advanced_search( query: str, vault_filter: str = "all", tag_filter: Optional[str] = None, limit: int = ADVANCED_SEARCH_DEFAULT_LIMIT, offset: int = 0, sort_by: str = "relevance", case_sensitive: bool = False, whole_word: bool = False, regex: bool = False, include_paths: Optional[str] = None, exclude_paths: Optional[str] = None, ) -> 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:``, ``ext:``), falls back remaining tokens to TF-IDF scored free-text search using the inverted index. Results include highlighted snippets with ```` tags and faceted counts for tags and vaults. Args: query: Raw query string (may include operators). vault_filter: Vault name or ``"all"`` (overridden by ``vault:`` op). tag_filter: Comma-separated tag names (merged with ``tag:`` ops). limit: Max results per page. offset: Pagination offset. sort_by: ``"relevance"`` or ``"modified"``. Returns: Dict with ``results``, ``total``, ``offset``, ``limit``, ``facets``, ``query_time_ms``. """ t0 = time.monotonic() query = query.strip() if query else "" parsed = _parse_advanced_query(query) # Merge explicit tag_filter with parsed tag: operators all_tags = list(parsed["tags"]) extra_tags = _normalize_tag_filter(tag_filter) for t in extra_tags: if t not in all_tags: all_tags.append(t) # Vault filter — parsed vault: overrides parameter effective_vault = parsed["vault"] or vault_filter # Tokenize free-text terms (splits on non-word chars like dots) # "192.168" → ["192", "168"] for proper inverted index matching query_terms_raw = [t for t in parsed["terms"] if t.strip()] query_terms = [] for t in query_terms_raw: query_terms.extend(tokenize(t)) # Also add stemmed forms for French stemming support query_stems = stem_tokens(query_terms) query_terms.extend(stem for stem in query_stems if stem not in query_terms) has_terms = len(query_terms) > 0 if not has_terms and not all_tags and not parsed["title"] and not parsed["path"] and not parsed["ext"]: 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", "")) } if parsed["ext"]: ext_filter = parsed["ext"] candidates = { dk for dk in candidates if ( inv.doc_info[dk].get("path", "").rsplit("/", 1)[-1].lower() == ext_filter or inv.doc_info[dk].get("path", "").rsplit("/", 1)[-1].lower().endswith(f".{ext_filter}") ) } # ------------------------------------------------------------------ # 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) # 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] score = 0.0 if has_terms: 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 # 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/ext): score = 1 score = 1.0 if score > 0: # --- Post-filters: case-sensitive, whole-word, regex, path filters --- if not _passes_search_filters( file_info, query_terms, query_terms_raw, " ".join(query_terms_raw) if query_terms_raw else query, case_sensitive, whole_word, regex, include_paths, exclude_paths ): continue # Build highlighted snippet content = file_info.get("content", "") if has_terms: if regex: raw_regex = " ".join(query_terms_raw) if query_terms_raw else "" snippet = _extract_regex_snippet(content, raw_regex) else: 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", ""), "extension": file_info.get("extension", file_info.get("path", "").rsplit(".", 1)[-1] if "." in file_info.get("path", "") else ""), } 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": scored_results.sort(key=lambda x: x[1].get("modified", ""), reverse=True) else: scored_results.sort(key=lambda x: -x[0]) 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], "total": total, "offset": offset, "limit": limit, "facets": { "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, } # --------------------------------------------------------------------------- # Suggestion helpers # --------------------------------------------------------------------------- def suggest_titles( prefix: str, vault_filter: str = "all", limit: int = SUGGEST_LIMIT, ) -> List[Dict[str, str]]: """Suggest file titles matching a prefix (accent-insensitive). Args: prefix: User-typed prefix string. vault_filter: Vault name or ``"all"``. limit: Maximum suggestions. Returns: List of ``{"vault", "path", "title"}`` dicts. """ if not prefix or len(prefix) < MIN_PREFIX_LENGTH: return [] inv = get_inverted_index() norm_prefix = normalize_text(prefix) results: List[Dict[str, str]] = [] seen: set = set() for norm_title, entries in inv.title_norm_map.items(): if norm_prefix in norm_title: for entry in entries: if vault_filter != "all" and entry["vault"] != vault_filter: continue key = f"{entry['vault']}::{entry['path']}" if key not in seen: seen.add(key) results.append(entry) if len(results) >= limit: return results return results def suggest_tags( prefix: str, vault_filter: str = "all", limit: int = SUGGEST_LIMIT, ) -> List[Dict[str, Any]]: """Suggest tags matching a prefix (accent-insensitive). Args: prefix: User-typed prefix (with or without leading ``#``). vault_filter: Vault name or ``"all"``. limit: Maximum suggestions. Returns: List of ``{"tag", "count"}`` dicts sorted by descending count. """ prefix = prefix.lstrip("#").strip() if not prefix or len(prefix) < MIN_PREFIX_LENGTH: return [] norm_prefix = normalize_text(prefix) all_tag_counts = get_all_tags(vault_filter) matches: List[Dict[str, Any]] = [] for tag, count in all_tag_counts.items(): norm_tag = normalize_text(tag) if norm_prefix in norm_tag: matches.append({"tag": tag, "count": count}) if len(matches) >= limit: break return matches # --------------------------------------------------------------------------- # Backward-compatible tag aggregation (unchanged API) # --------------------------------------------------------------------------- def get_all_tags(vault_filter: Optional[str] = None) -> Dict[str, int]: """Aggregate tag counts across vaults, sorted by descending count. Args: vault_filter: Optional vault name to restrict to a single vault. Returns: Dict mapping tag names to their total occurrence count. """ merged: Dict[str, int] = {} for vault_name, vault_data in index.items(): if vault_filter and vault_filter != "all" and vault_name != vault_filter: continue for tag, count in vault_data.get("tags", {}).items(): merged[tag] = merged.get(tag, 0) + count return dict(sorted(merged.items(), key=lambda x: -x[1]))