add net service alert

config/.env-template

@@ -18,3 +18,5 @@ OAA_CFG__TG__TOKEN=string
 OAA_CFG__TG__DASHBOARD_CHAT_ID=0
 OAA_CFG__TG__DASHBOARD_TRED_ID=0
 OAA_CFG__TG__CLOSE_ALERT_PATTERN=^Problem has been resolved
+OAA_CFG__TG__NET_CHAT_ID=0
+OAA_CFG__TG__NET_TRED_ID=0

config/config.py

@@ -51,6 +51,8 @@ class TelegramConfig(BaseModel):
     dashboard_chat_id: int
     dashboard_tred_id: int | None = None
     close_alert_pattern: str
+    net_chat_id: int
+    net_tred_id: int | None = None
 
 
 class SwaggerConfig(BaseModel):

osnova-api-alert.service

@@ -0,0 +1,12 @@
+	[Unit]
+	Description=Send alert from Osnova to telegram
+	After=multi-user.target
+
+	[Service]
+	User=linuxadm
+	Type=simple
+	Restart=always
+	ExecStart=/home/linuxadm/python-scripts/osnova-api-alert/.venv/bin/python3 /home/linuxadm/python-scripts/osnova-api-alert/main.py
+
+	[Install]
+	WantedBy=multi-user.target

routers/tg_send.py

@@ -1,7 +1,7 @@
 from fastapi import APIRouter, Depends
 import logging as log
 
-from schemas import TelegramMessageToDashboard
+from schemas import TelegramMessageToDashboard, TelegramMessageToNetwork
 from auth import verify_token_admin
 from telegram import send_message_to_dashboard, del_message_from_dashboard
 from redis_db import set_value, pop_value
@@ -31,4 +31,19 @@ async def send_message(
     if result["status"] == 200:
         log.info(f"Message sent to dashboard: {message.text}")
         await set_value(message.problem_id, result["msg_id"])
+    else:
+        log.warning(f"Failed to send message code: {result['status']}")
+    return
+
+
+@router.post("/send-net-service")
+async def send_message_net_service(
+    message: TelegramMessageToNetwork,
+    token: str = Depends(verify_token_admin),
+):
+    result = await send_message_to_dashboard(text=message.text)
+    if result["status"] == 200:
+        log.info(f"Message sent to network chat: {message.text}")
+    else:
+        log.warning(f"Failed to send message code: {result['status']}")
     return

schemas/__init__.py

@@ -1,5 +1,9 @@
-from .tg_send import TelegramMessageToDashboard
+from .tg_send import (
+    TelegramMessageToDashboard,
+    TelegramMessageToNetwork,
+)
 
 __all__ = [
     "TelegramMessageToDashboard",
+    "TelegramMessageToNetwork",
 ]

schemas/tg_send.py

@@ -4,3 +4,7 @@ from pydantic import BaseModel
 class TelegramMessageToDashboard(BaseModel):
     text: str
     problem_id: int
+
+
+class TelegramMessageToNetwork(BaseModel):
+    text: str

telegram/__init__.py

@@ -1,4 +1,4 @@
-from .msg import send_message_to_dashboard, del_message_from_dashboard
+from .zbx_msg import send_message_to_dashboard, del_message_from_dashboard
 
 __all__ = [
     "send_message_to_dashboard",

telegram/zbx_msg.py

@@ -40,3 +40,25 @@ async def del_message_from_dashboard(message_id):
             resp = await response.json()
             log.info(f"Message ID {message_id} deleted")
             return resp
+
+
+async def send_message_to_net_chat(text):
+    url = f"https://api.telegram.org/bot{conf.tg.token}/sendMessage"
+    params = {
+        "chat_id": conf.tg.net_chat_id,
+        "text": text,
+    }
+    if conf.tg.net_tred_id:
+        params["message_thread_id"] = conf.tg.net_tred_id
+    async with aiohttp.ClientSession() as session:
+        async with session.post(
+            url,
+            json=params,
+        ) as response:
+            log.info(f"Response status: {response.status}")
+            resp = await response.json()
+            log.info(f"Message ID: {resp['result']['message_id']}")
+            return {
+                "status": response.status,
+                "msg_id": resp["result"]["message_id"],
+            }