bruno
1673531b43
- ruff: 602→0 errors (428 auto-fixed, pyproject.toml ignores for FastAPI patterns) - bandit: skip B310 (urllib for vault file access is intentional) - Fixed SIM118 (dict.keys()→dict), PERF102, SIM113, SIM117 - Created pyproject.toml with ruff + bandit config - 285 tests still pass
72 lines
2.0 KiB
Python
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 logging
|
|
import shutil
|
|
import time
|
|
from pathlib import Path
|
|
from typing import 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
|