its life

auth/static_env.py

@@ -9,7 +9,7 @@ api_key_header = APIKeyHeader(name="X-API-KEY", auto_error=False)
 
 
 def verify_token_zabbix(token: str = Depends(api_key_header)):
-    if token != conf.zbx.token:
+    if token != conf.auth.from_zbx_token:
         log.warning("Invalid token")
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,

config/.env-template

@@ -12,9 +12,4 @@ OAA_CFG__REDIS__HOST=localhost
 OAA_CFG__REDIS__PORT=6379
 OAA_CFG__REDIS__PWD=P@ssw0rd!
 
-OAA_CFG__TG__BOT_TOKEN=string
+OAA_CFG__AUTH__FROM_ZBX_TOKEN=string
-OAA_CFG__TG__CHAT_ID=0
-OAA_CFG__TG__DASHBOARD_TRED_ID=0
-OAA_CFG__TG__NET_TRED_ID=0
-
-OAA_CFG__ZBX__TOKEN=string

config/config.py

@@ -32,7 +32,6 @@ class LogConfig(BaseModel):
 class PrefixConfig(BaseModel):
     swagger: str = "/docs"
     api_v1: str = "/api/v1"
-    tg_v1: str = api_v1 + "/tg"
     ping: str = "/ping"
     zbx: str = api_v1 + "/zbx"
 
@@ -43,17 +42,6 @@ class RedisConfig(BaseModel):
     pwd: str | None = None
 
 
-class TelegramConfig(BaseModel):
-    bot_token: str
-    chat_id: int
-    dashboard_tred_id: int | None = None
-    net_tred_id: int | None = None
-
-
-class FromZabbix(BaseModel):
-    token: str
-
-
 class SwaggerConfig(BaseModel):
     openapi_url: str = "/openapi.json"
     title: str = "API"
@@ -65,6 +53,10 @@ class SwaggerConfig(BaseModel):
     pwd: str
 
 
+class AuthConfig(BaseModel):
+    from_zbx_token: str
+
+
 class Settings(BaseSettings):
     model_config = SettingsConfigDict(
         env_file=(
@@ -80,8 +72,7 @@ class Settings(BaseSettings):
     log: LogConfig = LogConfig()
     prefix: PrefixConfig = PrefixConfig()
     redis: RedisConfig
-    tg: TelegramConfig
+    auth: AuthConfig
-    zbx: FromZabbix
 
 
 conf = Settings()

docker/docker-compose.yaml

@@ -27,11 +27,3 @@ services:
       - OAA_CFG__REDIS__HOST=localhost
       - OAA_CFG__REDIS__PORT=6379
       - OAA_CFG__REDIS__PWD=P@ssw0rd!
-
-      - OAA_CFG__TG__BOT_TOKEN=string
-      - OAA_CFG__TG__CHAT_ID=0
-      - OAA_CFG__TG__DASHBOARD_TRED_ID=0
-      - OAA_CFG__TG__NET_TRED_ID=0
-
-      - OAA_CFG__ZBX__TOKEN=string
-      - OAA_CFG__ZBX__CLOSE_ALERT_PATTERN=^Problem has been resolved

routers/from_zbx.py

@@ -6,8 +6,6 @@ from schemas import (
 )
 from auth import verify_token_zabbix
 from redis_db import set_value, pop_value
-from config import conf
-import re
 from telegram import send_message
 from telegram import del_message
 
@@ -16,23 +14,25 @@ router = APIRouter()
 
 @router.post("/send-to-dashboard")
 async def send_message_to_dashboard(
-    message: zbxMessageToDashboard,
+    data: zbxMessageToDashboard,
     token: str = Depends(verify_token_zabbix),
 ):
-    match = re.search(conf.zbx.close_alert_pattern, message.text)
+    if data.event_value == 0:
-    log.info(f"match: {match}")
+        msg_id = await pop_value(data.event_id)
-    if match:
-        msg_id = await pop_value(message.problem_id)
         if msg_id:
             msg_id = int(msg_id.decode("utf-8"))
-            await del_message(message_id=msg_id, chat_id=conf.tg.chat_id)
+            await del_message(
+                message_id=msg_id,
+                chat_id=data.tg_chat_id,
+                tg_token=data.tg_token,
+            )
         return
 
     result = await send_message(
-        text=message.subject + "\n\n" + message.text,
+        message=data.message,
-        chat_id=conf.tg.chat_id,
+        chat_id=data.tg_chat_id,
-        message_thread_id=conf.tg.dashboard_tred_id,
+        message_thread_id=data.tg_tread_id,
+        tg_token=data.tg_token,
     )
     if result and result["status"] == 200:
-        await set_value(message.problem_id, result["msg_id"])
+        await set_value(data.event_id, result["msg_id"])
-

schemas/from_zabbix.py

@@ -1,14 +1,11 @@
 from pydantic import BaseModel
+from typing import Optional
 
 
-class Message(BaseModel):
+class MessageToDashboard(BaseModel):
-    text: str
+    message: str
-
+    event_id: int
-
+    event_value: int
-class MessageToDashboard(Message):
+    tg_tread_id: int
-    subject: str
+    tg_chat_id: int
-    problem_id: int
+    tg_token: str
-
-
-class MessageToNetworkChat(Message):
-    subject: str

telegram/message.py

@@ -5,17 +5,18 @@ from config import conf
 
 
 async def send_message(
-    text: str,
+    message: str,
     chat_id: int,
-    message_thread_id: int | None = None,
+    tg_token: str,
+    message_thread_id: int,
 ) -> dict | None:
-    url = f"https://api.telegram.org/bot{conf.tg.bot_token}/sendMessage"
+    url = f"https://api.telegram.org/bot{tg_token}/sendMessage"
     params = {
         "chat_id": chat_id,
-        "text": text,
+        "message_thread_id": message_thread_id,
+        "text": message,
     }
-    if message_thread_id:
+
-        params["message_thread_id"] = message_thread_id
     async with aiohttp.ClientSession() as session:
         async with session.post(
             url,
@@ -34,9 +35,10 @@ async def send_message(
 
 async def del_message(
     message_id: int,
+    tg_token: str,
     chat_id: int,
 ) -> dict | None:
-    url = f"https://api.telegram.org/bot{conf.tg.bot_token}/deleteMessage"
+    url = f"https://api.telegram.org/bot{tg_token}/deleteMessage"
     async with aiohttp.ClientSession() as session:
         async with session.post(
             url,
@@ -50,4 +52,5 @@ async def del_message(
                 return {
                     "status": response.status,
                 }
-            log.warning(f"Response status: {response.status}")
+            else:
+                log.warning(f"Response status: {response.status}")