212 lines
7.8 KiB
Python
212 lines
7.8 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import ssl
|
|
import urllib.error
|
|
import urllib.request
|
|
from typing import Any
|
|
|
|
from sqlmodel import Session
|
|
|
|
from ...core.config import settings
|
|
from ...db.models import NotificationChannel
|
|
from ..secret_store import get_secret
|
|
|
|
SECRET_REF = "HOME_ASSISTANT_TOKEN"
|
|
|
|
|
|
def _api_value(key: str, default: Any = None) -> Any:
|
|
value = settings.apis.get(key)
|
|
return default if value is None else value
|
|
|
|
|
|
def _bool(value: Any, default: bool = False) -> bool:
|
|
if value is None:
|
|
return default
|
|
if isinstance(value, bool):
|
|
return value
|
|
return str(value).strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def _int(value: Any, default: int) -> int:
|
|
try:
|
|
return max(1, int(value))
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
def _clean_url(value: Any) -> str | None:
|
|
text = str(value or "").strip().rstrip("/")
|
|
return text or None
|
|
|
|
|
|
def normalize_config(raw: dict[str, Any] | None) -> dict[str, Any]:
|
|
config = dict(raw or {})
|
|
return {
|
|
"url": _clean_url(config.get("url") or _api_value("HOME_ASSISTANT_URL")),
|
|
"notifyDomain": str(config.get("notifyDomain") or _api_value("HOME_ASSISTANT_NOTIFY_DOMAIN", "notify") or "notify").strip() or "notify",
|
|
"notifyService": str(config.get("notifyService") or _api_value("HOME_ASSISTANT_NOTIFY_SERVICE", "notify") or "notify").strip() or "notify",
|
|
"target": str(config.get("target") or _api_value("HOME_ASSISTANT_NOTIFY_TARGET") or "").strip() or None,
|
|
"verifyTls": _bool(config.get("verifyTls"), _bool(_api_value("HOME_ASSISTANT_VERIFY_TLS"), False)),
|
|
"timeoutSeconds": _int(config.get("timeoutSeconds") or _api_value("HOME_ASSISTANT_TIMEOUT_SECONDS"), 10),
|
|
"titlePrefix": str(config.get("titlePrefix") or _api_value("HOME_ASSISTANT_NOTIFICATION_TITLE_PREFIX", "AppOps") or "AppOps").strip() or "AppOps",
|
|
"dryRun": _bool(config.get("dryRun"), _bool(_api_value("HOME_ASSISTANT_DRY_RUN"), False)),
|
|
}
|
|
|
|
|
|
def safe_config(config: dict[str, Any] | None) -> dict[str, Any]:
|
|
normalized = normalize_config(config)
|
|
return {
|
|
"url": normalized["url"],
|
|
"notifyDomain": normalized["notifyDomain"],
|
|
"notifyService": normalized["notifyService"],
|
|
"target": normalized["target"],
|
|
"verifyTls": normalized["verifyTls"],
|
|
"timeoutSeconds": normalized["timeoutSeconds"],
|
|
"titlePrefix": normalized["titlePrefix"],
|
|
"dryRun": normalized["dryRun"],
|
|
}
|
|
|
|
|
|
def _token(session: Session) -> str | None:
|
|
return get_secret(session, SECRET_REF) or (str(_api_value(SECRET_REF) or "").strip() or None)
|
|
|
|
|
|
def config_status(session: Session, channel: NotificationChannel) -> dict[str, Any]:
|
|
config = normalize_config(channel.config_json)
|
|
missing: list[str] = []
|
|
token = _token(session)
|
|
if not config["url"]:
|
|
missing.append("HOME_ASSISTANT_URL")
|
|
if not token:
|
|
missing.append("HOME_ASSISTANT_TOKEN")
|
|
if missing:
|
|
status = "missing_config"
|
|
elif not channel.enabled:
|
|
status = "disabled"
|
|
else:
|
|
status = "connected"
|
|
return {
|
|
"status": status,
|
|
"missing": missing,
|
|
"hasSecret": bool(token),
|
|
"safeConfig": safe_config(channel.config_json),
|
|
}
|
|
|
|
|
|
def format_message(event: dict[str, Any]) -> str:
|
|
severity = str(event.get("severity") or "info").upper()
|
|
source = str(event.get("source") or "appops")
|
|
target = str(event.get("target") or "global")
|
|
message = str(event.get("message") or event.get("title") or "Notification")
|
|
return f"{severity} - {source} - {target}\n{message}"
|
|
|
|
|
|
def _title(config: dict[str, Any], event: dict[str, Any]) -> str:
|
|
prefix = str(config.get("titlePrefix") or "AppOps")
|
|
title = str(event.get("title") or "Notification")
|
|
return f"{prefix}: {title}" if prefix else title
|
|
|
|
|
|
def _request_context(config: dict[str, Any]) -> ssl.SSLContext | None:
|
|
if config["verifyTls"]:
|
|
return None
|
|
return ssl._create_unverified_context()
|
|
|
|
|
|
def _call_home_assistant(config: dict[str, Any], token: str, method: str, path: str, payload: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
body = json.dumps(payload).encode("utf-8") if payload is not None else None
|
|
request = urllib.request.Request(
|
|
f"{config['url']}{path}",
|
|
data=body,
|
|
method=method,
|
|
headers={
|
|
"Authorization": f"Bearer {token}",
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json",
|
|
},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=config["timeoutSeconds"], context=_request_context(config)) as response:
|
|
return {"ok": True, "statusCode": response.status}
|
|
except urllib.error.HTTPError as exc:
|
|
if exc.code in {401, 403}:
|
|
return {"ok": False, "status": "auth_failed", "statusCode": exc.code}
|
|
return {"ok": False, "status": "error", "statusCode": exc.code}
|
|
except urllib.error.URLError:
|
|
return {"ok": False, "status": "unreachable"}
|
|
except OSError:
|
|
return {"ok": False, "status": "unreachable"}
|
|
|
|
|
|
def test_connection(session: Session, channel: NotificationChannel) -> dict[str, Any]:
|
|
status = config_status(session, channel)
|
|
if status["missing"]:
|
|
return {"success": False, "status": "missing_config", "missing": status["missing"]}
|
|
config = normalize_config(channel.config_json)
|
|
token = _token(session) or ""
|
|
if config["dryRun"]:
|
|
return {"success": True, "status": "connected", "dryRun": True}
|
|
result = _call_home_assistant(config, token, "GET", "/api/")
|
|
return {
|
|
"success": bool(result["ok"]),
|
|
"status": "connected" if result["ok"] else result.get("status", "error"),
|
|
"statusCode": result.get("statusCode"),
|
|
}
|
|
|
|
|
|
def _notify_payload(config: dict[str, Any], event: dict[str, Any]) -> dict[str, Any]:
|
|
payload: dict[str, Any] = {
|
|
"title": _title(config, event),
|
|
"message": format_message(event),
|
|
"data": {
|
|
"source": event.get("source"),
|
|
"severity": event.get("severity"),
|
|
"category": event.get("category"),
|
|
"target": event.get("target"),
|
|
"ruleId": event.get("ruleId"),
|
|
"eventId": event.get("id"),
|
|
},
|
|
}
|
|
if config["target"]:
|
|
payload["target"] = config["target"]
|
|
return payload
|
|
|
|
|
|
def send_notification(session: Session, channel: NotificationChannel, event: dict[str, Any]) -> dict[str, Any]:
|
|
if not channel.enabled:
|
|
return {"status": "suppressed", "reason": "disabled"}
|
|
status = config_status(session, channel)
|
|
if status["missing"]:
|
|
return {"status": "failed", "reason": "missing_config", "missing": status["missing"]}
|
|
config = normalize_config(channel.config_json)
|
|
if config["dryRun"]:
|
|
return {"status": "sent", "reason": "dry_run"}
|
|
token = _token(session) or ""
|
|
path = f"/api/services/{config['notifyDomain']}/{config['notifyService']}"
|
|
result = _call_home_assistant(config, token, "POST", path, _notify_payload(config, event))
|
|
if result["ok"]:
|
|
return {"status": "sent"}
|
|
return {"status": "failed", "reason": result.get("status", "error"), "statusCode": result.get("statusCode")}
|
|
|
|
|
|
def send_test_notification(session: Session, channel: NotificationChannel) -> dict[str, Any]:
|
|
event = {
|
|
"id": "test",
|
|
"ruleId": "notification.channel.test",
|
|
"source": "notification",
|
|
"event_type": "notification.channel.test",
|
|
"severity": "info",
|
|
"title": "Home Assistant test notification",
|
|
"message": "AppOps test notification.",
|
|
"category": "test",
|
|
"target": "home_assistant",
|
|
}
|
|
result = send_notification(session, channel, event)
|
|
return {
|
|
"success": result["status"] == "sent",
|
|
"status": result["status"],
|
|
"reason": result.get("reason"),
|
|
"missing": result.get("missing"),
|
|
}
|