187 lines
7 KiB
Python
187 lines
7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING, Any, cast
|
|
|
|
from app.features.core.migration import Migration as FeatureMigration
|
|
from app.features.notifications.schemas import NotificationEvents
|
|
from app.features.presets.service import Presets
|
|
from app.library.config import Config
|
|
from app.library.log import get_logger
|
|
|
|
if TYPE_CHECKING:
|
|
from app.features.notifications.repository import NotificationsRepository
|
|
|
|
LOG = get_logger()
|
|
|
|
|
|
class Migration(FeatureMigration):
|
|
name: str = "notifications"
|
|
|
|
def __init__(self, repo: NotificationsRepository, config: Config | None = None):
|
|
self._config: Config = config or Config.get_instance()
|
|
super().__init__(config=self._config)
|
|
self._repo: NotificationsRepository = repo
|
|
self._source_file: Path = Path(self._config.config_path) / "notifications.json"
|
|
|
|
async def should_run(self) -> bool:
|
|
return self._source_file.exists()
|
|
|
|
async def migrate(self) -> None:
|
|
if await self._repo.count() > 0:
|
|
LOG.warning("Notification targets already exist in the database; skipping migration.")
|
|
await self._move_file(self._source_file)
|
|
return
|
|
|
|
try:
|
|
items: list[dict] | None = json.loads(self._source_file.read_text())
|
|
except Exception as exc:
|
|
LOG.exception(
|
|
"Failed to read notifications migration file '%s'. Ignoring.",
|
|
self._source_file,
|
|
extra={"file_path": str(self._source_file), "exception_type": type(exc).__name__},
|
|
)
|
|
await self._move_file(self._source_file)
|
|
return
|
|
|
|
if items is None:
|
|
LOG.warning("No notification targets found in %s; skipping migration.", self._source_file)
|
|
await self._move_file(self._source_file)
|
|
return
|
|
|
|
inserted = 0
|
|
seen_names: dict[str, int] = {}
|
|
for index, item in enumerate(items):
|
|
normalized = self._normalize(item, index, seen_names)
|
|
if not normalized:
|
|
continue
|
|
try:
|
|
await self._repo.create(normalized)
|
|
inserted += 1
|
|
except Exception as exc:
|
|
LOG.exception(
|
|
"Failed to insert notification target '%s'.",
|
|
normalized.get("name"),
|
|
extra={"target_name": normalized.get("name"), "exception_type": type(exc).__name__},
|
|
)
|
|
|
|
LOG.info("Migrated %s notification target(s) from %s.", inserted, self._source_file)
|
|
await self._move_file(self._source_file)
|
|
|
|
def _normalize(self, item: Any, index: int, seen_names: dict[str, int]) -> dict[str, Any] | None:
|
|
if not isinstance(item, dict):
|
|
LOG.warning("Skipping notification at index %s due to invalid type.", index)
|
|
return None
|
|
|
|
assert isinstance(item, dict)
|
|
|
|
name: str | None = item.get("name")
|
|
if not name or not isinstance(name, str):
|
|
LOG.warning("Skipping notification at index %s due to missing name.", index)
|
|
return None
|
|
|
|
normalized_name = name.strip()
|
|
if not normalized_name:
|
|
LOG.warning("Skipping notification at index %s due to empty name.", index)
|
|
return None
|
|
|
|
name = self._unique_name(normalized_name, seen_names)
|
|
request = item.get("request") if isinstance(item.get("request"), dict) else {}
|
|
request = cast("dict[str, Any]", request)
|
|
url: str | None = request.get("url")
|
|
if not url or not isinstance(url, str):
|
|
LOG.warning("Skipping notification '%s' due to missing request url.", name)
|
|
return None
|
|
|
|
raw_method = request.get("method")
|
|
method = raw_method.upper() if isinstance(raw_method, str) else "POST"
|
|
if method not in {"POST", "PUT"}:
|
|
LOG.warning("Skipping notification '%s' due to invalid method '%s'.", name, method)
|
|
return None
|
|
|
|
raw_type = request.get("type")
|
|
req_type = raw_type.lower() if isinstance(raw_type, str) else "json"
|
|
if req_type not in {"json", "form"}:
|
|
LOG.warning("Skipping notification '%s' due to invalid type '%s'.", name, req_type)
|
|
return None
|
|
|
|
data_key = request.get("data_key") if isinstance(request.get("data_key"), str) else "data"
|
|
if not data_key:
|
|
data_key = "data"
|
|
|
|
headers = self._normalize_headers(request.get("headers"))
|
|
|
|
enabled_value = item.get("enabled")
|
|
enabled: bool = enabled_value if isinstance(enabled_value, bool) else True
|
|
events = self._normalize_events(item.get("on"))
|
|
if events is None:
|
|
LOG.warning("Skipping notification '%s' due to invalid events.", name)
|
|
return None
|
|
|
|
presets = self._normalize_presets(item.get("presets"))
|
|
if presets is None:
|
|
LOG.warning("Skipping notification '%s' due to invalid presets.", name)
|
|
return None
|
|
|
|
return {
|
|
"name": name,
|
|
"on": events,
|
|
"presets": presets,
|
|
"enabled": enabled,
|
|
"request_url": url,
|
|
"request_method": method,
|
|
"request_type": req_type,
|
|
"request_data_key": data_key,
|
|
"request_headers": headers,
|
|
}
|
|
|
|
def _normalize_events(self, events: Any) -> list[str] | None:
|
|
if events is None:
|
|
return []
|
|
if not isinstance(events, list):
|
|
return []
|
|
|
|
allowed = set(NotificationEvents.get_events().values())
|
|
valid = [event for event in events if event in allowed]
|
|
invalid = [event for event in events if event not in allowed]
|
|
|
|
if len(invalid) > 0 and len(valid) < 1:
|
|
return None
|
|
|
|
if len(invalid) > 0:
|
|
LOG.warning("Dropping invalid notification events: %s", ", ".join(invalid))
|
|
|
|
return valid
|
|
|
|
def _normalize_presets(self, presets: Any) -> list[str] | None:
|
|
if presets is None:
|
|
return []
|
|
if not isinstance(presets, list):
|
|
return []
|
|
|
|
allowed = {preset.name for preset in Presets.get_instance().get_all()}
|
|
valid = [preset for preset in presets if preset in allowed]
|
|
invalid = [preset for preset in presets if preset not in allowed]
|
|
|
|
if len(invalid) > 0 and len(valid) < 1:
|
|
return None
|
|
|
|
if len(invalid) > 0:
|
|
LOG.warning("Dropping invalid notification presets: %s", ", ".join(invalid))
|
|
|
|
return valid
|
|
|
|
def _normalize_headers(self, headers: Any) -> list[dict[str, str]]:
|
|
if not isinstance(headers, list):
|
|
return []
|
|
|
|
normalized: list[dict[str, str]] = []
|
|
for header in headers:
|
|
if not isinstance(header, dict):
|
|
continue
|
|
key = str(header.get("key", "")).strip()
|
|
value = str(header.get("value", "")).strip()
|
|
if key and value:
|
|
normalized.append({"key": key, "value": value})
|
|
return normalized
|