夜雨聆风Excel 上传到数据库智能增改的全流程实践
一、功能概述
📦 功能名称
业务数据同步与更新
📝 功能简介
该功能允许用户上传包含「业务编码」和「标准值」的Excel 文件。系统会自动读取文件内容，与数据库中现有的数据记录表进行比对，智能判断哪些数据是新增的、哪些发生了变更，并自动执行数据库的插入或更新操作。
⚙️ 核心处理逻辑
系统在后台会依次执行以下三步操作：
第一步：数据筛选与清洗
•有效性检查：仅处理 Excel 中「是否有效」列标记为「是」的数据。
•完整性检查：自动过滤掉「标准值」为空或空白的记录。
•格式校验：确保文件包含「业务编码」「标准值」「是否有效」这三列。
第二步：智能比对与分类
系统将上传的数据与数据库现有数据进行逐条比对（基于业务编码）：
•新增数据：数据库中没有该业务编码 → 标记为「待插入」。
•更新数据：业务编码已存在，但「标准值」不一致 → 标记为「待更新」。
•忽略数据：完全一致 → 跳过，不做任何操作。
第三步：执行写入
•插入：将新数据写入数据库。
•更新：更新现有数据的「标准值」，并自动将更新时间字段刷新为当前服务器时间。
📊 操作结果与反馈
操作完成后，系统会返回一个 Excel 文件供用户下载，文件名通常为「数据处理结果.xlsx」。
反馈文件包含用户上传的原始数据，并增加了一列「处理状态」：
•已插入 —— 该条数据在数据库中不存在，已成功新增。
•已更新 —— 该条数据已存在，且内容已根据新文件更新。
•未变动 —— 与数据库记录一致，未做任何修改。
🛠️ 异常处理
•文件格式错误：如果上传的文件缺少必要列或无法读取，系统会直接报错提示。
•数据库异常：如果在写入过程中发生错误（如主键冲突、连接断开），系统会终止操作并返回具体的错误信息。
二、代码实现
以下按照实际执行顺序，逐步展示各环节的核心代码。
2.1 接口路由定义
首先定义一个 POST 接口，接收用户上传的 Excel 文件：
@app.post("/import-excel-data")async def import_excel_data(upload_file: UploadFile = File(..., description="上传的excel文件")):
2.2 Excel 数据导入与清洗
读取上传的 Excel 文件后，执行三步清洗操作：筛选有效记录、过滤空值、保留目标列，并将列名映射为数据库字段名。
# 筛选"是否有效"为"是"的记录df_valid= df[df["是否有效"] == "是"].copy()#额外过滤：标准值不能为空（同时排除空字符串和NaN）df_valid= df_valid[df_valid["标准值"].notna()&(df_valid["标准值"].astype(str).str.strip()!= "")]# 仅保留目标列result= df_valid[["业务编码", "标准值", "是否有效"]].reset_index(drop=True)# 准备用于处理的数据，并重命名列以便与数据库字段对应df_upload= df_valid[["业务编码", "标准值"]].rename(columns={"业务编码":"biz_code", "标准值":"standard_value"})if df_upload.empty:return {"message": "没有符合条件的数据可供处理"}
2.3 获取数据库现有数据
查询目标数据表，获取当前已有的业务编码和标准值，为后续比对做准备：
# 查询目标表 t_data_recordquery_db= "SELECT biz_code, standard_value FROM t_data_record"try:df_db = Sqlquery(query_db)if not df_db.empty:df_db.columns = ["biz_code","standard_value"]except Exception as e:raise HTTPException(status_code=500,detail=f"数据库查询失败: {str(e)}")
2.4 数据比对与分类
将上传数据与数据库现有数据逐条比对，按业务编码分为「待插入」「待更新」两组：
# 将数据库中的业务编码存入集合，便于快速查找existing_ids= set(df_db["biz_code"].astype(str)) if not df_db.empty else set()insert_list= []update_list= []for _, row in df_upload.iterrows():biz_id = str(row["biz_code"])new_val = str(row["standard_value"]).strip()if biz_id not in existing_ids:# 情况1: 业务编码在数据库中不存在 → 需要插入insert_list.append(row)else:# 情况2 & 3: 业务编码已存在，需要检查标准值是否改变db_val = str(df_db[df_db["biz_code"].astype(str) == biz_id].iloc[0]["standard_value"]).strip()if db_val != new_val:# 情况3: 值已改变 → 需要更新update_list.append(row)# 情况2: 值未改变 → 不需要操作，自动跳过
2.5 执行数据库写入操作
根据比对结果，分别执行批量插入和批量更新。若任何一条失败，立即抛出异常并终止：
sql_insert = ("INSERT INTO t_data_record (biz_code,standard_value) VALUES (?, ?)")sql_update = ("UPDATE t_data_record SET standard_value = ?, update_time = GETDATE() WHERE biz_code = ?")try:# 执行批量插入for _, row in pd.DataFrame(insert_list).iterrows():res = SqlqueryUp(sql_insert,params=(str(row["biz_code"]),str(row["standard_value"])))if not res["success"]:raise Exception(f'插入失败 (业务编码: {row["biz_code"]}): {res["error"]}')# 执行批量更新for _, row in pd.DataFrame(update_list).iterrows():# 注意参数顺序：SET的值在前，WHERE的值在后res = SqlqueryUp(sql_update,params=(str(row["standard_value"]),str(row["biz_code"])))if not res["success"]:raise Exception(f'更新失败 (业务编码: {row["biz_code"]}): {res["error"]}')except Exception as e:raise HTTPException(status_code=500,detail=f"数据库操作失败: {str(e)}")
2.6 生成结果文件并返回
将处理结果写入 Excel 文件，为每条记录标注处理状态，并通过流式响应返回给前端下载：
# 为原始数据添加"处理状态"列df_result = df_valid.copy()df_result["处理状态"] = "未变动"# 标记已插入的记录insert_ids = [str(row["biz_code"]) for row in insert_list]df_result.loc[df_result["业务编码"].astype(str).isin(insert_ids),"处理状态"] = "已插入"# 标记已更新的记录update_ids = [str(row["biz_code"]) for row in update_list]df_result.loc[df_result["业务编码"].astype(str).isin(update_ids),"处理状态"] = "已更新"# 生成 Excel 文件output = io.BytesIO()with pd.ExcelWriter(output, engine="openpyxl") as writer:df_result.to_excel(writer, index=False,sheet_name="处理结果")output.seek(0)# 处理文件名编码（兼容中文文件名）filename = "数据处理结果.xlsx"encoded_filename = quote(filename)headers = {"Content-Disposition": (f"attachment; "f'filename="processed_data.xlsx"; 'f"filename*=UTF-8''{encoded_filename}")}return StreamingResponse(output,media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",headers=headers)
三、数据库工具函数
以上代码中用到的 Sqlquery 和 SqlqueryUp 是封装好的数据库操作函数，下面给出它们的完整实现。
3.1 数据库连接
集中管理数据库连接字符串，方便统一维护和修改：
💡以下连接信息均为示例占位符，请替换为你自己的数据库配置。切勿在代码中硬编码真实密码，建议使用环境变量管理。 |
def connection_string():server = "<your_db_server>" # 数据库服务器地址port = 1433 # 端口号database = "<your_database>" # 数据库名称user = "<your_username>" # 用户名password = "<your_password>" # 密码（建议用环境变量）# 构造连接字符串（使用FreeTDS ODBC 驱动）conn_string = (f"DRIVER=<path_to_odbc_driver>;"f"SERVER={server};"f"PORT={port};"f"DATABASE={database};"f"UID={user};"f"PWD={password};""TDS_VERSION=7.4;")return conn_string
3.2 数据查询方法
封装 SQL 查询方法，返回 pandas DataFrame 格式的查询结果，方便后续用 pandas 进行数据比对：
def Sqlquery(query, params=None):"""执行 SQL 查询语句，返回 DataFrame 结果"""conn_string = connection_string()conn = pyodbc.connect(conn_string)# 注册转换器：处理 GBK 编码的 VARCHAR 字段conn.add_output_converter(pyodbc.SQL_VARCHAR,lambda b: b.decode("gbk", errors="replace") if b else None)if params:df = pd.read_sql(query, conn, params=params)else:df = pd.read_sql(query, conn)conn.close()return df
3.3 数据更新方法
封装 SQL 更新方法（INSERT / UPDATE / DELETE），返回包含执行状态、受影响行数和错误信息的字典：
def SqlqueryUp(query, params=None):"""执行 SQL 更新语句（INSERT/UPDATE/DELETE），返回执行结果。返回值:dict: {"success": bool,"rows_affected": int, # 受影响行数（仅成功时有效）"error": str or None # 错误信息（失败时有值）}"""conn_string = connection_string()conn = Nonecursor = Nonetry:conn = pyodbc.connect(conn_string)conn.add_output_converter(pyodbc.SQL_VARCHAR,lambda b: b.decode("gbk", errors="replace") if b else None)cursor = conn.cursor()# 支持参数化查询，防止 SQL 注入if params:cursor.execute(query, params)else:cursor.execute(query)rows_affected = cursor.rowcountconn.commit()return {"success": True, "rows_affected": rows_affected, "error": None}except Exception as e:error_msg = str(e)print(f"数据库操作失败：{error_msg}")if conn:conn.rollback()return {"success": False, "rows_affected": 0, "error": error_msg}finally:if cursor:cursor.close()if conn:conn.close()
四、完整流程总结
将上述各环节串联起来，整个功能的处理流程如下：
步骤 | 操作 | 核心方法 | 输出结果 |
① | 接收上传文件 | UploadFile + File(...) | Excel 文件对象 |
② | 数据清洗 | pandas 筛选 + 过滤 | 有效数据 DataFrame |
③ | 查询现有数据 | Sqlquery(SELECT ...) | 数据库数据 DataFrame |
④ | 比对与分类 | 集合匹配 + 逐行比对 | insert_list / update_list |
⑤ | 执行写入 | SqlqueryUp(INSERT/UPDATE) | 写入结果（成功/失败） |
⑥ | 返回结果文件 | ExcelWriter + StreamingResponse | 带处理状态的 Excel |
🔑 关键设计要点
•参数化查询：所有 SQL 语句均使用参数化方式（? 占位符），有效防止 SQL 注入。
•错误即终止：批量写入过程中，任何一条记录失败都会立即抛出异常，避免数据不一致。
•结果可追溯：返回的 Excel 文件标注了每条记录的处理状态，用户可直观确认处理结果。
•连接安全：数据库操作函数在 finally 块中确保关闭连接，异常时自动回滚事务。
