Python+pandas查询SQL时如何传参并用循环复用合同日期异常处理代码

核心思路是将重复的查询、数据处理逻辑封装为可复用函数，通过统一的合同配置列表遍历执行，无需重复编写代码。

步骤1：定义全局配置与数据库连接 #

这部分只需执行一次，把你所有需要查询的合同的匹配参数、输出命名统一放在列表里：

import pandas as pd
import numpy as np
import pyodbc

# 数据库连接（只需初始化一次）
conn = pyodbc.connect('Driver={SQL Server};'
                      'Server=XXXXXXXX;'
                      'Database=XXXXXXX;'
                      'Trusted_Connection=yes;')

# 合同配置列表：按你的实际需求补全11份合同的参数，格式为(ContractName模糊匹配值, ClientCode模糊匹配值, 输出sheet名称)
contract_config = [
    ("***", "AAA", "First_DateReport"),
    ("第二个合同匹配值", "AAA", "Second_DateReport"),
    # 继续补充剩下9份合同的配置即可
]

# 存储所有处理后的合同df
contract_dfs = []

步骤2：封装查询与数据处理函数 #

把原来重复执行的步骤2、3封装为函数，入参为配置里的参数，返回处理好的异常数据df：

def get_contract_error_df(contract_like, client_like, sheet_name):
    # 动态替换SQL查询条件
    sql = f"""
    Select DISTINCT  
        ContractName , 
        clientcode, 
        TypeOfFile,
        (CASE 
            WHEN SentToBBDate > BBLoadedDate then 'ERROR - SENT DATE GREATER THAN LOAD DATE' 
            WHEN ReceivedDate > SentToBBDate then 'ERROR - RECEIVED DATE GREATER THAN SENT DATE'  
            WHEN ReceivedDate is NULL then 'ERROR - FILE NOT RECEIVED' 
            ELSE 'NORMAL' 
        END) AS Error_Message 
    from DataDashboard 
    Where ContractName like '%{contract_like}%' 
      AND ClientCode LIKE '%{client_like}%' 
    Order by TypeOfFile
    """
    # 执行查询
    raw_df = pd.read_sql_query(sql, conn)
    # 处理数据（已修正原代码中的Cigna变量名笔误）
    processed_df = raw_df.iloc[: , 1:]
    # 过滤掉正常数据，只保留异常
    processed_df = processed_df[processed_df['Error_Message'] != 'NORMAL'].reset_index(drop=True)
    # 可选：如果需要单独保存每个合同的Excel，取消下面注释即可
    # processed_df.to_excel(f"C:\\Users\\pd\\Desktop\\{sheet_name}.xlsx", index=False)
    return processed_df, sheet_name

步骤3：批量执行生成所有合同数据与汇总表 #

# 遍历所有合同配置，生成对应df
for config in contract_config:
    cl, cli, sn = config
    df, sheet_name = get_contract_error_df(cl, cli, sn)
    contract_dfs.append((df, sheet_name))

# 生成汇总表
consolidated_df = pd.concat([df for df, _ in contract_dfs], ignore_index=True)
# 把汇总表加到待写入列表的第一个位置
contract_dfs.insert(0, (consolidated_df, "Conso_DateReport"))

步骤4：统一写入Excel并设置格式 #

已修正原代码中格式设置的逻辑错误（原row_count取全局df，改为对应每个sheet的df）：

save_path = "C:\\Users\\pd\\Desktop\\Consolidated_DateReport.xlsx"
writer = pd.ExcelWriter(save_path, engine='xlsxwriter')

# 先写入所有sheet的内容
for df, sheet_name in contract_dfs:
    df.to_excel(writer, sheet_name, index=False)

# 定义表头格式
workbook = writer.book
header_format = workbook.add_format({
    'bold': True, 
    'size':11,
    'valign': 'top',
    'fg_color': '#c7e7ff',
    'border': 1
})

# 统一设置所有sheet的格式
for df, sheet_name in contract_dfs:
    worksheet = writer.sheets[sheet_name]
    row_count = len(df.index)
    col_count = len(df.columns)
    # 设置表头格式、冻结首行、自动筛选、列宽
    for col_num, value in enumerate(df.columns.values):
        worksheet.write(0, col_num, value, header_format)
    worksheet.freeze_panes(1, 0)
    if row_count > 0:
        worksheet.autofilter(0, 0, row_count, col_count-1)
    worksheet.set_column(0, col_count-1, 25)

# 保存文件并关闭数据库连接
writer.save()
conn.close()

注意事项 #

• 原代码中存在的问题已修正：包括SQL语句中TypeOfFile后缺失的逗号、变量名Cigna错误、Error_Message大小写不匹配、筛选行数计算错误的问题
• 后续如需新增合同查询，只需修改contract_config列表里的参数即可，无需修改核心逻辑

内容的提问来源于stack exchange，提问作者dh.purvi