本文介绍一种使用pandas高效实现“按指定列（第3、4、5列）判断唯一性，从排序后的源csv中选取首个不重复行，写入新文件并同步更新主表”的完整方案，解决原始脚本中误判重复、逻辑错位等核心问题。
在处理类似 tops.csv（主数据表）与 yesterday.csv（待筛选增量数据）的场景时，常见需求是：对 yesterday.csv 按关键列（如第11列降序 + 第2列升序）排序后，逐行检查其第3、4、5列（即索引为2、3、4的列）是否已在 tops.csv 中存在；一旦找到首个“三列组合”完全未出现过的行，就将其写入 for_email.csv 并追加至 tops.csv。
原始脚本存在多个关键缺陷：
- ❌ df_sorted.drop_duplicates(subset=df_sorted.columns[2:]) 错误地基于全部后续列去重，而非仅限目标三列（2,3,4）；
- ❌ tops_df[tops_df.columns[2:]].eq(row.iloc[2:]).all(axis=1).any() 的逐行比较逻辑未对齐索引，且未考虑列类型/空值鲁棒性；
- ❌ 循环中 break 位置过早（写入1行即终止），但未确保该行真实“未存在于 tops”——因前面的 drop_duplicates 已破坏原始排序优先级；
- ❌ 手动字符串拼接写入 CSV 绕过了 Pandas 的类型保持与转义机制，易引发格式错误。
✅ 推荐采用 pd.merge(..., how='outer', indicator=True) + 布尔索引 的声明式方案，兼具可读性、健壮性与性能：
QuickBooks Mileage CSV
下载
使用 qbo-mileage CLI 及用户凭证，从 Airtable、Outlook 或 Google Calendar 记录生成 QuickBooks Online 里程 CSV 文件。
import pandas as pd
# 1. 加载数据（无header，列索引从0开始）
tops = pd.read_csv('tops.csv', header=None)
yesterday = pd.read_csv('yesterday.csv', header=None)
# 2. 对 yesterday 按要求排序：第11列（索引10）降序，第2列（索引1）升序
yesterday_sorted = yesterday.sort_values(by=[10, 1], ascending=[False, True])
# 3. 外连接 + 指示器列，标记每行来源
merged = pd.merge(
tops,
yesterday_sorted,
how='outer',
indicator=True,
validate='one_to_one' # 可选：校验无重复匹配
)
# 4. 标识需保留的 "yesterday" 行：来源为 'right_only' 且 (2,3,4) 列组合在 tops 中未出现
# 注意：duplicated 默认保留首次出现，因此先标记所有重复项（含 tops 自身重复），再过滤
is_dup_in_key = merged.duplicated(subset=[2, 3, 4], keep=False) # 所有在key上重复的行
is_from_yesterday = merged['_merge'] == 'right_only'
# 保留：来自 yesterday 且其 key 在整个 merged 中首次出现（即未在 tops 中存在）
mask = is_from_yesterday & ~merged.duplicated(subset=[2, 3, 4], keep='first')
# 5. 提取目标行，并写入 for_email.csv
for_email_row = merged[mask].drop('_merge', axis=1)
for_email_row.to_csv('for_email.csv', index=False, header=False)
# 6. 将新行追加到 tops.csv（注意：若 tops 无header，append 模式需确保格式一致）
with open('tops.csv', 'a', newline='') as f:
for_email_row.to_csv(f, index=False, header=False)
print(f"已成功写入 {len(for_email_row)} 行至 for_email.csv 并追加至 tops.csv")? 关键设计说明：
- 排序前置：yesterday.sort_values(...) 确保最高优先级行排在最前，duplicated(..., keep='first') 才能准确捕获“首个未出现的行”；
- 精准去重键：显式指定 subset=[2,3,4]（对应第3、4、5列），避免列范围误用；
- 指示器驱动逻辑：_merge == 'right_only' 精确标识纯新增行，规避 tops 内部重复干扰；
- 原子性保障：to_csv(..., header=False) 确保追加时不写入列名，与原始 tops.csv 格式严格一致；
- 健壮性增强：keep='first' 在合并后数据中，将 tops 的原始行视为“首次出现”，yesterday 的同key行自然被标记为重复，从而被 ~duplicated(..., keep='first') 过滤。
? 扩展建议：
- 若需支持多行写入（如TOP-3），将 mask 替换为 mask.nlargest(3, columns=[10,1]) 或结合 cumsum() 限制数量；
- 生产环境建议添加异常处理（如文件不存在、编码错误）及日志记录；
- 对超大文件，可改用 dask 或流式处理避免内存压力。
该方案彻底规避了原始脚本的隐式索引错位与逻辑耦合问题，以清晰的数据流和向量化操作达成高可靠性与可维护性。
