Add search toggles, path filters, and find/replace functionality

This commit is contained in:
2026-05-26 13:35:38 -04:00
parent 6c282ac77f
commit 8fdcdaf412
5 changed files with 361 additions and 16 deletions
+98 -12
View File
@@ -1873,6 +1873,11 @@ async def api_advanced_search(
limit: int = Query(50, ge=1, le=200, description="Results per page"),
offset: int = Query(0, ge=0, description="Pagination offset"),
sort: str = Query("relevance", description="Sort by 'relevance' or 'modified'"),
case_sensitive: bool = Query(False, description="Match case"),
whole_word: bool = Query(False, description="Match whole words only"),
regex: bool = Query(False, description="Treat query as regex"),
include_paths: Optional[str] = Query(None, description="Comma-separated glob patterns to include"),
exclude_paths: Optional[str] = Query(None, description="Comma-separated glob patterns to exclude"),
current_user=Depends(require_auth),
):
"""Advanced full-text search with TF-IDF scoring, facets, and pagination.
@@ -1884,28 +1889,109 @@ async def api_advanced_search(
- ``path:<text>`` — filter by path substring
- ``ext:<type>`` — filter by file extension
- Remaining text is scored using TF-IDF with accent normalization.
- Toggles: case_sensitive, whole_word, regex
- Path filters: include_paths, exclude_paths (glob patterns)
Results include ``<mark>``-highlighted snippets and faceted tag/vault counts.
Args:
q: Query string with optional operators.
vault: Vault name or ``"all"``.
tag: Extra comma-separated tag names to require.
limit: Max results per page (1–200).
offset: Pagination offset.
sort: ``"relevance"`` (TF-IDF) or ``"modified"`` (date).
Returns:
``AdvancedSearchResponse`` with scored results, facets, and pagination info.
"""
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),
limit=limit, offset=offset, sort_by=sort,
case_sensitive=case_sensitive, whole_word=whole_word, regex=regex,
include_paths=include_paths, exclude_paths=exclude_paths),
)
@app.post("/api/search/replace")
async def api_search_replace(
body: dict = Body(...),
current_user=Depends(require_auth),
):
"""Find and replace across vault files."""
import re as re_mod
query = body.get("query", "")
replacement = body.get("replacement", "")
vault_filter = body.get("vault", "all")
case_sensitive = body.get("case_sensitive", False)
whole_word = body.get("whole_word", False)
regex_mode = body.get("regex", False)
include_paths = body.get("include_paths")
exclude_paths = body.get("exclude_paths")
replace_all = body.get("replace_all", False)
dry_run = body.get("dry_run", not replace_all)
if not query:
raise HTTPException(400, "Query is required")
search_results = advanced_search(
query, vault_filter=vault_filter,
case_sensitive=case_sensitive, whole_word=whole_word,
regex=regex_mode, include_paths=include_paths, exclude_paths=exclude_paths,
limit=500, sort_by="relevance",
)
if not search_results["results"]:
return {"matches": [], "total_matches": 0}
flags = 0 if case_sensitive else re_mod.IGNORECASE
if regex_mode:
pattern = re_mod.compile(query, flags)
elif whole_word:
pattern = re_mod.compile(rf"\b{re_mod.escape(query)}\b", flags)
else:
pattern = re_mod.compile(re_mod.escape(query), flags)
matches = []
total_replacements = 0
for result in search_results["results"]:
if not check_vault_access(result["vault"], current_user):
continue
vault_data = get_vault_data(result["vault"])
if not vault_data:
continue
file_path = _resolve_safe_path(Path(vault_data["path"]), result["path"])
if not file_path.exists():
continue
try:
original = file_path.read_text(encoding="utf-8", errors="replace")
except Exception:
continue
occurrences = list(pattern.finditer(original))
if not occurrences:
continue
if dry_run:
previews = []
for m in occurrences[:3]:
start = max(0, m.start() - 40)
end = min(len(original), m.end() + 40)
previews.append(f"...{original[start:end]}...")
matches.append({
"vault": result["vault"], "path": result["path"],
"title": result["title"], "match_count": len(occurrences),
"preview": previews,
})
total_replacements += len(occurrences)
else:
new_content, count = pattern.subn(replacement, original)
if count > 0:
_backup_file(file_path, result["vault"], result["path"])
file_path.write_text(new_content, encoding="utf-8")
log_file_save(current_user["username"], result["vault"], result["path"], len(new_content))
matches.append({
"vault": result["vault"], "path": result["path"],
"title": result["title"], "replacements": count,
})
total_replacements += count
await update_single_file(result["vault"], str(file_path))
if dry_run:
return {"matches": matches, "total_matches": total_replacements, "dry_run": True}
return {"replaced": matches, "total_replacements": total_replacements}
@app.get("/api/suggest", response_model=SuggestResponse)
async def api_suggest(
q: str = Query("", description="Prefix to search for in file titles"),
+73
View File
@@ -759,6 +759,67 @@ def _split_query_tokens(raw: str) -> List[str]:
return tokens
def _passes_search_filters(
file_info: dict,
query_terms: 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}"
# --- 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 # invalid regex
return _passes_path_filters(path, include_paths, exclude_paths)
# --- Case-sensitive ---
if case_sensitive and query_terms:
for term in query_terms:
if term not in search_text:
return False
# --- Whole-word ---
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):
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",
@@ -766,6 +827,11 @@ def advanced_search(
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.
@@ -923,6 +989,13 @@ def advanced_search(
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,
case_sensitive, whole_word, regex, include_paths, exclude_paths
):
continue
# Build highlighted snippet
content = file_info.get("content", "")
if has_terms: