爬虫开发中,请求—下载—解析—存储是最基础的四段流水线。请求和下载部分各语言方案大同小异,真正拉开效率差距的是解析层。BeautifulSoup 面对深层嵌套、条件筛选时力不从心;正则可读性差、维护成本高。XPath 是 W3C 标准查询语言,专为树结构设计,配合 Scrapy 的异步引擎,在大规模、跨平台爬虫项目中几乎没有对手。
一、Scrapy 项目初始化
pip install scrapy
scrapy startproject multispider && cd multispider
scrapy genspider technews example.com
在 items.py 中声明结构化字段:
import scrapy
class NewsItem(scrapy.Item):
title = scrapy.Field()
url = scrapy.Field()
author = scrapy.Field()
publish_date = scrapy.Field()
content = scrapy.Field()
tags = scrapy.Field()
source = scrapy.Field()
二、XPath 高频语法速查
场景 表达式 说明
全局搜索 //div[@class="list"] 不关心层级
相对定位 .//h2/a/@href 以当前节点为根,实战最关键
模糊匹配 contains(@class, "active") 多 class 场景必用
位置限定 //li[position()<=3] 取前 N 个
轴遍历 //h2/following-sibling::p 取兄弟节点
条件排除 //p[not(contains(@class,"ad"))] XPath 原生过滤广告
核心原则:循环遍历列表项时,子元素 XPath 必须以 . 开头(.//),否则会回到整个文档根节点全局搜索,导致数据错位。
三、核心爬虫:列表页 → 详情页两级解析
编辑 spiders/technews.py:
import scrapy
from multispider.items import NewsItem
class TechNewsSpider(scrapy.Spider):
name = 'technews'
allowed_domains = ['example.com']
start_urls = ['https://example.com/news']
def parse(self, response):
# 列表页:定位所有文章条目
for article in response.xpath('//div[@class="article-list"]/article'):
detail_url = article.xpath('.//h2/a/@href').get()
if detail_url:
yield response.follow(
detail_url,
callback=self.parse_detail,
meta={'list_title': article.xpath('.//h2/a/text()').get(default='').strip()}
)
# 翻页
next_page = response.xpath('//a[contains(@class,"next")]/@href').get()
if next_page:
yield response.follow(next_page, callback=self.parse)
def parse_detail(self, response):
item = NewsItem()
item['title'] = (
response.xpath('//h1[@class="article-title"]/text()').get(default='').strip()
or response.meta.get('list_title', '')
)
item['url'] = response.url
item['author'] = response.xpath('//span[@class="author-name"]/text()').get(default='匿名').strip()
item['publish_date'] = response.xpath('//time[@class="publish-date"]/@datetime').get()
item['tags'] = response.xpath('//div[@class="tags"]//a/text()').getall()
item['source'] = 'technews'
# 正文提取:排除广告/推荐节点
paragraphs = response.xpath(
'//div[@class="article-body"]'
'//p[not(contains(@class,"ad")) and not(contains(@class,"recommend"))]'
'/text()'
).getall()
item['content'] = '\n'.join(p.strip() for p in paragraphs if p.strip())
yield item
四个关键技巧:
- .// 相对路径:循环体内必须用 . 开头,避免跨条目误抓
- get(default=''):防止 NoneType 错误,提供安全兜底
- response.follow():自动补全相对 URL,无需手动拼域名
- meta 透传:列表页元数据传递到详情页,做 fallback 容错
四、跨平台适配:规则配置与爬虫逻辑解耦
不同站点 HTML 结构不同,但数据模型和清洗逻辑完全可复用。核心思路是将 XPath 规则抽成配置字典:
SITE_RULES = {
'siteA': {
},'start_urls': ['https://site-a.com/news'], 'list_item': '//div[@class="news-item"]', 'detail_link': './/a[@class="title"]/@href', 'title': '//h1[@class="post-title"]/text()', 'author': '//span[@itemprop="author"]/text()', 'publish_date': '//meta[@property="article:published_time"]/@content', 'content': '//div[@class="post-content"]//p/text()', 'tags': '//div[@class="tag-list"]//a/text()', 'next_page': '//a[@rel="next"]/@href',
'siteB': {
},# ... 另一个站点的规则
}
class MultiSiteSpider(scrapy.Spider):
name = 'multisite'
def start_requests(self):
for site_name, rules in SITE_RULES.items():
for url in rules['start_urls']:
yield scrapy.Request(url, callback=self.parse_list,
meta={'site_name': site_name, 'rules': rules})
def parse_list(self, response):
rules = response.meta['rules']
for article in response.xpath(rules['list_item']):
link = article.xpath(rules['detail_link']).get()
if link:
yield response.follow(link, callback=self.parse_detail, meta=response.meta)
# 翻页
next_page = response.xpath(rules['next_page']).get()
if next_page:
yield response.follow(next_page, callback=self.parse_list, meta=response.meta)
def parse_detail(self, response):
rules = response.meta['rules']
item = NewsItem()
item['url'] = response.url
item['source'] = response.meta['site_name']
item['title'] = response.xpath(rules['title']).get(default='').strip()
item['author'] = response.xpath(rules['author']).get(default='匿名').strip()
item['content'] = '\n'.join(p.strip() for p in response.xpath(rules['content']).getall() if p.strip())
item['tags'] = response.xpath(rules['tags']).getall()
yield item
新增站点只需加一段规则配置,核心代码零改动——这是 Scrapy 跨平台扩展的工程优势。
五、接入代理 IP:突破反爬封锁
跨平台大规模爬虫必然触发目标站点的 IP 频率限制。以亿牛云爬虫代理为例,在 Scrapy 中接入代理只需编写一个下载器中间件。
新建 middlewares.py:
import base64
import random
def base64ify(bytes_or_str):
"""生成 Proxy-Authorization 认证头"""
input_bytes = bytes_or_str.encode('utf8') if isinstance(bytes_or_str, str) else bytes_or_str
return base64.urlsafe_b64encode(input_bytes).decode('ascii')
class ProxyMiddleware(object):
def process_request(self, request, spider):
# 亿牛云爬虫代理参数(官网 www.16yun.cn)
proxyHost = "t.16yun.cn"
proxyPort = "31111"
proxyUser = "username" # 替换为你的用户名
proxyPass = "password" # 替换为你的密码
# 设置代理地址
request.meta['proxy'] = f"http://{proxyHost}:{proxyPort}"
# 添加认证头(Scrapy 2.6.2+ 可省略,会自动设置)
request.headers['Proxy-Authorization'] = 'Basic ' + base64ify(f"{proxyUser}:{proxyPass}")
# 设置 Proxy-Tunnel:相同随机数 = 相同出口 IP(适合需要登录态保持的场景)
tunnel = random.randint(1, 10000)
request.headers['Proxy-Tunnel'] = str(tunnel)
# 如需每个请求强制切换 IP,关闭连接复用
request.headers['Connection'] = "Close"
在 settings.py 中启用中间件并配置重试策略:
DOWNLOADER_MIDDLEWARES = {
'multispider.middlewares.ProxyMiddleware': 100,
}
代理认证失败(407)时自动重试
RETRY_ENABLED = True
RETRY_TIMES = 3
RETRY_HTTP_CODES = [500, 502, 503, 504, 407, 408, 429]
并发与限速
CONCURRENT_REQUESTS = 16
DOWNLOAD_DELAY = 0.5
DOWNLOAD_TIMEOUT = 15
代理 IP 使用要点:
场景 配置方式 说明
每次请求换 IP Connection: Close
- 随机 Tunnel 最常用,适合批量抓取
保持同一 IP 固定 Proxy-Tunnel
值 适合需要登录/Cookie 缓存的流程
HTTPS 站点 使用库原生代理认证 避免手动 Proxy-Authorization
被转发到目标站
407 错误 检查域名/端口/用户名/密码 认证信息错误
429 错误 降低并发或增加延迟 请求速率超出订单上限
六、数据清洗管道
编辑 pipelines.py,将清洗逻辑与爬虫逻辑分离:
import re
import json
from datetime import datetime
from itemadapter import ItemAdapter
from scrapy.exceptions import DropItem
class DataCleaningPipeline:
def process_item(self, item, spider):
adapter = ItemAdapter(item)
# 去除控制字符和首尾空白
for field in ['title', 'author', 'content']:
val = adapter.get(field, '')
if val:
val = re.sub(r'[\x00-\x1f\x7f-\x9f\u00a0]', '', val).strip()
adapter[field] = val if val else None
# 标签去重
tags = adapter.get('tags', [])
seen, cleaned = set(), []
for tag in (t.strip() for t in tags if t.strip()):
key = tag.lower()
if key not in seen:
seen.add(key)
cleaned.append(tag)
adapter['tags'] = cleaned[:10]
# 必填校验
if not adapter.get('title'):
raise DropItem("Missing title")
return item
class JsonExportPipeline:
def open_spider(self, spider):
self.file = open('output.jsonl', 'w', encoding='utf-8')
def process_item(self, item, spider):
self.file.write(json.dumps(dict(item), ensure_ascii=False) + '\n')
return item
def close_spider(self, spider):
self.file.close()
settings.py
ITEM_PIPELINES = {
'multispider.pipelines.DataCleaningPipeline': 100,
'multispider.pipelines.JsonExportPipeline': 200,
}
七、调试与运行
Scrapy Shell 验证 XPath(写代码前必做)
scrapy shell 'https://example.com/news'
response.xpath('//h1[@class="article-title"]/text()').get()
'Python 3.12 新特性解析'
运行爬虫
scrapy crawl technews -o results.json
八、XPath 避坑指南
陷阱 错误写法 正确写法
全局搜索误抓 article.xpath('//h2/text()') article.xpath('.//h2/text()')
多 class 失配 @class="item active" contains(@class, "active")
空白未处理 .get()
直接用 .get(default='').strip()
编码乱码 默认编码 FEED_EXPORT_ENCODING='utf-8'
九、总结
Scrapy + XPath 的工程价值集中在三个层面:
- 解析层:XPath 的树结构查询能力远超 BeautifulSoup,深层嵌套、多条件筛选、跨轴遍历是原生优势
- 架构层:异步引擎 + 中间件 + Pipeline 天然支持大规模、跨平台扩展。规则配置与爬虫逻辑解耦,新增站点边际成本趋近于零
- 反爬层:通过代理 IP 中间件(如亿牛云爬虫代理)无缝接入 IP 池,Proxy-Tunnel 机制精确控制 IP 切换时机,配合 407 重试策略保障稳定性
实际项目中,先用 scrapy shell 验证 XPath 表达式再写代码;清洗逻辑统一收敛到 Pipeline;代理中间件根据业务场景选择随机 IP 或固定 IP 模式。这三点做到位,爬虫的可维护性和稳定性会有质的提升。