udochudo/Telezab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(alerts): improve active problem fetching and message formatting

Browse Source 
Reworked the logic for retrieving data from Zabbix API to make it more efficient and filter-aware. Message generation for Telegram bot was refactored and decoupled from data retrieval logic to improve structure, readability, and reuse.

Signed-off-by: UdoChudo <stream@udochudo.ru>
This commit is contained in:
Udo Chudo 2025-06-17 23:46:39 +05:00
parent 0169bf5d6b
commit d9df449a17
1 changed files with 27 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
app/bot/utils/tg_formatter.py
View File

@ -1,35 +1,44 @@
from datetime import datetime from datetime import datetime
from pytz import timezone from pytz import timezone
from app.bot.utils.tg_escape_chars import escape_telegram_chars from app.bot.utils.tg_escape_chars import escape_telegram_chars
from config import ZABBIX_URL, ZABBIX_TZ
def format_trigger_message(trigger, zabbix_url: str) -> str:
tz = timezone('Europe/Moscow') def format_trigger_for_tg(trigger):
"""
Формирует текст сообщения для одного триггера
и возвращает (text, inline_buttons_data)
"""
priority_map = {'4': 'HIGH', '5': 'DISASTER'} priority_map = {'4': 'HIGH', '5': 'DISASTER'}
priority_map_emoji = {'4': '⚠️','5': '⛔️'}
event_time_epoch = int(trigger.get('lastEvent', {}).get('clock', trigger.get('lastchange', 0))) event_time_epoch = int(trigger['lastEvent']['clock'])
event_time = datetime.fromtimestamp(event_time_epoch, tz=tz) event_time = datetime.fromtimestamp(event_time_epoch, tz=timezone(ZABBIX_TZ))
event_time_formatted = event_time.strftime('%Y-%m-%d %H:%M:%S Мск') event_time_formatted = event_time.strftime('%Y-%m-%d %H:%M:%S Мск')
host = trigger.get('hosts', [{}])[0].get('name', 'Неизвестно') description = escape_telegram_chars(trigger['description'])
priority = priority_map.get(str(trigger.get('priority')), 'Неизвестно') host = trigger['hosts'][0]['name']
description = escape_telegram_chars(trigger.get('description', '')).replace("{HOST.NAME}", host) priority = priority_map.get(trigger['priority'], 'Неизвестно')
icon = priority_map_emoji.get(trigger['priority'], 'Неизвестно')
items = trigger.get('items', []) description = description.replace("{HOST.NAME}", host)
item_ids = [item['itemid'] for item in items] for i, item in enumerate(trigger['items']):
for i, item in enumerate(items):
placeholder = f"{{ITEM.LASTVALUE{i + 1}}}" placeholder = f"{{ITEM.LASTVALUE{i + 1}}}"
if placeholder in description: if placeholder in description:
description = description.replace(placeholder, item.get('lastvalue', '?')) description = description.replace(placeholder, item['lastvalue'])
batchgraph_link = f"{zabbix_url}/history.php?action=batchgraph&" item_ids = [item['itemid'] for item in trigger['items']]
batchgraph_link = f"{ZABBIX_URL}/history.php?action=batchgraph&"
batchgraph_link += "&".join([f"itemids[{item_id}]={item_id}" for item_id in item_ids]) batchgraph_link += "&".join([f"itemids[{item_id}]={item_id}" for item_id in item_ids])
batchgraph_link += "&graphtype=0" batchgraph_link += "&graphtype=0"
return ( text = (
f"<b>Host</b>: {host}\n" f"<b>Host</b>: {host}\n"
f"<b>Описание</b>: {description}\n" f"<b>Описание</b>: {description}\n"
f"<b>Критичность</b>: {priority}\n" f"<b>Критичность</b>: {icon} {priority}\n"
f"<b>Время создания</b>: {event_time_formatted}\n" f"<b>Время создания</b>: {event_time_formatted}\n"
f'<b>URL</b>: <a href="{batchgraph_link}">Ссылка на график</a>' # f'<b>URL</b>: <a href="{batchgraph_link}">Ссылка на график</a>'
) )
# Возвращаем текст и ссылку, чтобы потом сделать inline кнопку
return text, batchgraph_link