批量URL爬取中断求助：处理千条后停止无报错

你的判断完全准确：批量处理1000-1500条URL后程序静默停止，大概率是目标网站触发了反爬机制，临时限制了你的IP访问（没有返回错误码，直接断开连接或不响应请求）。单独运行时请求量小，所以不会触发限制。下面是几个实用的解决方案，按实现难度和效果排序：

1. 添加随机请求延迟（最简单有效的第一步） #

网站会通过请求频率识别机器人，给每个请求加随机延迟，模拟人类浏览节奏，能大幅降低被限制的概率。

修改你的article_link函数，加入延迟：

import time
import random

def article_link(URL):
    try:
        # 添加1-3秒的随机延迟，避免固定间隔被识别
        time.sleep(random.uniform(1, 3))
        soup = url_parse(URL)
        # 找到第一个符合条件的链接就返回，减少不必要的循环
        for i in soup.find_all("a", class_="button entry-content__button entry-content__button--smaller"):
            return i['href']
        return None
    except Exception as e:
        # 替换宽泛的pass，打印错误信息方便排查
        print(f"处理URL {URL} 时出错: {str(e)}")
        return None

2. 优化请求工具与异常处理 #

原来用urllib的urlopen灵活性不足，换成requests库（先安装：pip install requests），它更方便处理会话、超时和具体异常，避免因请求挂起导致程序停滞：

import requests
from bs4 import BeautifulSoup

def url_parse(site):
    hdr = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36'
    }
    try:
        # 设置10秒超时，避免请求一直挂着占用资源
        response = requests.get(site, headers=hdr, timeout=10)
        # 检查请求是否成功（比如403、500等状态码会触发异常）
        response.raise_for_status()
        soup = BeautifulSoup(response.text, 'html.parser')
        return soup
    except requests.exceptions.RequestException as e:
        print(f"请求URL {site} 失败: {str(e)}")
        return None

3. 加入重试机制 #

偶尔的请求失败（比如网络波动）很正常，给失败的请求加重试逻辑，避免因一次失败就中断批量任务：

from tenacity import retry, stop_after_attempt, wait_exponential

# 使用tenacity库实现智能重试，先安装：pip install tenacity
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def url_parse_with_retry(site):
    hdr = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36'
    }
    response = requests.get(site, headers=hdr, timeout=10)
    response.raise_for_status()
    return BeautifulSoup(response.text, 'html.parser')

# 修改article_link使用带重试的解析函数
def article_link(URL):
    try:
        time.sleep(random.uniform(1, 3))
        soup = url_parse_with_retry(URL)
        if soup:
            for i in soup.find_all("a", class_="button entry-content__button entry-content__button--smaller"):
                return i['href']
        return None
    except Exception as e:
        print(f"最终处理URL {URL} 失败: {str(e)}")
        return None

4. 使用代理IP（如果延迟和重试仍无效） #

如果网站严格限制单个IP的请求次数，可以用代理IP轮换请求。你可以找免费代理池，或者使用付费代理服务，下面是简单的代理使用示例：

def url_parse_with_proxy(site, proxies=None):
    hdr = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36'
    }
    try:
        response = requests.get(site, headers=hdr, timeout=10, proxies=proxies)
        response.raise_for_status()
        return BeautifulSoup(response.text, 'html.parser')
    except requests.exceptions.RequestException as e:
        print(f"代理请求URL {site} 失败: {str(e)}")
        return None

# 批量处理时轮换代理
proxies_list = [
    {"http": "http://proxy1:port", "https": "https://proxy1:port"},
    {"http": "http://proxy2:port", "https": "https://proxy2:port"},
    # 补充更多代理...
]

for idx, rows in data.iterrows():
    # 按索引轮换代理，避免单个代理请求过多
    proxy = proxies_list[idx % len(proxies_list)]
    rows['article_source'] = article_link(rows['url'], proxy)

5. 批量处理的小优化 #

• 不要直接在iterrows里修改DataFrame，用apply效率更高：

def process_row(row):
    row['article_source'] = article_link(row['url'])
    return row

data = data.apply(process_row, axis=1)

• 把失败的URL记录到日志文件，方便后续手动处理：

import logging
logging.basicConfig(filename='failed_urls.log', level=logging.INFO)

# 在异常处理中添加日志记录
logging.info(f"Failed URL: {URL}")

优先从添加延迟和优化异常处理开始，这两个改动最小，效果最明显。调整后你的程序应该能顺利处理完所有URL了。

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