使用Python解析含整数与字符串的CSV并通过Insert语句写入Oracle表的报错排查与数据类型处理

我来帮你拆解问题并给出可行的解决方案：

为什么会出现ORA-01036错误？ #

你的核心问题在于混淆了SQL表达式与参数化查询的用法：

• 你的CSV数据里包含seq_addnoteid.nextval、sysdate这类Oracle原生SQL表达式，但你用?占位符把它们当作普通字符串参数传递给数据库。
• Oracle不会将参数值解析为可执行的SQL表达式，反而会把这些值当成非法的变量/字符串，同时还会触发字段类型不匹配的问题，最终抛出ORA-01036错误。
• 另外你同时维护了「获取max ID自增」和「使用序列」两种ID生成逻辑，这本身就是冲突的，必须二选一。

推荐解决方案：使用数据库序列（更可靠，避免并发冲突） #

序列是Oracle官方推荐的自增ID生成方式，比自己查max ID更安全（不会出现多线程插入时的主键冲突）。下面是修改后的代码：

import csv
from datetime import datetime
import cx_Oracle  # 假设你使用cx_Oracle连接Oracle

# 假设你已经建立好数据库连接conn
cursor = conn.cursor()

with open('sample.csv', 'r') as f:
    reader = csv.reader(f)
    columns = next(reader)  # 获取表头
    
    for data in reader:
        placeholders = []
        params = []
        for col, val in zip(columns, data):
            # 区分Oracle表达式和普通参数
            if val in ['seq_addnoteid.nextval', 'sysdate']:
                # 表达式直接写入SQL，不使用占位符
                placeholders.append(val)
            elif col == 'CREATION_DATE':
                # 处理自定义日期字符串，显式转换避免格式问题
                placeholders.append("TO_DATE(?, 'DD-MON-RR')")
                params.append(val)
            else:
                # 普通字段用参数化占位符，防止SQL注入
                placeholders.append('?')
                params.append(val)
        
        # 动态构建插入语句
        insert_query = f"""
            INSERT INTO NOTES({','.join(columns)})
            VALUES ({','.join(placeholders)})
        """
        cursor.execute(insert_query, params)

conn.commit()
print("Records inserted successfully")
cursor.close()
conn.close()

各字段数据类型适配说明 #

• ID字段：直接使用seq_addnoteid.nextval，数据库会自动生成数值类型的主键，无需额外处理。
• CREATION_DATE字段：
  • 如果是sysdate，直接让Oracle执行该表达式获取当前时间；
  • 如果是19-MAY-22这类字符串，用TO_DATE(?, 'DD-MON-RR')显式转换，避免因数据库日期格式设置导致的转换失败。
• ISIMPORTANT字段：CSV中的Y是字符串类型，直接传递给数据库的字符型字段即可。
• TYPE/ENTITY_TYPE字段：数字和字符类型会被cx_Oracle自动映射为对应的Oracle数据类型，无需手动转换。

如果你坚持使用自增max ID（不推荐，存在并发风险） #

如果一定要用自己维护的自增ID，需要去掉CSV里的seq_addnoteid.nextval，改为代码生成ID：

import csv
from datetime import datetime

cursor = conn.cursor()
# 获取最大ID，处理表为空的情况
cursor.execute("SELECT MAX(ID) from NOTES")
max_id = cursor.fetchone()[0]
max_id = int(max_id) if max_id is not None else 0

with open('sample.csv', 'r') as f:
    reader = csv.reader(f)
    columns = next(reader)
    id_index = columns.index('ID')
    date_index = columns.index('CREATION_DATE')
    
    # 构建通用占位符语句
    placeholders = ','.join('?' * len(columns))
    insert_query = f"INSERT INTO NOTES({','.join(columns)}) VALUES ({placeholders})"
    
    for data in reader:
        # 替换ID为自增值
        data[id_index] = max_id + 1
        max_id += 1
        # 处理日期字段
        if data[date_index] == 'sysdate':
            data[date_index] = datetime.now()
        cursor.execute(insert_query, data)

conn.commit()
print("Records inserted successfully")
cursor.close()
conn.close()

⚠️ 注意：这种方式在多线程/多进程插入场景下会出现主键冲突，仅适合单线程低并发的场景。

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