72 lines
2.7 KiB
Python
72 lines
2.7 KiB
Python
import subprocess
|
||
import requests
|
||
import logging
|
||
import time
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
class Notifier:
|
||
def __init__(self, wechat_tokens, wecom_tokens=None):
|
||
self.wechat_tokens = wechat_tokens
|
||
self.wecom_tokens = wecom_tokens or {}
|
||
self.ip = self._get_ip_address()
|
||
|
||
def _get_ip_address(self):
|
||
"""获取本机IP地址（优先获取192.168开头的IP）"""
|
||
import socket
|
||
try:
|
||
hostname = socket.gethostname()
|
||
ip_addresses = socket.gethostbyname_ex(hostname)[2]
|
||
|
||
for ip in ip_addresses:
|
||
if ip.startswith('192.168.2.'):
|
||
return ip
|
||
|
||
except Exception as e:
|
||
print(f"获取IP地址失败: {e}，使用hostname")
|
||
|
||
# 如果没有找到192.168.2.开头的IP或发生异常，使用hostname作为后备
|
||
return hostname
|
||
|
||
def send_wechat_alert(self, message):
|
||
"""发送微信告警"""
|
||
title = f"设备告警：{self.ip}"
|
||
if not self.wechat_tokens:
|
||
logger.warning("未配置微信推送 Token，跳过发送。")
|
||
return
|
||
|
||
for name, token in self.wechat_tokens.items():
|
||
url = f'https://wx.xtuis.cn/{token}.send'
|
||
try:
|
||
requests.post(url, data={'text': title, 'desp': f"设备 {self.ip} {message}"}, timeout=10)
|
||
logger.info(f"告警已发送给: {name}")
|
||
time.sleep(3) # 避免发送太快导致失败
|
||
except Exception as e:
|
||
logger.warning(f"发送告警失败: {e}")
|
||
|
||
def send_weCom_alert(self, message):
|
||
"""发送企业微信告警"""
|
||
title = f"{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}, {self.ip}"
|
||
content = f"{title}\n\n{message}"
|
||
|
||
if not self.wecom_tokens:
|
||
logger.warning("未配置企业微信 Webhook Key，跳过发送。")
|
||
return
|
||
|
||
for name, key in self.wecom_tokens.items():
|
||
url = f'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={key}'
|
||
payload = {
|
||
"msgtype": "text",
|
||
"text": {
|
||
"content": content
|
||
}
|
||
}
|
||
try:
|
||
response = requests.post(url, json=payload, timeout=10)
|
||
if response.status_code == 200:
|
||
logger.info(f"企业微信告警已发送给: {name}")
|
||
else:
|
||
logger.warning(f"发送企业微信告警到 {name} 失败: HTTP {response.status_code}, {response.text}")
|
||
except Exception as e:
|
||
logger.warning(f"发送企业微信告警到 {name} 失败: {e}")
|