当前位置：首页 > news >正文

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

news 2025/6/28 9:35:03

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 = %sORDER 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_codedate,  # 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  # MEETINGworksheet.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 = 20print(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')