本文介绍一种使用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) + 布尔索引 的声明式方案，兼具可读性、健壮性与性能：

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 或流式处理避免内存压力。

该方案彻底规避了原始脚本的隐式索引错位与逻辑耦合问题，以清晰的数据流和向量化操作达成高可靠性与可维护性。