Python 如何提取郵件內容

歡迎關注【無量測試之道】公眾號，回復【領取資源】,
Python編程學習資源干貨、
Python+Appium框架APP的UI自動化、
Python+Selenium框架Web的UI自動化、
Python+Unittest框架API自動化、

資源和代碼 免費送啦~
文章下方有公眾號二維碼，可直接微信掃一掃關注即可。

今天分享的文章主要講解如何從郵件里面提取用戶返回的線上問題內容並做解析，通過拿到的數據信息進行分析整理，然后進行封裝請求禪道里的接口進行提交，提交請求過程中會對數據庫中是否存在進行一次判斷處理，如果沒有存在的就提交，如果數據庫中存在就不用再提交，基於這個思路來看下今天的分享。

基礎信息准備

import imaplib, email,re,requests,time,pymysql
imapserver = 'smtp.office365.com'
emailuser = "[e-mail]"
emailpasswd = "test123"
 
#beta環境禪道地址
beta_loginhost="http://zen.beta.com/index.php?m=user&f=login"
beta_add_bughost="http://zen.beta.com/index.php?m=bug&f=create&productID=10&branch=0&extra=moduleID=0"
 
#live環境禪道地址
live_loginhost="https://zen.live.com/index.php?m=user&f=login"
live_add_bughost="https://zen.live.com/index.php?m=bug&f=create&productID=10&branch=0&extra=moduleID=0"
 
envs="live" #定義使用的環境

 

數據庫連接信息

 
#連接數據庫相關的信息：
beta_dicts={
 "HOST" : '10.8.2.3',
 "PORT" : 3306,
 "USER": 'zentao',
 "PASSWORD" : 'test123',
 "NAME":"zentao"
}
live_dicts={
 "HOST" : '10.7.1.7',
 "PORT" : 3306,
 "USER": 'zentao',
 "PASSWORD" : 'test123',
 "NAME":"zentao"
}

 

數據庫查詢

 
#數據庫查詢操作
def executesql(query,envs):
 try:
 if(envs=="beta"):
 conn = pymysql.connect(beta_dicts['HOST'], beta_dicts['USER'], beta_dicts['PASSWORD'], beta_dicts['NAME'], int(beta_dicts['PORT']),charset='utf8')
 print(beta_dicts)
 else:
 conn = pymysql.connect(live_dicts['HOST'], live_dicts['USER'], live_dicts['PASSWORD'], live_dicts['NAME'], int(live_dicts['PORT']),charset='utf8')
 print(live_dicts)
 cursor = conn.cursor()
 cursor.execute(query)
 result =cursor.fetchall()
 print("execute successfully!!!")
 if(len(result)==0):
 return 0
 else:
 return result[0][0]
 except Exception as e:
 print(e)
 print("execute failed")
 finally:
 cursor.close()
 conn.close()

 

建立連接與檢索

 
#建立連接與檢索匹配的郵件
def search():
 print("start to connect")
 conn = imaplib.IMAP4_SSL(imapserver)
 conn.login(emailuser, emailpasswd)
 conn.select('INBOX') # 選擇收件箱（默認）
 print(conn)
 now = time.localtime()
 nowt = time.strftime("%d-%b-%Y", now)
 print(nowt)
 results , data = conn.search(None,'(FROM "Liang.Wu")','(ON "'+str(nowt)+'")')
 mailidlist = data[0].split()
 print(mailidlist)
 try:
 for id in mailidlist:
 print(id)
 resultss, data = conn.fetch(id, '(RFC822)') # 通過郵件id獲取郵件，data是fetch到的郵件具體內容
 e = email.message_from_bytes(data[0][1])

 

解釋說明與Print

'''
 Header()類：
 email.header.Header(s=None, charset=None, maxlinelen=None, header_name=None, continuation_ws=' ', errors='strict')
 其中參數的含義理解如下：
 s：標頭的值，也就是對應 From、To、Subject 的值； 
 charset：字符集格式，默認是 ASCII，但是一般指定 UTF-8 格式以兼容更多字符； 
 header_name:標頭名，就是 From、To、Subject、Time 等；
 '''
 subject = email.header.make_header(email.header.decode_header(e['SUBJECT']))
 mail_from = email.header.make_header(email.header.decode_header(e['From']))
 print("郵件的subject是%s" % subject)
 print("郵件的發件人是%s" % mail_from)
 body = str(get_body(e), encoding='ISO-8859-1') # utf-8 gb2312 GB18030解析中文日文英文
 print("郵件內容是%s" % body)
 parse1(body)
 print("good job")
 except Exception as e:
 print("we catch an error!!!",e)
 finally:
 print("logout is success")
 print("the finally of operation!!!")
 conn.logout()

 

獲取郵件主體信息

#獲取郵件主體信息
def get_body(msg):
 if msg.is_multipart ():#Return True if the message’s payload is a list of sub-Message objects, otherwise return False. When is_multipart() returns False, the payload should be a string object.
 return get_body(msg.get_payload(0))
 else:
 '''Return the current payload, which will be a list of Message objects when is_multipart() is True, 
 or a string when is_multipart() is False. If the payload is a list and you mutate the list object,
 you modify the message’s payload in place.'''
 return msg.get_payload(None , decode=True)

 

解析郵件內容並提交禪道

 
# 解析郵件內容並調用禪道提交（上一篇文章結合來看）
def parse1(body):
 pattern = re.compile('Dear Colleagues,<br>(.*?)Thanks and Regards,<br>', re.S)
 pattern1 = re.compile('black">(.*?)<o:p>', re.S)
 pattern2=re.compile(';">\r(.*?);\r<',re.S)
 lists = re.findall(pattern, body)
 print("*"*10)
 lists = str(lists[0]).replace("\n", "").split("<br>")
 print(lists)
 resultlist = []
 for i in range(len(lists)):
 if (len(lists[i]) > 1):
 resultlist.append(lists[i])
 print(resultlist)
 id = resultlist[1]
 ids=str(str(resultlist[1]).split(":")[1]).lstrip()
 Subject = resultlist[2]
 Subjects="[FeedBack-"+str(str(resultlist[1]).split(":")[1]).lstrip() + "]--"+str(str(resultlist[2]).split(":")[1])
 Creator = resultlist[3]
 Creators = str(str(resultlist[3]).split(":")[1])
 Category = resultlist[4]
 IssueCategory = resultlist[5]
 if ("Low" in resultlist[6]):
 Severity = "4"
 Severity_desc = "Severity: Low (Limited business impact)"
 if ("Medium" in resultlist[6]):
 Severity = "3"
 Severity_desc = "Severity: Medium (Functional but impact operations)"
 if ("High" in resultlist[6]):
 Severity = "2"
 Severity_desc = "Severity: High (Major system outage)"
 Module = resultlist[7]
 if('black">' in resultlist[8] and '<o:p>' in resultlist[8]):
 Details = str(re.findall(pattern1, resultlist[8])[0]).replace("&quot;", "\"")
 if(';">\r' in resultlist[8] and ';\r<' in resultlist[8]):
 Details = str(re.findall(pattern2, resultlist[8])[0]).replace("&quot;", "\"")
 link = resultlist[9]
 steps = id + "<br>" + Subject + "<br>" + Creator + "<br>" + Category + "<br>" + IssueCategory + "<br>" + Severity_desc + "<br>" + Module + "<br>" + Details + "<br>" + link
 print(steps.replace("<br>", "\n"))
 sql="SELECT * FROM zt_bug WHERE title LIKE \"[FeedBack-"+str(ids)+"%\""
 print(sql)
 if(executesql(sql,envs)>=1):
 print("there is an record exists!!!")
 #add_bug(Subjects, Creators, Severity, steps,envs)
 else:
 add_bug(Subjects,Creators,Severity,steps,envs)

 

提交bug至禪道

#提交bug到禪道的方法
def add_bug(a,b,c,d,e): #此方法可以與上一遍文章結合在一起提交到禪道
 pass

以上內容就是今天分享的全部內容，這個最后的方法也是空着的，所以這里也就回答了上一篇文章中大家提到的疑問—->自動提交bug到禪道的使用場景會是怎么樣的。

備注：我的個人公眾號已正式開通，致力於測試技術的分享，包含：大數據測試、功能測試，測試開發，API接口自動化、測試運維、UI自動化測試等，微信搜索公眾號：“無量測試之道”，或掃描下方二維碼：

 添加關注，讓我們一起共同成長！