使用Pandas操作CSV时如何避免重复条目并正确更新计数？

我需要用Python Pandas在CSV文件中记录success_count（成功次数）和failure_count（失败次数），要求每个email条目唯一，根据传入参数递增对应计数。但实际输出的CSV中出现了重复的email条目，尽管代码里已经检查了email是否存在，问题依然存在。

以下是可复现问题的完整代码：

import pandas as pd
from datetime import datetime


def save_data(email: str, is_success: int, is_failed: int) -> None:
    csv_file_path = "data.csv"

    # 检查CSV文件是否存在并处理空文件情况
    try:
        df = pd.read_csv(csv_file_path)
    except FileNotFoundError:
        df = pd.DataFrame(
            columns=["email", "success_count", "failure_count", "last_updated_on"]
        )

    # 检查重复并更新计数
    if email in df["email"].values:
        index = df[df["email"] == email].index[0]
        df.at[index, "failure_count"] += is_failed
        df.at[index, "success_count"] += is_success
        df.at[index, "last_updated_on"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    else:
        # 添加新条目
        new_entry = {
            "email": email,
            "success_count": is_success,
            "failure_count": is_failed,
            "last_updated_on": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        }
        df = df._append(new_entry, ignore_index=True)

    # 写入CSV文件
    try:
        df.to_csv(csv_file_path, index=False)
    except Exception as e:
        print("写入CSV文件时出错:", e)


if __name__ == "__main__":
    arr = [
        ("123456", 1, 0),
        ("456789", 0, 1),
        ("789012", 1, 0),
        #
        ("123456", 0, 1),
        ("456789", 1, 0),
        ("789012", 0, 1),
    ]

    for data in arr:
        email, is_success, is_failed = data
        save_data(email=email, is_success=is_success, is_failed=is_failed)

问题原因 #

核心问题是Pandas自动类型转换导致的匹配失败：
第一次写入CSV时，email列的内容是纯数字字符串（如"123456"），但第二次调用pd.read_csv时，Pandas会自动推断将该列解析为整数类型。后续传入的email是字符串类型，与DataFrame中存储的整数类型不匹配，导致email in df["email"].values判断为False，从而重复创建新条目。

另外，代码中使用了Pandas的私有方法_append，虽然能运行，但并非官方推荐的公共API，存在潜在兼容性问题。

修复方案 #

1. 强制指定email列类型为字符串：读取CSV时通过dtype参数明确指定email列类型，避免自动类型转换。
2. 替换私有方法为公共API：用pd.concat替代_append（Pandas 2.0+中append方法已被弃用，concat是更稳妥的选择）。

修改后的代码 #

import pandas as pd
from datetime import datetime


def save_data(email: str, is_success: int, is_failed: int) -> None:
    csv_file_path = "data.csv"

    # 读取CSV时强制指定email列为字符串类型，避免自动类型转换
    try:
        df = pd.read_csv(csv_file_path, dtype={"email": str})
    except FileNotFoundError:
        df = pd.DataFrame(
            columns=["email", "success_count", "failure_count", "last_updated_on"]
        )

    # 检查email是否存在并更新计数
    if email in df["email"].values:
        index = df[df["email"] == email].index[0]
        df.at[index, "failure_count"] += is_failed
        df.at[index, "success_count"] += is_success
        df.at[index, "last_updated_on"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    else:
        # 用pd.concat替代私有_append方法
        new_entry = pd.DataFrame([{
            "email": email,
            "success_count": is_success,
            "failure_count": is_failed,
            "last_updated_on": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        }])
        df = pd.concat([df, new_entry], ignore_index=True)

    # 写入CSV
    try:
        df.to_csv(csv_file_path, index=False)
    except Exception as e:
        print("写入CSV文件时出错:", e)


if __name__ == "__main__":
    arr = [
        ("123456", 1, 0),
        ("456789", 0, 1),
        ("789012", 1, 0),
        ("123456", 0, 1),
        ("456789", 1, 0),
        ("789012", 0, 1),
    ]

    for data in arr:
        email, is_success, is_failed = data
        save_data(email=email, is_success=is_success, is_failed=is_failed)

验证结果 #

运行修改后的代码，data.csv中每个email只会保留唯一条目，对应的success_count和failure_count会正确累加，例如：

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