import smtplib

import os

from email.mime.text import MIMEText

from email.mime.multipart import MIMEMultipart

import traceback

import time

def send(self,content=None):

'''''

發送郵件

'''

server = None

index = 1 #郵件重試

while index < 3:

index += 1

try:

server = smtplib.SMTP_SSL("smtp.exmail.qq.com",port=465)

server.login(self.user,self.passwd)

server.sendmail("" % self.user, self.to_list + self.cc_list, self.get_attach(content=content))

server.close()

print("send email successful")

break

except Exception as e:

traceback.print_exc()

time.sleep(5)

print("send email failed")

finally:

try:

if server is not None:

server.close()

except Exception as e:

traceback.print_exc()

def get_attach(self,content=None,subtype='html'):

'''''

構造郵件內容

'''

attach = MIMEMultipart()

#添加郵件內容

txt = MIMEText(content, _subtype=subtype, _charset='utf-8')

attach.attach(txt)

if self.subject is not None:

#主題,最上面的一行

attach["Subject"] = self.subject

if self.user is not None:

#顯示在發件人

attach["From"] = "Data Team" % self.user

if self.to_list:

attach["To"] = ";".join(self.to_list)#收件人列表

if self.cc_list:

attach["Cc"] = ";".join(self.cc_list)#抄送列表

if self.doc:

#估計任何文件都可以用base64，比如rar等

#文件名漢字用gbk編碼代替

name = os.path.basename(self.doc).encode("gbk")

f = open(self.doc,"rb")

doc = MIMEText(f.read(), "base64", "gb2312")

doc["Content-Type"] = 'application/octet-stream'

doc["Content-Disposition"] = 'attachment; filename="' + name +'"'

attach.attach(doc)

f.close()

return attach.as_string()

def get_jobs_descript(self):

"""

獲取任務的執行str

:return:

"""

html =''

html += ''

html += '

'

html += ''

html += '

調度報告'

html += ''

html +=''

html +='

'

till = str(datetime.datetime.now() - self.start_time)

till = till[0:till.index(".")] #精確到秒級小數點，然后要自己弄好，

html +='

啟動時間:[' + self.start_time.strftime("%Y-%m-%d %H:%M:%S") +'] 結束時間:[' + self.get_end_time_str() +'] 持續時間:'+till+'

'

ss = 0

er = 0

skip = 0

run_c = 0

pl = ''

for job_name in self.job_run_:

node = self.dag_job.get_node(job_name=job_name)

if node.status == 1:

if node.is_active == False:

skip += 1

else:

ss += 1

elif node.status == -1:

er += 1

elif node.status == 0:

run_c += 1

pl += node.get_job_mail_str()

html +="

統計 [成功:" + str(ss) +'\t錯誤:' + str(er)+'\t跳過:'+str(skip)+'\t未執行:' + str(self.job_size - ss - er - skip)+'] 總任務數: ' + str(self.job_size)+'

'

html +='

郵件發送時間:' + str(datetime.datetime.now().strftime('%Y-%m-%d %H:%M')) +'

'

html += '

調度報告

'

html +='

html +=''

html +='

'

html +='

狀態'

html +='

任務名'

html +='

創建時間'

html +='

結束時間'

html +='

持續時間'

html +='

重試次數'

html +='

重試上限'

html +='

是否跳過'

html +='

一定正確'

html +='

優先級'

html +='

'

html +='

'

html +='

'

html += pl

html +='

'

return html

def get_job_mail_str(self):

"""

獲取發送郵件的行數

:return:

"""

sz = "

"

if self.start_time is None:

sz += "

SKIP" + self.name +''

else:

if self.status == 1:

sz += "

SUCCEED" + self.name+''

elif self.status ==-1:

sz += "

ERROR" + self.name+''

elif self.status ==0:

sz += "

RUNNING" + self.name+''

elif self.status is None:

sz += "

NOT_RUN" + self.name+''

if self.start_time is not None:

sz +='

' + self.start_time.strftime('%Y-%m-%d %H:%M:%S')+''

else:

sz +='

'

if self.end_time is not None:

sz +='

'+self.end_time.strftime('%Y-%m-%d %H:%M:%S') +''

else:

sz +='

'

if self.start_time is None or self.end_time is None:

sz +='

'

else:

plz = str(self.end_time - self.start_time)

if plz.index(':') == 1:

sz += '

0' + str(self.end_time - self.start_time)[0:7] +''

else:

sz += '

' + str(self.end_time - self.start_time)[0:8] +''

sz += '

' + str(self.retry_time) + ''

sz += '

' + str(self.retry_limit) + ''

if self.is_active:

sz +='

'

else:

sz += '

' + str(self.is_active == False) +''

if self.forver_right:

sz += '

' + str(self.forver_right == True) +''

else:

sz += '

'

sz += '

' + str(self.priority) +''

sz += ''

return sz

總結

以上是生活随笔為你收集整理的python的email模块_python email 模块的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。