用了 Scrapy-Redis 分布式爬虫后，一个能打的都没有！

关于 Python 系列其他文章的传送门

关于 Redis 运维 系列其他文章的传送门

使用 Scrapy + Redis – 爬取豆瓣读书¶

需求¶

爬取豆瓣读书，并提取后续链接加入待爬取队列。
通过 Redis，构建分布式爬虫，抓取最新 top 1 电影，分析其影评。

分析网站链接¶

经过对豆瓣的分页链接分析得知，start=?? 为分页参数，每页 20 条。

第一页：https://book.douban.com/tag/编程?start=0&type=T
第二页：https://book.douban.com/tag/编程?start=20&type=T

一、实现需求 1 – 核心项目开发¶

创建¶

1	scrapy startproject monsterspider . # 在已有 python 项目中构建爬虫

配置 settings.py ¶

BOT_NAME = 'monsterspider'

SPIDER_MODULES = ['monsterspider.spiders']

NEWSPIDER_MODULE = 'monsterspider.spiders'

USER_AGENT = "Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/55.0.2883.75 Safari/537.36"

ROBOTSTXT_OBEY = False
COOKIES_ENABLED = False

使用 crawl 模板创建爬虫¶

使用模板创建爬虫：

1	scrapy genspider -t crawl book douban.com

会得到如下代码：

import scrapy
from scrapy.linkextractors import LinkExtractor
from scrapy.spiders import CrawlSpider, Rule


class BookSpider(CrawlSpider):
    name = 'book'
    allowed_domains = ['douban.com']
    start_urls = ['http://douban.com/']

    rules = (
        Rule(LinkExtractor(allow=r'Items/'), callback='parse_item', follow=True),
    )

    def parse_item(self, response):
        item = {}
        #item['domain_id'] = response.xpath('//input[@id="sid"]/@value').get()
        #item['name'] = response.xpath('//div[@id="name"]').get()
        #item['description'] = response.xpath('//div[@id="description"]').get()
        return item

关于模板：

scrapy genspider [-t template] <name> <domain>
-t：这个选项可以使用一个模板来创建爬虫类，常用模板有 basic、crawl。
- basic：之前创建过 basic 模板，爬虫是 scrapy.spiders.Spider 类；
- crawl：这次使用的 crawl 模板，爬虫是 scrapy.spiders.crawl.CrawlSpider 类，是 scrapy.spiders.Spider 的子类，增强了功能。在其中可以使用 Rule 和 LinkExtractor。

规则 Rule 和 LinkExtractor 的定义：

rules 这个元组里，使用 Rule 定义了多条规则，用规则来方便的跟进链接；
LinkExtractor 从 response 中提取链接：
- allow 需要一个对象或可迭代对象，其中配置正则表达式，表示匹配什么链接，注意，它只关心 <a> 标签。
callback 定义匹配链接后执行的回调，特别注意不要使用 parse 这个名称。返回一个包含 Item 或 Request 对象的列表；
follow：为 True 是会跟进链接。

接下来在生成的模板内，补充爬虫的规则，获取电影的 title 和 rate，这里的处理和之前的示例一样：
spiders/book.py：

import scrapy
from scrapy.linkextractors import LinkExtractor
from scrapy.spiders import CrawlSpider, Rule
from ..items import BookItem

class BookSpider(CrawlSpider):
    name = 'book'
    allowed_domains = ['douban.com']

    # 起点
    start_urls = ['https://book.douban.com/tag/编程']

    rules = (
        Rule(LinkExtractor(allow=r'start=\d+'), callback='parse_item', follow=False),
    )  # follow 为 False 将不跟进

    def parse_item(self, response):
        """获取电影的 title 和 rate，这里的处理和之前的示例一样"""
        for subject in response.xpath('//li[@class="subject-item"]'):
            item = BookItem()
            item['title'] = subject.xpath('.//h2/a/text()')[0].extract().strip()
            rate = subject.xpath('.//span[@class="rating_nums"]/text()').extract()
            if rate:  # 有可能没有评分
                item['rate'] = rate[0]
            else:
                item['rate'] = '0'
            yield item

注意再注意：返回的 item 必须是 scrapy.Item 的子类，一般是先在 items 中定义好的。如果直接 return 字符串，会导致数据无法传给 pipelines 处理，并且控制台出现类型这样的报错 ERROR: Spider must return request, item, or None, got 'str' in <GET http://XXXXXXXX

编写 item¶

import scrapy


class BookItem(scrapy.Item):
    title = scrapy.Field()  # 书名
    rate = scrapy.Field()  # 分数

    def __repr__(self):
        return '<{} {}>'.format(self.__class__.__name__, dict(self))

开启 pipeline 存入 json 文件¶

在 settings.py 中开启 pipeline：

1
2
3

ITEM_PIPELINES = {
   'monsterspider.pipelines.MonsterspiderPipeline': 300,
}

将数据存入 /tmp/books.json 文件：

from itemadapter import ItemAdapter
import json


class MonsterspiderPipeline:
    def process_item(self, item, spider):
    # item 获取的 item；   spider 获取该 item 的 spider
        self.jsonfile.write(json.dumps(dict(item)) + ',\n')
        return item # 向后处理

    def open_spider(self, spider):
        filename = '/tmp/books.json'
        self.jsonfile = open(filename, 'w')
        self.jsonfile.write('[\n')

    def close_spider(self, spider):
        if self.jsonfile:
            self.jsonfile.write(']')
            self.jsonfile.close()

爬取¶

1	scrapy crawl book

目前只会爬取 start_urls 指定页中，匹配 r'start=\d+' 的链接。

即第 1 - 9 页的内容。

如果将 follow 改为 True 试一试？

所有获取的 url 中，只要匹配 r'start=\d+'，就跟进继续爬取。
会从第 1 页一直往后爬取，直到匹配不到才停止。

使用代理¶

在爬取过程中，如果遇到了反爬策略，可能会出现拒绝访问或者提示 302 Moved Temporarily 跳转到登录页，要求登录后才可以访问。

这相当于封了 IP，所以可以在爬取时使用代理变换 IP 来解决。

思路：在发起 HTTP 请求之前，会经过下载中间件，所以可以自定义一个下载中间件，在其中临时获取一个代理地址，然后再发起 HTTP 请求。

从 http://www.xicidaili.com/ 代理上找到免费代理，测试通过后，可以加入到代码中。（免费的，貌似已经挂了）

或者使用商业代理，每天有免费的 20 个 ip 可以使用，这次使用的这个：
http://h.wandouip.com/get

1、编写 downloader 中间件¶

仿照 middlewares.py 中的下载中间件写，编写 process_request，返回 None。

class ProxyDownloaderMiddleware:
    # 增加代理 IP 池，可以从网络搜罗，可以从配置文件中读取
    proxy_ip = "125.122.169.114"  # 代理 ip，通过 http://h.wandouip.com/get 获得
    proxy_port = "36410"  # 代理端口号
    proxies = [
        f'http://{proxy_ip}:{proxy_port}'
    ]

    def process_request(self, request: Request, spider):
        request.meta['proxy'] = random.choice(self.proxies)  # 增加 proxy
        print(request.url, request.meta['proxy'])
        print('-' * 30)

2、配置¶

在 settings.py 中启动中间件：

# 3秒
DOWNLOAD_DELAY = 3

DOWNLOADER_MIDDLEWARES = {
   'monsterspider.middlewares.ProxyDownloaderMiddleware': 123,
}

3、测试¶

在 book.py 内增加一个测试的 spider 类 IPTestSpider，用来测试代理功能。

访问 http://myip.ipip.net 可以看到自己的 IP。所以如果使用代理成功，那么该爬虫返回的内容应该是当前代理的 IP 信息。

class IPTestSpider(Spider):
    name = 'test'
    allowed_domains = ['ipip.net']
    # 起点
    start_urls = ['http://myip.ipip.net/']


    def parse(self, response):
        text = response.text
        print(text, '=======================')
        item = BookItem()
        item['title'] = text
        return item

注意：使用豌豆代理时，要把自己的 IP 申请到白名单，不然会返回 407 拒绝
添加白名单接口(登录状态下调用)：

https://h.wandouip.com/get/add-white-list?iplist=你的IP

测试爬取：

1	]$ scrapy crawl test

查看控制台打印的内容，是不是代理的地址信息。如果代理测试成功，/tmp/books.json 文件中也应该有类似这样的 IP 信息：

1
2
3

[
{"title": "\u5f53\u524d IP\uff1a114.97.201.169  \u6765\u81ea\u4e8e\uff1a\u4e2d\u56fd \u5b89\u5fbd \u5408\u80a5  \u7535\u4fe1\n"},
]

二、实现需求 2 – 使用 Scrapy-redis 组件构建分布式爬虫¶

概述¶

这是一个能给 Scrapy 框架引入分布式的组件。

分布式由 Redis 提供支持，可以在不同节点上运行爬虫，共用同一个 Redis 实例。

大家将待爬取的 URLs，和结果 Items 都储存在同一个 Redis 中。

在以下方面做了增强：

Scheduler：
- 本质上将原来的普通队列，变成了 redis 以提供多爬虫多进程共享，并行能力增强。
Duplication Filter：
- scrapy 使用 python 中的 set 类型来去重；
- scrapy-redis 使用 redis 的 set 类型去重。
Item Pipeline：
- 在 Item Pipeline 中增加一个处理，即将数据 items 存入 redis 的 items queue 中。
Base Spiders：
- 提供使用了 RedisMixin 的 RedisSpider 和 RedisCrawlSpider 类，从 Redis 中读取 Url。

Redis 是服务，爬虫就是它的客户端，这样客户端就可以扩展出并行的很多爬虫一起爬取。

安装¶

1	pip install scrapy-redis

豆瓣影评项目分析¶

抓取内容分析：模板是抓取最新 top 1 的电影，分析其影评。

在豆瓣电影，点击“全部正在热映”：跳转至 https://movie.douban.com/cinema/nowplaying/beijing/ 。

这部分内容是直接通过网页 HTML 返回的，提取 影片 id 的 xpath 为：

1	//div[@id="nowplaying"]//li[1]/@id

影片页：

点击 top 1 的电影，打开影片页，我这里是：https://movie.douban.com/subject/27073752/?from=playing_poster
在页面下面的短评处点击【全部(n) 条】进入影片评论页面
从而得到 影评页面的链接：https://movie.douban.com/subject/27073752/comments?start=0&limit=20&status=P&sort=new_score

经过对链接测试后发现：

只有 start 有用，limit 不能控制返回的条目数；
当 start 测试到 220 时，发现不能返回数据了。其实很多网站都有这种策略，显示的数据可能有很多页，但是人一般不可能看那么多页，能看查回来的 10 页结果就不错了。

提取影评的 xpath：

1	//div[@class="comment"]//span[@class="short"]

创建 Scrapy 项目¶

1	scrapy startproject review moviereview

配置使用 scrapy-redis¶

在 setting.py 中增加下面配置，来使用 scrapy-redis：

BOT_NAME = 'review'

SPIDER_MODULES = ['review.spiders']
NEWSPIDER_MODULE = 'review.spiders'

USER_AGENT = "Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/55.0.2883.75 Safari/537.36"

ROBOTSTXT_OBEY = False
DOWNLOAD_DELAY = 3
COOKIES_ENABLED = False

# ---------- 以上修改 ----------
# ---------- 以下新增 ----------

# Enables scheduling storing requests queue in redis.
SCHEDULER = "scrapy_redis.scheduler.Scheduler"

# Ensure all spiders share same duplicates filter through redis.
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"

# Don't cleanup redis queues, allows to pause/resume crawls.
#SCHEDULER_PERSIST = True

# The item pipeline serializes and stores the items in this redis key.
# 这个key很重要
#REDIS_ITEMS_KEY = '%(spider)s:items'

# Default start urls key for RedisSpider and RedisCrawlSpider.
# 起始 url 的 key
#REDIS_START_URLS_KEY = '%(name)s:start_urls'

# Store scraped item in redis for post-processing.
# 开启这个才会往 redis 里存
ITEM_PIPELINES = {
   'review.pipelines.ReviewPipeline': 300,
   'scrapy_redis.pipelines.RedisPipeline': 300
}

# Specify the host and port to use when connecting to Redis (optional).
REDIS_HOST = 'localhost'
REDIS_PORT = 6379

#LOG_LEVEL = 'DEBUG'

构建 Item¶

import scrapy

class ReviewItem(scrapy.Item):
    review = scrapy.Field()

构建爬虫¶

创建爬虫：

1	scrapy genspider -t crawl dbreview douban.com

按照下面的代码写完，使用 scrapy crawl dbreview 先测试一下。

如果成功，再将类改为 RedisCrawlSpider，就代表启用 scrapy-redis 了。

import scrapy
from scrapy.linkextractors import LinkExtractor
from scrapy.spiders import CrawlSpider, Rule
from ..items import ReviewItem
from scrapy_redis.spiders import RedisCrawlSpider


class DbreviewSpider(RedisCrawlSpider):  # 使用这个类，就代表使用 scrapy-redis 了
    name = 'dbreview'
    allowed_domains = ['douban.com']

    # 影评页面的链接，在没有使用 redis 时使用
    # start_urls = ['https://movie.douban.com/subject/27073752/comments?start=0&limit=20&status=P&sort=new_score']

    """Spider that reads urls from redis queue (myspider:start_urls)."""
    redis_key = 'dbreview:start_urls'  # 从 redis 获取数据时使用的默认 key

    rules = (
        Rule(LinkExtractor(allow=r'start=\d+'), callback='parse_item', follow=False),
    )

    def parse_item(self, response):
        print('-' * 30)

        comment = '//div[@class="comment"]//span[@class="short"]/text()'  # 提取影评的 xpath

        reviews = response.xpath(comment).extract()
        for review in reviews:
            item = ReviewItem()
            item['review'] = review.strip()

            # print('----------* ', item['review'])

            yield item

启动爬虫¶

1	scrapy crawl dbreview

会发现程序会卡住，这是因为在等待起始 URL。

在 redis 添加开始 url¶

注意：以下操作可能会清空 redis 的库。

在 redis 中推入起始 url：

1
2
3

# key 的格式在 setting 中定义了
# REDIS_START_URLS_KEY = '%(name)s:start_urls'
lpush dbreview:start_urls https://movie.douban.com/subject/27073752/comments?start=0&limit=20&status=P&sort=new_score

如果成功，在爬虫就会开始工作。而分析好的 items 会被存到 redis 中。

在 redis 中，会被创建两条：

dbreview:dupefilter：set 类型，保存了待爬取 url 的散列值，会被去重；
dbreview:items：list 类型，保存了处理好的 item。

三、数据分析和展示¶

使用爬虫，爬取所有数据，然后使用 redis 中的数据开始分析

jieba 分词¶

官网：https://github.com/fxsjy/jieba

安装：

1	pip install jieba

测试代码：

# encoding=utf-8
import jieba

seg_list = jieba.cut("我来到北京清华大学", cut_all=True)
print("Full Mode: " + "/ ".join(seg_list))  # 全模式

seg_list = jieba.cut("我来到北京清华大学", cut_all=False)
print("Default Mode: " + "/ ".join(seg_list))  # 精确模式

seg_list = jieba.cut("他来到了网易杭研大厦")  # 默认是精确模式
print(", ".join(seg_list))

seg_list = jieba.cut_for_search("小明硕士毕业于中国科学院计算所，后在日本京都大学深造")  # 搜索引擎模式
print(", ".join(seg_list))

s = jieba.lcut("他来到了网易杭研大厦")  # 直接返回列表
print(s)

s = jieba.cut("他来到了网易杭研大厦")  # 返回生成器
print(s)

stopword 停用词¶

数据清洗：把脏数据洗掉。检测出并去除掉数据中无效或无关的数据。例如，空值、非法值的检测，重复数据检测等。

对于一条条影评来说，我们分析的数据中包含了很多无效的数据，比如标点符号、英文的冠词、中文"的"等等，需要把它们清除掉。使用停用词来去除这些无效的数据。

项目地址之一：https://github.com/goto456/stopwords

1	git clone https://github.com/goto456/stopwords.git

wordcloud 词云¶

官网：https://amueller.github.io/word_cloud/index.html

依赖：numpy、matplotlib

安装：

1	pip install wordcloud

方法	说明
`fit_words(frequencies)`	Create a word_cloud from words and frequencies
`generate(text)`	Generate wordcloud from text
`generate_from_frequencies(frequencies[, …])`	Create a word_cloud from words and frequencies
`generate_from_text(text)`	Generate wordcloud from text
`process_text(text)`	Splits a long text into words, eliminates the stopwords
`recolor([random_state, color_func, colormap])`	Recolor existing layout
`to_array()`	Convert to numpy array
`to_file(filename)`	Export to image file

代码实现¶

from redis import Redis
import json
import jieba

# 创建中文的停用词
stopwords = set()
with open('/private/tmp/stopwords/cn_stopwords.txt', encoding='utf8') as f:
    for line in f:
        # print(line.rstrip('\r\n'))
        stopwords.add(line.rstrip('\r\n'))
print(len(stopwords))
print(stopwords)

# 从 redis 拿回处理好的 items
redis = Redis()
items = redis.lrange('dbreview:items', 0, -1)
print(type(items))  # list

# 提取出 item，进行分词
words = {}
for item in items:
    val = json.loads(item)['review']
    for word in jieba.cut(val):
        if word not in stopwords:  # 如果分后的词，在停用词中，就不要了
            words[word] = words.get(word, 0) + 1  # 词为 key，出现次数为 value
total = len(words)
print(total)

# 计算词出现的比值
frenq = {k: v / total for k, v in words.items()}
print(sorted(frenq.items(), key=lambda x: x[1], reverse=True))
# print(sorted(words.items(), key=lambda x: x[1], reverse=True))


# 制作词云
from wordcloud import WordCloud
import matplotlib.pyplot as plt

wordcloud = WordCloud(font_path='/System/Library/Fonts/STHeiti Medium.ttc', background_color='white',
                      max_font_size=200, width=800, height=400, min_font_size=8)

plt.figure(2)
wordcloud.fit_words(frenq)
plt.imshow(wordcloud)
plt.axis('off')  # 去掉坐标系
plt.show()