逐条更新数据，统计成功和失败数目。
脚本内容，
# 虚拟环境：py3_8_20_patent_env # conda install -c conda-forge rdkit # pip install openpyxl # 一个接一个更新表 my_dataset 内容，根据name更新 alpha_xx_au，alpha_yy_au，alpha_zz_au，alpha_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(level=logging.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_name=self.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} 秒")
