本文实例讲述了Python实现的查询mysql数据库并通过邮件发送信息功能。分享给大家供大家参考，具体如下：
这里使用Python查询mysql数据库，并通过邮件发送宕机信息。
Python代码如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
|
#-*- coding: UTF-8 -*-#!/usr/bin/env python'''''author:qlzhongCreated on 2015-6-29征途宕机日志统计汇总'''import MySQLdbimport timeimport datetimeimport smtplibfrom email.mime.text import MIMETextmailto_list=["mail@mail.com"]#mailto_list=["zhongqilong@ztgame.com"]mail_host="smtp.qq.com" #设置服务器mail_user="" #用户名mail_pass="" #口令mail_postfix="" #发件箱的后缀def send_mail(to_list,sub,content): me="hello"+"<"+mail_user+"@"+mail_postfix+">" msg = MIMEText(content,_subtype='plain',_charset='utf-8') msg['Subject'] = sub msg['From'] = me msg['To'] = ";".join(to_list) try: server = smtplib.SMTP() server.connect(mail_host) server.login(mail_user,mail_pass) server.sendmail(me, to_list, msg.as_string()) server.close() return True except Exception, e: print str(e) return Falseclass MySQLHelper: #配置数据库信息并连接 def __init__(self,host="****",user="****",password="****",port=3306,charset="utf8"): self.host=host self.user=user self.password=password self.port=port self.charset=charset try: self.conn=MySQLdb.connect(host=self.host,user=self.user,passwd=self.password,port=self.port) self.conn.set_character_set(self.charset) self.cur=self.conn.cursor() print("==================connect success====================") except MySQLdb.Error as e: print("Mysql Error %d: %s" % (e.args[0], e.args[1])) #取出需要统计的数据库名称 def db_name(self): un_db_name = ['information_schema','cz','ecshop','edutone','gz','mysql','newparent','parent','performance_schema','test','xx','yyhd'] name = [] try: self.cur.execute('show databases') for row in self.cur.fetchall(): for i in row: if i not in un_db_name: name.append(i) return name except MySQLdb.Error as e: print("Mysql Error %d: %s" % (e.args[0], e.args[1])) #指定查询的数据库名称 def selectDb(self,db): try: self.conn.select_db(db) except MySQLdb.Error as e: print("Mysql Error %d: %s" % (e.args[0], e.args[1])) #使用该语句来直接查询昨天和今天的差异 def monion_today_yesddiff(self, today, yestoday): try: strresult = "" strsql = 'SELECT address, charversion, sum(today) as today, sum(yesterday) as yesterday ' strsql += 'FROM (SELECT address, "" as today, tmp as yesterday, charversion FROM ( SELECT count(*) As tmp, address, charversion From `' + yestoday strsql += '` WHERE charversion like \'1.0.0.3%\' GROUP BY address) As TEST WHERE tmp>=50 ' strsql += ' union all ' strsql += 'SELECT address, tmp as today, "" as yesterday, charversion FROM (SELECT count(*) As tmp, address, charversion From `' strsql += today strsql += '` WHERE charversion like \'1.0.0.3%\' GROUP BY address) As TEST WHERE tmp>=50 ) As Diff GROUP BY address, charversion' print(strsql + "\n") self.cur.execute(strsql) name_list = [tuple[0] for tuple in self.cur.description] strresult += str(name_list) + "\n" # for row in self.cur.fetchall(): # return row s = self.cur.fetchall() todaynum = 0 yestodaynum = 0 for col in s: strresult += str(col[0]) + " " + str(col[1]) + " " + str(col[2]) + " " + str(col[3]) + "\n" todaynum += int(col[2]) yestodaynum += int(col[3]) strresult += "今日宕机总数：" + str(todaynum) + " 昨日宕机总数：" + str(yestodaynum) + " 同昨日相比增加: " + str(todaynum - yestodaynum) + "\n" return strresult except MySQLdb.Error as e: print("Mysql Error:%s\n" %(e)) def close(self): self.cur.close() self.conn.close() todayrang = 0; yestodayrang = 0; #按照范围查询 def monion_rang_today_yesddiff(self, today, yestoday, num1, num2): try: strresult = "" strsql = 'SELECT sum(today) as today, sum(yesterday) as yesterday FROM (SELECT "" as today, tmp as yesterday FROM ( SELECT count(*) As tmp From `' + yestoday strsql += '` WHERE charversion like \'1.0.0.3%\' GROUP BY address) As TEST WHERE tmp<' + str(num2) + ' AND tmp>=' + str(num1) + ' union all ' strsql += 'SELECT tmp as today, "" as yesterday FROM (SELECT count(*) As tmp From `' + today + '` WHERE charversion like \'1.0.0.3%\' GROUP BY address) As TEST WHERE tmp<' + str(num2) + ' AND tmp>=' + str(num1) + ' ) As Diff' print(strsql + "\n") self.cur.execute(strsql) name_list = [tuple[0] for tuple in self.cur.description] #strresult += str(name_list) + "\n" # for row in self.cur.fetchall(): # return row s = self.cur.fetchall() todaynum = 0 yestodaynum = 0 for col in s: strresult += str(num1) + " <= tmp < " + str(num2) + " " + str(col[0]) + " " + str(col[1]) + "\n" self.todayrang += int(col[0]) self.yestodayrang += int(col[1]) return strresult except MySQLdb.Error as e: print("Mysql Error:%s\n" %(e)) def close(self): self.cur.close() self.conn.close() #宕机数地址50以下最多的版本 def monion_rang_today_diff(self, today, num): try: strresult = "" strsql = 'SELECT charversion, sum(today) as today FROM (SELECT tmp as today, "" as yesterday, charversion FROM (SELECT count(*) As tmp, charversion From `' + today strsql += '` WHERE charversion like \'1.0.0.3%\' GROUP BY address) As TEST WHERE tmp< ' + str(num) + ') As Diff GROUP BY charversion' print(strsql + "\n") self.cur.execute(strsql) name_list = [tuple[0] for tuple in self.cur.description] #strresult += str(name_list) + "\n" # for row in self.cur.fetchall(): # return row s = self.cur.fetchall() for col in s: strresult += str(col[0]) + " " + str(col[1]) + "\n" return strresult except MySQLdb.Error as e: print("Mysql Error:%s\n" %(e)) def close(self): self.cur.close() self.conn.close()if __name__ == '__main__': textbody="" textbody = textbody + "征途宕机日志查询汇总" + "\n" #时间 timenow = datetime.datetime.now() textbody = textbody + "时间：" + timenow.strftime('%Y-%m-%d %H:%M:%S') + "\n" #连接 ipadress="192.168.100.38" port=3306 dbHelper = MySQLHelper(ipadress, "gameerror", "errorpasswd", port) textbody = textbody + "服务器地址：" + ipadress + ":" + str(port) + "\n" dbHelper.selectDb("GAMEERROR") #操作 dbname = dbHelper.db_name() textbody = textbody + "数据库：" + str(dbname[0]) + "\n" time1 = timenow + datetime.timedelta(days = -1) time2 = timenow + datetime.timedelta(days = -2) strtime1 = time1.strftime('%Y%m%d') tabletoday = "ErrorDump" + strtime1 strtime2 = time2.strftime('%Y%m%d') tableyestoday = "ErrorDump" + strtime2 textbody = textbody + "table name: today: " + tabletoday + " yestoday: " + tableyestoday + "\n" textbody = textbody + "\n昨天和今天的差异 宕机地址 版本号 今天宕机次数 昨天宕机次数" + "\n" textbody = textbody + str(dbHelper.monion_today_yesddiff(tabletoday, tableyestoday)) + "\n" textbody = textbody + "50以下地址(tmp代表某个宕机地址的个数) 今天 昨天:" + "\n" textbody = textbody + str(dbHelper.monion_rang_today_yesddiff(tabletoday, tableyestoday, 30, 50)) textbody = textbody + str(dbHelper.monion_rang_today_yesddiff(tabletoday, tableyestoday, 10, 30)) textbody = textbody + str(dbHelper.monion_rang_today_yesddiff(tabletoday, tableyestoday, 0, 10)) textbody = textbody + "50以上地址总和" + " 今天： " + str(dbHelper.todayrang) + " 昨天: " + str(dbHelper.yestodayrang) + " 今天比昨天增加: " + str(dbHelper.todayrang - dbHelper.yestodayrang) + "\n" num=50 textbody = textbody + "\n宕机数地址" + str(num) + "以下最多的版本 版本号 次数" + "\n" textbody = textbody + str(dbHelper.monion_rang_today_diff(tabletoday, num)) file_object = open('ztdumptip.txt') try: all_the_text = file_object.read() finally: file_object.close() textbody += all_the_text print(textbody) if send_mail(mailto_list,"征途客户端宕机日志统计",textbody): print "发送成功" else: print "发送失败" dbHelper.close() |
希望本文所述对大家Python程序设计有所帮助。
原文链接：https://blog.csdn.net/jesse__zhong/article/details/79696171
