Python爬虫批量抓取URL数据如何按固定条数拆分输出CSV文件

现有代码存在的核心问题 #

• 全局存储的ranks/names/prices列表没有在每批次写入CSV后清空，会导致后续批次的CSV包含前面所有批次的爬取数据，不符合每批次单独存储的需求
• 循环逻辑冗余，同时使用while+切片的嵌套循环，可读性低，也容易出现索引越界问题
• 没有做异常捕获，单个URL爬取失败会导致整个程序崩溃，7000个URL的场景下容错性极差

优化后的实现方案 #

直接修改批次配置参数即可适配正式场景，演示阶段BATCH_SIZE设为2，正式使用改为1000即可：

import requests
from bs4 import BeautifulSoup
import time
from random import randint
import pandas as pd
import os

# 配置参数，可按需修改
BATCH_SIZE = 2  # 每批次处理的URL数量，正式使用改成1000即可
URL_LIST = ['www.1.com', 'www.2.com', 'www.3.com', 'www.4.com', 'www.5.com', 'www.6.com', 'www.7.com', 'www.8.com', 'www.9.com', 'www.10.com']
SAVE_PATH = './crawl_result'  # CSV保存路径，自行修改
os.makedirs(SAVE_PATH, exist_ok=True)

def crawl_single_url(url):
    """单个URL的爬取逻辑封装"""
    try:
        r = requests.get(url, timeout=10)
        r.raise_for_status()
        soup = BeautifulSoup(r.text, 'html.parser')
        
        # 提取排名相关信息
        rank = [item.text for item in soup.find('div', class_ = 'sc-16r8icm-0 bILTHz')]
        # 提取名称相关信息
        name = [ticker.text for ticker in soup.find('h2', class_ = 'sc-1q9q90x-0 jCInrl h1')]
        # 提取价格相关信息
        price = [price_tag.text for price_tag in soup.find('div', class_ = 'sc-16r8icm-0 kjciSH priceTitle')]
        return rank + name + price
    except Exception as e:
        print(f"爬取URL {url} 失败，错误信息：{e}")
        return None

if __name__ == "__main__":
    total_urls = len(URL_LIST)
    # 按批次拆分URL列表
    for batch_idx in range(0, total_urls, BATCH_SIZE):
        batch_urls = URL_LIST[batch_idx: batch_idx + BATCH_SIZE]
        batch_data = []
        print(f"开始处理第{batch_idx//BATCH_SIZE + 1}批次，共{len(batch_urls)}个URL")
        
        for count, url in enumerate(batch_urls, start=1):
            print(f'批次内第{count}个URL开始爬取：{url}')
            row = crawl_single_url(url)
            if row:
                batch_data.append(row)
            # 随机休眠
            sleep_interval = randint(1, 2)
            print(f'休眠 {sleep_interval} 秒')
            time.sleep(sleep_interval)
        
        # 本批次处理完成，写入CSV
        if batch_data:
            final_table = pd.DataFrame(batch_data, columns=['rank', 'type', 'watchlist', 'name', 'symbol', 'price', 'changes'])
            save_name = f'summary_{batch_idx + BATCH_SIZE}.csv'
            final_table.to_csv(os.path.join(SAVE_PATH, save_name), index=False, encoding='utf-8-sig')
            print(f"第{batch_idx//BATCH_SIZE + 1}批次处理完成，结果已保存至{save_name}")

优化点说明 #

• 把单URL爬取逻辑封装成独立函数，代码更清晰，便于后续修改维护
• 每批次单独存储爬取数据，批次处理完成写入CSV后自动销毁临时数据，不会出现数据累加的问题
• 增加了异常捕获逻辑，单个URL爬取失败只会跳过该URL，不会中断整个爬取任务
• 用range的步长参数直接拆分URL批次，无需额外的while循环和索引计数，逻辑更简洁不易出错
• 自动创建保存目录，避免路径不存在的报错，CSV默认保存为utf-8-sig编码，用Excel打开不会乱码

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