Python实现报警信息实时发送至邮箱功能(实例代码)

# -*- coding: utf-8 -*-import psutilimport timefrom emailsender import txtMailfrom log import myloggersimport gcclass mycpumonitor(): # up是cpu监控的阈值，默认是90% def __init__(self, up=None):  self.up = 90 if up is None else up def cpu_monitor(self):  cpu_percent = psutil.cpu_percent()  now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())  if cpu_percent > self.up:   filename = 'cpu.txt'   with open(filename, 'w') as f: # 如果filename不存在会自动创建， 'w'表示写数据，写之前会清空文件中的原有数据！    f.write(str(now) + "CPU使用率超过" + str(self.up) + "!!!!./n")    f.write(str(now) + "当前CPU使用率" + str(cpu_percent) + "!!!/n")   mail = txtMail()   try:    mail.txt_send_mail(filename="test.config", alarm=filename)    temp_msg = "CPU超标，当前CPU使用率：" + str(cpu_percent)    logger1 = myloggers(temp_msg)    logger1.maillogging()    del mail    gc.collect()   except:    print("文本文件格式不正确")

# -*- coding: utf-8 -*-import psutilimport timefrom emailsender import txtMailfrom log import myloggersimport gcclass mymemmonitor(): # up是内存监控的阈值，默认是90% def __init__(self, up=None):  self.up = 90 if up is None else up def mem_monitor(self):  mem_percent = psutil.virtual_memory()[2]  now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())  if mem_percent > self.up:   filename = 'mem.txt'   with open(filename, 'w') as f: # 如果filename不存在会自动创建， 'w'表示写数据，写之前会清空文件中的原有数据！    f.write(str(now) + "内存使用率超过" + str(self.up) + "!!!!./n")    f.write(str(now) + "当前内存使用率" + str(mem_percent) + "!!!/n")   mail = txtMail()   try:    mail.txt_send_mail(filename="test.config", alarm=filename)    temp_msg = "内存超标，当前内存使用率：" + str(mem_percent)    logger1 = myloggers(temp_msg)    logger1.maillogging()    del mail    gc.collect()   except:    print("文本文件格式不正确")

# -*- coding: utf-8 -*-# Author: WuYang# Date-modified: 07Nov2019# Function: send alarm data through SMTP protocol# Architecture: cpu.py, mem.py, etc. is to monitor performance of server# Lang: Python3.7# Env: CentOS7/WindowServer 2012R2import timefrom cpu import mycpumonitorfrom mem import mymemmonitorimport gcif __name__ == "__main__": while True:  memObj = mymemmonitor(50)  memObj.mem_monitor()  cpuObj = mycpumonitor(10)  cpuObj.cpu_monitor()  del memObj  gc.collect()  time.sleep(10)

#-*- coding: utf-8 -*-import smtplibimport chardetimport codecsimport osfrom email.mime.text import MIMETextfrom email.header import Headerfrom email.mime.multipart import MIMEMultipart# 第三方SMTP服务class txtMail(object): def __init__(self, host=None, auth_user=None, auth_password=None):  self.host = "smtp.163.com" if host is None else host # 设置发送邮件服务使用专用报警账户的用户名  self.auth_user = "xxxxxx@163.com" if auth_user is None else auth_user # 上线时使用专用报警账户的用户名  self.auth_password = (   "xxxxxxxx" if auth_password is None else auth_password  ) # 上线时使用专用报警账户的密码  self.sender = "xxxxxxxx@163.com" def send_mail(self, subject, msg_str, recipient_list, attachment_list=None):  message = MIMEMultipart()  message["From"] = self.sender  message["To"] = Header(";".join(recipient_list), "utf-8")  message["Subject"] = Header(subject, "utf-8")  message.attach(MIMEText(msg_str, "plain", "utf-8"))  # 如果有附件，则添加附件  if attachment_list:   for att in attachment_list:    attachment = MIMEText(open(att, "rb").read(), "base64", "utf-8")    attachment["Content-Type"] = "application/octet-stream"    # 这里filename可以任意写，写什么名字，邮件中显示什么名字    filename = os.path.basename(att)    attachment.add_header(     "Content-Disposition",     "attachment",     filename=("utf-8", "", filename),    )    message.attach(attachment)  smtpObj = smtplib.SMTP_SSL(self.host)  smtpObj.connect(self.host, smtplib.SMTP_SSL_PORT)  smtpObj.login(self.auth_user, self.auth_password)  smtpObj.sendmail(self.sender, recipient_list, message.as_string())  smtpObj.quit()  print("邮件发送成功")  print(attachment_list) def guess_chardet(self, filename):  """  :param filename: 传入一个文本文件  :return: 返回文本文件的编码格式  """  encoding = None  try:   # 由于本需求所解析的文本文件都不大，可以一次性读入内存   # 如果是大文件，则读取固定字节数   raw = open(filename, "rb").read()   if raw.startswith(codecs.BOM_UTF8): # 处理UTF-8 with BOM    encoding = "utf-8-sig"   else:    result = chardet.detect(raw)    encoding = result["encoding"]  except:   pass  return encoding def txt_send_mail(self, filename, alarm):  """  :param filename:  :return:  将指定格式的txt文件发送至邮箱，txt文件样例如下  # 收件人，逗号分隔  # 附件，逗号分隔  """  with open(filename, encoding=self.guess_chardet(filename)) as f:   lines = f.readlines()  recipient_list = lines[0].strip().split(",")  attachment_list = []  for file in lines[-1].strip().split(","):   if os.path.isfile(file):    attachment_list.append(file)  # 如果没有附件，则为None  if attachment_list == []:   attachment_list = None  with open(alarm, encoding=self.guess_chardet(alarm)) as f1:   lines1 = f1.readlines()  subject = lines1[0].strip()  msg_str = "".join(lines1[1:])  self.send_mail(   subject=subject,   msg_str=msg_str,   recipient_list=recipient_list,   attachment_list=attachment_list,  )

XXXXX@qq.com,XXXXX@163.comlibvirt.log,qemu.log