Python爬虫写入CSV文件时如何检查指定title的行是否已存在

问题原因 #

• 以a+模式打开文件后，文件指针默认在文件末尾，直接调用csv.reader(f)读取不到任何已有内容
• csv.reader返回的每一行是包含所有字段的列表，你直接判断字符串ad['title']是否在列表里肯定匹配失败
• 新创建csv文件时没有写入表头，后续读取字段会错位
• main函数里每解析一条数据就调用一次write_csv并传入整个累计的ads_data，会导致重复写入校验和重复写入操作

修复后的完整代码 #

from pathlib import Path
import time
import csv
from selenium import webdriver
from bs4 import BeautifulSoup
import pandas as pd

DRIVER_PATH = str(Path('geckodriver').resolve())


def write_csv(ads):
    filename = 'results.csv'
    file_exists = Path(filename).exists()
    existing_titles = set()
    
    # 先读取已有的所有title存入集合，查询效率更高
    if file_exists:
        with open(filename, 'r', encoding='utf-8') as f:
            reader = csv.DictReader(f)
            for row in reader:
                existing_titles.add(row['title'])
    
    # 追加写入不重复的新数据
    with open(filename, 'a+', encoding='utf-8', newline='') as f:
        fields = ['title', 'url']
        writer = csv.DictWriter(f, fieldnames=fields)
        # 新文件首次写入先加表头
        if not file_exists:
            writer.writeheader()
        
        for ad in ads:
            # 跳过空title的无效数据，同时校验是否重复
            if ad['title'] and ad['title'] not in existing_titles:
                writer.writerow(ad)
                # 新写入的title加入集合，避免本次批量写入时重复
                existing_titles.add(ad['title'])
                print('success')
            else:
                print('fail')


def get_html(url):
    browser = webdriver.Firefox(executable_path=DRIVER_PATH)
    browser.get(url)
    return browser.page_source


def scrapde_data(card):
    try:
        h2 = card.h2
    except:
        title = ''
        url = ''
    else:
        title = h2.text.strip()

    try:
        url = card.find('a').get('href')
    except:
        url = ''

    data = {'title': title, 'url': url}

    return data


def main():
    while True:
        url = '#'
        html = get_html(url)
        soup = BeautifulSoup(html, 'lxml')
        cards = soup.find_all('div', {"class": "produto--comprar"})

        ads_data = []

        for card in cards:
            data = scrapde_data(card)
            ads_data.append(data)
        
        # 所有数据解析完成后统一调用写入，减少重复IO操作
        write_csv(ads_data)
        time.sleep(5)

if __name__ == '__main__':
    main()

关键修改说明 #

• 提前判断文件是否存在，首次创建时自动写入表头，避免字段错位
• 单独读取已有文件的所有title存入集合，查询效率远高于逐行遍历
• 写入新数据前先校验title是否在已有的集合中，校验通过才写入，同时更新集合避免本次批量数据重复
• 调整write_csv调用时机，统一批量写入减少IO操作和重复校验

内容的提问来源于stack exchange，提问作者Guilherme Gonçalves Rodrigues