96 lines
3.8 KiB
Python
96 lines
3.8 KiB
Python
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())
|
||
}
|
||
)
|