209 lines
6.8 KiB
Python
209 lines
6.8 KiB
Python
"""Persistent JSON store for Notification Store."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import uuid
|
|
from copy import deepcopy
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from homeassistant.core import HomeAssistant
|
|
|
|
from .const import DEFAULT_STORE_PATH, LEVEL_INFO, LEVELS
|
|
|
|
|
|
def _now_iso() -> str:
|
|
"""Return a local ISO timestamp without microseconds."""
|
|
return datetime.now().isoformat(timespec="seconds")
|
|
|
|
|
|
def _empty_data() -> dict[str, list[dict[str, Any]]]:
|
|
"""Return an empty store structure."""
|
|
return {"notifications": []}
|
|
|
|
|
|
def _normalize_notification(item: Any) -> dict[str, Any] | None:
|
|
"""Normalize a notification item from disk."""
|
|
if not isinstance(item, dict):
|
|
return None
|
|
|
|
notification = dict(item)
|
|
notification.setdefault("id", str(uuid.uuid4()))
|
|
notification.setdefault("title", "")
|
|
notification.setdefault("message", "")
|
|
notification.setdefault("level", LEVEL_INFO)
|
|
notification.setdefault("tag", "")
|
|
notification.setdefault("source", "")
|
|
notification.setdefault("icon", "")
|
|
notification.setdefault("status", "unread")
|
|
notification.setdefault("created", _now_iso())
|
|
|
|
if notification.get("level") not in LEVELS:
|
|
notification["level"] = LEVEL_INFO
|
|
|
|
if notification.get("status") not in ["unread", "read"]:
|
|
notification["status"] = "unread"
|
|
|
|
return notification
|
|
|
|
|
|
def _normalize_data(data: Any) -> dict[str, list[dict[str, Any]]]:
|
|
"""Normalize store data from disk."""
|
|
if not isinstance(data, dict):
|
|
return _empty_data()
|
|
|
|
notifications = data.get("notifications", [])
|
|
if not isinstance(notifications, list):
|
|
notifications = []
|
|
|
|
normalized = []
|
|
for item in notifications:
|
|
notification = _normalize_notification(item)
|
|
if notification is not None:
|
|
normalized.append(notification)
|
|
|
|
return {"notifications": normalized}
|
|
|
|
|
|
class NotificationStore:
|
|
"""Read and write notification data from a JSON file."""
|
|
|
|
def __init__(self, hass: HomeAssistant, store_path: str | None = None) -> None:
|
|
"""Initialize the store."""
|
|
self.hass = hass
|
|
self.path = self._resolve_path(store_path or DEFAULT_STORE_PATH)
|
|
|
|
def _resolve_path(self, store_path: str) -> str:
|
|
"""Resolve a relative Home Assistant config path or absolute path."""
|
|
path = Path(store_path).expanduser()
|
|
if path.is_absolute():
|
|
return str(path)
|
|
return self.hass.config.path(str(path))
|
|
|
|
def _load_sync(self) -> dict[str, list[dict[str, Any]]]:
|
|
"""Load the store synchronously."""
|
|
if not os.path.exists(self.path):
|
|
return _empty_data()
|
|
|
|
try:
|
|
with open(self.path, "r", encoding="utf-8") as file:
|
|
return _normalize_data(json.load(file))
|
|
except Exception:
|
|
return _empty_data()
|
|
|
|
def _save_sync(self, data: dict[str, Any]) -> None:
|
|
"""Save the store synchronously."""
|
|
os.makedirs(os.path.dirname(self.path), exist_ok=True)
|
|
normalized = _normalize_data(data)
|
|
with open(self.path, "w", encoding="utf-8") as file:
|
|
json.dump(normalized, file, ensure_ascii=False, indent=2)
|
|
|
|
async def async_load(self) -> dict[str, list[dict[str, Any]]]:
|
|
"""Load the store asynchronously."""
|
|
data = await self.hass.async_add_executor_job(self._load_sync)
|
|
return deepcopy(data)
|
|
|
|
async def async_save(self, data: dict[str, Any]) -> None:
|
|
"""Save the store asynchronously."""
|
|
await self.hass.async_add_executor_job(self._save_sync, data)
|
|
|
|
async def async_add_notification(
|
|
self,
|
|
title: str,
|
|
message: str,
|
|
level: str = LEVEL_INFO,
|
|
tag: str = "",
|
|
source: str = "",
|
|
icon: str = "",
|
|
) -> dict[str, Any]:
|
|
"""Add a notification and return it."""
|
|
data = await self.async_load()
|
|
notification = {
|
|
"id": str(uuid.uuid4()),
|
|
"title": title,
|
|
"message": message,
|
|
"level": level if level in LEVELS else LEVEL_INFO,
|
|
"tag": tag,
|
|
"source": source,
|
|
"icon": icon,
|
|
"status": "unread",
|
|
"created": _now_iso(),
|
|
}
|
|
data.setdefault("notifications", []).insert(0, notification)
|
|
await self.async_save(data)
|
|
return notification
|
|
|
|
async def async_mark_all_read(self) -> int:
|
|
"""Mark all notifications as read and return changed count."""
|
|
data = await self.async_load()
|
|
changed = 0
|
|
for item in data.get("notifications", []):
|
|
if item.get("status", "unread") == "unread":
|
|
item["status"] = "read"
|
|
item["read_at"] = _now_iso()
|
|
changed += 1
|
|
await self.async_save(data)
|
|
return changed
|
|
|
|
async def async_mark_one_read(self, notification_id: str) -> bool:
|
|
"""Mark one notification as read."""
|
|
data = await self.async_load()
|
|
found = False
|
|
for item in data.get("notifications", []):
|
|
if item.get("id") == notification_id:
|
|
item["status"] = "read"
|
|
item["read_at"] = _now_iso()
|
|
found = True
|
|
break
|
|
await self.async_save(data)
|
|
return found
|
|
|
|
async def async_mark_index_read(self, index: int) -> str | None:
|
|
"""Mark one unread notification by unread index as read."""
|
|
data = await self.async_load()
|
|
unread = [
|
|
item
|
|
for item in data.get("notifications", [])
|
|
if item.get("status", "unread") == "unread"
|
|
]
|
|
|
|
if index < 0 or index >= len(unread):
|
|
return None
|
|
|
|
notification_id = unread[index].get("id")
|
|
if not notification_id:
|
|
return None
|
|
|
|
for item in data.get("notifications", []):
|
|
if item.get("id") == notification_id:
|
|
item["status"] = "read"
|
|
item["read_at"] = _now_iso()
|
|
break
|
|
|
|
await self.async_save(data)
|
|
return notification_id
|
|
|
|
async def async_clear_read(self) -> int:
|
|
"""Delete read notifications and return removed count."""
|
|
data = await self.async_load()
|
|
notifications = data.get("notifications", [])
|
|
kept = [
|
|
item
|
|
for item in notifications
|
|
if item.get("status", "unread") == "unread"
|
|
]
|
|
removed = len(notifications) - len(kept)
|
|
data["notifications"] = kept
|
|
await self.async_save(data)
|
|
return removed
|
|
|
|
async def async_clear_all(self) -> int:
|
|
"""Delete all notifications and return removed count."""
|
|
data = await self.async_load()
|
|
removed = len(data.get("notifications", []))
|
|
await self.async_save(_empty_data())
|
|
return removed
|