141 lines
4.5 KiB
Python
141 lines
4.5 KiB
Python
import poplib
|
||
import ssl
|
||
import email
|
||
from email.header import decode_header
|
||
import time
|
||
import pymysql
|
||
|
||
# 关闭SSL校验，解决老Python握手重置
|
||
ssl._create_default_https_context = ssl._create_unverified_context
|
||
|
||
# ========== 邮箱配置 自行修改 ==========
|
||
MAIL_HOST = "pop.163.com"
|
||
MAIL_PORT = 995
|
||
MAIL_USER = "lyudream@163.com"
|
||
MAIL_PWD = "DVZxyF5NKCkR6fL5"
|
||
|
||
# ========== 本地MariaDB配置 ==========
|
||
DB_HOST = "127.0.0.1"
|
||
DB_USER = "root"
|
||
DB_PWD = "hp93000"
|
||
DB_NAME = "alert_mail_stat"
|
||
|
||
# 解码邮件标题、正文
|
||
def decode_str(s):
|
||
value, charset = decode_header(s)[0]
|
||
if charset:
|
||
value = value.decode(charset)
|
||
return value
|
||
|
||
# 获取邮件正文
|
||
def get_email_content(msg):
|
||
content = ""
|
||
if msg.is_multipart():
|
||
for part in msg.walk():
|
||
content_type = part.get_content_type()
|
||
content_disposition = str(part.get("Content-Disposition"))
|
||
if content_type == 'text/plain' and 'attachment' not in content_disposition:
|
||
payload = part.get_payload(decode=True)
|
||
charset = part.get_charset()
|
||
if charset is None:
|
||
charset = 'utf-8'
|
||
content = payload.decode(charset, errors='ignore')
|
||
break
|
||
else:
|
||
payload = msg.get_payload(decode=True)
|
||
charset = msg.get_charset()
|
||
if charset is None:
|
||
charset = 'utf-8'
|
||
content = payload.decode(charset, errors='ignore')
|
||
return content
|
||
|
||
# 简单解析告警关键字（适配Alertmanager中文邮件模板：告警名称、实例、级别）
|
||
def parse_alert_info(text):
|
||
alert_name = ""
|
||
instance = ""
|
||
severity = "unknown"
|
||
lines = text.splitlines()
|
||
for line in lines:
|
||
if "告警名称" in line:
|
||
alert_name = line.split("：")[-1].strip()
|
||
if "实例" in line:
|
||
instance = line.split("：")[-1].strip()
|
||
if "级别" in line:
|
||
s_val = line.split("：")[-1].strip()
|
||
if s_val.lower() in ["critical", "严重"]:
|
||
severity = "critical"
|
||
elif s_val.lower() in ["warning", "警告"]:
|
||
severity = "warning"
|
||
return alert_name, instance, severity
|
||
|
||
# 主抓取逻辑
|
||
def crawl_mail():
|
||
# 连接数据库
|
||
db = pymysql.connect(host=DB_HOST, user=DB_USER, password=DB_PWD, database=DB_NAME, charset='utf8mb4')
|
||
cursor = db.cursor()
|
||
|
||
# 连接POP3邮箱
|
||
pop_conn = poplib.POP3_SSL(MAIL_HOST, MAIL_PORT)
|
||
pop_conn.user(MAIL_USER)
|
||
pop_conn.pass_(MAIL_PWD)
|
||
total_num, total_size = pop_conn.stat()
|
||
print(f"当前邮箱总邮件数量: {total_num}")
|
||
|
||
if total_num == 0:
|
||
print("暂无新邮件")
|
||
pop_conn.quit()
|
||
db.close()
|
||
return
|
||
|
||
# 从最新一封开始遍历
|
||
for idx in range(total_num, 0, -1):
|
||
# 获取邮件原始内容
|
||
resp, raw_lines, octet = pop_conn.retr(idx)
|
||
raw_msg = b'\r\n'.join(raw_lines).decode('utf-8', errors='ignore')
|
||
msg = email.message_from_string(raw_msg)
|
||
|
||
# 生成唯一标识去重
|
||
mail_uid = msg.get("Message-ID", str(time.time()))
|
||
|
||
# 查重，避免重复入库
|
||
cursor.execute("SELECT id FROM alert_log WHERE mail_uid=%s", (mail_uid,))
|
||
if cursor.fetchone():
|
||
continue
|
||
|
||
# 解析标题、发件人
|
||
subject = decode_str(msg.get("Subject", ""))
|
||
from_addr = msg.get("From", "")
|
||
# 过滤非告警发件人（自行修改匹配你的Alertmanager发件地址）
|
||
if "prometheusalert" not in from_addr.lower():
|
||
continue
|
||
|
||
# 正文解析告警字段
|
||
body = get_email_content(msg)
|
||
alert_name, instance, severity = parse_alert_info(body)
|
||
if not alert_name:
|
||
print(f"邮件{idx}未识别到告警，跳过")
|
||
continue
|
||
|
||
# 区分触发/恢复告警
|
||
alert_type = 1
|
||
if "恢复" in subject or "resolved" in subject.lower():
|
||
alert_type = 2
|
||
|
||
now = time.strftime("%Y-%m-%d %H:%M:%S")
|
||
sql = """
|
||
INSERT INTO alert_log
|
||
(mail_uid, alert_type, alert_name, instance, severity, starts_at, ends_at, content, receive_time)
|
||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||
"""
|
||
cursor.execute(sql, (mail_uid, alert_type, alert_name, instance, severity, now, None, body, now))
|
||
db.commit()
|
||
print(f"入库成功 | 告警:{alert_name} 实例:{instance}")
|
||
|
||
pop_conn.quit()
|
||
cursor.close()
|
||
db.close()
|
||
print("本次抓取完成\n")
|
||
|
||
if __name__ == "__main__":
|
||
crawl_mail()
|