Python Pandas读取Excel并生成SQL WHERE条件

import pandas as pd
import numpy as np
from math import ceil

def build_condition(field, value):
    """根据字段值和类型构造单个条件片段"""
    if pd.isna(value):
        return f"{field} IS NULL"
    
    # 处理日期时间类型
    if isinstance(value, (pd.Timestamp, np.datetime64)):
        # 转换为 SQL 标准格式的字符串
        val_str = pd.to_datetime(value).strftime('%Y-%m-%d %H:%M:%S')
        return f"{field} = '{val_str}'"
    
    # 处理字符串类型
    if isinstance(value, str):
        # 转义单引号：替换为一个单引号为两个单引号
        escaped = value.replace("'", "''")
        return f"{field} = '{escaped}'"
    
    # 处理数值类型（int, float, np.number）
    if isinstance(value, (int, float, np.number)):
        return f"{field} = {value}"
    
    # 其他情况（例如布尔值等）作为字符串处理
    val_str = str(value).replace("'", "''")
    return f"{field} = '{val_str}'"

def generate_where_batches(df, batch_size=100):
    """
    将 DataFrame 的每一行转换为 AND 连接的条件组，
    每 batch_size 个组用 OR 连接为一批，返回批次的列表
    """
    fields = df.columns.tolist()
    row_conditions = []
    
    for _, row in df.iterrows():
        # 构建单个行的条件列表
        conds = []
        for field in fields:
            cond = build_condition(field, row[field])
            conds.append(cond)
        # 用 AND 连接，并加上括号形成一个条件组
        group = "(" + " AND ".join(conds) + ")"
        row_conditions.append(group)
    
    # 分批，每 batch_size 个组用 OR 连接
    batches = []
    for i in range(0, len(row_conditions), batch_size):
        batch_groups = row_conditions[i:i+batch_size]
        batch_str = " OR ".join(batch_groups)
        batches.append(batch_str)
    
    return batches

# ------------------- 使用示例 -------------------
if __name__ == "__main__":
    # 请替换为实际的 Excel 文件路径
    excel_file = "data.xlsx"
    
    # 读取 Excel（默认读取第一个 sheet）
    df = pd.read_excel(excel_file)
    
    # 生成批次条件（每 100 条记录为一批）
    batches = generate_where_batches(df, batch_size=100)
    
    # 打印输出每一批
    for idx, batch in enumerate(batches, start=1):
        print(f"--- 第 {idx} 批 (WHERE 条件部分) ---")
        print(batch)
        print()  # 空行分隔

pip install pandas openpyxl

--- 第 1 批 (WHERE 条件部分) ---
(name = 'John' AND age = 30 AND created_at = '2023-01-01 10:00:00') OR (name IS NULL AND age = 25 AND created_at IS NULL) OR (name = 'O''Brien' AND age = 40 AND created_at = '2023-02-01 12:30:00')

WHERE id IN (101, 102, 103, ...)

pip install pandas openpyxl

import pandas as pd
import math
def generate_batch_where_clauses(
    excel_path: str,
    sheet_name: str = 0,
    column_name: str = 'id',
    batch_size: int = 1000,
    output_file: str = None
):
    """
    读取 Excel 指定列，按批次生成 SQL WHERE 条件
    参数:
        excel_path: Excel 文件路径
        sheet_name: 工作表名称或索引，默认第一个
        column_name: 需要提取的列名
        batch_size: 每批包含的值个数
        output_file: 可选，输出到文本文件
    返回:
        list: 每个批次生成的 WHERE 子句字符串
    """
    # 1. 读取 Excel
    df = pd.read_excel(excel_path, sheet_name=sheet_name)
    # 2. 检查列是否存在
    if column_name not in df.columns:
        raise ValueError(f"列 '{column_name}' 不存在于 Excel 中，可用的列有: {df.columns.tolist()}")
    # 3. 提取该列，去除空值，去重，转换为 Python 列表
    values = df[column_name].dropna().unique().tolist()
    if not values:
        print("警告：没有有效的值，请检查数据。")
        return []
    # 4. 分批
    total = len(values)
    batch_count = math.ceil(total / batch_size)
    where_clauses = []
    for i in range(batch_count):
        start = i * batch_size
        end = min((i + 1) * batch_size, total)
        batch_values = values[start:end]
        # 将值转换为字符串，注意处理字符串类型需要加引号
        # 这里简单处理：如果是数值类型，直接转为字符串，否则加单引号
        first_val = batch_values[0]
        if isinstance(first_val, (int, float)):
            value_strs = [str(v) for v in batch_values]
        else:
            # 如果包含字符串，需要转义单引号（如 O'Reilly -> O''Reilly）
            value_strs = [f"'{str(v).replace("'", "''")}'" for v in batch_values]
        in_clause = ", ".join(value_strs)
        where_clause = f"WHERE {column_name} IN ({in_clause})"
        where_clauses.append(where_clause)
        # 可选打印
        print(f"批次 {i+1}/{batch_count}: {len(batch_values)} 个值")
        print(where_clause)
        print("-" * 80)
    # 5. 若指定输出文件，则写入
    if output_file:
        with open(output_file, 'w', encoding='utf-8') as f:
            for clause in where_clauses:
                f.write(clause + ";\n")   # 加个分号便于直接使用
        print(f"所有 WHERE 子句已保存到: {output_file}")
    return where_clauses
# 使用示例
if __name__ == "__main__":
    generate_batch_where_clauses(
        excel_path="data.xlsx",
        sheet_name="Sheet1",
        column_name="user_id",
        batch_size=500,
        output_file="where_clauses.sql"
    )

def generate_composite_where_clauses(excel_path, sheet_name=0, columns=None, batch_size=1000):
    """
    根据多列生成组合 WHERE 条件，每行一组条件
    columns: 要使用的列名列表，如 ['region', 'status']
    """
    df = pd.read_excel(excel_path, sheet_name=sheet_name)
    if columns is None:
        columns = df.columns.tolist()
    # 构建条件列表（每行一个元组）
    conditions = []
    for _, row in df.iterrows():
        # 跳过含空值的行
        if row[columns].isnull().any():
            continue
        # 生成类似 "col1='A' AND col2=1" 的字符串
        parts = []
        for col in columns:
            val = row[col]
            if isinstance(val, str):
                val_escaped = val.replace("'", "''")
                parts.append(f"{col} = '{val_escaped}'")
            else:
                parts.append(f"{col} = {val}")
        conditions.append(" AND ".join(parts))
    # 去重
    conditions = list(set(conditions))
    # 分批
    total = len(conditions)
    batch_count = math.ceil(total / batch_size)
    where_clauses = []
    for i in range(batch_count):
        batch = conditions[i*batch_size:(i+1)*batch_size]
        # 组合为 OR 连接
        or_clause = " OR ".join([f"({cond})" for cond in batch])
        where_clause = f"WHERE {or_clause}"
        where_clauses.append(where_clause)
    return where_clauses