Python 每日定时查询数据库生成Excel报表，并群发邮件

最近在做游戏打点数据的一些统计处理，写了个Python脚本完成每日定时自动查询生成Excel报表并群发邮件的小功能。

拆解几个需求点：

一.连接数据库并查询

以下是一个查询总注册人数的示例代码，host port db user password等填入自己数据库的参数。

mysql就不多说了，根据统计需求自己撸吧

import pymysqlhost = 'xxxx'
port = xxx
db = 'xxx'
user = 'xxx'
password = 'xxx'# ---- 用pymysql 操作数据库
def get_connection():conn = pymysql.connect(host=host, port=port, db=db, user=user, password=password)return connconn = get_connection()# 使用 cursor() 方法创建一个 dict 格式的游标对象 cursor
cursor = conn.cursor(pymysql.cursors.DictCursor)#总注册
cursor.execute("select count(distinct userId) as total from Statistics where ops = 'register' ")
data = cursor.fetchone()
allTotalRegisterNum = data['total']
print("-- 总注册人数: %s " % (data['total']))

二.查询的数据生成Excel报表

以下为统计关卡数据并生成Excel的小例子，startPersonNum，finishPersonNum，ratio都是之前查库处理好的数据，生成的Excel名称为地区+数据统计+日期，region为统计的地区，yesterday为昨天日期

import xlwt
workbook = xlwt.Workbook(encoding = 'ascii')
worksheet = workbook.add_sheet('关卡数据')#关卡统计数据
print("-- 开始关卡数据统计 -- ")
worksheet.col(1).width = 3000
worksheet.col(2).width = 3000
worksheet.col(3).width = 3000
worksheet.write(0, 1, "开始关卡人数") 
worksheet.write(0, 2, "结束关卡人数") 
worksheet.write(0, 3, "单关完成率") 
worksheet.write(i, 1, startPersonNum) 
worksheet.write(i, 2, finishPersonNum) 
worksheet.write(i, 3, ratio) workbook.save("%s_数据统计_%s.xls" % (region, str(yesterday))) # 保存文件

三.发送邮件

sender 发送发，receivers 接收方数组， mail_host 邮件服务器，mail_user="xxx" #用户名
mail_pass="xxx" #口令

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.header import Header
from email.mime.base import MIMEBase
from email import encoderssender = 'xxx.com'
receivers = ['xxx.com'，'xxx2.com']  # 接收邮件，可设置为你的QQ邮箱或者其他邮箱def SendEmail(): # 今天日期today = datetime.date.today() # 昨天时间yesterday = today - datetime.timedelta(days=1)for receiver in receivers:#创建一个带附件的实例message = MIMEMultipart()message['From'] = sendermessage["To"] = receiversubject = '%s数据统计'%(str(yesterday))message['Subject'] = '%s数据统计'%(str(yesterday))#邮件正文内容message.attach(MIMEText('%s数据统计'%(str(yesterday)), 'plain', 'utf-8'))for channel in channels:if(os.path.exists("%s_数据统计_%s.xls"%(region, str(yesterday)))):        att1 = MIMEBase('application', "octet-stream")att1.set_payload(open("%s_数据统计_%s.xls"%(channel, str(yesterday)),'rb').read())encoders.encode_base64(att1)att1.add_header('Content-Disposition', 'attachment; filename="%s-%s.xls"'%(channel, str(yesterday)))message.attach(att1)try:# 第三方 SMTP 服务mail_host="smtp.yeah.net"  #设置服务器mail_user="xxx"    #用户名mail_pass="xxx"   #口令 smtpObj = smtplib.SMTP()smtpObj.connect(mail_host,25)    # 25 为 SMTP 端口号smtpObj.login(mail_user, mail_pass)  smtpObj.sendmail(sender, [receiver], message.as_string())print("邮件发送成功")except smtplib.SMTPException as e:print("Error: 无法发送邮件%s"% e)

四.定时功能

以下为每日1点启动发送邮件函数的例子

from apscheduler.schedulers.blocking import BlockingSchedulerdef job():SendEmail()def start():# BlockingSchedulerscheduler = BlockingScheduler()scheduler.add_job(job, 'cron', day_of_week='0-6', hour=1, minute=00)scheduler.start()if __name__ == '__main__':start()