spug/spug_api/libs/spug.py

62 lines
2.1 KiB
Python

# Copyright: (c) OpenSpug Organization. https://github.com/openspug/spug
# Copyright: (c) <spug.dev@gmail.com>
# Released under the MIT License.
from apps.alarm.models import Group, Contact
from apps.setting.utils import AppSetting
from apps.notify.models import Notify
from libs.mail import Mail
import requests
import json
spug_server = 'http://spug-wx.qbangmang.com'
notify_source = 'info-circle'
def _parse_args(n_grp):
spug_key = AppSetting.get_default('spug_key')
if not spug_key:
Notify.make_notify(notify_source, '1', '发送报警信息失败', '未配置报警服务调用凭据,请在系统管理/系统设置/报警服务设置中配置。')
return None, None
return spug_key, sum([json.loads(x.contacts) for x in Group.objects.filter(id__in=n_grp)], [])
def notify_by_wx(event, subject, n_grp):
spug_key, u_ids = _parse_args(n_grp)
if u_ids is None:
return
users = set(x.wx_token for x in Contact.objects.filter(id__in=u_ids, wx_token__isnull=False))
if users:
data = {
'token': spug_key,
'event': event,
'subject': subject,
'users': list(users)
}
requests.post(f'{spug_server}/apis/notify/wx/', json=data)
def notify_by_email(event, subject, n_grp):
spug_key, u_ids = _parse_args(n_grp)
if u_ids is None:
return
users = set(x.email for x in Contact.objects.filter(id__in=u_ids, email__isnull=False))
if users:
mail_service = json.loads(AppSetting.get_default('mail_service', '{}'))
if mail_service.get('server'):
event_map = {'1': '告警', '2': '恢复'}
subject = f'{event_map[event]}-{subject}'
mail = Mail(**mail_service)
mail.send_text_mail(users, subject, f'{subject}\r\n\r\n自动发送,请勿回复。')
else:
data = {
'token': spug_key,
'event': event,
'subject': subject,
'users': list(users)
}
requests.post(f'{spug_server}/apis/notify/mail/', json=data)
def notify_by_sms():
pass