import pandas as pd from datetime import datetime from pathlib import Path from loguru import logger import hashlib import time def QhWeiYiZhiMd5(QhDf, QhUniqueColumns, QhEncoding="utf-8-sig"): """ 高效率生成MD5 作者：阙辉 """ try: QhCombined = QhDf[QhUniqueColumns].astype(str).agg(''.join, axis=1) QhDf["QhUniqueId"] = QhCombined.apply(lambda x: hashlib.md5(x.encode(QhEncoding)).hexdigest()) return QhDf except Exception as e: logger.exception(f"MD5生成失败,QueHui!") return QhDf def QhDfUpsertToCsv( QhWenJianPath, # 文件保存路径 QhNewJieGuoDf, # 新结果DataFrame QhUniqueColumns, # 唯一键组合列 QhSortColumns=None, # 排序列 QhDateFormat="%Y-%m-%d", # 日期格式 QhEncoding="utf-8-sig", # 文件编码 ): """ 高效UPSERT：存在更新/不存在插入 支持排序、日期自动格式化 作者：阙辉 """ QhStartTime = time.time() logger.info("===== 开始执行CSV Upsert操作 QueHui=====") # 自动补全.csv后缀 QhWenJianPath = str(Path(QhWenJianPath).with_suffix(".csv")) try: # 检查传入数据是否有效 if not isinstance(QhNewJieGuoDf, pd.DataFrame): logger.error(f"传入数据不是DataFrame，终止操作,QueHui!") return None if QhNewJieGuoDf.empty: logger.warning(f"新数据为空，无需执行更新,QueHui!") return QhNewJieGuoDf # 检查唯一键列是否存在 for QhCol in QhUniqueColumns: if QhCol not in QhNewJieGuoDf.columns: logger.error(f"唯一键列不存在：{QhCol}，终止操作,QueHui!") return None # 新数据生成MD5 QhNewJieGuoDf = QhWeiYiZhiMd5(QhNewJieGuoDf, QhUniqueColumns, QhEncoding=QhEncoding) logger.info("新数据MD5生成完成,QueHui!") # 新数据内部去重（修复你说的BUG） QhNewJieGuoDf = QhNewJieGuoDf.drop_duplicates(subset=["QhUniqueId"], keep="last") logger.info(f"新数据内部去重完成，去重后数据量：{len(QhNewJieGuoDf)}！QueHui!") # 生成版本号 QhBanBenHao = f"Q{datetime.now().strftime('%Y%m%d%H%M%S%f')[:-3]}" QhNewJieGuoDf["QhBanBenHao"] = QhBanBenHao logger.info(f"新数据版本号生成完成：【{QhBanBenHao}】,QueHui!") # 调整列顺序 QhNewCols = list(QhNewJieGuoDf.columns) if "QhUniqueId" in QhNewCols: QhNewCols.remove("QhUniqueId") QhNewCols.insert(0, "QhUniqueId") if "QhBanBenHao" in QhNewCols: QhNewCols.remove("QhBanBenHao") QhNewCols.append("QhBanBenHao") QhNewJieGuoDf = QhNewJieGuoDf[QhNewCols] logger.info("列顺序调整完成：QhUniqueId置顶，QhBanBenHao置尾,QueHui!") # 读取旧文件 try: QhOldDf = pd.read_csv(QhWenJianPath, encoding=QhEncoding) logger.info(f"读取旧文件成功：【{QhWenJianPath}】，数据量：{len(QhOldDf)} 行,QueHui!") # 旧数据无QhUniqueId则自动生成 if "QhUniqueId" not in QhOldDf.columns: logger.warning(f"旧数据未找到QhUniqueId列，根据组合键自动生成,QueHui!") # 检查字段是否齐全 QhMissingCols = [c for c in QhUniqueColumns if c not in QhOldDf.columns] if QhMissingCols: logger.error(f"旧数据缺少拼接字段：{QhMissingCols}，无法生成,QueHui!") return None # 自动生成MD5 QhOldDf = QhWeiYiZhiMd5(QhOldDf, QhUniqueColumns, QhEncoding=QhEncoding) logger.success(f"旧数据MD5自动生成完成,QueHui!") # 旧数据内部去重（加固） QhOldDf = QhOldDf.drop_duplicates(subset=["QhUniqueId"], keep="last") logger.info(f"旧数据内部去重完成，去重后数据量：{len(QhOldDf)}！QueHui!") except FileNotFoundError: logger.warning(f"文件不存在，即将创建新文件：{QhWenJianPath},QueHui!") # 排序处理 if QhSortColumns: QhExistSortCols = [c for c in QhSortColumns if c in QhNewJieGuoDf.columns] for QhCol in QhExistSortCols: try: QhNewJieGuoDf[QhCol] = pd.to_datetime(QhNewJieGuoDf[QhCol], format=QhDateFormat, errors="coerce") except Exception as e: logger.exception(f"日期转换失败：{QhCol},QueHui!") if QhExistSortCols: QhNewJieGuoDf = QhNewJieGuoDf.sort_values(by=QhExistSortCols, ascending=True) logger.info(f"新数据排序完成：{QhExistSortCols},QueHui!") # 保存新文件 try: QhNewJieGuoDf.to_csv(QhWenJianPath, index=False, encoding=QhEncoding) except Exception as e: logger.exception(f"新文件保存失败,QueHui!") return None QhTotalTime = round(time.time() - QhStartTime, 2) logger.success(f"新文件创建保存成功，总耗时：{QhTotalTime} 秒,QueHui!") return QhNewJieGuoDf # 过滤旧数据中与新数据重复的数据 QhNewKeys = QhNewJieGuoDf["QhUniqueId"].values QhOldDf = QhOldDf[~QhOldDf["QhUniqueId"].isin(QhNewKeys)] logger.info(f"旧数据过滤完成，移除与新数据重复的数据！QueHui!") # 合并数据 QhJieGuoDf = pd.concat([QhOldDf, QhNewJieGuoDf], ignore_index=True) logger.info(f"新旧数据合并完成，合并后总行数：{len(QhJieGuoDf)} 行,QueHui!") # 统一列顺序 QhNewCols = list(QhJieGuoDf.columns) if "QhUniqueId" in QhNewCols: QhNewCols.remove("QhUniqueId") QhNewCols.insert(0, "QhUniqueId") if "QhBanBenHao" in QhNewCols: QhNewCols.remove("QhBanBenHao") QhNewCols.append("QhBanBenHao") QhJieGuoDf = QhJieGuoDf[QhNewCols] # 最终排序 if QhSortColumns: QhExistSortCols = [c for c in QhSortColumns if c in QhJieGuoDf.columns] for QhCol in QhExistSortCols: try: QhJieGuoDf[QhCol] = pd.to_datetime(QhJieGuoDf[QhCol], format=QhDateFormat, errors="coerce") except Exception as e: logger.exception(f"日期转换失败：{QhCol},QueHui!") if QhExistSortCols: QhJieGuoDf = QhJieGuoDf.sort_values(by=QhExistSortCols, ascending=True) logger.info(f"最终数据排序完成：{QhExistSortCols},QueHui!") # 保存文件 try: QhJieGuoDf.to_csv(QhWenJianPath, index=False, encoding=QhEncoding) except Exception as e: logger.exception(f"最终文件保存失败,QueHui!") return None # 耗时统计 QhTotalTime = round(time.time() - QhStartTime, 2) logger.success(f"===== Upsert操作全部完成，总耗时：{QhTotalTime} 秒 QueHui=====") return QhJieGuoDf except Exception as e: logger.exception(f"Upsert执行发生全局异常,QueHui!") return None
