直接上源代码：
#!/usr/bin/env python3 # -*- coding:utf-8 -*- """ Created on : 2020/7/20 12:38 @Author : Miracle @blog : https://blog.csdn.net/weixin_39633383 @github: https://github.com/Mr-Miracle """ import os import datetime from mailbox import Message, Mailbox import pandas as pd import pypyodbc as mdb import wx import time from exchangelib import * from exchangelib.protocol import BaseProtocol, NoVerifyHTTPAdapter import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) BaseProtocol.HTTP_ADAPTER_CLS = NoVerifyHTTPAdapter # 读取MDB文件内容 def read_mdb(file): # 连接mdb文件 conn_str = (r'Driver={Microsoft Access Driver (*.mdb)};DBQ='+file+';' ) conn = mdb.win_connect_mdb(conn_str) # 创建游标 port_info = [] cur = conn.cursor() sql_str = "SELECT (SELECT FCODE FROM BaseInFo) AS PORTCODE,FDATE,(IIF ((SELECT COUNT(1) FROM FCWVCH WHERE FKMH IS NULL)=0, 0, 1)) AS RECORDS FROM GZB GROUP BY FDATE;" cur.execute(sql_str) gzb = cur.fetchall() for i in gzb: str1 = str(i[1])[0:10].replace('-', '') # 对日期的格式进行调整 port_info.append([i[0], str1, i[2]]) cur.close() conn.close() return port_info # noinspection PyBroadException def send_email(to, subject, body): """ 电子邮件发送 :param to: 收件人邮箱 :param subject: 邮件主题 :param body: 邮件正文 :return:""" cred = Credentials( username="###", # 创建、申明邮箱账户 password="###", ) config = Configuration( server="###", # 发件服务器地址 credentials=cred, auth_type=NTLM, ) account = Account( primary_smtp_address="###", config=config, autodiscover=False, access_type=DELEGATE, ) message = Message( account=account, # 发件人、申明的账户 subject=subject, # 邮件主题 body=HTMLBody(body), # 邮件的内容 to_recipients=[Mailbox(email_address=to), ], # 收件人 ) try: message.send_and_save() print('邮件发送成功！') except: print('邮件发送失败！') # 定义用户窗体的类 class AppFrame(wx.Frame): def __init__(self, parent, title): super(AppFrame, self).__init__(parent, title=title, size=(500, 480)) panel = wx.Panel(self) # 定义用户的文本输入框等组件 self.text0 = wx.StaticText(panel, -1, label="MDB根目录:") # id = -1，表示窗口id自动生成 self.text1 = wx.StaticText(panel, -1, label="最新文件个数:") self.text2 = wx.StaticText(panel, -1, label="托管行名称:") self.text3 = wx.StaticText(panel, -1, label="年金计划:") self.input0 = wx.TextCtrl(panel, -1, 'F:\TGH') self.input1 = wx.TextCtrl(panel, -1, '7') self.banks = ['所有托管', '交通银行', '浦发银行', '招商银行', '建设银行', '农业银行', '工商银行'] self.input2 = wx.ComboBox(panel, -1, choices=self.banks, value=self.banks[0]) self.input3 = wx.TextCtrl(panel, -1, '江西 山东 中央 湖南 安徽 新疆维吾尔 黑龙江') self.execute_button = wx.Button(panel, -1, label="执行") self.exit_button = wx.Button(panel, -1, label="退出") # 创建文本域 self.multiText = wx.TextCtrl(panel, -1, size=(200, 100), style=wx.TE_MULTILINE | wx.TE_READONLY) # 创建一个文本控件，并设置只读 self.multiText.SetInsertionPoint(0) # 设置插入点 # and a status bar self.CreateStatusBar() self.SetStatusText("Created by Miracle on 2020.10") # 定义各个组件的页面布局 v_box = wx.BoxSizer(wx.VERTICAL) # 实例化一个垂直盒子 h_box = wx.BoxSizer(wx.HORIZONTAL) # 实例化一个水平盒子 sizer = wx.FlexGridSizer(5, 2, 15, 30) # 设置5行2列，垂直间距15像素，水平间距20像素 sizer.AddMany([[self.text0, wx.ALIGN_LEFT], (self.input0, wx.EXPAND), (self.text1, wx.ALIGN_LEFT), (self.input1, wx.EXPAND), (self.text2, wx.ALIGN_LEFT), (self.input2, wx.EXPAND), (self.text3, wx.ALIGN_LEFT), (self.input3, wx.EXPAND), (self.execute_button, wx.Center), (self.exit_button, wx.Center), ]) h_box.Add(sizer, proportion=1, flag=wx.ALL | wx.EXPAND, border=5) # 将这组件插入水平盒子 v_box.Add(h_box, 0, wx.ALL | wx.CENTER, 5) # 在垂直盒子里添加水平盒子 v_box.Add(self.multiText, 1, wx.ALL | wx.EXPAND, 5) # 在垂直盒子里添加文本域 self.SetSizer(v_box) # 启用所有布局 self.Show() # 显示用户页面的窗体 self.Bind(wx.EVT_BUTTON, self.execute_event, self.execute_button) # 定义按钮的点击事件 self.Bind(wx.EVT_BUTTON, self.exit_event, self.exit_button) # 定义按钮的点击事件 def execute_event(self, event): # 年金计划组合信息 ports = [###] ## 此部分为计划组合信息，数据安全原因，未列示 # 获取用户的输入值 root = self.input0.GetValue() num = self.input1.GetValue() annuity_bank = self.input2.GetValue() annuity_plan = self.input3.GetValue() # 判断托管行根目录路径是否存在，否则不执行 if os.path.exists(root): check_plan0 = [] check_plan1 = [] for port in ports: for s in port: if annuity_plan: for i in annuity_plan.split(" "): if i in s: check_plan0.append(port) else: # 否则查询所有计划 check_plan0 = ports if annuity_bank == '所有托管': check_plan1 = ports else: # 否则查询所有计划 for j in annuity_bank.split(" "): if j in s: check_plan1.append(port) # 将2个查询条件的结果取交集，生成需要进行查询的计划组合list check_plan = [val for val in check_plan0 if val in check_plan1] result_data = [] for plan in check_plan: self.write_log_to_text("开始检查："+','.join(plan)) data = self.mdb_check(root, plan, int(num)) # 取到MDB的检查结果列表 result_data.append(data) # 对list进行表格化的处理 msg_list = pd.DataFrame(result_data) msg_list.columns = ['托管行', '年金计划', '组合名称', '组合代码', '日期', '到达情况', '检查结果'] msg = msg_list.to_html(bold_rows=True) # 发送异常的提醒邮件 send_email('###', '周度数据MDB文件检查_异常数据' +self.get_now_time(), msg) self.write_log_to_text("异常提醒邮件发送完成！") else: wx.MessageBox("托管行MDB文件的目录不存在!\n"+ root, "输入有误", wx.OK | wx.YES_DEFAULT) event.Skip() # MDB文件检查的主要逻辑 def mdb_check(self, mdb_root, port, num): """ :param mdb_root: MDB文件的根目录-/TGH :param port: 需要检查的组合的基本信息的list :param num: 检查的最新文件的个数 :return: 检查结果list """ check_list = [] # 程序的返回值，异常数据的list file_load = os.path.join(mdb_root, port[1], port[5]) mdb_list = [] for root, sub, files in os.walk(file_load): for file in files: if file.endswith(".MDB"): mdb_file = os.path.join(root, file) mdb_list.append(mdb_file) mdb_list.sort(reverse=True) # 如果文件夹下有mdb文件 day_result = [] if len(mdb_list) > 0: for i in range(num): # 取前num个文件来进行数据读取 print(datetime.datetime.now(), '读取MDB文件：', mdb_list[i]) self.write_log_to_text('读取MDB文件：' + mdb_list[i]) result = read_mdb(mdb_list[i]) for result in result: day_result.append((result[1], result[2])) print(datetime.datetime.now(), '取到的数据：\n', day_result) else: print(datetime.datetime.now(), '取到的数据： 找不到此组合的MDB文件!!!') self.write_log_to_text('取到的数据： 找不到此组合的MDB文件!!!') for i in self.get_last_week(): if (i, 0) in day_result: # 当且仅当估值表中取到的日期和上周日期一致，估值表的科目字段不为空时显示正常 check_list = [port[0], str(port[2])[0:2], port[4], port[5], i, '已到达 ', '正常'] print(check_list) self.write_log_to_text(' '.join(check_list)) elif (i, 1) in day_result: check_list = [port[0], str(port[2])[0:2], port[4], port[5], i, '已到达 ', '异常'] print(check_list) self.write_log_to_text(' '.join(check_list)) else: check_list = [port[0], str(port[2])[0:2], port[4], port[5], i, '未到达 ', '异常'] print(check_list) self.write_log_to_text(' '.join(check_list)) return check_list def exit_event(self, event): """Close the frame, terminating the application.""" self.Close(True) event.Skip() # 获取当前时间 @staticmethod def get_now_time(): current_time = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())) return current_time # 日志动态打印 def write_log_to_text(self, log_msg): current_time = self.get_now_time() log_msg_in = "INFO:" + str(current_time) + " " + log_msg + "\n" # 换行 self.multiText.AppendText(log_msg_in) # 取上周7天的数 @staticmethod def get_last_week(): now = datetime.datetime.now() day_list = [] for i in range(7): time_stamp = now - datetime.timedelta(days=now.weekday() + (7 - i)) str_day = time_stamp.strftime("%Y%m%d") day_list.append(str_day) return day_list if __name__ == '__main__': app = wx.App() # 创建应用程序对象 frame = AppFrame(None, title="周度MDB文件数据检查_v1.0") frame.Center() # 设置主窗口居中对齐 app.MainLoop() 项目效果图：