scrapy-redis 使 redis 不止保存 url（例如：json）

你的名字 2023-02-11 03:25 2阅读 0赞

scrapy-redis 使 redis 不止保存 url：[https://my.oschina.net/u/4382439/blog/3712637][https_my.oschina.net_u_4382439_blog_3712637]

Scrapy-redis 和 Scrapyd 用法详解：[https://zhuanlan.zhihu.com/p/44564597][https_zhuanlan.zhihu.com_p_44564597]

Scrapy-redis GitHub 地址：[https://github.com/rmax/scrapy-redis][https_github.com_rmax_scrapy-redis]

scrapy-redis中url队列类型的控制(zset、list)：[https://blog.csdn.net/ryuhfxz/article/details/85782467][https_blog.csdn.net_ryuhfxz_article_details_85782467]

先看 scrapy-redis 源码 （ spider.py ）:

from scrapy import signals
    from scrapy.exceptions import DontCloseSpider
    from scrapy.spiders import Spider, CrawlSpider
    
    from . import connection, defaults
    from .utils import bytes_to_str
    
    
    class RedisMixin(object):
        """Mixin class to implement reading urls from a redis queue."""
        redis_key = None
        redis_batch_size = None
        redis_encoding = None
    
        # Redis client placeholder.
        server = None
    
        def start_requests(self):
            """Returns a batch of start requests from redis."""
            return self.next_requests()
    
        def setup_redis(self, crawler=None):
            """Setup redis connection and idle signal.
    
            This should be called after the spider has set its crawler object.
            """
            if self.server is not None:
                return
    
            if crawler is None:
                # We allow optional crawler argument to keep backwards
                # compatibility.
                # XXX: Raise a deprecation warning.
                crawler = getattr(self, 'crawler', None)
    
            if crawler is None:
                raise ValueError("crawler is required")
    
            settings = crawler.settings
    
            if self.redis_key is None:
                self.redis_key = settings.get(
                    'REDIS_START_URLS_KEY', defaults.START_URLS_KEY,
                )
    
            self.redis_key = self.redis_key % {'name': self.name}
    
            if not self.redis_key.strip():
                raise ValueError("redis_key must not be empty")
    
            if self.redis_batch_size is None:
                # TODO: Deprecate this setting (REDIS_START_URLS_BATCH_SIZE).
                self.redis_batch_size = settings.getint(
                    'REDIS_START_URLS_BATCH_SIZE',
                    settings.getint('CONCURRENT_REQUESTS'),
                )
    
            try:
                self.redis_batch_size = int(self.redis_batch_size)
            except (TypeError, ValueError):
                raise ValueError("redis_batch_size must be an integer")
    
            if self.redis_encoding is None:
                self.redis_encoding = settings.get('REDIS_ENCODING', defaults.REDIS_ENCODING)
    
            self.logger.info("Reading start URLs from redis key '%(redis_key)s' "
                             "(batch size: %(redis_batch_size)s, encoding: %(redis_encoding)s",
                             self.__dict__)
    
            self.server = connection.from_settings(crawler.settings)
            # The idle signal is called when the spider has no requests left,
            # that's when we will schedule new requests from redis queue
            crawler.signals.connect(self.spider_idle, signal=signals.spider_idle)
    
        def next_requests(self):
            """Returns a request to be scheduled or none."""
            use_set = self.settings.getbool('REDIS_START_URLS_AS_SET', defaults.START_URLS_AS_SET)
            fetch_one = self.server.spop if use_set else self.server.lpop
            # XXX: Do we need to use a timeout here?
            found = 0
            # TODO: Use redis pipeline execution.
            while found < self.redis_batch_size:
                data = fetch_one(self.redis_key)
                if not data:
                    # Queue empty.
                    print('task is none')
                    break
                req = self.make_request_from_data(data)
                if req:
                    yield req
                    found += 1
                else:
                    self.logger.debug("Request not made from data: %r", data)
    
            if found:
                self.logger.debug("Read %s requests from '%s'", found, self.redis_key)
    
        def make_request_from_data(self, data):
            """Returns a Request instance from data coming from Redis.
    
            By default, ``data`` is an encoded URL. You can override this method to
            provide your own message decoding.
    
            Parameters
            ----------
            data : bytes
                Message from redis.
    
            """
            url = bytes_to_str(data, self.redis_encoding)
            return self.make_requests_from_url(url)
    
        def schedule_next_requests(self):
            """Schedules a request if available"""
            # TODO: While there is capacity, schedule a batch of redis requests.
            for req in self.next_requests():
                self.crawler.engine.crawl(req, spider=self)
    
        def spider_idle(self):
            """Schedules a request if available, otherwise waits."""
            # XXX: Handle a sentinel to close the spider.
            self.schedule_next_requests()
            raise DontCloseSpider
    
    
    class RedisSpider(RedisMixin, Spider):
        """Spider that reads urls from redis queue when idle.
    
        Attributes
        ----------
        redis_key : str (default: REDIS_START_URLS_KEY)
            Redis key where to fetch start URLs from..
        redis_batch_size : int (default: CONCURRENT_REQUESTS)
            Number of messages to fetch from redis on each attempt.
        redis_encoding : str (default: REDIS_ENCODING)
            Encoding to use when decoding messages from redis queue.
    
        Settings
        --------
        REDIS_START_URLS_KEY : str (default: "<spider.name>:start_urls")
            Default Redis key where to fetch start URLs from..
        REDIS_START_URLS_BATCH_SIZE : int (deprecated by CONCURRENT_REQUESTS)
            Default number of messages to fetch from redis on each attempt.
        REDIS_START_URLS_AS_SET : bool (default: False)
            Use SET operations to retrieve messages from the redis queue. If False,
            the messages are retrieve using the LPOP command.
        REDIS_ENCODING : str (default: "utf-8")
            Default encoding to use when decoding messages from redis queue.
    
        """
    
        @classmethod
        def from_crawler(self, crawler, *args, **kwargs):
            obj = super(RedisSpider, self).from_crawler(crawler, *args, **kwargs)
            obj.setup_redis(crawler)
            return obj
    
    
    class RedisCrawlSpider(RedisMixin, CrawlSpider):
        """Spider that reads urls from redis queue when idle.
    
        Attributes
        ----------
        redis_key : str (default: REDIS_START_URLS_KEY)
            Redis key where to fetch start URLs from..
        redis_batch_size : int (default: CONCURRENT_REQUESTS)
            Number of messages to fetch from redis on each attempt.
        redis_encoding : str (default: REDIS_ENCODING)
            Encoding to use when decoding messages from redis queue.
    
        Settings
        --------
        REDIS_START_URLS_KEY : str (default: "<spider.name>:start_urls")
            Default Redis key where to fetch start URLs from..
        REDIS_START_URLS_BATCH_SIZE : int (deprecated by CONCURRENT_REQUESTS)
            Default number of messages to fetch from redis on each attempt.
        REDIS_START_URLS_AS_SET : bool (default: True)
            Use SET operations to retrieve messages from the redis queue.
        REDIS_ENCODING : str (default: "utf-8")
            Default encoding to use when decoding messages from redis queue.
    
        """
    
        @classmethod
        def from_crawler(self, crawler, *args, **kwargs):
            obj = super(RedisCrawlSpider, self).from_crawler(crawler, *args, **kwargs)
            obj.setup_redis(crawler)
            return obj

## 方法 1： ##

仔细看完的话会发现：make\_request\_from\_data(self, data) 这个方法是从 redis 中返回一个请求实例，默认是一个 url 接下来重写一下这个方法直接传入到 self.make\_requests\_from\_url 一个 json 串就好了，在这个方法里面可以把这个串解析了请求 url 或者生产 url。代码如下

# -*- coding: utf-8 -*-
    
    
    from scrapy.http import Request
    from scrapy_redis.utils import bytes_to_str
    from scrapy_redis.spiders import RedisSpider
    
    
    class MySpider(RedisSpider):
        name = 'my_spider'
        redis_key = f'start_urls:{name}'
    
        def make_request_from_data(self, data):
            """
            :param data: params data bytes, Message from redis
            :return:
            """
            company = bytes_to_str(data, self.redis_encoding)
            return self.make_requests_from_url(company)
    
        def make_requests_from_url(self, company):
            data = eval(company)
            url = data["url"]
            headers = {
                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 "
                              "(KHTML, like Gecko) Chrome/70.0.3538.67 Safari/537.36",
                "Accept": "*/*"
            }
            return Request(url, self.parse, meta={"data": data}, dont_filter=True, headers=headers)
    
        def parse(self, response):
            pass
    
        
    if __name__ == '__main__':
        from scrapy import cmdline
        cmdline.execute('scrapy crawl my_spider'.split())
        pass

所以，只需要在重写  make\_request\_from\_data 方法，然后在里面解析 data 即可。

## 方法 2：重写 next\_requests 方法 ##

def next_requests(self):
        redis_ip = self.settings.get('REDIS_HOST')
        redis_port = self.settings.get('REDIS_PORT')
        redis_pw = self.settings.get('REDIS_PARAMS').get('password')
        redis_pool = redis.ConnectionPool(host=redis_ip, port=redis_port, password=redis_pw)
        self.redis_conn = redis.Redis(connection_pool=redis_pool)
    
        found = 0
    
        while found < self.redis_batch_size:
            data_raw = self.redis_conn.spop(self.redis_key)  # 从redis中取出内容
    
            if not data_raw:
                break
    
            data = json.loads(data_raw)  # 存入redis的内容是json，需要转化
    
            if "source_url" not in data:
                break
            req = scrapy.Request(url=data['source_url'],
                          meta=data['meta'])  # 发出请求
            if req:
                yield req
                found += 1
            else:
                self.logger.debug("Request not made from data: %s", data)
    
        if found:
            self.logger.debug("Read %s requests from '%s'", found, self.redis_key)

示例代码（ 为了演示，这个代码有点冗余，可以根据自己情况重写和删除 ）：

# -*- coding: utf-8 -*-
    # @Author  :
    # @File    : simple_spider_base_class.py
    # @Software: PyCharm
    # @description : XXX
    
    
    import json
    import datetime
    from abc import ABC
    from scrapy_redis import defaults
    from scrapy_redis.spiders import RedisSpider
    from scrapy.utils.project import get_project_settings
    
    
    class SpiderBaseClass(RedisSpider, ABC):
    
        def __init__(self):
            super(SpiderBaseClass, self).__init__()
            self.temp = None
            self.__server_pipeline = None
            self.task_string = None
    
        def __del__(self):
            pass
    
        def __get_server_pipeline(self):
            if not self.__server_pipeline:
                self.__server_pipeline = self.server.pipeline()
            return self.__server_pipeline
    
        def __get_custom_redis_key(self):
            custom_redis_key = self.redis_key.split(':')[1]
            return custom_redis_key
    
        def return_url(self, task_string=None):
            """
                用来重写，返回值是一个 url
            :param task_string:
            :return: url
            """
            self.temp = None
            return 'http://www.examplewebsite.com'
    
        def redis_sort_set_pop(self, r_k=None):
            """
                Pop a request
                timeout not support in this queue class
            """
            t_k = r_k if r_k else self.redis_key
            # use atomic range/remove using multi/exec
            pipe = self.__get_server_pipeline()
            pipe.multi()
            pipe.zrange(t_k, 0, 0).zremrangebyrank(t_k, 0, 0)
            results, count = pipe.execute()
            return results, count
    
        def next_requests(self):
            """Returns a request to be scheduled or none."""
            use_set = self.settings.getbool('REDIS_START_URLS_AS_SET', defaults.START_URLS_AS_SET)
            fetch_one = self.server.spop if use_set else self.server.lpop
            # XXX: Do we need to use a timeout here?
            found = 0
            # TODO: Use redis pipeline execution.
            while found < self.redis_batch_size:
                try:
                    data = fetch_one(self.redis_key)
                except BaseException as e:
                    results, count = self.redis_sort_set_pop()
                    if len(results):
                        task_string = results[0].decode('utf-8')  # 字节类型
                        self.task_string = task_string
                        msg = 'get task : {0}'.format(task_string)
                        print(msg)
                        self.logger.info(msg)
                        task_dict = json.loads(task_string)
                        if task_dict.get('spider_url', None):
                            data = task_dict['spider_url']
                            print('get url : {0}'.format(data))
                            self.logger.info('get url : {0}'.format(data))
                        else:
                            data = self.return_url(task_string=task_string)
                if not data:
                    # Queue empty.
                    print('task is none')
                    break
    
                # 添加时间标识
                # dict_data = collections.OrderedDict(dict_data)
                # dict_data['add_time'] = str(datetime.datetime.now())
                # task_string = json.dumps(dict_data, ensure_ascii=False)
                # try:
                #     self.server.zadd('mirror:{0}'.format(self.name), {task_string: row_score})
                # except BaseException as e:
                #     pass
    
                req = self.make_request_from_data(data)
                req.meta['task_string'] = self.task_string
                req.meta['start_time'] = str(datetime.datetime.now().replace(microsecond=0))
    
                if req:
                    yield req
                    found += 1
                else:
                    self.logger.debug("Request not made from data: %r", data)
    
            if found:
                self.logger.debug("Read %s requests from '%s'", found, self.redis_key)
    
        def __custom_get_task(self):
            row_str = None
            row_score = None
            pipe = self.__get_server_pipeline()
            loop = True
            while loop:
                try:
                    # 升序
                    temp = pipe.zrange(self.redis_key, 0, 0, withscores=True)[0]
                    # temp = pipe.zrange(self.custom_redis_key, 0, 0, withscores=True)[0]
                    # 降序
                    # temp = pipe.zrevrange(self.redis_key, 0, 0, withscores=True)[0]
                    row_value_score = pipe.execute()[0][0]  # <class 'tuple'>: (b'99', 99.0)
                    row_str, row_score = row_value_score
                    pipe.watch(self.redis_key)
                    pipe.multi()
                    results = pipe.zrem(self.redis_key, row_str.decode('utf-8')).execute()
                    # results = pipe.zrem(self.custom_redis_key, row_str.decode('utf-8')).execute()
                    pipe.unwatch()
                    if results[0]:  # results 是一个 list, 第一个元素代表成功与失败。1：成功。0:失败
                        # 删除成功。退出循环
                        loop = False  # 没有出现异常，说明不同线程或进程操作正确，跳出循环
                        break
                    else:
                        # 删除失败，说明数据已经被别的进程取到并且删除，需要继续取任务和删除任务
                        continue
                except WatchError as watch_error:
                    # 释放锁，继续下次循环
                    pipe.unwatch()
                    continue
                except IndexError as i_e:
                    # 当 redis 里面 没有任务时，再操作 redis 时报 IndexError 异常
                    # 没有任务时，结束循环，令 row_str=None
                    loop = False
                    row_str = None
                except BaseException as e:
                    row_str = None
                    break
            return row_str, row_score
    
        def next_requests_old(self):
            """
                Returns a request to be scheduled or none.
            :return:
            """
            use_set = self.settings.getbool('REDIS_START_URLS_AS_SET', defaults.START_URLS_AS_SET)
            # fetch_one = self.server.spop if use_set else self.server.lpop
            all_settings = get_project_settings()
            flag = all_settings.getbool('REDIS_START_URLS_AS_SET_TAG')
            # XXX: Do we need to use a timeout here?
            found = 0
            # TODO: Use redis pipeline execution.
            while found < self.redis_batch_size:
                if flag:
                    row_str, row_score = self.__custom_get_task()
                else:
                    fetch_one = self.server.spop if use_set else self.server.lpop
    
                json_data = fetch_one(self.redis_key) if not flag else row_str
    
                source = None
                if json_data:
                    task_string = json_data.decode('utf-8')  # 字节类型
                    self.task_string = task_string
                    self.logger.info('get task : {0}'.format(task_string))
    
                    dict_data = json.loads(task_string)
                    source = dict_data.get('source', None)
                    task_id = dict_data.get('id', None)
                    # LOG.INFO('get task is {0}'.format(self.task_string))
                    if dict_data.get('spider_url', None):
                        data = dict_data['spider_url']
                        print('get url : {0}'.format(data))
                        self.logger.info('get url : {0}'.format(data))
                    else:
                        data = self.return_url(task_string=task_string)
                else:
                    data = None
                    # LOG.INFO('get task is None')
                    print('get task is None')
                    self.logger.info('get task is None')
                if not data:
                    # Queue empty.
                    break
    
                if source:
                    try:
                        filter_conditions = {'_id': ObjectId(task_id)}
                    except BaseException as e:
                        filter_conditions = {'_id': task_id}
                    item = {"$set": {"status": 'running'}}
    
                    db_template_name = all_settings.get('DB_TEMPLATE_NAME', None)
                    if db_template_name:
                        db_name = db_config.get(db_template_name).get('database', None)
                        tb_name = db_config.get(db_template_name).get('echo_tb', None)
    
                        is_success = self.wb.update_one(
                            db_name=db_name, tb_name=tb_name,
                            filter_conditions=filter_conditions, item=item
                        )
                        if 0 == is_success:
                            self.logger.info('update status : running')
                        else:
                            self.logger.info('update status fail')
                # 添加时间标识
                # dict_data = collections.OrderedDict(dict_data)
                # dict_data['add_time'] = str(datetime.datetime.now())
                task_string = json.dumps(dict_data, ensure_ascii=False)
    
                # try:
                #     self.server.zadd('mirror:{0}'.format(self.name), {task_string: row_score})
                # except BaseException as e:
                #     pass
    
                req = self.make_request_from_data(data)
                req.meta['task_string'] = task_string
                req.meta['start_time'] = str(datetime.datetime.now().replace(microsecond=0))
                if req:
                    yield req
                    found += 1
                else:
                    self.logger.debug("Request not made from data: %r", data)
    
            if found:
                self.logger.debug("Read %s requests from '%s'", found, self.redis_key)
    
    
    if __name__ == '__main__':
        pass

[https_my.oschina.net_u_4382439_blog_3712637]: https://my.oschina.net/u/4382439/blog/3712637
[https_zhuanlan.zhihu.com_p_44564597]: https://zhuanlan.zhihu.com/p/44564597
[https_github.com_rmax_scrapy-redis]: https://github.com/rmax/scrapy-redis
[https_blog.csdn.net_ryuhfxz_article_details_85782467]: https://blog.csdn.net/ryuhfxz/article/details/85782467