如何避免CSV文件重复写入相同数据？技术实现求助

面对这种定时运行、不能重复写入的需求，我给你两个最常用的解决思路，看哪种更适配你的场景：

方案一：数据库标记法（最推荐） #

既然你的数据来自数据库，而且查询条件是check=0，那最根本的办法就是写入CSV后，把这些已处理的记录标记为已完成，这样下次查询就不会再拿到它们了。

修改后的代码大概是这样：

import csv
import psycopg2  # 假设你用的是PostgreSQL，其他数据库逻辑类似

# 初始化数据库连接
conn = psycopg2.connect(host="your_host", dbname="your_db", user="your_user", password="your_pwd")
try:
    conn.autocommit = False  # 开启事务，保证写入和标记操作的原子性
    read_cursor = conn.cursor()
    # 查询所有未处理的记录
    read_cursor.execute("select item, check from items where check=0")
    records = list(read_cursor)  # 先把查询结果存到内存，方便后续处理

    if records:
        # 写入CSV：如果要保留所有历史记录用"a"追加，若只存最新批次用"w"覆盖
        with open('quantity_checker.csv', "a", newline="") as f:
            writer = csv.writer(f, delimiter=';')
            # 判断文件是否为空，为空则先写表头
            import os
            if os.stat('quantity_checker.csv').st_size == 0:
                writer.writerow(['item', 'check'])
            writer.writerows(records)
        
        # 批量标记这些记录为已处理（比如把check设为1）
        item_list = [record[0] for record in records]
        update_cursor = conn.cursor()
        update_cursor.execute(
            "update items set check=1 where item = any(%s)",
            (item_list,)
        )
    conn.commit()  # 事务提交
except Exception as e:
    conn.rollback()  # 出错则回滚所有操作
    print(f"处理失败：{str(e)}")
finally:
    conn.close()  # 关闭数据库连接

为什么首推这个方案？因为数据库是数据的唯一可信来源，用它标记处理状态最可靠，不会因为CSV文件损坏、丢失导致重复判断出错。而且事务机制能保证：要么CSV写入成功且数据库标记完成，要么全部回滚，避免数据不一致的情况。

方案二：本地CSV校验法（适合不能修改数据库的场景） #

如果因为权限或业务限制，没法修改数据库的check字段，那只能通过本地读取已有的CSV，对比后过滤掉重复记录再写入。

思路是：先提取CSV里已有的item（或item+check组合，根据你的业务唯一性规则），再从数据库查询结果中过滤掉已存在的记录，最后写入CSV。

代码示例：

import csv
import os

csv_path = 'quantity_checker.csv'
# 读取已有CSV，收集已存在的item
existing_items = set()
if os.path.exists(csv_path):
    with open(csv_path, "r", newline="") as f:
        reader = csv.reader(f, delimiter=';')
        next(reader)  # 跳过表头
        for row in reader:
            if row:  # 避免空行干扰
                existing_items.add(row[0])

# 查询数据库未处理记录
conn = ...  # 你的数据库连接代码
read_cursor = conn.cursor()
read_cursor.execute("select item, check from items where check=0")
# 过滤掉已存在的记录
new_records = [row for row in read_cursor if row[0] not in existing_items]

# 写入新记录到CSV
if new_records:
    with open(csv_path, "a", newline="") as f:
        writer = csv.writer(f, delimiter=';')
        # 文件为空时先写表头
        if not os.path.exists(csv_path) or os.stat(csv_path).st_size == 0:
            writer.writerow(['item', 'check'])
        writer.writerows(new_records)

这个方案的缺点是：如果CSV被手动修改、删除，就会导致重复写入；而且如果多进程同时运行，可能出现竞争问题（不过你是每小时运行一次，大概率不会有这个问题）。

额外注意点 #

• 如果item不是唯一标识（比如同一个item可能多次出现check=0的情况），那就要用item+check组合，或者数据库主键来判断重复。
• 定时运行建议用系统定时任务（比如Linux的cron、Windows的任务计划），不要让代码一直循环sleep，这样更稳定可靠。

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