使用 Python 将 Excel 数据批量导入到SQLite数据库中

pip install FreeSpire.XLS

from spire.xls import Workbook
import sqlite3
def excel_to_sqlite(excel_path, db_path):
    # 1. 加载 Excel 文件
    workbook = Workbook()
    workbook.LoadFromFile(excel_path)
    # 2. 连接数据库
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    # 3. 遍历每个工作表
    for sheet_index in range(workbook.Worksheets.Count):
        sheet = workbook.Worksheets.get_Item(sheet_index)
        sheet_name = sheet.Name.replace(" ", "")          # 表名中去掉空格
        # 4. 读取表头（第一行）
        header = []
        for col in range(sheet.AllocatedRange.ColumnCount):
            raw_value = sheet.Range[1, col + 1].Value
            # 字段名中去掉空格，并避免空字段
            field_name = raw_value.replace(" ", "") if raw_value else f"col_{col}"
            header.append(field_name)
        # 5. 创建数据库表（所有字段暂定为 TEXT 类型）
        create_sql = f"""
        CREATE TABLE IF NOT EXISTS {sheet_name} (
            {', '.join([f'[{h}] TEXT' for h in header])}
        )
        """
        cursor.execute(create_sql)
        # 6. 逐行插入数据（跳过表头行）
        for row in range(1, sheet.AllocatedRange.RowCount):  # row=1 对应 Excel 第二行
            row_data = []
            for col in range(sheet.AllocatedRange.ColumnCount):
                cell_value = sheet.Range[row + 1, col + 1].Value
                row_data.append(cell_value)
            # 使用参数化查询防止 SQL 注入
            placeholders = ','.join(['?' for _ in row_data])
            insert_sql = f"INSERT INTO {sheet_name} ({','.join(header)}) VALUES ({placeholders})"
            cursor.execute(insert_sql, row_data)
    # 7. 提交并清理
    conn.commit()
    conn.close()
    workbook.Dispose()
if __name__ == "__main__":
    excel_to_sqlite("Sample.xlsx", "output/Report.db")

import pymysql
conn = pymysql.connect(host='localhost', user='root', password='123456', db='test')
cursor = conn.cursor()
# 建表时将 [field] 改为 `field`

sqlite3 database.db
.mode csv
.import data.csv table_name

import sqlite3
import pandas as pd
# 读取CSV并导入
df = pd.read_csv('data.csv')
conn = sqlite3.connect('database.db')
df.to_sql('table_name', conn, if_exists='replace', index=False)

sqlite3 database.db < dump.sql