更新mysql数据python脚本

逐条更新数据统计成功和失败数目。脚本内容# 虚拟环境py3_8_20_patent_env # conda install -c conda-forge rdkit # pip install openpyxl # 一个接一个更新表 my_dataset 内容根据name更新 alpha_xx_aualpha_yy_aualpha_zz_aualpha_iso_au # 错误数据不会影响正常数据 import logging import time import warnings import numpy as np import pandas as pd import pymysql from pymysql.constants import CLIENT class MoleGeneDataImporter: def __init__(self): self.current_date pd.Timestamp.now().strftime(%Y%m%d%H%M) self.setup_logging() self.load_config() def setup_logging(self): 配置日志系统 logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) warnings.filterwarnings(ignore) def load_config(self): 加载配置信息 self.mysql_config { host: localhost, user: root, password: root, database: db-1, port: 3306, client_flag: CLIENT.MULTI_STATEMENTS } self.excel_file ./data/2025-9-26/xxx.xlsx self.sheet_name 数据 def read_excel_data(self): 读取Excel文件数据 try: df pd.read_excel(self.excel_file, sheet_nameself.sheet_name) logging.info(Excel文件读取成功) return df except Exception as e: logging.error(f读取Excel文件失败: {e}) raise staticmethod def update_data_to_db(dataframe, connection): table_name my_dataset # dataframe中的nan替换为None dataframe dataframe.replace(np.nan, None) 逐条更新数据库遇到错误跳过 success_count 0 error_count 0 error_records [] try: with connection.cursor() as cursor: update_sql f UPDATE {table_name} SET alpha_xx_au %s, alpha_yy_au %s, alpha_zz_au %s, alpha_iso_au %s WHERE name %s for _, row in dataframe.iterrows(): try: # 执行单条更新 cursor.execute(update_sql, ( row[Alpha_xx_au], row[Alpha_yy_au], row[Alpha_zz_au], row[Alpha_iso_au], row[Name] )) # 每条记录成功更新后立即提交 connection.commit() success_count 1 # 每成功100条记录输出一次日志 if success_count % 100 0: logging.info(f已成功更新 {success_count} 条记录) except Exception as e: # 如果发生异常回滚当前记录实际上由于每条记录独立提交回滚操作可能没有意义因为已经提交的记录无法回滚 # connection.rollback() error_count 1 error_records.append(row[Name]) logging.warning(f更新记录 {row[Name]} 时出错: {e}) # 继续处理下一条记录 continue logging.info(f数据更新完成成功: {success_count} 条失败: {error_count} 条) if error_count 0: logging.warning(f更新失败的记录: {error_records}) except Exception as e: connection.rollback() logging.error(f数据更新过程中发生错误: {e}) raise def run(self): 主运行方法 try: # 读取和处理数据 df self.read_excel_data() logging.info(f共读取到 {len(df)} 条记录) # 连接数据库并插入数据 connection pymysql.connect(**self.mysql_config) logging.info(MySQL数据库连接成功) logging.info(数据准备完成开始逐条更新数据库...) self.update_data_to_db(df, connection) except Exception as e: logging.error(f程序运行出错: {e}) finally: if connection in locals() and connection: connection.close() logging.info(MySQL数据库连接已关闭。) if __name__ __main__: start_time time.time() importer MoleGeneDataImporter() importer.run() end_time time.time() logging.info(f逐条更新数据到mysql数据库完成程序运行耗时: {end_time - start_time} 秒)