安装必要的库
确保你已经安装了Python和相关的包：
pip install scrapy beautifulsoup4
如果需要一个完整的环境，可以创建一个 requirements.txt 文件：
scrapy beautifulsoup4
然后安装它们：
pip install -r requirements.txt
配置请求头
在抓取网页时，模拟浏览器行为是关键，设置请求头，包括 User-Agent、Referer 和 Accept-Language：
from scrapy.http.request import Request
from scrapy.spider import Spider
class MySpider(Spider):
def start_requests(self):
headers = {
'User-Agent': 'Mozilla/5. (Windows NT 10.; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58..3029.110 Safari/537.3',
'Referer': 'https://example.com/',
'Accept-Language': 'en-US,en;q=.9',
}
yield Request('https://target-website.com', headers=headers, callback=self.parse)
处理重定向
使用 Response.url 来跟踪重定向路径,确保抓取最终资源：
from scrapy.spiders import Crawler
from scrapy.http.response import Response
class MySpider(Crawler):
def handle(self, request, response=None, spider=None):
if response:
if response.status == 302:
# 跟踪重定向
new_url = response.url
self.log('Redirecting to {}'.format(new_url))
# 重定向到新URL
self.crawler.engine.send_request(request, new_url, spider)
else:
# 首次请求
yield self.crawler.engine.send_request(request, request.url, spider)
模拟浏览器行为
- 设置下载延迟：避免过于频繁的请求,模拟真实用户：
from scrapy.http.request import Request
from scrapy.spider import Spider
class MySpider(Spider):
def __init__(self):
super().__init__()
self.download_delay = 1. # 设置下载延迟
def start_requests(self):
yield Request('https://target-website.com', callback=self.parse)
- 模拟点击或提交表单：
from scrapy.http.request import FormRequest
from scrapy.spider import Spider
class MySpider(Spider):
def parse(self, response):
formdata = {
'username': 'johndoe',
'password': 'password123',
}
yield FormRequest(
url='https://login.example.com',
formdata=formdata,
callback=self-after_login,
don_load=True
)
- 模拟滚动：使用JavaScript执行滚动动作：
from scrapy.contrib.loader import JsError
from scrapy.contrib.spiders import CrawlableSpider
from scrapy.http.response import TextResponse
class MySpider(CrawlableSpider):
def parse(self, response):
try:
response.execute_script('window.scrollTo(, 100);')
# 等待加载完成
response.wait()
except JsError as e:
self.log('JavaScript错误：%s' % e)
# 提取数据
elements = response.xpath('//div[@class="some-class"]').extract()
for el in elements:
yield {
'id': el.xpath('@id').extract(),
'text': el.xpath('text()').extract(),
}
处理异常和错误
使用中间件来处理常见的错误,如503服务不可用：
from scrapy import Spider, Request
from scrapy.contrib.spiders import SitemapSpider
from scrapy.contrib.downloadermiddlewares import RedirectMiddleware
class MySpider(SitemapSpider):
def __init__(self):
super().__init__()
def start_requests(self):
yield Request('https://example.com/sitemap.xml', callback=self.parse_sitemap)
def parse_sitemap(self, response):
if response.status == 503:
self.log('Service Unavailable')
# 重试逻辑
yield self.request('https://example.com/sitemap.xml', callback=self.parse_sitemap)
else:
# 处理正常响应
sitemap_links = response.xpath('//loc').extract()
for link in sitemap_links:
yield Request(link, callback=self.parse_page)
使用代理防止封禁
配置代理,避免IP被封禁：
from scrapy.http.proxy import ProxyMiddleware
from scrapy.spiders import Crawler
class MySpider(Crawler):
def __init__(self):
super().__init__()
self.proxy = 'http://proxy.example.com:3128'
def start_requests(self):
yield self.crawler.engine.send_request(
Request('https://target-website.com',
priority=1,
meta={'proxy': self.proxy}),
spider=self)
保持会话和使用HTTPS
使用 SessionMiddleware 来保持会话,确保登录状态：
from scrapy import Spider
from scrapy.contrib.spiders import CrawlableSpider
from scrapy.contrib.middlewares import SessionMiddleware
class MySpider(CrawlableSpider):
def __init__(self):
super().__init__()
self.session = None
def start_requests(self):
self.crawler.engine.bypass_request_hook = True
self.crawler.add_pipeline(
SessionMiddleware(self),
self
)
yield Request('https://login.example.com', callback=self.parse_login, meta={'dont_load': True})
加速速度
- 使用压缩：启用Brotli或Gzip压缩,减少数据传输时间：
from scrapy.http.request import Request
from scrapy.spider import Spider
class MySpider(Spider):
def start_requests(self):
headers = {'Range': 'bytes=-500'}
yield Request(
'https://target-website.com/largefile.html',
headers=headers,
priority=1,
meta={'compressed': True}
)
- 并行下载：避免过多请求导致封锁：
from scrapy.http.request import Request
from scrapy.spider import Spider
from scrapy.contrib.spiders import CrawlableSpider
class MySpider(CrawlableSpider):
def __init__(self):
super().__init__()
def parse(self, response):
yield from response.download(
callback=self.parse_file,
priority=1,
meta={'filename': response.url.split('/')[-1]}
)
编写代码实现
使用Scrapy创建爬虫项目：
from scrapy import Spider
from scrapy.spiders import CrawlableSpider
class MySpider(CrawlableSpider):
def parse(self, response):
yield {
'url': response.url,
'content': response.text,
}
使用BeautifulSoup：
from bs4 import BeautifulSoup
from scrapy.http.response import Response
from scrapy.spider import Spider
class MySpider(Spider):
def parse(self, response):
soup = BeautifulSoup(response.text, 'html.parser')
elements = soup.find_all('div', class_='class-name')
for el in elements:
yield {
'id': el.get('id'),
'text': el.get_text(),
}
完整示例
创建一个完整的Scrapy项目,包括中间件和配置：
from scrapy import Spider
from scrapy.contrib.spiders import CrawlableSpider
from scrapy.contrib.downloadermiddlewares import RedirectMiddleware
from scrapy.contrib.httpcache import HttpCacheMiddleware
from scrapy.contrib.middleware.memories import MemoriesMiddleware
from scrapy.contrib.middleware.request import RequestMiddleware
from scrapy.contrib.middleware.response import ResponseMiddleware
class MySpider(CrawlableSpider):
def __init__(self):
super().__init__()
self.name = 'myspider'
self.allowed_domains = ['example.com']
self.start_url = 'https://example.com'
def start_requests(self):
yield from self.crawler.engine.send_request(
Request(self.start_url, callback=self.parse),
spider=self
)
def parse(self, response):
yield from response.following
yield {
'url': response.url,
'content': response.text,
}
class MySpiderSpider(CrawlableSpider):
def __init__(self):
super().__init__()
self.name = 'myspider2'
self.allowed_domains = ['another.com']
self.start_url = 'https://another.com'
def start_requests(self):
yield from self.crawler.engine.send_request(
Request(self.start_url, callback=self.parse),
spider=self
)
def parse(self, response):
yield from response.following
yield {
'url': response.url,
'content': response.text
