136 lines
5.1 KiB
Python
136 lines
5.1 KiB
Python
"""
|
|
Router for Testing and Managing Multi-Channel Alerting (Telegram, Slack, Discord, SITOMAT).
|
|
Adheres to Dokumen 34 Architectural Blueprint.
|
|
"""
|
|
import os
|
|
import json
|
|
import logging
|
|
import urllib.request
|
|
from typing import Optional, Dict, Any, List
|
|
from fastapi import APIRouter, HTTPException, status
|
|
from pydantic import BaseModel, Field
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/alerts", tags=["Alerting & Notifications"])
|
|
|
|
|
|
class AlertTestRequest(BaseModel):
|
|
channel: str = Field(..., description="Channel to test: 'telegram', 'slack', 'discord', 'sitomat'")
|
|
webhook_url: Optional[str] = Field(None, description="Webhook URL for Slack, Discord, or SITOMAT")
|
|
bot_token: Optional[str] = Field(None, description="Telegram Bot Token")
|
|
chat_id: Optional[str] = Field(None, description="Telegram Chat ID")
|
|
message: Optional[str] = Field(None, description="Custom test message content")
|
|
|
|
|
|
class AlertChannelsStatusResponse(BaseModel):
|
|
telegram_configured: bool
|
|
slack_configured: bool
|
|
discord_configured: bool
|
|
sitomat_configured: bool
|
|
default_coalescing_window_seconds: int = 300
|
|
|
|
|
|
@router.get("/status", response_model=AlertChannelsStatusResponse)
|
|
def get_alert_channels_status():
|
|
"""
|
|
Menampilkan status konfigurasi saluran alerting pada environment runtime.
|
|
"""
|
|
return AlertChannelsStatusResponse(
|
|
telegram_configured=bool(os.getenv("ALERT_TELEGRAM_BOT_TOKEN") and os.getenv("ALERT_TELEGRAM_CHAT_ID")),
|
|
slack_configured=bool(os.getenv("ALERT_SLACK_WEBHOOK_URL")),
|
|
discord_configured=bool(os.getenv("ALERT_DISCORD_WEBHOOK_URL")),
|
|
sitomat_configured=bool(os.getenv("ALERT_SITOMAT_GATEWAY_URL")),
|
|
default_coalescing_window_seconds=300
|
|
)
|
|
|
|
|
|
@router.post("/test", status_code=status.HTTP_200_OK)
|
|
def test_alert_dispatch(req: AlertTestRequest):
|
|
"""
|
|
Menguji konektivitas pengiriman notifikasi ke saluran target (Telegram, Slack, Discord, atau SITOMAT).
|
|
"""
|
|
channel = req.channel.lower().strip()
|
|
msg_text = req.message or "🧪 MERICA Test Alert: Multi-Channel Webhook Connectivity Verified."
|
|
|
|
if channel == "telegram":
|
|
token = req.bot_token or os.getenv("ALERT_TELEGRAM_BOT_TOKEN")
|
|
chat_id = req.chat_id or os.getenv("ALERT_TELEGRAM_CHAT_ID")
|
|
if not token or not chat_id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Telegram Bot Token and Chat ID must be provided or configured in environment."
|
|
)
|
|
url = f"https://api.telegram.org/bot{token}/sendMessage"
|
|
payload = {"chat_id": chat_id, "text": f"<b>{msg_text}</b>", "parse_mode": "HTML"}
|
|
|
|
elif channel == "slack":
|
|
url = req.webhook_url or os.getenv("ALERT_SLACK_WEBHOOK_URL")
|
|
if not url:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Slack Webhook URL must be provided or configured in environment."
|
|
)
|
|
payload = {
|
|
"text": msg_text,
|
|
"blocks": [
|
|
{
|
|
"type": "section",
|
|
"text": {"type": "mrkdwn", "text": f"🧪 *MERICA Studio Test Alert*\n{msg_text}"}
|
|
}
|
|
]
|
|
}
|
|
|
|
elif channel == "discord":
|
|
url = req.webhook_url or os.getenv("ALERT_DISCORD_WEBHOOK_URL")
|
|
if not url:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Discord Webhook URL must be provided or configured in environment."
|
|
)
|
|
payload = {
|
|
"content": f"🧪 **MERICA Test Alert**\n{msg_text}"
|
|
}
|
|
|
|
elif channel == "sitomat":
|
|
url = req.webhook_url or os.getenv("ALERT_SITOMAT_GATEWAY_URL")
|
|
if not url:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="SITOMAT Gateway URL must be provided or configured in environment."
|
|
)
|
|
payload = {
|
|
"incidentType": "TEST_CONNECTIVITY",
|
|
"message": msg_text,
|
|
"source": "MERICA Studio UI Inspector"
|
|
}
|
|
|
|
else:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Unsupported alert channel '{channel}'. Supported: 'telegram', 'slack', 'discord', 'sitomat'."
|
|
)
|
|
|
|
# Dispatch HTTP POST
|
|
try:
|
|
http_req = urllib.request.Request(
|
|
url,
|
|
data=json.dumps(payload).encode("utf-8"),
|
|
headers={"Content-Type": "application/json"},
|
|
method="POST"
|
|
)
|
|
with urllib.request.urlopen(http_req, timeout=5) as resp:
|
|
status_code = getattr(resp, "status", 200)
|
|
return {
|
|
"status": "success",
|
|
"channel": channel,
|
|
"http_status": status_code,
|
|
"message": f"Successfully delivered test alert to {channel}."
|
|
}
|
|
except Exception as e:
|
|
logger.error("Failed to test alert for channel %s: %s", channel, e)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_502_BAD_GATEWAY,
|
|
detail=f"Failed to deliver test alert to {channel}: {str(e)}"
|
|
)
|