Initial commit: Pangolin Sentinel Docker & Unraid Hub

backend/notifier.py

@@ -0,0 +1,95 @@
+import asyncio
+import time
+import httpx
+from typing import Dict, Any, Optional
+from backend.config import get_setting
+
+async def send_notification(
+    title: str,
+    message: str,
+    status: str = "info",  # "success", "error", "warning", "info"
+    details: Optional[Dict[str, Any]] = None
+) -> Dict[str, Any]:
+    webhook_url = get_setting("webhook_url", "")
+    webhook_type = get_setting("webhook_type", "discord")
+    bot_token = get_setting("telegram_bot_token", "")
+    chat_id = get_setting("telegram_chat_id", "")
+
+    if webhook_type == "telegram":
+        if not bot_token or not chat_id:
+            return {"success": False, "error": "Telegram Bot Token and Chat ID required"}
+    elif not webhook_url:
+        return {"success": False, "error": "Webhook URL not configured"}
+
+    color_map = {
+        "success": 0x10B981,  # Emerald green
+        "error": 0xEF4444,    # Red
+        "warning": 0xF59E0B,  # Amber
+        "info": 0x3B82F6      # Blue
+    }
+    color = color_map.get(status, 0x3B82F6)
+
+    try:
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            if webhook_type == "discord":
+                embed = {
+                    "title": title,
+                    "description": message,
+                    "color": color,
+                    "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+                    "footer": {"text": "Pangolin VPS Sentinel"}
+                }
+                if details:
+                    fields = []
+                    for k, v in details.items():
+                        fields.append({"name": str(k).capitalize(), "value": str(v), "inline": True})
+                    embed["fields"] = fields
+
+                payload = {
+                    "username": "Pangolin Sentinel",
+                    "embeds": [embed]
+                }
+                resp = await client.post(webhook_url, json=payload)
+                return {"success": resp.is_success, "status_code": resp.status_code, "text": resp.text}
+
+            elif webhook_type == "telegram":
+                tg_url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
+                icon = "✅" if status == "success" else "❌" if status == "error" else "ℹ️"
+                text = f"<b>{icon} {title}</b>\n\n{message}\n"
+                if details:
+                    text += "\n"
+                    for k, v in details.items():
+                        text += f"• <b>{k.capitalize()}:</b> {v}\n"
+                
+                resp = await client.post(tg_url, json={
+                    "chat_id": chat_id,
+                    "text": text,
+                    "parse_mode": "HTML"
+                })
+                return {"success": resp.is_success, "status_code": resp.status_code, "text": resp.text}
+
+            else:  # Generic webhook
+                payload = {
+                    "event": "pangolin_event",
+                    "title": title,
+                    "message": message,
+                    "status": status,
+                    "details": details or {},
+                    "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
+                }
+                resp = await client.post(webhook_url, json=payload)
+                return {"success": resp.is_success, "status_code": resp.status_code, "text": resp.text}
+
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+async def test_webhook() -> Dict[str, Any]:
+    return await send_notification(
+        title="🔔 Test Notification from Pangolin Sentinel",
+        message="Your webhook configuration is active and working properly on your Pangolin VPS.",
+        status="info",
+        details={
+            "vps": get_setting("app_host", "Pangolin-VPS"),
+            "time": time.strftime("%Y-%m-%d %H:%M:%S UTC", time.gmtime())
+        }
+    )