以下是一个将 CSV 文件导入 SQLite 数据库的 Python 脚本，并附带详细讲解。

---

1. 脚本功能概述

该脚本的功能是将 CSV 文件中的数据导入 SQLite 数据库。具体步骤如下：

1. 创建 SQLite 数据库连接。
2. 根据 CSV 文件的内容创建表。
3. 将 CSV 文件中的数据插入到表中。

---

2. 完整代码

import sqlite3 
import csv 
import sys 
from typing import Optional 
 
def create_connection(db_file: str) -> Optional [sqlite3.Connection]:
    """ 创建 SQLite 数据库连接 """
    try:
        conn = sqlite3.connect(db_file)
        print("成功连接到数据库！")
        return conn 
    except sqlite3.Error as e:
        print(f"连接数据库时出错：{e}")
        return None 
 
def create_table(conn: sqlite3.Connection, table_name: str, headers: list):
    """ 根据 CSV 文件头创建表 """
    try:
        # 将字段名转换为 SQL 语句中的列定义 
        columns = ", ".join([f'"{header}" TEXT' for header in headers])
        sql = f'CREATE TABLE IF NOT EXISTS "{table_name}" ({columns})'
        conn.execute(sql)
        print(f"成功创建表 {table_name}！")
    except sqlite3.Error as e:
        print(f"创建表时出错：{e}")
 
def insert_data(conn: sqlite3.Connection, table_name: str, rows: list, headers: list):
    """ 插入数据到表中 """
    try:
        # 准备占位符 
        placeholders = ", ".join(["?"] * len(headers))
        # 准备 SQL 插入语句 
        sql = f'INSERT INTO "{table_name}" VALUES ({placeholders})'
        
        # 执行批量插入 
        conn.executemany(sql, rows)
        conn.commit()
        print(f"成功插入 {len(rows)} 条记录！")
    except sqlite3.Error as e:
        print(f"插入数据时出错：{e}")
 
def main(csv_file: str, db_file: str, table_name: str):
    """ 主函数 """
    try:
        # 读取 CSV 文件 
        with open(csv_file, 'r', encoding='utf-8') as file:
            reader = csv.reader(file)
            headers = next(reader)  # 获取表头 
            rows = list(reader)     # 获取所有数据行 
            
        # 创建数据库连接 
        conn = create_connection(db_file)
        if conn is None:
            return 
        
        # 创建表 
        create_table(conn, table_name, headers)
        
        # 插入数据 
        insert_data(conn, table_name, rows, headers)
        
        # 关闭连接 
        conn.close()
        print("数据库连接已关闭！")
        
    except FileNotFoundError as e:
        print(f"文件未找到：{e}")
    except Exception as e:
        print(f"程序运行时出错：{e}")
 
if __name__ == "__main__":
    # 解析命令行参数 
    if len(sys.argv) != 4:
        print("使用方法：python csv_to_sqlite.py <CSV文件路径> <数据库文件路径> <表名>")
        sys.exit(1)
    
    csv_file = sys.argv[1]
    db_file = sys.argv[2]
    table_name = sys.argv[3]
    
    main(csv_file, db_file, table_name)

---

3. 调用方式

3.1 命令行调用

在终端中运行脚本，需要提供三个参数：

1. CSV 文件路径
2. SQLite 数据库文件路径
3. 表名

示例：

python csv_to_sqlite.py example.csv mydatabase.db mytable 

3.2 参数说明

• example.csv：待导入的 CSV 文件路径。
• mydatabase.db：生成的 SQLite 数据库文件路径。
• mytable：数据库中存储数据的表名。

---

4. 代码讲解

4.1 导入库

import sqlite3 
import csv 
import sys 
from typing import Optional 

• sqlite3：用于操作 SQLite 数据库。
• csv：用于读取 CSV 文件。
• sys：用于处理命令行参数。
• Optional：用于类型提示。

4.2 创建数据库连接

def create_connection(db_file: str) -> Optional[sqlite3.Connection]:
    """ 创建 SQLite 数据库连接 """
    try:
        conn = sqlite3.connect(db_file)
        print("成功连接到数据库！")
        return conn 
    except sqlite3.Error as e:
        print(f"连接数据库时出错：{e}")
        return None 

• 使用 sqlite3.connect() 创建数据库连接。
• 如果连接失败，返回 None。

4.3 创建表

def create_table(conn: sqlite3.Connection, table_name: str, headers: list):
    """ 根据 CSV 文件头创建表 """
    try:
        # 将字段名转换为 SQL 语句中的列定义 
        columns = ", ".join([f'"{header}" TEXT' for header in headers])
        sql = f'CREATE TABLE IF NOT EXISTS "{table_name}" ({columns})'
        conn.execute(sql)
        print(f"成功创建表 {table_name}！")
    except sqlite3.Error as e:
        print(f"创建表时出错：{e}")

• 根据 CSV 文件的表头动态生成 SQL 创建表的语句。
• 每个字段默认为 TEXT 类型。

4.4 插入数据

def insert_data(conn: sqlite3.Connection, table_name: str, rows: list, headers: list):
    """ 插入数据到表中 """
    try:
        # 准备占位符 
        placeholders = ", ".join(["?"] * len(headers))
        # 准备 SQL 插入语句 
        sql = f'INSERT INTO "{table_name}" VALUES ({placeholders})'
        
        # 执行批量插入 
        conn.executemany(sql, rows)
        conn.commit()
        print(f"成功插入 {len(rows)} 条记录！")
    except sqlite3.Error as e:
        print(f"插入数据时出错：{e}")

• 使用 executemany() 批量插入数据以提高效率。
• 使用 ? 占位符避免 SQL 注入。

4.5 主函数

def main(csv_file: str, db_file: str, table_name: str):
    """ 主函数 """
    try:
        # 读取 CSV 文件 
        with open(csv_file, 'r', encoding='utf-8') as file:
            reader = csv.reader(file)
            headers = next(reader)  # 获取表头 
            rows = list(reader)     # 获取所有数据行 
            
        # 创建数据库连接 
        conn = create_connection(db_file)
        if conn is None:
            return 
        
        # 创建表 
        create_table(conn, table_name, headers)
        
        # 插入数据 
        insert_data(conn, table_name, rows, headers)
        
        # 关闭连接 
        conn.close()
        print("数据库连接已关闭！")
        
    except FileNotFoundError as e:
        print(f"文件未找到：{e}")
    except Exception as e:
        print(f"程序运行时出错：{e}")

• 读取 CSV 文件并提取表头和数据行。
• 调用其他函数完成数据库操作。

4.6 命令行参数解析

if __name__ == "__main__":
    # 解析命令行参数 
    if len(sys.argv) != 4:
        print("使用方法：python csv_to_sqlite.py <CSV文件路径> <数据库文件路径> <表名>")
        sys.exit(1)
    
    csv_file = sys.argv[1]
    db_file = sys.argv[2]
    table_name = sys.argv[3]
    
    main(csv_file, db_file, table_name)

• 检查命令行参数数量是否正确。
• 提取 CSV 文件路径、数据库文件路径和表名。

---

5. 注意事项

1. 编码问题：确保 CSV 文件的编码为 UTF-8，否则可能会出现乱码。
2. 字段类型：当前脚本将所有字段设为 TEXT 类型。如果需要其他类型（如 INTEGER 或 REAL），可以根据实际需求修改。
3. 性能优化：对于大数据量的 CSV 文件，可以分批插入数据以提高效率。
4. 异常处理：脚本已经包含基本的异常处理，但可以根据实际需求进一步完善。