55 lines
2.0 KiB
Python
55 lines
2.0 KiB
Python
import requests
|
|
from utils.network_proxy import apply_global_proxy_from_env, get_send_msg_proxy_url
|
|
from utils.project_logger import get_project_logger
|
|
|
|
|
|
apply_global_proxy_from_env()
|
|
logger = get_project_logger("send_msg")
|
|
|
|
class LarkNotifier:
|
|
"""飞书消息发送工具类"""
|
|
|
|
def __init__(self, bot_name="weibo", receive_id="on_f972f5f8c4de6ac967574b97dc8c1d82"):
|
|
# 将基础配置放在初始化中，方便后续修改或通过环境变量传入
|
|
self.url = "https://zhiyinzhe.xyz/sendMsg/api/send"
|
|
self.bot_name = bot_name
|
|
self.receive_id = receive_id
|
|
|
|
# 使用 Session 可以复用底层 TCP 连接，提升发送效率
|
|
self.session = requests.Session()
|
|
|
|
proxy_url = get_send_msg_proxy_url()
|
|
if proxy_url:
|
|
self.session.proxies.update({
|
|
"http": proxy_url,
|
|
"https": proxy_url,
|
|
})
|
|
logger.info("send_msg 使用独立代理: %s", proxy_url)
|
|
else:
|
|
logger.info("send_msg 未启用代理")
|
|
|
|
def send_alert(self, title: str, content: str):
|
|
"""
|
|
发送通用文本告警
|
|
:param title: 告警标题 (如 "⚠️ 市场获取失败")
|
|
:param content: 告警具体内容 (如 "Slug: xxx \n 原因: xxx")
|
|
"""
|
|
payload = {
|
|
"bot_name": self.bot_name,
|
|
"receive_id": self.receive_id,
|
|
"receive_id_type": "union_id",
|
|
"content": f"{title}\n{content}"
|
|
}
|
|
|
|
try:
|
|
# 使用 data=payload 发送 Form Data
|
|
response = self.session.post(self.url, data=payload, timeout=5)
|
|
response.raise_for_status()
|
|
logger.info("告警发送成功: %s", title)
|
|
except Exception as e:
|
|
# 告警系统自身的错误只记录日志，绝不抛出异常影响主程序
|
|
logger.error("发送飞书告警失败: %s", e)
|
|
|
|
# 实例化一个全局单例供其他模块直接导入使用
|
|
lark_notifier = LarkNotifier()
|