import csv import os import tempfile import pymysql from pywebio.input import file_upload from pywebio.output import put_text, put_table, put_processbar, set_processbar, put_buttons from pywebio.session import download # MySQL数据库配置 DB_CONFIG = { 'host': 'localhost', 'user': 'testDB', 'password': 'WxMKtsNG8BRmWxDX', 'database': 'testdb', 'charset': 'utf8mb4', 'cursorclass': pymysql.cursors.DictCursor } class CSVProcessor: def __init__(self): try: print("正在连接数据库...") print(f"数据库配置信息: 主机={DB_CONFIG['host']}, 数据库={DB_CONFIG['database']}") # 首先尝试连接MySQL服务器（不指定数据库） try: print("尝试连接MySQL服务器...") config_without_db = DB_CONFIG.copy() del config_without_db['database'] print("connecting to MySQL server...") # 添加连接超时设置 config_without_db['connect_timeout'] = 10 temp_conn = pymysql.connect(**config_without_db) print("成功连接到MySQL服务器") temp_cursor = temp_conn.cursor() temp_cursor.close() temp_conn.close() print("成功连接MySQL服务器") except pymysql.Error as e: print(str(e)) raise e # 连接指定的数据库 self.conn = pymysql.connect(**DB_CONFIG) self.cursor = self.conn.cursor() print("数据库连接成功") except pymysql.Error as e: error_msg = f"数据库连接失败: {str(e)}" if "Can't connect to MySQL server" in str(e): error_msg = "\n错误：MySQL服务未运行\n解决方案：\n1. 以管理员身份打开命令提示符\n2. 输入命令 'net start MySQL80' 启动服务\n3. 如果启动失败，请检查MySQL服务是否正确安装" elif e.args[0] == 2003: error_msg += "\n解决方案：\n1. 检查MySQL服务是否启动\n2. 确认主机名是否正确（当前为localhost）\n3. 检查防火墙设置是否允许MySQL连接" elif e.args[0] == 1045: error_msg += "\n解决方案：\n1. 验证用户名和密码是否正确\n2. 确认该用户是否有权限访问数据库\n3. 如忘记密码，请联系数据库管理员重置" elif e.args[0] == 1049: error_msg += "\n解决方案：\n1. 确认数据库名称是否正确（当前为testdb）\n2. 使用以下SQL命令创建数据库：\n CREATE DATABASE testdb;" print(error_msg) put_text(error_msg) # 在Web界面显示错误信息 self.conn = None self.cursor = None def upload_csv(self): try: print("等待用户上传CSV文件...") upfile = file_upload("请选择CSV文件:", accept=".csv", multiple=False) if not upfile: put_text("请上传CSV文件") print("用户未选择文件") return file_content = upfile['content'] file_name = upfile['filename'] print(f"文件已上传: {file_name}") # 将上传的文件内容保存到临时文件 import tempfile import os temp_dir = tempfile.gettempdir() temp_file_path = os.path.join(temp_dir, file_name) with open(temp_file_path, 'wb') as f: f.write(file_content) self.process_csv(temp_file_path) except Exception as e: error_msg = f"文件上传错误: {str(e)}" put_text(error_msg) print(error_msg) def process_csv(self, file_name): try: print("开始处理CSV文件...") # 读取CSV文件 data = self.read_csv(file_name) # 显示结果 if data: success_msg = f"文件读取成功，共{len(data)}行数据" put_text("文件读取成功") put_table([['文件名', '行数'], [file_name, len(data)]]) print(success_msg) # 处理充电枪编码数据 if len(data) > 1: # 确保有数据行 self.process_gun_codes(data) except Exception as e: error_msg = f"数据处理错误: {str(e)}" put_text(error_msg) print(error_msg) def read_csv(self, filename): try: print(f"正在读取文件: {filename}") if not os.path.exists(filename): raise FileNotFoundError(f"文件不存在: {filename}") data = [] encodings = ['utf-8', 'gbk', 'gb2312', 'utf-16'] for encoding in encodings: try: with open(filename, 'r', encoding=encoding) as file: print(f"尝试使用 {encoding} 编码读取文件...") csv_reader = csv.reader(file) row_count = 0 data = [] # 读取第一行以验证CSV格式 header = next(csv_reader, None) if not header: raise ValueError("CSV文件为空") data.append(header) print(f"CSV文件头: {header}") for row in csv_reader: if not any(row): # 跳过空行 continue data.append(row) row_count += 1 if row_count % 100 == 0: print(f"已读取 {row_count} 行...") print(f"文件读取完成，使用 {encoding} 编码成功") return data except UnicodeDecodeError: print(f"{encoding} 编码读取失败，尝试其他编码...") continue except Exception as e: raise Exception(f"使用 {encoding} 编码读取时出错: {str(e)}") raise Exception("尝试所有编码都失败，无法读取文件") except Exception as e: error_msg = f"读取文件 {filename} 时出错: {str(e)}" put_text(error_msg) print(error_msg) return None def process_gun_codes(self, data): if not self.cursor: put_text("数据库未连接，无法处理数据") return try: # 假设CSV第一列是充电枪编码 gun_codes = [row[1] for row in data[1:]] # 跳过标题行 results = [] # 存储查询结果 total_codes = len(gun_codes) # 创建进度条 put_processbar('progress', auto_close=True) # 处理每个充电枪编码 for index, each_code in enumerate(gun_codes): sql_query = """ SELECT max(temp_count) AS count1 FROM ( SELECT a.实际支付, @count := IF(@met_one = 0 AND a.实际支付 = 0, @count + 1, @count) AS temp_count, @met_one := IF(a.实际支付 != 0 OR @met_one = 1, 1, 0) AS temp_met FROM testdb.dingdan2025 a CROSS JOIN (SELECT @count := 0, @met_one := 0) AS vars where a.充电终端 = %s ORDER BY a.下单时间 DESC ) AS subquery """ try: print(f"执行SQL查询，充电枪编码: {each_code}") print(f"SQL语句: {sql_query}") print(f"参数值: {each_code}") # 执行查询 self.cursor.execute(sql_query, (each_code,)) result = self.cursor.fetchone() print(f"查询结果: {result}") if result is None: print(f"警告：充电枪编码 {each_code} 未找到匹配记录") # 更新进度条 set_processbar('progress', (index + 1) / total_codes) except pymysql.Error as e: error_msg = f"SQL执行错误 (充电枪编码: {each_code}): {str(e)}" print(error_msg) put_text(error_msg) if e.args[0] == 1146: # 表不存在错误 print("错误：表'dingdan2025'不存在，请确认数据库表是否正确创建") raise e results.append(result['count1'] if result and result['count1'] is not None else 0) # 准备表格数据 headers = data[0] + ['连续异常积累单数'] table_data = [headers] # 添加表头 # 合并CSV数据和查询结果 for i, row in enumerate(data[1:]): table_data.append(row + [str(results[i])]) # 在Web界面显示结果 put_text("数据分析结果：") put_table(table_data) # 准备下载的CSV数据 output_csv = tempfile.NamedTemporaryFile(mode='w', newline='', suffix='.csv', delete=False, encoding='utf-8-sig') csv_writer = csv.writer(output_csv) csv_writer.writerows(table_data) output_csv.close() # 添加下载按钮 def download_callback(): with open(output_csv.name, 'rb') as f: content = f.read() download("分析结果.csv", content) os.unlink(output_csv.name) # 删除临时文件 put_buttons(['下载分析结果'], onclick=[download_callback]) print("数据处理完成") except Exception as e: error_msg = f"处理充电枪编码时出错: {str(e)}" put_text(error_msg) print(error_msg) def __del__(self): if self.cursor: self.cursor.close() if self.conn: self.conn.close() print("数据库连接已关闭") if __name__ == '__main__': try: print("正在初始化应用...") app = CSVProcessor() from pywebio.platform.tornado_http import start_server print("正在启动Web服务器...") start_server( applications=app.upload_csv, port=8077, host='localhost', debug=False, cdn=False, auto_open_webbrowser=True ) print("Web服务器启动成功，请访问 http://localhost:8077") except ImportError as e: print(f"错误：缺少必要的库，请确保已安装pywebio。错误信息：{e}") except OSError as e: if "address already in use" in str(e).lower(): print("错误：端口8077已被占用，请关闭占用该端口的程序后重试。") else: print(f"错误：服务器启动失败。错误信息：{e}") except Exception as e: print(f"错误：服务器启动失败。错误信息：{e}") print("提示：请检查端口8077是否被占用，或尝试更换其他端口。")