爬虫中常见的网络请求异常包括连接错误、超时和HTTP状态码异常,需通过try-
class='language-default'>except分层捕获并针对性处理。

在爬虫项目中,Python的异常处理机制绝不是可有可无的装饰品,它简直就是保障爬虫生命力与稳定性的核心骨架。没有它,你的爬虫就像在薄冰上跳舞,任何一点风吹草动——网络波动、目标网站结构微调、IP被封——都可能让它瞬间崩塌,功亏一篑。真正有效的异常处理,能让爬虫从容应对这些“不确定性”,哪怕遭遇挫折也能优雅地恢复,继续它的使命,确保数据收集的连续性和完整性。
解决方案
要让爬虫变得“皮实”起来,我们得系统地运用
class='language-default'>try-class='language-default'>except-finally-else
结构。这不仅仅是捕获错误那么简单,它更像是一种风险管理策略。我的经验是,先预判那些最常出现的“雷区”,比如网络连接中断、请求超时、HTTP状态码异常,以及数据解析时的各种意外。针对这些预判,我们用特定的
class='language-default'>except
块去精准拦截。
比如,当发起一个网络请求时,可能会遇到服务器无响应、DNS解析失败或者代理挂掉。这些都属于
class='language-default'>class='language-default'>requests.class='language-default'>exceptions.RequestException
的范畴。如果直接用一个大而全的
class='language-default'>except Exception as e:
去捕获,虽然能防止程序崩溃,但你丢失了错误发生时的具体上下文,也就难以对症下药。因此,我们应该先捕获更具体的异常,再逐步放宽到更通用的异常。
立即学习“Python免费学习笔记(深入)”;
一个健壮的爬虫,其异常处理逻辑应该包含:
- 请求层面的异常:针对
class='language-default'>class='language-default'>requests库可能抛出的各种错误,如
class='language-default'>ConnectionError、
class='language-default'>Timeout、
class='language-default'>HTTPError等。
- 解析层面的异常:当使用BeautifulSoup、lxml或json库解析数据时,可能出现选择器失效、键不存在、JSON格式错误等问题。
- 业务逻辑异常:比如在数据校验时发现数据不符合预期,或者某些字段缺失。
- 重试机制:对于瞬时性的网络问题,简单的重试往往就能解决。但要注意,重试不能是无限次的,并且最好配合指数退避(exponential backoff),给服务器一点喘息的时间。
- 日志记录:每一次异常的发生,都应该被详细记录下来,包括发生时间、URL、异常类型、堆栈信息等,这对于后续的调试和问题排查至关重要。
importclass='language-default'>class='language-default'>requests import time import random importlogginglogging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') def fetch_url_with_retry(url, retries=3, backoff_factor=0.5): for i in range(retries): try: response =class='language-default'>class='language-default'>requests.get(url, timeout=10)class='language-default'>response.raise_for_status() # Raisesclass='language-default'>HTTPError for bad responses (4xx or 5xx) return responseclass='language-default'>exceptclass='language-default'>class='language-default'>requests.class='language-default'>exceptions.class='language-default'>Timeout:logging.warning(f"请求超时,URL: {url},尝试重试 {i+1}/{retries}...")class='language-default'>exceptclass='language-default'>class='language-default'>requests.class='language-default'>exceptions.class='language-default'>ConnectionError:logging.warning(f"连接错误,URL: {url},尝试重试 {i+1}/{retries}...")class='language-default'>exceptclass='language-default'>class='language-default'>requests.class='language-default'>exceptions.class='language-default'>HTTPError as e: if e.response.status_code == 404:logging.error(f"页面未找到 (404),URL: {url}") returnclass='language-default'>class='language-default'>class='language-default'>class='language-default'>class='language-default'>None # 404通常不需要重试logging.warning(f"HTTP错误 {e.response.status_code},URL: {url},尝试重试 {i+1}/{retries}...")class='language-default'>exceptclass='language-default'>class='language-default'>requests.class='language-default'>exceptions.RequestException as e:logging.error(f"未知请求异常,URL: {url},错误: {e},尝试重试 {i+1}/{retries}...") if i < retries - 1: sleep_time = backoff_factor * (2 ** i) + random.uniform(0, 1) # 指数退避加随机抖动logging.info(f"等待 {sleep_time:.2f} 秒后重试...") time.sleep(sleep_time)logging.error(f"多次重试失败,URL: {url} 无法获取。") returnclass='language-default'>class='language-default'>class='language-default'>class='language-default'>class='language-default'>None # 示例使用 # response = fetch_url_with_retry("http://www.example.com/nonexistent") # if response: # print(response.text[:100])
爬虫中常见的网络请求异常有哪些,以及如何针对性地捕获和处理?
在爬虫的世界里,网络请求异常简直是家常便饭。我的经验是,大部分爬虫的“崩溃”都始于此。最常见的几种,无非就是连接不上、请求超时、以及HTTP状态码不正常。
首先是
class='language-default'>class='language-default'>requests.class='language-default'>exceptions.class='language-default'>ConnectionError
。这通常意味着你的程序无法建立到目标服务器的连接。可能是目标网站宕机了,也可能是你的网络有问题,或者DNS解析失败。这种错误,第一时间想到的应该是重试。但别傻乎乎地立刻重试,给它一点时间,比如等个几秒钟,再尝试。如果还是不行,那可能就不是瞬时问题了,得考虑是不是IP被封了,或者目标网站真的挂了。
接着是
class='language-default'>class='language-default'>requests.class='language-default'>exceptions.class='language-default'>Timeout
。当你发送请求后,在指定的时间内没有收到服务器的响应,就会抛出这个异常。超时可能是因为服务器处理请求太慢,也可能是网络延迟高。对于这种,重试同样是有效手段,但可能需要调整超时时间,或者尝试更换代理。我通常会设置一个合理的超时时间,比如5-10秒,而不是无限等待。
然后是
class='language-default'>class='language-default'>requests.class='language-default'>exceptions.class='language-default'>HTTPError
。这发生在服务器返回了非200的HTTP状态码时,比如404(页面未找到)、403(禁止访问)、500(服务器内部错误)等等。
class='language-default'>class='language-default'>requests
库的
class='language-default'>response.raise_for_status()
方法就是为此而生。对于404,通常意味着这个URL是无效的,不需要重试,直接记录并跳过就好。403往往是反爬机制在作祟,此时你需要考虑更换User-Agent、使用代理IP,甚至模拟登录。而500错误,可能是服务器暂时性故障,重试几次往往能解决。
处理这些异常,关键在于“针对性”。我们应该利用Python的异常继承链,先捕获最具体的异常,再捕获更通用的。这就像你生病了,医生会先诊断是感冒还是肺炎,而不是直接给你开个“万能药”。
importclass='language-default'>class='language-default'>requests import time import random importlogginglogging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') def robust_get(url, retries=3, delay_base=1): for attempt in range(retries): try: # 模拟代理切换或User-Agent轮换 headers = {'User-Agent': f'Mozilla/5.0 (Windows NT 10.0; Win64; x64) appleWebKit/537.36 (KHTML, like Gecko) Chrome/{random.randint(80, 100)}.0.0.0 Safari/537.36'} response =class='language-default'>class='language-default'>requests.get(url, timeout=15, headers=headers)class='language-default'>response.raise_for_status() # 检查HTTP状态码 return responseclass='language-default'>exceptclass='language-default'>class='language-default'>requests.class='language-default'>exceptions.class='language-default'>Timeout:logging.warning(f"请求超时,URL: {url} (尝试 {attempt + 1}/{retries})")class='language-default'>exceptclass='language-default'>class='language-default'>requests.class='language-default'>exceptions.class='language-default'>ConnectionError:logging.warning(f"连接错误,URL: {url} (尝试 {attempt + 1}/{retries})")class='language-default'>exceptclass='language-default'>class='language-default'>requests.class='language-default'>exceptions.class='language-default'>HTTPError as e: status_code = e.response.status_code if status_code == 404:logging.error(f"资源未找到 (404),URL: {url}。跳过。") returnclass='language-default'>class='language-default'>class='language-default'>class='language-default'>class='language-default'>None elif status_code == 403:logging.warning(f"访问被拒绝 (403),URL: {url}。可能需要更换IP或User-Agent。") elif status_code >= 500:logging.warning(f"服务器内部错误 ({status_code}),URL: {url}。") else:logging.warning(f"HTTP错误 ({status_code}),URL: {url}。")class='language-default'>exceptclass='language-default'>class='language-default'>requests.class='language-default'>exceptions.RequestException as e:logging.error(f"发生未知请求错误: {e},URL: {url}") if attempt < retries - 1: sleep_time = delay_base * (2 ** attempt) + random.uniform(0, 1)logging.info(f"等待 {sleep_time:.2f} 秒后重试...") time.sleep(sleep_time)logging.error(f"多次重试失败,无法获取 URL: {url}") returnclass='language-default'>class='language-default'>class='language-default'>class='language-default'>class='language-default'>None # 示例: # resp = robust_get("https://httpbin.org/status/403") # if resp: # print(resp.text)
通过这种分层、精细化的处理,我们能让爬虫在面对网络世界的各种“恶意”时,表现得更加从容和专业。
数据解析阶段的异常处理,如何避免因数据结构变化导致爬虫崩溃?
爬虫最脆弱的环节之一,就是数据解析。我见过太多爬虫,前一秒还在欢快地抓取数据,后一秒就因为目标网站HTML结构或者API响应格式的微小变动,直接“猝死”。这种感觉就像你精心搭建的乐高城堡,被一阵突如其来的风吹散了。
常见的解析异常,主要集中在以下几类:
-
class='language-default'>IndexError和
:当你试图访问一个不存在的列表索引或者字典键时。比如,你期望某个HTML元素下有第三个子元素,结果只有两个;或者某个JSON字段突然不见了。class='language-default'>class='language-default'>KeyError -
:在使用BeautifulSoup或lxml时,如果你尝试访问一个不存在的标签属性或者解析结果对象上没有的方法。
class='language-default'>AttributeError -
:数据类型不匹配,比如你期望一个字符串,结果却是个
class='language-default'>TypeErrorclass='language-default'>class='language-default'>class='language-default'>class='language-default'>class='language-default'>None,然后你尝试对
class='language-default'>class='language-default'>class='language-default'>class='language-default'>class='language-default'>None调用字符串方法。
-
:当API返回的响应体不是一个合法的JSON字符串时。
class='language-default'>json.JSONDecodeError
避免这些问题,核心思路是“防御性编程”:永远不要假设数据结构是完美的、不变的。
- 安全访问字典和列表:对于字典,使用
class='language-default'>.get(key, default_value)方法,而不是直接
class='language-default'>dict[key]。这样即使键不存在,也不会抛出
class='language-default'>class='language-default'>KeyError,而是返回你设定的默认值(通常是
class='language-default'>class='language-default'>class='language-default'>class='language-default'>class='language-default'>None)。对于列表,在访问索引前,先检查列表的长度,或者使用
try-
class='language-default'>exceptclass='language-default'>IndexError。
- 检查
class='language-default'>class='language-default'>class='language-default'>class='language-default'>class='language-default'>None值
:在对解析结果进行操作前,务必检查它是否为class='language-default'>class='language-default'>class='language-default'>class='language-default'>class='language-default'>None。比如,
if element is not
class='language-default'>class='language-default'>class='language-default'>class='language-default'>class='language-default'>None: element.text。
- 使用健壮的选择器:CSS选择器或XPath表达式应该尽可能地具有鲁棒性,避免过度依赖层级关系。比如,优先使用
class='language-default'>id、
class或者
data-*属性,而不是
div > div > span这种脆弱的结构。
- 捕获特定解析异常:将解析代码包裹在
try-
class='language-default'>except块中,捕获
class='language-default'>IndexError、
class='language-default'>class='language-default'>KeyError、
class='language-default'>AttributeError、
class='language-default'>json.JSONDecodeError等。当这些异常发生时,记录下错误信息和对应的URL,然后跳过当前项,而不是让整个爬虫停下来。
- 数据校验:在数据入库前,进行一次最终的校验。比如,确保某个字段是数字类型,某个字符串长度符合要求等。
from bs4 import BeautifulSoup import json importlogginglogging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') def parse_html_data(html_content, url): data = {} try: soup = BeautifulSoup(html_content, 'lxml') # 示例1: 安全访问元素及其文本 title_element = soup.select_one('h1.product-title') # 使用更具体的选择器 data['title'] = title_element.text.strip() if title_element elseclass='language-default'>class='language-default'>class='language-default'>class='language-default'>class='language-default'>None # 示例2: 安全访问属性 image_element = soup.select_one('img.product-image') data['image_url'] = image_element.get('src') if image_element elseclass='language-default'>class='language-default'>class='language-default'>class='language-default'>class='language-default'>None # 示例3: 处理可能缺失的列表项 price_list = soup.select('span.price-item') try: data['main_price'] = price_list[0].text.strip() if price_list elseclass='language-default'>class='language-default'>class='language-default'>class='language-default'>class='language-default'>None data['discount_price'] = price_list[1].text.strip() if len(price_list) > 1 elseclass='language-default'>class='language-default'>class='language-default'>class='language-default'>class='language-default'>Noneclass='language-default'>exceptclass='language-default'>IndexError:logging.warning(f"解析价格列表时索引越界,URL: {url}") data['main_price'] =class='language-default'>class='language-default'>class='language-default'>class='language-default'>class='language-default'>None data['discount_price'] =class='language-default'>class='language-default'>class='language-default'>class='language-default'>class='language-default'>Noneclass='language-default'>exceptclass='language-default'>AttributeErroras e:logging.error(f"解析HTML时属性错误,URL: {url},错误: {e}") returnclass='language-default'>class='language-default'>class='language-default'>class='language-default'>class='language-default'>Noneclass='language-default'>except Exception as e: # 捕获其他未预料的解析错误logging.error(f"解析HTML时发生未知错误,URL: {url},错误: {e}") returnclass='language-default'>class='language-default'>class='language-default'>class='language-default'>class='language-default'>None return data def parse_json_data(json_string, url): try: data = json.loads(json_string) # 安全访问字典键 product_name = data.get('product', {}).get('name') product_price = data.get('product', {}).get('details', {}).get('price') if product_name isclass='language-default'>class='language-default'>class='language-default'>class='language-default'>class='language-default'>None:logging.warning(f"JSON数据中缺少 'product.name' 字段,URL: {url}") return {'name': product_name, 'price': product_price}class='language-default'>exceptclass='language-default'>json.JSONDecodeErroras e:logging.error(f"JSON解析错误,URL: {url},错误: {e}") returnclass='language-default'>class='language-default'>class='language-default'>class='language-default'>class='language-default'>Noneclass='language-default'>except Exception as e:logging.error(f"解析JSON时发生未知错误,URL: {url},错误: {e}") returnclass='language-default'>class='language-default'>class='language-default'>class='language-default'>class='language-default'>None # 示例使用 # html_example = "<html><body><h1class='product-title'>Test Product</h1><imgclass='product-image' src='test.jpg'><spanclass='price-item'>$100</span></body></html>" # parsed_html = parse_html_data(html_example, "http://example.com/product/1") # print(parsed_html) # json_example = '{"product": {"name": "Laptop", "details": {"price": 1200}}}' # parsed_json = parse_json_data(json_example, "http://example.com/api/product/1") # print(parsed_json)
通过这些手段,我们能够大幅提升爬虫在面对目标网站结构变化时的韧性,让它不至于因为一点小变动就“罢工”。
构建健壮爬虫时,除了捕获异常,还有哪些策略可以提升系统的容错性和稳定性?
单纯地捕获异常,只是“治标不治本”。一个真正健壮的爬虫系统,需要一系列组合拳来提升其容错性和稳定性。这就像建造一座大楼,地基要稳固,结构要合理,还得有消防系统和应急通道。
-
完善的日志系统:这不仅仅是记录异常,而是记录爬虫运行的方方面面。请求URL、响应状态码、解析结果、入库情况,甚至每次重试的详情。使用
logging模块,设置不同的日志级别(DEBUG, INFO, WARNING, ERROR, CRITICAL),将日志输出到文件,并定期归档。这样,当问题发生时,你才能有迹可循,快速定位问题。
-
智能的重试机制与指数退避:前面已经提到,对于瞬时性的网络错误,重试是有效的。但关键在于“智能”。不要立即重试,而是等待一段时间,并且每次重试的等待时间逐渐增加(指数退避),同时加入随机抖动,避免“死循环”或给目标网站造成更大压力。设置最大重试次数,超过后放弃当前任务。
-
代理IP池与User-Agent轮换:这是应对反爬机制的利器。当IP被封禁或某个User-Agent被识别时,系统能自动切换到下一个可用的代理或User-Agent。一个健康的代理池需要有检测机制,定期清理失效代理。
-
限速与请求间隔:对目标网站的访问频率进行控制,模拟人类的浏览行为。设置一个随机的请求间隔(例如2到5秒),可以有效降低被封禁的风险,也体现了对目标网站的“尊重”。
-
任务队列与持久化:对于大规模爬虫,使用消息队列(如Redis、RabbitMQ)来管理待抓取URL,并将已抓取和待抓取的任务状态进行持久化。这样,即使爬虫程序意外中断,也能从上次中断的地方恢复,避免重复抓取或数据丢失。
-
监控与告警:这是最容易被忽视,但却至关重要的一环。实时监控爬虫的运行状态,比如抓取速度、错误率、代理IP可用率、数据入库量等。当某个指标超出预设阈值时,通过邮件、短信或即时通讯工具发送告警,让你能第一时间介入处理。
-
数据校验与清洗:在数据入库前,对抓取到的数据进行严格的校验和清洗。例如,检查字段是否缺失、数据类型是否正确、是否存在异常值。不符合要求的数据,可以记录下来进行人工复查,而不是直接丢弃或入库。
-
模块化与解耦:将爬虫的不同功能(请求、解析、存储、调度)模块化,降低耦合度。这样,当某个模块出现问题时,更容易隔离和修复,而不影响整个系统的运行。
这些策略的引入,能让爬虫从一个简单的脚本,升级为一个能够自我修复、稳定运行的系统。它不再只是被动地捕获错误,而是主动地预防错误,并具备从错误中恢复的能力,这才是真正意义上的“健壮”。
python 爬虫 css redis html js json windows app 工具 safari 栈 ai Python rabbitmq json css html beautifulsoup 数据类型 if try Error Logging 字符串 循环 数据结构 继承 栈 堆 class finally 数字类型 对象 选择器 redis http


