用 Python 和 SQLite 检查 CSV 中的重复 ID

CSV 的每一条记录都符合格式要求，也可能出现同一个样本 ID 重复多次的问题。在连接标签、合并批次或统计独立样本数之前，需要单独检查。小文件可以用 Python set 记录已经见过的 ID，但每增加一个不同的 ID，内存中也会多一项。

这篇教程把 ID 存入临时 SQLite 数据库，逐条读取 CSV，输出重复 ID 的出现次数，以及首次和末次出现的记录位置。之前的 CSV 流式处理教程按标签统计记录数；这里补上独立的唯一性检查。

1. 先定义什么算重复

输入必须是 UTF-8 CSV，可以带字节顺序标记 BOM。表头必须包含 sample_id，其他列及列顺序不限。列名必须非空且互不重复，每条记录的字段数必须与表头一致。

ID 按原始文本精确比较。001 与 1 不同，A 与 a 也不同。脚本会拒绝空 ID 和首尾带空白的 ID，不会悄悄删除空白，也不会做 Unicode 规范化。其他字段的取值不在本次检查范围内。

报告只按 ID 分组。同一个 ID 即使标签不同，也会被列为重复。重复 ID 可能来自合理的重复测量、批次合并错误或元数据冲突，需要先核查，再决定是否删除记录。

2. 下载并检查环境

将 find_duplicate_ids.py 下载到新的工作目录。需要 Python 3.10 或更新版本，并且能够导入 sqlite3；SQLite 需要 3.24.0 或更新版本以支持 UPSERT。示例在 macOS、Python 3.14.7、SQLite 3.53.4 下验证，不需要 pip 包或独立数据库服务器。

python3 --version
python3 -c "import sqlite3; print(sqlite3.sqlite_version)"
python3 find_duplicate_ids.py --help

Python 的 sqlite3 模块可以直接打开磁盘数据库。部分自定义 Python 构建不包含该模块；如果导入失败，换用包含它的 Python 发行版本。

3. 执行重复检查

下面的 bash 或 zsh 命令创建一份虚构的小型输入和报告。请在临时工作目录中操作：cat > 和 > 会覆盖同名文件。不要把报告重定向到输入文件路径。

cat > samples.csv <<'CSV'
sample_id,label
001,1
002,0
001,0
003,1
002,0
002,1
CSV
python3 find_duplicate_ids.py samples.csv > duplicates.csv
cat duplicates.csv

脚本把下面的汇总写入标准错误：

records=6 unique_ids=3 duplicate_ids=2

标准输出中的 CSV 报告为：

sample_id,occurrences,first_record,last_record
001,2,1,3
002,3,2,6

这里的 duplicate_ids=2 表示有两个不同的 ID 出现了不止一次。每个 ID 保留首次出现之后，多出来的记录共有三条：6 - 3 = 3。这两个数对应不同的问题。

记录编号从表头之后的第一条数据开始计为 1。编号对应 CSV 解析后的记录，不是物理行号：带引号的字段可以包含换行。报告按 SQLite 的 BINARY 文本顺序排列 ID，不按数值大小或首次出现顺序排列。

格式有效且没有重复时，报告只包含表头。只有表头、没有数据的输入也有效。检查成功时返回退出状态 0，即使发现重复也一样；退出状态 1 表示输入、数据库或 I/O 错误。

4. 完整脚本

"""Find exact duplicate sample_id values in a UTF-8 CSV. Python 3.10+."""
import argparse
from contextlib import closing
import csv
from pathlib import Path
import sqlite3
import sys
import tempfile

UPSERT = """
INSERT INTO ids VALUES (?, 1, ?, ?)
ON CONFLICT(sample_id) DO UPDATE SET
    occurrences = ids.occurrences + 1,
    last_record = excluded.last_record
"""


def audit(input_path, output, *, work_dir=None):
    """Validate all input before writing a CSV report; return audit counts."""
    with tempfile.TemporaryDirectory(prefix="csv-ids-", dir=work_dir) as directory:
        with closing(sqlite3.connect(Path(directory) / "ids.sqlite3")) as db:
            db.execute("""CREATE TABLE ids (
                sample_id TEXT PRIMARY KEY NOT NULL,
                occurrences INTEGER NOT NULL,
                first_record INTEGER NOT NULL,
                last_record INTEGER NOT NULL
            )""")
            records = 0
            with db, open(input_path, encoding="utf-8-sig", newline="") as handle:
                reader = csv.reader(handle, strict=True)
                header = next(reader, None)
                if not header or any(not name for name in header):
                    raise ValueError("Expected a nonempty CSV header")
                if len(set(header)) != len(header) or "sample_id" not in header:
                    raise ValueError("Header must be unique and include sample_id")
                column = header.index("sample_id")
                for records, row in enumerate(reader, start=1):
                    if len(row) != len(header):
                        raise ValueError(f"Record {records}: wrong field count")
                    sample_id = row[column]
                    if not sample_id or sample_id != sample_id.strip():
                        raise ValueError(f"Record {records}: empty or padded sample_id")
                    db.execute(UPSERT, (sample_id, records, records))
            unique_ids = db.execute("SELECT COUNT(*) FROM ids").fetchone()[0]
            writer = csv.writer(output, lineterminator="\n")
            writer.writerow(["sample_id", "occurrences", "first_record", "last_record"])
            duplicate_ids = 0
            for row in db.execute("""SELECT * FROM ids WHERE occurrences > 1
                                     ORDER BY sample_id COLLATE BINARY"""):
                writer.writerow(row)
                duplicate_ids += 1
    return records, unique_ids, duplicate_ids


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("input", type=Path)
    parser.add_argument("--work-dir", type=Path, help="existing directory for scratch database")
    args = parser.parse_args()
    try:
        records, unique_ids, duplicates = audit(args.input, sys.stdout, work_dir=args.work_dir)
    except (OSError, UnicodeError, ValueError, csv.Error, sqlite3.Error) as exc:
        parser.exit(1, f"error: {exc}\n")
    print(f"records={records} unique_ids={unique_ids} duplicate_ids={duplicates}", file=sys.stderr)


if __name__ == "__main__":
    main()

CSV 解析器负责处理引号和字段内换行。使用 newline="" 打开文件，让解析器处理 CSV 换行；utf-8-sig 则接受文件开头可选的 BOM。具体行为见 Python 的 CSV 文档。

数据库每个不同的 ID 只保留一项。SQLite 的 UPSERT 语法在主键已存在时增加计数，并更新末次记录编号，首次记录编号保持不变。SQL 占位符将 ID 作为值传入，含引号的 ID 也能正常处理。

读取输入的循环在一个事务内执行。数据库上下文管理器在成功时提交，异常时回滚；外层的 closing(...) 另行关闭连接，然后才删除临时目录。参见连接上下文管理器文档。

脚本在全部输入完成解析和校验后才开始输出报告，避免后部坏记录留下看似完整的检查结果。不过，输出过程中发生写入错误，仍可能留下不完整报告。shell 重定向也会在 Python 启动之前创建或清空报告文件，因此这不是原子写入输出文件的实现。

5. 验证计数与错误处理

将下面的代码保存为 check_report.py，与 duplicates.csv 放在一起，运行 python3 check_report.py：

import csv

with open("duplicates.csv", encoding="utf-8", newline="") as handle:
    rows = list(csv.DictReader(handle))
assert rows == [
    {"sample_id": "001", "occurrences": "2", "first_record": "1", "last_record": "3"},
    {"sample_id": "002", "occurrences": "3", "first_record": "2", "last_record": "6"},
]
print("PASS: duplicate counts and record positions match")

再创建一份格式错误的输入，直接执行检查：

cat > invalid.csv <<'CSV'
sample_id,label
001,1
002
CSV
python3 find_duplicate_ids.py invalid.csv

预期错误如下，退出状态为 1，标准输出不产生 CSV：

error: Record 2: wrong field count

第二条数据只有一个字段，而表头要求两个。修正输入后重新运行，不要把重定向得到的空报告当成检查通过。

6. 选择临时数据库的位置

系统临时目录可能位于空间很小的分区，或者内存文件系统。Linux 上的 /tmp 有时是 tmpfs，写入其中的文件仍会消耗内存。如果目标是把 ID 索引移出内存，应明确选择位于合适磁盘文件系统的工作目录。

下面使用当前文件系统中的临时工作目录：

mkdir csv-scratch
python3 find_duplicate_ids.py samples.csv --work-dir csv-scratch > duplicates-second.csv

目录必须已经存在。每次运行会在其中新建自己的 csv-ids-... 子目录，正常结束或处理输入错误后会将其清理。强制终止进程可能留下临时文件。临时数据库不作为审计结果保存；需要追溯时，应保留 CSV 报告和输入文件校验清单。

这种写法避免了随不同 ID 数量增长的 Python 集合，但当前记录、表头、SQLite 页缓存和 I/O 缓冲仍会占用内存，不是严格的内存上限控制。数据库及索引的存储需求会随不同 ID 的数量增加。100,000 条记录的测试用于验证正确性，不代表速度或峰值内存测量。

7. 常见问题

问题	处理方法
缺少或重复的 `sample_id` 表头	修正列名，大小写与拼写必须一致。
ID 为空或首尾有空白	修正源数据，或在上游明确规范化规则。
字段数错误或 CSV 解析错误	检查引号与分隔符；脚本要求逗号分隔的 CSV。
数据库或磁盘空间不足	选择有足够空间容纳数据库和报告的工作目录。
超大字段被拒绝	检查输入，确有需要时明确调整 `csv.field_size_limit`。
同一 ID 对应冲突标签	核查原始记录；报告不判断哪个标签正确。

请使用内容稳定的输入文件，脚本不会锁定文件或创建快照。如果数据量小到可以轻松放入内存，内存中的方法可能更简单。本文不宣称 SQLite 一定更快。

8. 小结

逐条读取 CSV，把不断增长的 ID 索引放到合适文件系统上的临时 SQLite 数据库。先检查程序退出状态，再根据报告中的记录位置核查重复 ID。检测重复与决定如何处理重复，应分开完成。