Python实现本地文件夹CSV批量及自动导入MySQL数据库方案咨询

前置依赖安装 #

先装需要的第三方库：

pip install pandas pymysql sqlalchemy watchdog

提前在MySQL中创建好对应的数据表，表字段顺序、类型要和CSV的列保持一致，避免导入出错。

1 存量CSV批量导入功能实现 #

直接上可运行代码，修改配置参数即可使用：

import os
import pandas as pd
from sqlalchemy import create_engine

# --------------------------配置项修改这里--------------------------
CSV_FOLDER = "/path/to/your/csv/folder"  # 本地CSV文件夹路径
MYSQL_CONFIG = {
    "host": "127.0.0.1",
    "port": 3306,
    "user": "your_username",
    "password": "your_password",
    "database": "your_database",
    "table": "your_target_table"  # 导入的目标表名
}
# -------------------------------------------------------------------

# 创建MySQL连接引擎
engine = create_engine(
    f"mysql+pymysql://{MYSQL_CONFIG['user']}:{MYSQL_CONFIG['password']}@{MYSQL_CONFIG['host']}:{MYSQL_CONFIG['port']}/{MYSQL_CONFIG['database']}?charset=utf8mb4"
)

def import_csv_to_mysql(csv_path):
    """单个CSV文件导入MySQL的公共方法"""
    try:
        # 读取CSV，encoding可根据实际文件编码修改为gbk/utf-8-sig等
        df = pd.read_csv(csv_path, encoding="utf-8")
        # 导入数据库，if_exists设为append表示追加，replace为覆盖
        df.to_sql(
            name=MYSQL_CONFIG["table"],
            con=engine,
            if_exists="append",
            index=False,  # 不导入pandas的索引列
            chunksize=1000  # 大文件可分批导入，每次1000行，避免内存溢出
        )
        print(f"文件 {os.path.basename(csv_path)} 导入成功")
    except Exception as e:
        print(f"文件 {os.path.basename(csv_path)} 导入失败，错误信息：{str(e)}")

if __name__ == "__main__":
    # 遍历文件夹下所有CSV文件
    for filename in os.listdir(CSV_FOLDER):
        if filename.lower().endswith(".csv"):
            csv_full_path = os.path.join(CSV_FOLDER, filename)
            import_csv_to_mysql(csv_full_path)

2 新增CSV自动监听导入功能实现 #

基于watchdog库实现文件夹实时监听，新增CSV文件后自动触发导入逻辑，复用上面的import_csv_to_mysql方法：

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

# 复用上面的配置项和import_csv_to_mysql方法

class CSVFileHandler(FileSystemEventHandler):
    def on_created(self, event):
        # 只处理CSV文件，忽略文件夹创建事件
        if not event.is_directory and event.src_path.lower().endswith(".csv"):
            # 等待文件写入完成，避免大文件复制过程中读取报错
            time.sleep(3)
            print(f"检测到新增CSV文件：{event.src_path}")
            import_csv_to_mysql(event.src_path)

if __name__ == "__main__":
    # 先执行一次存量导入
    for filename in os.listdir(CSV_FOLDER):
        if filename.lower().endswith(".csv"):
            csv_full_path = os.path.join(CSV_FOLDER, filename)
            import_csv_to_mysql(csv_full_path)
    
    # 启动文件夹监听
    event_handler = CSVFileHandler()
    observer = Observer()
    observer.schedule(event_handler, CSV_FOLDER, recursive=False)  # recursive设为True会监听子文件夹
    observer.start()
    print(f"已启动文件夹{CSV_FOLDER}监听，等待新增CSV文件...")
    
    # 保持程序运行
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

优化建议 #

• 新增文件去重：可将已导入文件的文件名/哈希值存入专门的MySQL记录表或者本地文本文件，每次导入前先校验，避免同一个文件重复导入
• 异常落盘：将导入失败的文件路径、错误信息写入日志文件，方便后续排查
• 权限校验：提前确保程序对CSV文件夹有读取权限，对MySQL表有写入权限

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