如何使用Python合并存在自增主键ID冲突的两个SQLite数据库并保留唯一数据行

嘿，这个分布式SQLite合并去重的问题我之前也帮人处理过，完全贴合你的场景——各自独立生成自增ID导致合并时冲突，还不想硬编码列名（毕竟未来可能加更多列）对吧？别担心，有两个成熟的方案，不管是纯SQL操作还是Python自动化脚本都能完美解决：

方案1：纯SQL合并（适合直接操作数据库，无需改代码） #

如果你的SQLite版本在3.35.0及以上（这个版本之后支持了灵活的列排除语法），直接用SQL就能搞定，不用手动列所有字段：

1. 先打开其中一个数据库（比如userA的），把另一个数据库（userB的）附加进来：

ATTACH DATABASE 'userB_test_score.db' AS userb_db;

2. 执行插入语句，只把userB数据库中除ID外所有字段组合唯一的行插入到主数据库，让主数据库自动生成新的自增ID：

INSERT INTO scores
SELECT * EXCEPT(id)
FROM userb_db.scores
WHERE NOT EXISTS (
    SELECT 1 FROM scores
    WHERE (scores.* EXCEPT(id)) = (userb_db.scores.* EXCEPT(id))
);

这个* EXCEPT(id)语法会自动排除id列，不管未来你加多少列，都不用修改这条SQL，完美解决你不想逐列比较的痛点！

如果你的SQLite版本比较旧，不支持* EXCEPT，那可以用下面的Python方案，动态获取列名来生成SQL。

方案2：Python自动化脚本（适合批量处理或旧版本SQLite） #

这个脚本会自动读取表的列结构，动态生成去重条件，未来加列也不用改代码，完全适配你的需求：

import sqlite3

def merge_deduplicate_scores(main_db_path, merge_db_path):
    # 连接主数据库（比如userA的数据库）
    main_conn = sqlite3.connect(main_db_path)
    main_cursor = main_conn.cursor()
    
    # 附加要合并的数据库（userB的数据库）
    main_cursor.execute(f"ATTACH DATABASE '{merge_db_path}' AS merge_db;")
    
    # 获取scores表的所有列名，排除id列
    main_cursor.execute("PRAGMA table_info(scores);")
    all_columns = [col[1] for col in main_cursor.fetchall()]
    non_id_columns = [col for col in all_columns if col != 'id']
    
    # 动态生成列列表和去重比较条件
    column_str = ', '.join(non_id_columns)
    compare_conditions = ' AND '.join([f"scores.{col} = merge_db.scores.{col}" for col in non_id_columns])
    
    # 执行去重插入：只插入主数据库不存在的唯一行
    insert_sql = f"""
        INSERT INTO scores ({column_str})
        SELECT {column_str}
        FROM merge_db.scores
        WHERE NOT EXISTS (
            SELECT 1 FROM scores
            WHERE {compare_conditions}
        );
    """
    main_cursor.execute(insert_sql)
    
    # 提交更改并关闭连接
    main_conn.commit()
    main_conn.close()

# 调用示例：把userB的数据库合并到userA的数据库里
merge_deduplicate_scores('userA_test_score.db', 'userB_test_score.db')

另外，顺便提一下你原来的Python插入代码里的小问题：你写了on.commit()应该是con.commit()，而且with con:上下文管理器会自动帮你提交事务、关闭游标和连接，所以可以简化成这样：

import sqlite3 as lite

with lite.connect('test_score.db') as con:
    cur = con.cursor()
    # 插入时不用传NULL给id，SQLite自增主键会自动生成ID
    cur.execute("INSERT INTO scores (first, last, score) VALUES (?,?,?)", (first, last, score))
# 离开with块后，连接会自动关闭，事务自动提交

验证合并结果 #

执行完上面的方法后，查询主数据库的scores表：

SELECT * FROM scores;

就能得到你预期的结果：

1,Adam,Smith,68
2,John,Snow,76
3,Jim,Green,88
4,Jim,Green,91
5,Tom,Hanks,15
6,Chris,Prat,99
7,Tom,Hanks,09

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