一个py文件搞定mysql查询+Json转换+表数据提取+根据数据条件生成excel文件+打包运行一条龙

import os import argparse import pymssql import json import pandas as pd from datetime import datetime from pandas.io.formats.excel import ExcelFormatter import openpyxl # 投注类型映射字典 BET_MAPPING = { 1: 'WIN', 2: 'PLA', 3: 'QIN', 4: 'QPL', 5: 'DBL', 6: 'TCE', 7: 'QTT', 8: 'D-Q', 9: 'TBL', 10: 'T-T', 11: '6UP', 12: 'D-T', 13: 'TRI', 14: 'FCT', 17: 'F-F' } Meetingloc = {'ST':1, 'HV':2, 'S1':5, 'S2':6, 'S3':7, 'S4':8, 'S5':9} def convert_pool_bitmap(pool_list): """将数字列表转换为投注类型""" return [BET_MAPPING.get(num, f'未知({num})') for num in pool_list] def convert_meeting_loc(venue_list): """将数字列表转换为 meeting 类型""" return Meetingloc.get(venue_list) def mains(date): # 数据库配置（需要根据实际情况修改） server = 'your server' user = 'your user' password = 'your pwd' database = 'your DB name' port = 'your port' # 准备数据容器 output_data = [] try: # 建立数据库连接 conn = pymssql.connect( server=server, user=user, password=password, database=database, port=port ) sql = """ SELECT * FROM wc2wca_col_log WHERE msg_details LIKE %s AND msg_code = %s AND biz_date = %s AND msg_type = %s ORDER BY msg_time desc """ # 定义查询参数 for i in Meetingloc.keys(): like_pattern = '%"meetingID":{"meetingLoc":' + str(Meetingloc.get(i)) + ',"meetingDayCode":[1-9]},"raceNo":1%' # like_pattern = '%"meetingID":{"meetingLoc":8,"meetingDayCode":[1-9]},"raceNo":1%' params = ( like_pattern, # LIKE参数 '1120', # msg_code date, # biz_date '1' # msg_type ) # 创建游标执行查询 with conn.cursor(as_dict=True) as cursor: cursor.execute(sql,params) for row in cursor: try: # 解析JSON数据 json_data = json.loads(row['msg_details']) # 提取目标字段 race_no = json_data['Data']['raceNo'] pool_bitmap = json_data['Data']['poolBitmap'] msg_time = row['msg_time'] # 转换投注类型 converted_pool = convert_pool_bitmap(pool_bitmap) # 存储到数据集 output_data.append({ 'meeting':i, 'time': msg_time.strftime('%Y-%m-%d %H:%M:%S.%f'), # 格式化时间 'interval': '', 'raceNo': race_no, 'pool': ', '.join(converted_pool) # 列表转字符串 }) except (KeyError, json.JSONDecodeError) as e: print(f"数据处理异常: {str(e)}") continue # 数据处理： 将 S1 S2 S3 S4 S5 S6 ST HV 数据分组 if output_data: groups = {} for item in output_data: meeting = item.get('meeting') if meeting in Meetingloc.key(): if meeting not in groups: groups[meeting] = [] groups[meeting].append(item) # 自定义输出路径 output_dir = "./reports" # 自定义输出路径 os.makedirs(output_dir, exist_ok=True) filename = f"RaceData_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx" # 生成文件名（带时间戳） filename = os.path.join(output_dir, filename) # 创建Excel文件 with pd.ExcelWriter(filename, engine='openpyxl') as writer: for meeting_name, records in groups.items(): df = pd.DataFrame(records, columns=['meeting','time','interval', 'raceNo', 'pool']) #指定顺序 df = df[['meeting','time','interval', 'RaceNo', 'Pool']] df["time"] = pd.to_datetime(df["time"]) # 转换为时间类型 # df = df.sort_values(by="time").reset_index(drop=True) # 计算时间差（秒） df["Next_Time"] = df["time"].shift(-1) df["interval"] = (df["Next_Time"] - df["time"]).dt.total_seconds() df.drop("Next_Time", axis=1, inplace=True) # 处理最后一条记录的NaN值 df["interval"] = df["interval"].fillna(0) df.to_excel( writer, sheet_name=meeting_name+str(date), index=False, startrow=0, startcol=0) # 获取工作表对象writer # worksheet = writer.sheets.get(meeting_name) worksheet = writer.sheets[meeting_name+str(date)] # 设置列宽（单位：字符） worksheet.column_dimensions['A'].width = 10 # MEETING worksheet.column_dimensions['B'].width = 30 # 时间列 worksheet.column_dimensions['C'].width = 10 # 间隔 worksheet.column_dimensions['D'].width = 10 # RaceNo列 worksheet.column_dimensions['E'].width = 60 # Pool类型列 # 设置标题行样式 header_style = { 'font': {'bold': True, 'color': 'FFFFFF'}, 'fill': {'fill_type': 'solid', 'start_color': '4F81BD'}, 'alignment': {'horizontal': 'center'} } for cell in worksheet[1]: # 第一行为标题行 cell.font = openpyxl.styles.Font(**header_style['font']) cell.fill = openpyxl.styles.PatternFill(**header_style['fill']) cell.alignment = openpyxl.styles.Alignment(**header_style['alignment']) # 自动调整行高 for row in worksheet.iter_rows(): for cell in row: worksheet.row_dimensions[cell.row].height = 20 print(f"数据已成功导出到 {filename}") except pymssql.Error as e: print(f"数据库错误: {str(e)}") finally: if conn: conn.close() def process_date(date_str): """ 处理日期参数的函数 :param date_str: 必传参数（格式示例：20240325） """ # 验证参数格式（可选） if len(date_str) != 8 or not date_str.isdigit(): raise ValueError("日期格式应为8位数字，例如：20240325") # 示例处理结果 year = date_str[:4] month = date_str[4:6] day = date_str[6:] print(f"解析结果：{year}年{month}月{day}日") # 这里添加你的业务逻辑 print(f"成功接收日期参数：{date_str}") print("正在处理...") # 这里添加你的业务逻辑 mains(date_str) def main(): # [!] 关键点：确保这里没有参数 # 创建命令行参数解析器 parser = argparse.ArgumentParser(description='日期处理程序') parser.add_argument('date', type=str, help='必传的日期参数（8位数字，示例：20240325）') args = parser.parse_args() process_date(args.date) if __name__ == "__main__": process_date('20240328')

最后运行：

一、cmd直接运行脚本测试：

python date_app.py 20240325

二、打包：打包问题：‘“indexerror: tuple index out of range” 可以参考连接 , 一般问题可以参考连接

pyinstaller --onefile date_app.py

三、生成dist文件后，会生成exe文件 在dist文件夹里面新建reports文件夹：报告文件夹

四、测试打包后的程序 1、新建start.bat空文件 2、放入代码：

start PMUCOL_LOG.exe 20240328

3、运行start.bat

五、生成报告：.\reports\xxxx.xlsx

备注：至于生成Excel文件的代码，想看数据格式的。可以参考