66 lines
1.8 KiB
Python
66 lines
1.8 KiB
Python
"""Discord a další notifikace pro provoz EMS."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
|
||
import httpx
|
||
|
||
from app.config import get_settings
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
async def send_discord(message: str, level: str = "info") -> bool:
|
||
"""
|
||
Pošle notifikaci na Discord webhook.
|
||
level: 'info', 'warning', 'error', 'critical'
|
||
Vrátí True při úspěchu.
|
||
"""
|
||
settings = get_settings()
|
||
webhook_url = settings.discord_webhook_url
|
||
if not webhook_url:
|
||
logger.debug("Discord webhook not configured, skipping notification")
|
||
return False
|
||
|
||
emoji = {"info": "ℹ️", "warning": "⚠️", "error": "❌", "critical": "🚨"}.get(level, "ℹ️")
|
||
|
||
try:
|
||
async with httpx.AsyncClient(timeout=10) as client:
|
||
resp = await client.post(
|
||
webhook_url,
|
||
json={
|
||
"content": f"{emoji} **EMS Alert** [{level.upper()}]\n{message}",
|
||
},
|
||
)
|
||
resp.raise_for_status()
|
||
return True
|
||
except Exception as e:
|
||
logger.warning("Discord notification failed: %s", e)
|
||
return False
|
||
|
||
|
||
async def notify_modbus_mismatch(
|
||
asset_code: str,
|
||
register: int,
|
||
register_name: str,
|
||
value_written: int,
|
||
value_verified: int,
|
||
attempt: int,
|
||
) -> None:
|
||
msg = (
|
||
f"Modbus mismatch na **{asset_code}**\n"
|
||
f"Registr: `0x{register:04X}` ({register_name})\n"
|
||
f"Zapsáno: `{value_written}` | Přečteno: `{value_verified}`\n"
|
||
f"Pokus č. {attempt}"
|
||
)
|
||
await send_discord(msg, level="error")
|
||
|
||
|
||
async def notify_self_sustain_activated(site_code: str, reason: str) -> None:
|
||
msg = (
|
||
f"Přepnutí na **SELF_SUSTAIN** – lokalita `{site_code}`\n"
|
||
f"Důvod: {reason}"
|
||
)
|
||
await send_discord(msg, level="critical")
|