zbx-tg-bot/zabbix/zabbix_api.py

83 lines
2.3 KiB
Python

import logging as log
from pyexpat.errors import messages
from zabbix_utils import ZabbixAPI
from config import conf
from datetime import datetime, timedelta
def get_active_problems() -> dict:
api = ZabbixAPI(url=conf.zabbix.url, token=conf.zabbix.token)
try:
problems = api.problem.get(
output=[
"eventid",
"suppressed",
"severity",
],
)
event_ids = []
for problem in problems:
if (
problem["suppressed"] == "0"
and int(problem["severity"]) >= conf.zabbix.min_severity
):
event_ids.append(problem["eventid"])
events = api.event.get(
selectHosts=["host"],
eventids=event_ids,
output=[
"eventid",
"name",
"severity",
],
)
events_dict = {"event_ids": []}
for event in events:
event["host"] = event.pop("hosts", None)[0]["host"]
events_dict[event["eventid"]] = event
events_dict["event_ids"].append(event["eventid"])
return events_dict
except:
log.warning("Get event from zabbix error")
def event_acknowledge(event_id: int, mute_time: int, muted_by: str):
api = ZabbixAPI(url=conf.zabbix.url, token=conf.zabbix.token)
if mute_time == 0:
mute_to = 0
else:
mute_to = int((datetime.now() + timedelta(hours=mute_time)).timestamp())
try:
response = api.event.acknowledge(
eventids=event_id,
action=38,
suppress_until=mute_to,
message=muted_by,
)
if response:
log.info(f"Event {event_id} acknowledged")
return True
except:
log.warning(f"Acknowledge event {event_id} from zabbix error")
return False
def event_close(event_id: int, closed_by: str):
api = ZabbixAPI(url=conf.zabbix.url, token=conf.zabbix.token)
try:
response = api.event.acknowledge(
eventids=event_id,
action=5,
message=closed_by,
)
if response:
log.info(f"Event {event_id} closed")
return True
except:
log.warning(f"Closed event {event_id} from zabbix error")
return False