113 lines
3.6 KiB
Python
113 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import threading
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .config import Config
|
|
from .models import UpdateManifest
|
|
|
|
DEFAULT_SETTINGS: dict[str, Any] = {
|
|
"app_name": "NotifyPulse",
|
|
"active_usecase": "",
|
|
"paused": False,
|
|
"hotkey": "F13",
|
|
"startup_toast": True,
|
|
"notify_sound": True,
|
|
"auto_open_browser": True,
|
|
"minimize_to_tray": True,
|
|
"run_on_startup": False,
|
|
"confirm_delete": True,
|
|
"entry_display_mode": "percent",
|
|
"notification_duration": 5,
|
|
"overlay_opacity": 0.4,
|
|
"overlay_duration": 6,
|
|
"overlay_stretch": False,
|
|
"overlay_monitor": 0,
|
|
"wallpaper_fit": "fill",
|
|
"pwa_bg_blur": 18,
|
|
"pwa_bg_opacity": 0.72,
|
|
"log_max_entries": 100,
|
|
"webui_refresh_ms": 1000,
|
|
}
|
|
|
|
DEFAULT_USECASES: list[dict[str, Any]] = []
|
|
|
|
DEFAULT_UPDATE_MANIFEST: dict[str, Any] = {
|
|
"windows": UpdateManifest().model_dump(),
|
|
}
|
|
|
|
|
|
class JsonStore:
|
|
def __init__(self, data_dir: Path):
|
|
self._data_dir = data_dir
|
|
self._lock = threading.RLock()
|
|
self._settings_file = data_dir / "settings.json"
|
|
self._usecases_file = data_dir / "usecases.json"
|
|
self._update_file = data_dir / "update_manifest.json"
|
|
self._init_defaults()
|
|
|
|
def _init_defaults(self) -> None:
|
|
with self._lock:
|
|
self._ensure_file(self._settings_file, DEFAULT_SETTINGS)
|
|
self._ensure_file(self._usecases_file, DEFAULT_USECASES)
|
|
self._ensure_file(self._update_file, DEFAULT_UPDATE_MANIFEST)
|
|
|
|
@staticmethod
|
|
def _write_json(path: Path, payload: Any) -> None:
|
|
path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
|
|
|
|
def _ensure_file(self, path: Path, payload: Any) -> None:
|
|
if not path.exists():
|
|
self._write_json(path, payload)
|
|
|
|
@staticmethod
|
|
def _read_json(path: Path, fallback: Any) -> Any:
|
|
try:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
return fallback
|
|
|
|
def get_settings(self) -> dict[str, Any]:
|
|
with self._lock:
|
|
raw = self._read_json(self._settings_file, {})
|
|
return {**DEFAULT_SETTINGS, **raw}
|
|
|
|
def update_settings(self, patch: dict[str, Any]) -> dict[str, Any]:
|
|
with self._lock:
|
|
current = self.get_settings()
|
|
current.update(patch)
|
|
self._write_json(self._settings_file, current)
|
|
return current
|
|
|
|
def get_usecases(self) -> list[dict[str, Any]]:
|
|
with self._lock:
|
|
raw = self._read_json(self._usecases_file, DEFAULT_USECASES)
|
|
return raw if isinstance(raw, list) else []
|
|
|
|
def set_usecases(self, payload: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
with self._lock:
|
|
self._write_json(self._usecases_file, payload)
|
|
return payload
|
|
|
|
def get_update_manifest(self, platform: str) -> UpdateManifest:
|
|
with self._lock:
|
|
raw = self._read_json(self._update_file, DEFAULT_UPDATE_MANIFEST)
|
|
platform_raw = raw.get(platform) if isinstance(raw, dict) else None
|
|
if not isinstance(platform_raw, dict):
|
|
return UpdateManifest()
|
|
return UpdateManifest.model_validate(platform_raw)
|
|
|
|
def set_update_manifest(self, platform: str, manifest: UpdateManifest) -> None:
|
|
with self._lock:
|
|
raw = self._read_json(self._update_file, DEFAULT_UPDATE_MANIFEST)
|
|
if not isinstance(raw, dict):
|
|
raw = {}
|
|
raw[platform] = manifest.model_dump()
|
|
self._write_json(self._update_file, raw)
|
|
|
|
|
|
Config.ensure_dirs()
|
|
store = JsonStore(Config.data_dir)
|