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

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> Python知识库 -> Python 每日定时查询数据库生成Excel报表，并群发邮件 -> 正文阅读

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

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

拆解几个需求点：

一.连接数据库并查询

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

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

import pymysql

host = '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 conn

conn = 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 encoders


sender = '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'] = sender
        message["To"] = receiver
        subject = '%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 BlockingScheduler

def job():
    SendEmail()

def start():
    # BlockingScheduler
    scheduler = BlockingScheduler()
    scheduler.add_job(job, 'cron', day_of_week='0-6', hour=1, minute=00)
    scheduler.start()

if __name__ == '__main__':
    start()