Python3+PyMySQL从CSV插入MySQL数据时如何避免SQL注入？

嘿，针对你每天用Python3.6 + PyMySQL把电话统计CSV插入MariaDB5.5.56的场景，参数化查询是彻底避免SQL注入的核心方案，这也是Python操作关系型数据库的标准安全做法。我结合你的CSV格式给你梳理一套可落地的实现步骤：

核心原则：绝对不要字符串拼接SQL #

SQL注入的根源就是直接把外部数据拼接到SQL语句里，比如这种写法绝对不能用：

# 危险！会导致SQL注入
sql = f"INSERT INTO call_stats (extension) VALUES ('{extension}')"

只要用PyMySQL的占位符绑定参数，就能让数据库自动处理数据转义，彻底阻断注入风险。MariaDB/MySQL的占位符用%s（注意不是Python的格式化符号，是数据库参数绑定的专用占位符）。

具体实现步骤 #

1. 先处理CSV文件，提取有效数据 #

你的CSV前面有几行说明文本（类型、时间范围）和表头，需要先跳过这些无效行，提取真正的统计数据。用Python内置的csv模块处理最稳妥：

import csv

def parse_call_csv(csv_path):
    data_rows = []
    with open(csv_path, 'r', encoding='utf-8') as f:
        reader = csv.reader(f)
        # 跳过前4行无效内容（根据你的CSV实际行数调整）
        for _ in range(4):
            next(reader)
        # 提取统计日期（可优化：从CSV的时间范围行自动解析，这里先固定示例值）
        stat_date = '2018-04-17'
        # 遍历数据行，清洗并整理成参数格式
        for row in reader:
            if not row[0].strip():  # 跳过空行
                continue
            # 提取各字段值，做好空值处理
            extension = row[0].strip()
            in_answered = int(row[2]) if row[2].strip() else 0
            in_unanswered = int(row[3]) if row[3].strip() else 0
            out_answered = int(row[4]) if row[4].strip() else 0
            out_unanswered = int(row[5]) if row[5].strip() else 0
            total_answered = int(row[6]) if row[6].strip() else 0
            total_unanswered = int(row[7]) if row[7].strip() else 0
            total_duration = row[8].strip()
            # 把一行数据整理成元组，方便后续批量插入
            data_rows.append((
                extension, in_answered, in_unanswered,
                out_answered, out_unanswered, total_answered,
                total_unanswered, total_duration, stat_date
            ))
    return data_rows

2. 用参数化查询批量插入数据库 #

用executemany批量插入比单条插入效率更高，同时保持安全性：

import pymysql

def insert_to_db(data_rows):
    # 建立数据库连接（替换成你的实际配置）
    conn = pymysql.connect(
        host='localhost',
        user='your_db_user',
        password='your_db_pwd',
        database='your_db_name',
        charset='utf8'
    )
    try:
        with conn.cursor() as cursor:
            # 构造参数化的INSERT语句，所有变量都用%s占位
            insert_sql = """
            INSERT INTO call_stats (
                extension, in_answered, in_unanswered,
                out_answered, out_unanswered, total_answered,
                total_unanswered, total_duration, stat_date
            ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
            """
            # 批量插入，第二个参数是数据列表（每个元素对应占位符的元组）
            cursor.executemany(insert_sql, data_rows)
        # 提交事务，确保数据写入
        conn.commit()
        print(f"成功插入{len(data_rows)}条统计数据")
    except Exception as e:
        # 出错回滚事务，避免部分数据写入导致不一致
        conn.rollback()
        print(f"插入失败：{str(e)}")
    finally:
        # 无论成功失败都关闭连接
        conn.close()

3. 调用执行 #

把两个函数结合起来，每天运行一次即可：

if __name__ == "__main__":
    csv_data = parse_call_csv("每日电话统计.csv")
    insert_to_db(csv_data)

关键注意事项 #

• 占位符统一用%s：不管字段是字符串、数字还是日期，都用%s，PyMySQL会自动根据字段类型处理转义，不用自己加引号。
• 数据清洗要到位：CSV里可能有空值、空格，提前转成数据库接受的格式（比如空字符串转0），避免插入报错。
• 避免重复插入：因为是每天生成的统计，最好在插入前加判断，比如检查stat_date和extension的组合是否已存在，防止重复数据。
• 事务必须处理：批量插入时如果中间出错，用rollback回滚，避免部分数据写入数据库导致数据不一致。

内容的提问来源于stack exchange，提问作者Kai Kudou