Add saved searches with CRUD API and UI sidebar
Add extension field to search results and display it Add active filter badges and save button to search header
This commit is contained in:
@@ -0,0 +1,71 @@
|
||||
"""
|
||||
Saved search definitions — persisted search queries with their active filters.
|
||||
|
||||
Stored in data/saved_searches.json per user.
|
||||
"""
|
||||
|
||||
import json
|
||||
import time
|
||||
import logging
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
from typing import List, Dict, Any, Optional
|
||||
|
||||
logger = logging.getLogger("obsigate.saved_searches")
|
||||
|
||||
DATA_DIR = Path("data")
|
||||
|
||||
|
||||
def _get_file(username: str) -> Path:
|
||||
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
||||
return DATA_DIR / f"{username}_saved_searches.json"
|
||||
|
||||
|
||||
def _read(file: Path) -> List[Dict[str, Any]]:
|
||||
if not file.exists():
|
||||
return []
|
||||
try:
|
||||
return json.loads(file.read_text(encoding="utf-8"))
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
|
||||
def _write(file: Path, data: List[Dict[str, Any]]):
|
||||
tmp = file.with_suffix(".tmp")
|
||||
tmp.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
|
||||
shutil.move(str(tmp), str(file))
|
||||
|
||||
|
||||
def get_saved(username: str) -> List[Dict[str, Any]]:
|
||||
return _read(_get_file(username))
|
||||
|
||||
|
||||
def save_search(username: str, definition: dict) -> dict:
|
||||
"""Save a search definition. Returns the saved entry."""
|
||||
searches = _read(_get_file(username))
|
||||
|
||||
entry = {
|
||||
"id": str(int(time.time() * 1000)),
|
||||
"query": definition.get("query", ""),
|
||||
"vault": definition.get("vault", "all"),
|
||||
"case_sensitive": definition.get("case_sensitive", False),
|
||||
"whole_word": definition.get("whole_word", False),
|
||||
"regex": definition.get("regex", False),
|
||||
"include_paths": definition.get("include_paths", ""),
|
||||
"exclude_paths": definition.get("exclude_paths", ""),
|
||||
"created_at": time.time(),
|
||||
}
|
||||
searches.insert(0, entry)
|
||||
# Keep last 50
|
||||
searches = searches[:50]
|
||||
_write(_get_file(username), searches)
|
||||
return entry
|
||||
|
||||
|
||||
def delete_saved(username: str, search_id: str) -> bool:
|
||||
searches = _read(_get_file(username))
|
||||
new_list = [s for s in searches if s["id"] != search_id]
|
||||
if len(new_list) == len(searches):
|
||||
return False
|
||||
_write(_get_file(username), new_list)
|
||||
return True
|
||||
Reference in New Issue
Block a user