refactor config and services

auth/__init__.py

@@ -1,6 +1,6 @@
-from .static_env import verify_token_admin, verify_user_pwd
+from .static_env import verify_user_pwd, verify_token_zabbix
 
 __all__ = [
-    "verify_token_admin",
     "verify_user_pwd",
+    "verify_token_zabbix",
 ]

auth/static_env.py

@@ -8,8 +8,8 @@ security = HTTPBasic()
 api_key_header = APIKeyHeader(name="X-API-KEY", auto_error=False)
 
 
-def verify_token_admin(token: str = Depends(api_key_header)):
+def verify_token_zabbix(token: str = Depends(api_key_header)):
-    if token != conf.token.admin:
+    if token != conf.zbx.token:
         log.warning("Invalid token")
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,

config/.env-template

@@ -5,9 +5,6 @@ OAA_CFG__RUN__RELOAD=True
 OAA_CFG__LOG__LEVEL=30
 OAA_CFG__LOG__LEVEL_TO_FILE=30
 
-OAA_CFG__TOKEN__ADMIN=string
-OAA_CFG__TOKEN__USER=string
-
 OAA_CFG__SWAGGER__LOGIN=admin
 OAA_CFG__SWAGGER__PWD=P@ssw0rd!
 
@@ -15,10 +12,10 @@ 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__TOKEN=string
+OAA_CFG__TG__CHAT_ID=0
-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
+
+OAA_CFG__ZBX__TOKEN=string
+OAA_CFG__ZBX__CLOSE_ALERT_PATTERN=^Problem has been resolved

config/__init__.py

@@ -1,5 +1,6 @@
 from .config import conf, STATIC_DIR
 
+
 __all__ = [
     "conf",
     STATIC_DIR,

config/config.py

@@ -34,11 +34,7 @@ class PrefixConfig(BaseModel):
     api_v1: str = "/api/v1"
     tg_v1: str = api_v1 + "/tg"
     ping: str = "/ping"
-
+    zbx: str = api_v1 + "/zbx"
-
-class TokenConfig(BaseModel):
-    admin: str
-    user: str
 
 
 class RedisConfig(BaseModel):
@@ -48,14 +44,17 @@ class RedisConfig(BaseModel):
 
 
 class TelegramConfig(BaseModel):
-    token: str
+    bot_token: str
-    dashboard_chat_id: int
+    chat_id: int
     dashboard_tred_id: int | None = None
-    close_alert_pattern: str
-    net_chat_id: int
     net_tred_id: int | None = None
 
 
+class FromZabbix(BaseModel):
+    token: str
+    close_alert_pattern: str
+
+
 class SwaggerConfig(BaseModel):
     openapi_url: str = "/openapi.json"
     title: str = "API"
@@ -81,9 +80,9 @@ class Settings(BaseSettings):
     swagger: SwaggerConfig
     log: LogConfig = LogConfig()
     prefix: PrefixConfig = PrefixConfig()
-    token: TokenConfig
     redis: RedisConfig
     tg: TelegramConfig
+    zbx: FromZabbix
 
 
 conf = Settings()

routers/__init__.py

@@ -2,7 +2,7 @@ from fastapi import APIRouter
 
 from .swagger import router as swagger_router
 from .ping import router as ping_router
-from .tg_send import router as tg_send_router
+from .from_zbx import router as zbx_router
 from config import conf
 
 
@@ -17,8 +17,9 @@ router.include_router(
     prefix=conf.prefix.ping,
     tags=["Ping"],
 )
+
 router.include_router(
-    tg_send_router,
+    zbx_router,
-    prefix=conf.prefix.tg_v1,
+    prefix=conf.prefix.zbx,
-    tags=["Telegram"],
+    tags=["From Zabbix"],
 )

routers/from_zbx.py

@@ -0,0 +1,50 @@
+from fastapi import APIRouter, Depends
+import logging as log
+
+from schemas import (
+    zbxMessageToDashboard,
+    zbxMessageToNetworkChat,
+)
+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
+
+router = APIRouter()
+
+
+@router.post("/send-to-dashboard")
+async def send_message_to_dashboard(
+    message: zbxMessageToDashboard,
+    token: str = Depends(verify_token_zabbix),
+):
+    match = re.search(conf.zbx.close_alert_pattern, message.text)
+    log.info(f"match: {match}")
+    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)
+        return
+
+    result = await send_message(
+        text=message.subject + "\n\n" + message.text,
+        chat_id=conf.tg.chat_id,
+        message_thread_id=conf.tg.dashboard_tred_id,
+    )
+    if result and result["status"] == 200:
+        await set_value(message.problem_id, result["msg_id"])
+
+
+@router.post("/send-to-net-chat")
+async def send_message_to_net_chat(
+    message: zbxMessageToNetworkChat,
+    token: str = Depends(verify_token_zabbix),
+):
+    await send_message(
+        text=message.subject + "\n\n" + message.text,
+        chat_id=conf.tg.chat_id,
+        message_thread_id=conf.tg.net_tred_id,
+    )

routers/tg_send.py

@@ -1,57 +0,0 @@
-from fastapi import APIRouter, Depends
-import logging as log
-
-from schemas import TelegramMessageToDashboard, TelegramMessageToNetwork
-from auth import verify_token_admin
-from telegram import (
-    send_message_to_dashboard,
-    del_message_from_dashboard,
-    send_message_to_net_chat,
-)
-from redis_db import set_value, pop_value
-from config import conf
-import re
-
-router = APIRouter()
-
-
-@router.post("/send")
-async def send_message(
-    message: TelegramMessageToDashboard,
-    token: str = Depends(verify_token_admin),
-):
-
-    match = re.search(conf.tg.close_alert_pattern, message.text)
-
-    log.info(f"match: {match}")
-    if match:
-        msg_id = await pop_value(message.problem_id)
-        if msg_id:
-            msg_id = int(msg_id.decode("utf-8"))
-            await del_message_from_dashboard(message_id=msg_id)
-        return
-
-    result = await send_message_to_dashboard(
-        text=message.subject + "\n\n" + message.text
-    )
-    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_net_chat(
-        text=message.subject + "\n\n" + 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,9 +1,9 @@
-from .tg_send import (
+from .from_zabbix import (
-    TelegramMessageToDashboard,
+    MessageToDashboard as zbxMessageToDashboard,
-    TelegramMessageToNetwork,
+    MessageToNetworkChat as zbxMessageToNetworkChat,
 )
 
 __all__ = [
-    "TelegramMessageToDashboard",
+    "zbxMessageToDashboard",
-    "TelegramMessageToNetwork",
+    "zbxMessageToNetworkChat",
 ]

schemas/from_zabbix.py

@@ -1,12 +1,14 @@
 from pydantic import BaseModel
 
 
-class TelegramMessageToDashboard(BaseModel):
+class Message(BaseModel):
     text: str
+
+
+class MessageToDashboard(Message):
+    subject: str
     problem_id: int
-    subject: str
 
 
-class TelegramMessageToNetwork(BaseModel):
+class MessageToNetworkChat(Message):
-    text: str
     subject: str

telegram/__init__.py

@@ -1,11 +1,9 @@
-from .zbx_msg import (
+from .message import (
-    send_message_to_dashboard,
+    send_message,
-    del_message_from_dashboard,
+    del_message,
-    send_message_to_net_chat,
 )
 
 __all__ = [
-    "send_message_to_dashboard",
+    "send_message",
-    "del_message_from_dashboard",
+    "del_message",
-    "send_message_to_net_chat",
 ]

telegram/message.py

@@ -0,0 +1,53 @@
+import logging as log
+
+import aiohttp
+from config import conf
+
+
+async def send_message(
+    text: str,
+    chat_id: int,
+    message_thread_id: int | None = None,
+) -> dict | None:
+    url = f"https://api.telegram.org/bot{conf.tg.bot_token}/sendMessage"
+    params = {
+        "chat_id": chat_id,
+        "text": text,
+    }
+    if message_thread_id:
+        params["message_thread_id"] = message_thread_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()
+            if response.status == 200:
+                log.info(f"Message with ID: {resp['result']['message_id']} send")
+                return {
+                    "status": response.status,
+                    "msg_id": resp["result"]["message_id"],
+                }
+            log.warning(f"Message not send. Response status: {response.status}")
+
+
+async def del_message(
+    message_id: int,
+    chat_id: int,
+) -> dict | None:
+    url = f"https://api.telegram.org/bot{conf.tg.bot_token}/deleteMessage"
+    async with aiohttp.ClientSession() as session:
+        async with session.post(
+            url,
+            json={
+                "chat_id": chat_id,
+                "message_id": message_id,
+            },
+        ) as response:
+            if response.status == 200:
+                log.info(f"Message ID {message_id} deleted")
+                return {
+                    "status": response.status,
+                }
+            log.warning(f"Response status: {response.status}")

telegram/zbx_msg.py

@@ -1,64 +0,0 @@
-import logging as log
-
-import aiohttp
-from config import conf
-
-
-async def send_message_to_dashboard(text):
-    url = f"https://api.telegram.org/bot{conf.tg.token}/sendMessage"
-    params = {
-        "chat_id": conf.tg.dashboard_chat_id,
-        "text": text,
-    }
-    if conf.tg.dashboard_tred_id:
-        params["message_thread_id"] = conf.tg.dashboard_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"],
-            }
-
-
-async def del_message_from_dashboard(message_id):
-    url = f"https://api.telegram.org/bot{conf.tg.token}/deleteMessage"
-    async with aiohttp.ClientSession() as session:
-        async with session.post(
-            url,
-            json={
-                "chat_id": conf.tg.dashboard_chat_id,
-                "message_id": message_id,
-            },
-        ) as response:
-            log.info(f"Response status: {response.status}")
-            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"],
-            }