用 Excel 导入增改数据库表功能实现

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 = None    cursor = None    try:        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.rowcount        conn.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 块中确保关闭连接，异常时自动回滚事务。