给定年月自动批量导入CSV至SQLite数据库的可复用脚本实现

问题分析 #

现有脚本需要手动调整日期格式的日数字前缀（如0{}/{}），无法自动覆盖当月所有日期，且缺乏增量导入能力。以下是优化后的可复用解决方案：

实现代码 #

import os
import pandas as pd
import sqlite3
from datetime import datetime, timedelta

def import_monthly_csv_to_sqlite(year_month, db_path='data.db', table_name='test', csv_dir='./'):
    # 解析年月参数（格式示例：'2020-01'）
    year, month = map(int, year_month.split('-'))
    
    # 生成当月首尾日期
    start_date = datetime(year, month, 1)
    end_date = (datetime(year, month+1, 1) if month != 12 else datetime(year+1, 1, 1)) - timedelta(days=1)
    
    # 生成当月所有日期对应的CSV文件名
    current_date = start_date
    all_month_files = []
    while current_date <= end_date:
        all_month_files.append(current_date.strftime('%Y-%m-%d.csv'))
        current_date += timedelta(days=1)
    
    # 连接SQLite数据库
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    
    # 创建已导入文件记录表（首次运行自动生成）
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS imported_files (
            file_name TEXT PRIMARY KEY,
            import_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    ''')
    conn.commit()
    
    # 获取已导入文件列表
    cursor.execute('SELECT file_name FROM imported_files')
    imported_files = set(row[0] for row in cursor.fetchall())
    
    # 筛选本地存在且未导入的文件
    pending_import = []
    for file in all_month_files:
        full_path = os.path.join(csv_dir, file)
        if os.path.exists(full_path) and file not in imported_files:
            pending_import.append(full_path)
    
    if not pending_import:
        print(f"{year_month} 无新增CSV文件需导入")
        conn.close()
        return
    
    # 批量读取并导入数据
    df_list = [pd.read_csv(file) for file in pending_import]
    combined_df = pd.concat(df_list, ignore_index=True)
    combined_df.to_sql(table_name, conn, if_exists='append', index=False)
    
    # 记录已导入文件
    for file_path in pending_import:
        cursor.execute('INSERT INTO imported_files (file_name) VALUES (?)', (os.path.basename(file_path),))
    conn.commit()
    
    print(f"成功导入 {len(pending_import)} 个文件至 {table_name} 表")
    conn.close()

使用示例 #

# 导入2020年1月的新增CSV文件（指定CSV目录）
import_monthly_csv_to_sqlite('2020-01', csv_dir='./your_csv_directory')

# 自定义数据库路径和表名
import_monthly_csv_to_sqlite('2023-06', db_path='./my_db.db', table_name='sales_data')

核心特性 #

• 自动日期格式处理：通过strftime('%Y-%m-%d')自动生成带补0的日期文件名，无需手动拆分单/双日
• 增量导入：利用imported_files表跟踪已导入文件，避免重复操作
• 高可复用性：通过参数指定年月、数据库路径、目标表名和CSV目录，适配不同场景
• 容错检查：自动跳过不存在的文件，无新增文件时给出明确提示

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