66 lines
1.5 KiB
Python
66 lines
1.5 KiB
Python
|
from datetime import datetime, timedelta
|
||
|
|
|
||
|
|
from zabbix_utils import AsyncZabbixAPI
|
||
|
|
from config import settings
|
||
|
|
|
||
|
|
|
||
|
|
async def login_api():
|
||
|
|
api = AsyncZabbixAPI(
|
||
|
|
url=settings.zbx.url,
|
||
|
|
)
|
||
|
|
await api.login(
|
||
|
|
token=settings.zbx.token,
|
||
|
|
)
|
||
|
|
return api
|
||
|
|
|
||
|
|
|
||
|
|
async def get_host_by_id(host_id: int) -> str:
|
||
|
|
api = await login_api()
|
||
|
|
response = await api.host.get(
|
||
|
|
hostids=host_id,
|
||
|
|
)
|
||
|
|
devise_name = response[0]["name"]
|
||
|
|
await api.logout()
|
||
|
|
return devise_name
|
||
|
|
|
||
|
|
|
||
|
|
async def maintenance_by_host_id(host_id: int, period: int, host_name: str):
|
||
|
|
now = datetime.now()
|
||
|
|
active_since = int(now.timestamp())
|
||
|
|
active_till = int((now + timedelta(seconds=period)).timestamp())
|
||
|
|
api = await login_api()
|
||
|
|
|
||
|
|
maintenance = await api.maintenance.create(
|
||
|
|
{
|
||
|
|
"name": f"{host_name} {active_since}",
|
||
|
|
"active_since": active_since,
|
||
|
|
"active_till": active_till,
|
||
|
|
"hostids": [host_id],
|
||
|
|
"timeperiods": {"period": period},
|
||
|
|
}
|
||
|
|
)
|
||
|
|
await api.logout()
|
||
|
|
return maintenance
|
||
|
|
|
||
|
|
|
||
|
|
async def event_acknowledge(event_id: int):
|
||
|
|
api = await login_api()
|
||
|
|
response = await api.event.acknowledge(
|
||
|
|
eventids=event_id,
|
||
|
|
action=10,
|
||
|
|
severity=0,
|
||
|
|
)
|
||
|
|
await api.logout()
|
||
|
|
return response
|
||
|
|
|
||
|
|
|
||
|
|
async def add_event_comment(event_id: int, msg: str):
|
||
|
|
api = await login_api()
|
||
|
|
response = await api.event.acknowledge(
|
||
|
|
eventids=event_id,
|
||
|
|
message=msg,
|
||
|
|
action=4,
|
||
|
|
)
|
||
|
|
await api.logout()
|
||
|
|
return response
|