本文详解如何在使用 pandas 持续更新 csv 日志文件时，确保邮箱字段唯一、计数准确，彻底解决因并发检查与写入逻辑缺陷导致的重复条目问题。
本文详解如何在使用 pandas 持续更新 csv 日志文件时，确保邮箱字段唯一、计数准确，彻底解决因并发检查与写入逻辑缺陷导致的重复条目问题。
在日志类场景（如统计用户操作的成功/失败次数）中，常需以邮箱为唯一标识持续更新 CSV 文件。但原代码存在一个根本性逻辑缺陷：它依赖 email in df["email"].values 做存在性判断，再执行更新或追加——这看似合理，却忽略了两个关键风险：
- 竞态条件（Race Condition）：若多个进程/线程或高频调用 save_data()，两次读取 CSV 的间隔中，另一调用可能已写入同邮箱，导致两次都判定“不存在”而各自追加；
- _append() 已弃用且非原子操作：DataFrame._append() 在 Pandas 2.0+ 中已被弃用，且追加后未做去重，无法修复已有重复。
✅ 正确解法不是“事后去重”，而是在合并阶段就保证语义正确性：使用 pd.concat() + groupby().agg() 或更推荐的 set_index().combine_first()，实现“读取→合并→去重→聚合→写入”原子流程。
✅ 推荐方案：基于 set_index 的幂等更新（推荐）
该方法将邮箱设为索引，利用 combine_first() 自动处理新旧数据合并，并天然去重：
import pandas as pd
from datetime import datetime
import os
def save_data(email: str, is_success: int, is_failed: int) -> None:
csv_file_path = "data.csv"
# 构建当前记录（单行 DataFrame）
new_df = pd.DataFrame([{
"email": email,
"success_count": is_success,
"failure_count": is_failed,
"last_updated_on": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
}]).set_index("email")
# 读取现有数据（若存在），并设 index 为 email
if os.path.exists(csv_file_path):
try:
existing_df = pd.read_csv(csv_file_path).set_index("email")
# 合并：new_df 覆盖 existing_df 中同 email 行，其余保留 existing_df
# 注意：combine_first 是左优先，此处需先将 new_df 作为右参数，故用 existing_df.combine_first(new_df)
# 但我们需要的是「新值累加」而非覆盖，因此改用 groupby 累加
merged_df = pd.concat([existing_df, new_df]).groupby(level=0).sum(numeric_only=True)
# 手动更新 last_updated_on 为最新时间（取 new_df 的时间）
merged_df["last_updated_on"] = new_df["last_updated_on"].reindex(merged_df.index).fillna(
existing_df["last_updated_on"]
)
except Exception as e:
print(f"Warning: Failed to read existing CSV, will overwrite. Error: {e}")
merged_df = new_df.reset_index()
else:
merged_df = new_df.reset_index()
# 写入 CSV（确保列顺序一致）
merged_df = merged_df.reset_index(drop=True)[
["email", "success_count", "failure_count", "last_updated_on"]
]
merged_df.to_csv(csv_file_path, index=False)
# 测试调用（同原示例）
if __name__ == "__main__":
arr = [
("123456", 1, 0),
("456789", 0, 1),
("789012", 1, 0),
("123456", 0, 1), # 重复邮箱：应累加 success=1, failure=1
("456789", 1, 0), # 应累加为 success=1, failure=1
("789012", 0, 1), # 应累加为 success=1, failure=1
]
for data in arr:
email, is_success, is_failed = data
save_data(email=email, is_success=is_success, is_failed=is_failed)
⚠️ 关键注意事项
- 不要依赖 drop_duplicates() 事后清理：它只能删除已有重复，无法解决“重复插入”的根源，且在高并发下仍会短暂产生脏数据；
- 避免 _append() 和 loc/iloc 手动索引更新：易出错、不可扩展、不兼容新版 Pandas；
- groupby().sum() 是累加核心：对数值列（success_count, failure_count）自动聚合，无需手动查找索引；
- 时间戳处理需显式逻辑：last_updated_on 非数值型，不能 sum()，需用 reindex() + fillna() 保证每行取最新时间；
- 生产环境建议加锁或换用数据库：CSV 本质是文件，无事务支持；极高频写入请改用 SQLite 或 Redis。
✅ 最终效果验证
运行后 data.csv 将严格包含 3 行唯一邮箱，对应计数均为 success_count=1, failure_count=1，last_updated_on 为第二次写入的时间，完全符合业务预期。
坚持“以索引为键、以聚合为操作、以原子写入为终点”的设计原则，即可稳健规避 CSV 重复陷阱。
