使用Pandas合并CSV并插入SQLite时的重复行更新问题

问题概述 #

使用Pandas合并多个CSV文件并插入SQLite表，首次运行正常，但后续更新时总是出现行重复；可以添加新的合并列，但无法在表的现有行末尾补充新内容。表结构包含name_ID、Number_ID列，以及对应一年中每一天的365个列。

当前代码实现 #

读取CSV文件并获取指定列 #

import pandas as pd
import sqlite3

pd.set_option('display.max_columns', 6)
# 修正原代码列索引的语法错误：需双层方括号
dia1 = pd.read_csv('dia0705.csv', header=1, sep=";", dtype='unicode')[["EO", "NOME ACIONISTA", "CPF/CNPJ"]]
dia2 = pd.read_csv('dia0712.csv', header=1, sep=";", dtype='unicode')[["EO", "NOME ACIONISTA", "CPF/CNPJ"]]

dia1['CPF/CNPJ'] = dia1['CPF/CNPJ'].astype(str)
dia2['CPF/CNPJ'] = dia2['CPF/CNPJ'].astype(str)

dia1['NOME ACIONISTA'] = dia1['NOME ACIONISTA'].astype(str)
dia2['NOME ACIONISTA'] = dia2['NOME ACIONISTA'].astype(str)

合并数据并修改列名 #

merge1 = pd.merge(dia1, dia2, how='outer', on=["NOME ACIONISTA", "CPF/CNPJ"])  # indicator=True)

merge1.rename(columns={"EO_x": "dia0705"}, inplace=True)
merge1.rename(columns={"EO_y": "dia0712"}, inplace=True)
merge1.rename(columns={"NOME ACIONISTA": "Nome_Acionista"}, inplace=True)
# 修正原代码缺失的闭合括号
merge1.rename(columns={"CPF/CNPJ": "CPF_CNPJ"}, inplace=True)

重复合并多个CSV文件 #

merge2 = pd.merge(merge1, dia3, how='outer', on=["Nome_Acionista", "CPF_CNPJ"])

dia4 = pd.read_csv('220913_completo.csv', header=1, sep=";", dtype='unicode')[["EO", "NOME ACIONISTA", "CPF/CNPJ"]]
dia4.rename(columns={"NOME ACIONISTA": "Nome_Acionista"}, inplace=True)
dia4.rename(columns={"CPF/CNPJ": "CPF_CNPJ"}, inplace=True)
dia4.rename(columns={"EO": "dia0913"}, inplace=True)

merge3 = pd.merge(merge2, dia4, how='outer', on=["Nome_Acionista", "CPF_CNPJ"])

# 原代码重复读取dia4，此处保留原样
dia4 = pd.read_csv('220913_completo.csv', header=1, sep=";", dtype='unicode')[["EO", "NOME ACIONISTA", "CPF/CNPJ"]]
dia4.rename(columns={"NOME ACIONISTA": "Nome_Acionista"}, inplace=True)
dia4.rename(columns={"CPF/CNPJ": "CPF_CNPJ"}, inplace=True)
dia4.rename(columns={"EO": "dia0913"}, inplace=True)

连接SQLite并插入数据 #

connection = sqlite3.connect('2022.db')
c = connection.cursor()

merge3.to_sql(
        name='acoes',
        con=connection,
        if_exists='append',
        index=False,
    )

问题原因与解决方案 #

重复行的根源 #

if_exists='append'会直接将新数据追加到表中，不做重复校验。后续重复运行合并插入时，已有行会被再次添加，导致重复。

无法更新现有行的原因 #

SQLite的to_sql无直接的"更新插入（Upsert）"能力，单纯追加只会新增行，无法为已有行补充新日期列的数据。

具体解决步骤 #

1. 设置唯一主键约束
   以Nome_Acionista和CPF_CNPJ（或表中已有的name_ID/Number_ID）作为唯一标识，在SQLite表中添加主键约束，从根源阻止重复行：

   -- 若表已存在，添加主键约束
   ALTER TABLE acoes ADD PRIMARY KEY (Nome_Acionista, CPF_CNPJ);
   -- 若新建表，直接指定主键
   CREATE TABLE acoes (
       name_ID INTEGER,
       Number_ID INTEGER,
       Nome_Acionista TEXT,
       CPF_CNPJ TEXT,
       dia0705 TEXT,
       dia0712 TEXT,
       -- 其他日期列...
       PRIMARY KEY (Nome_Acionista, CPF_CNPJ)
   );
   

2. 使用Upsert逻辑实现更新插入
   借助SQLite的INSERT OR REPLACE语法，先将新数据写入临时表，再合并到主表：

   # 将新合并的数据写入临时表
   merge3.to_sql(name='temp_acoes', con=connection, if_exists='replace', index=False)
   
   # 执行Upsert：更新已有行的新日期列，插入全新行
   upsert_query = """
   INSERT OR REPLACE INTO acoes (Nome_Acionista, CPF_CNPJ, dia0913 /* 补充其他新增列 */)
   SELECT Nome_Acionista, CPF_CNPJ, dia0913 /* 对应新增列 */
   FROM temp_acoes
   """
   c.execute(upsert_query)
   connection.commit()
   
   # 删除临时表
   c.execute("DROP TABLE temp_acoes")
   

3. 优化合并流程
   无需每次重新合并所有历史CSV，直接读取SQLite现有数据与新CSV合并：

   # 读取表中已有数据
   existing_df = pd.read_sql("SELECT * FROM acoes", connection)
   # 读取并处理新CSV
   new_dia = pd.read_csv('new_dia.csv', header=1, sep=";", dtype='unicode')[["EO", "NOME ACIONISTA", "CPF/CNPJ"]]
   new_dia.rename(columns={
       "NOME ACIONISTA": "Nome_Acionista",
       "CPF/CNPJ": "CPF_CNPJ",
       "EO": "new_dia_date"  # 替换为实际日期列名
   }, inplace=True)
   # 合并现有数据与新数据
   updated_df = pd.merge(existing_df, new_dia, how='outer', on=["Nome_Acionista", "CPF_CNPJ"])
   # 覆盖写入主表（需确保主键约束已设置）
   updated_df.to_sql(name='acoes', con=connection, if_exists='replace', index=False)
   

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