ObsiGate/backend/saved_searches.py
Bruno Charest 6fc43e2485
Some checks failed
CI / lint (push) Failing after 11s
CI / test (push) Has been skipped
CI / build (push) Has been skipped
CI / security (push) Successful in 7s
fix: ruff lint errors + bandit false positives + pip-audit non-blocking
2026-05-28 12:41:31 -04:00

72 lines
2.0 KiB
Python

"""
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
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