1. 程式人生 > 程式設計 >Python實現報警資訊實時傳送至郵箱功能(例項程式碼)

Python實現報警資訊實時傳送至郵箱功能(例項程式碼)

Python實現報警資訊實時傳送至郵箱功能,具體內容如下所示:

程式設計

實現程式碼

cpu.py

# -*- coding: utf-8 -*-
import psutil
import time
from emailsender import txtMail
from log import myloggers
import gc
class 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("文字檔案格式不正確")

mem.py

# -*- coding: utf-8 -*-
import psutil
import time
from emailsender import txtMail
from log import myloggers
import gc
class 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("文字檔案格式不正確")

watchpc.py

# -*- 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 2012R2
import time
from cpu import mycpumonitor
from mem import mymemmonitor
import gc
if __name__ == "__main__":
 while True:
  memObj = mymemmonitor(50)
  memObj.mem_monitor()
  cpuObj = mycpumonitor(10)
  cpuObj.cpu_monitor()
  del memObj
  gc.collect()
  time.sleep(10)

emailsender.py

#-*- coding: utf-8 -*-
import smtplib
import chardet
import codecs
import os
from email.mime.text import MIMEText
from email.header import Header
from 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 = "[email protected]" if auth_user is None else auth_user # 上線時使用專用報警賬戶的使用者名稱
  self.auth_password = (
   "xxxxxxxx" if auth_password is None else auth_password
  ) # 上線時使用專用報警賬戶的密碼
  self.sender = "[email protected]"
 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,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,)

test.config

[email protected],[email protected]
libvirt.log,qemu.log

注意事項

需進入郵箱設定,開啟POP3/SMTP/IMAP。

總結

以上所述是小編給大家介紹的Python實現報警資訊實時傳送至郵箱功能,希望對大家有所幫助,如果大家有任何疑問請給我留言,小編會及時回覆大家的。在此也非常感謝大家對我們網站的支援!
如果你覺得本文對你有幫助,歡迎轉載,煩請註明出處,謝謝!