# 这里引入了redis模块，这个是redis-python库的接口，用于通过python访问redis数据库，
# 这个文件主要是实现连接redis数据库的功能，这些连接接口在其他文件中经常被用到

import redis
import six

from scrapy.utils.misc import load_object

DEFAULT_REDIS_CLS = redis.StrictRedis

# 可以在settings文件中配置套接字的超时时间、等待时间等
# Sane connection defaults.
DEFAULT_PARAMS = {
    ‘socket_timeout‘: 30,
    ‘socket_connect_timeout‘: 30,
    ‘retry_on_timeout‘: True,
}

# 要想连接到redis数据库，和其他数据库差不多，需要一个ip地址、端口号、用户名密码（可选）和一个整形的数据库编号
# Shortcut maps ‘setting name‘ -> ‘parmater name‘.
SETTINGS_PARAMS_MAP = {
    ‘REDIS_URL‘: ‘url‘,
    ‘REDIS_HOST‘: ‘host‘,
    ‘REDIS_PORT‘: ‘port‘,
}


def get_redis_from_settings(settings):
    """Returns a redis client instance from given Scrapy settings object.
    This function uses ``get_client`` to instantiate the client and uses
    ``DEFAULT_PARAMS`` global as defaults values for the parameters. You can
    override them using the ``REDIS_PARAMS`` setting.
    Parameters
    ----------
    settings : Settings
        A scrapy settings object. See the supported settings below.
    Returns
    -------
    server
        Redis client instance.
    Other Parameters
    ----------------
    REDIS_URL : str, optional
        Server connection URL.
    REDIS_HOST : str, optional
        Server host.
    REDIS_PORT : str, optional
        Server port.
    REDIS_PARAMS : dict, optional
        Additional client parameters.
    """
    params = DEFAULT_PARAMS.copy()
    params.update(settings.getdict(‘REDIS_PARAMS‘))
    # XXX: Deprecate REDIS_* settings.
    for source, dest in SETTINGS_PARAMS_MAP.items():
        val = settings.get(source)
        if val:
            params[dest] = val

    # Allow ``redis_cls`` to be a path to a class.
    if isinstance(params.get(‘redis_cls‘), six.string_types):
        params[‘redis_cls‘] = load_object(params[‘redis_cls‘])

    # 返回的是redis库的Redis对象，可以直接用来进行数据操作的对象
    return get_redis(**params)


# Backwards compatible alias.
from_settings = get_redis_from_settings


def get_redis(**kwargs):
    """Returns a redis client instance.
    Parameters
    ----------
    redis_cls : class, optional
        Defaults to ``redis.StrictRedis``.
    url : str, optional
        If given, ``redis_cls.from_url`` is used to instantiate the class.
    **kwargs
        Extra parameters to be passed to the ``redis_cls`` class.
    Returns
    -------
    server
        Redis client instance.
    """
    redis_cls = kwargs.pop(‘redis_cls‘, DEFAULT_REDIS_CLS)
    url = kwargs.pop(‘url‘, None)


    if url:
        return redis_cls.from_url(url, **kwargs)
    else:
        return redis_cls(**kwargs)

import logging

import time

 

from scrapy.dupefilters import BaseDupeFilter

from scrapy.utils.request import request_fingerprint

 

from .connection import get_redis_from_settings

 

 

DEFAULT_DUPEFILTER_KEY = "dupefilter:%(timestamp)s"

 

logger = logging.getLogger(__name__)

 

 

# TODO: Rename class to RedisDupeFilter.

class RFPDupeFilter(BaseDupeFilter):

    """Redis-based request duplicates filter.

    This class can also be used with default Scrapy‘s scheduler.

    """

 

    logger = logger

 

    def __init__(self, server, key, debug=False):

        """Initialize the duplicates filter.

        Parameters

        ----------

        server : redis.StrictRedis

            The redis server instance.

        key : str

            Redis key Where to store fingerprints.

        debug : bool, optional

            Whether to log filtered requests.

        """

        self.server = server

        self.key = key

        self.debug = debug

        self.logdupes = True

 

    @classmethod

    def from_settings(cls, settings):

        """Returns an instance from given settings.

        This uses by default the key ``dupefilter:<timestamp>``. When using the

        ``scrapy_redis.scheduler.Scheduler`` class, this method is not used as

        it needs to pass the spider name in the key.

        Parameters

        ----------

        settings : scrapy.settings.Settings

        Returns

        -------

        RFPDupeFilter

            A RFPDupeFilter instance.

        """

        server = get_redis_from_settings(settings)

        # XXX: This creates one-time key. needed to support to use this

        # class as standalone dupefilter with scrapy‘s default scheduler

        # if scrapy passes spider on open() method this wouldn‘t be needed

        # TODO: Use SCRAPY_JOB env as default and fallback to timestamp.

        key = DEFAULT_DUPEFILTER_KEY % {‘timestamp‘: int(time.time())}

        debug = settings.getbool(‘DUPEFILTER_DEBUG‘)

        return cls(server, key=key, debug=debug)

 

    @classmethod

    def from_crawler(cls, crawler):

        """Returns instance from crawler.

        Parameters

        ----------

        crawler : scrapy.crawler.Crawler

        Returns

        -------

        RFPDupeFilter

            Instance of RFPDupeFilter.

        """

        return cls.from_settings(crawler.settings)

 

    def request_seen(self, request):

        """Returns True if request was already seen.

        Parameters

        ----------

        request : scrapy.http.Request

        Returns

        -------

        bool

        """

        fp = self.request_fingerprint(request)

        # This returns the number of values added, zero if already exists.

        added = self.server.sadd(self.key, fp)

        return added == 0

 

    def request_fingerprint(self, request):

        """Returns a fingerprint for a given request.

        Parameters

        ----------

        request : scrapy.http.Request

        Returns

        -------

        str

        """

        return request_fingerprint(request)

 

    def close(self, reason=‘‘):

        """Delete data on close. Called by Scrapy‘s scheduler.

        Parameters

        ----------

        reason : str, optional

        """

        self.clear()

 

    def clear(self):

        """Clears fingerprints data."""

        self.server.delete(self.key)

 

    def log(self, request, spider):

        """Logs given request.

        Parameters

        ----------

        request : scrapy.http.Request

        spider : scrapy.spiders.Spider

        """

        if self.debug:

            msg = "Filtered duplicate request: %(request)s"

            self.logger.debug(msg, {‘request‘: request}, extra={‘spider‘: spider})

        elif self.logdupes:

            msg = ("Filtered duplicate request %(request)s"

                   " - no more duplicates will be shown"

                   " (see DUPEFILTER_DEBUG to show all duplicates)")

            msg = "Filtered duplicate request: %(request)s"

            self.logger.debug(msg, {‘request‘: request}, extra={‘spider‘: spider})

            self.logdupes = False

"""A pickle wrapper module with protocol=-1 by default."""

 

try:

    import cPickle as pickle  # PY2

except ImportError:

    import pickle

 

 

def loads(s):

    return pickle.loads(s)

 

 

def dumps(obj):

    return pickle.dumps(obj, protocol=-1)

from scrapy.utils.misc import load_object

from scrapy.utils.serialize import ScrapyJSONEncoder

from twisted.internet.threads import deferToThread

 

from . import connection

 

 

default_serialize = ScrapyJSONEncoder().encode

 

 

class RedisPipeline(object):

    """Pushes serialized item into a redis list/queue"""

 

    def __init__(self, server,

                 key=‘%(spider)s:items‘,

                 serialize_func=default_serialize):

        self.server = server

        self.key = key

        self.serialize = serialize_func

 

    @classmethod

    def from_settings(cls, settings):

        params = {

            ‘server‘: connection.from_settings(settings),

        }

        if settings.get(‘REDIS_ITEMS_KEY‘):

            params[‘key‘] = settings[‘REDIS_ITEMS_KEY‘]

        if settings.get(‘REDIS_ITEMS_SERIALIZER‘):

            params[‘serialize_func‘] = load_object(

                settings[‘REDIS_ITEMS_SERIALIZER‘]

            )

 

        return cls(**params)

 

    @classmethod

    def from_crawler(cls, crawler):

        return cls.from_settings(crawler.settings)

 

    def process_item(self, item, spider):

        return deferToThread(self._process_item, item, spider)

 

    def _process_item(self, item, spider):

        key = self.item_key(item, spider)

        data = self.serialize(item)

        self.server.rpush(key, data)

        return item

 

    def item_key(self, item, spider):

        """Returns redis key based on given spider.

        Override this function to use a different key depending on the item

        and/or spider.

        """

        return self.key % {‘spider‘: spider.name}

from scrapy.utils.reqser import request_to_dict, request_from_dict

 

from . import picklecompat

 

 

class Base(object):

    """Per-spider queue/stack base class"""

 

    def __init__(self, server, spider, key, serializer=None):

        """Initialize per-spider redis queue.

        Parameters:

            server -- redis connection

            spider -- spider instance

            key -- key for this queue (e.g. "%(spider)s:queue")

        """

        if serializer is None:

            # Backward compatibility.

            # TODO: deprecate pickle.

            serializer = picklecompat

        if not hasattr(serializer, ‘loads‘):

            raise TypeError("serializer does not implement ‘loads‘ function: %r"

                            % serializer)

        if not hasattr(serializer, ‘dumps‘):

            raise TypeError("serializer ‘%s‘ does not implement ‘dumps‘ function: %r"

                            % serializer)

 

        self.server = server

        self.spider = spider

        self.key = key % {‘spider‘: spider.name}

        self.serializer = serializer

 

    def _encode_request(self, request):

        """Encode a request object"""

        obj = request_to_dict(request, self.spider)

        return self.serializer.dumps(obj)

 

    def _decode_request(self, encoded_request):

        """Decode an request previously encoded"""

        obj = self.serializer.loads(encoded_request)

        return request_from_dict(obj, self.spider)

 

    def __len__(self):

        """Return the length of the queue"""

        raise NotImplementedError

 

    def push(self, request):

        """Push a request"""

        raise NotImplementedError

 

    def pop(self, timeout=0):

        """Pop a request"""

        raise NotImplementedError

 

    def clear(self):

        """Clear queue/stack"""

        self.server.delete(self.key)

 

 

class SpiderQueue(Base):

    """Per-spider FIFO queue"""

 

    def __len__(self):

        """Return the length of the queue"""

        return self.server.llen(self.key)

 

    def push(self, request):

        """Push a request"""

        self.server.lpush(self.key, self._encode_request(request))

 

    def pop(self, timeout=0):

        """Pop a request"""

        if timeout > 0:

            data = self.server.brpop(self.key, timeout)

            if isinstance(data, tuple):

                data = data[1]

        else:

            data = self.server.rpop(self.key)

        if data:

            return self._decode_request(data)

 

 

class SpiderPriorityQueue(Base):

    """Per-spider priority queue abstraction using redis‘ sorted set"""

 

    def __len__(self):

        """Return the length of the queue"""

        return self.server.zcard(self.key)

 

    def push(self, request):

        """Push a request"""

        data = self._encode_request(request)

        score = -request.priority

        # We don‘t use zadd method as the order of arguments change depending on

        # whether the class is Redis or StrictRedis, and the option of using

        # kwargs only accepts strings, not bytes.

        self.server.execute_command(‘ZADD‘, self.key, score, data)

 

    def pop(self, timeout=0):

        """

        Pop a request

        timeout not support in this queue class

        """

        # use atomic range/remove using multi/exec

        pipe = self.server.pipeline()

        pipe.multi()

        pipe.zrange(self.key, 0, 0).zremrangebyrank(self.key, 0, 0)

        results, count = pipe.execute()

        if results:

            return self._decode_request(results[0])

 

 

class SpiderStack(Base):

    """Per-spider stack"""

 

    def __len__(self):

        """Return the length of the stack"""

        return self.server.llen(self.key)

 

    def push(self, request):

        """Push a request"""

        self.server.lpush(self.key, self._encode_request(request))

 

    def pop(self, timeout=0):

        """Pop a request"""

        if timeout > 0:

            data = self.server.blpop(self.key, timeout)

            if isinstance(data, tuple):

                data = data[1]

        else:

            data = self.server.lpop(self.key)

 

        if data:

            return self._decode_request(data)

 

 

__all__ = [‘SpiderQueue‘, ‘SpiderPriorityQueue‘, ‘SpiderStack‘]

import importlib
import six

from scrapy.utils.misc import load_object

from . import connection


# TODO: add SCRAPY_JOB support.
class Scheduler(object):
    """Redis-based scheduler"""

    def __init__(self, server,
                 persist=False,
                 flush_on_start=False,
                 queue_key=‘%(spider)s:requests‘,
                 queue_cls=‘scrapy_redis.queue.SpiderPriorityQueue‘,
                 dupefilter_key=‘%(spider)s:dupefilter‘,
                 dupefilter_cls=‘scrapy_redis.dupefilter.RFPDupeFilter‘,
                 idle_before_close=0,
                 serializer=None):
        """Initialize scheduler.
        Parameters
        ----------
        server : Redis
            The redis server instance.
        persist : bool
            Whether to flush requests when closing. Default is False.
        flush_on_start : bool
            Whether to flush requests on start. Default is False.
        queue_key : str
            Requests queue key.
        queue_cls : str
            Importable path to the queue class.
        dupefilter_key : str
            Duplicates filter key.
        dupefilter_cls : str
            Importable path to the dupefilter class.
        idle_before_close : int
            Timeout before giving up.
        """
        if idle_before_close < 0:
            raise TypeError("idle_before_close cannot be negative")

        self.server = server
        self.persist = persist
        self.flush_on_start = flush_on_start
        self.queue_key = queue_key
        self.queue_cls = queue_cls
        self.dupefilter_cls = dupefilter_cls
        self.dupefilter_key = dupefilter_key
        self.idle_before_close = idle_before_close
        self.serializer = serializer
        self.stats = None

    def __len__(self):
        return len(self.queue)

    @classmethod
    def from_settings(cls, settings):
        kwargs = {
            ‘persist‘: settings.getbool(‘SCHEDULER_PERSIST‘),
            ‘flush_on_start‘: settings.getbool(‘SCHEDULER_FLUSH_ON_START‘),
            ‘idle_before_close‘: settings.getint(‘SCHEDULER_IDLE_BEFORE_CLOSE‘),
        }

        # If these values are missing, it means we want to use the defaults.
        optional = {
            # TODO: Use custom prefixes for this settings to note that are
            # specific to scrapy-redis.
            ‘queue_key‘: ‘SCHEDULER_QUEUE_KEY‘,
            ‘queue_cls‘: ‘SCHEDULER_QUEUE_CLASS‘,
            ‘dupefilter_key‘: ‘SCHEDULER_DUPEFILTER_KEY‘,
            # We use the default setting name to keep compatibility.
            ‘dupefilter_cls‘: ‘DUPEFILTER_CLASS‘,
            ‘serializer‘: ‘SCHEDULER_SERIALIZER‘,
        }
        for name, setting_name in optional.items():
            val = settings.get(setting_name)
            if val:
                kwargs[name] = val

        # Support serializer as a path to a module.
        if isinstance(kwargs.get(‘serializer‘), six.string_types):
            kwargs[‘serializer‘] = importlib.import_module(kwargs[‘serializer‘])

        server = connection.from_settings(settings)
        # Ensure the connection is working.
        server.ping()

        return cls(server=server, **kwargs)

    @classmethod
    def from_crawler(cls, crawler):
        instance = cls.from_settings(crawler.settings)
        # FIXME: for now, stats are only supported from this constructor
        instance.stats = crawler.stats
        return instance

    def open(self, spider):
        self.spider = spider

        try:
            self.queue = load_object(self.queue_cls)(
                server=self.server,
                spider=spider,
                key=self.queue_key % {‘spider‘: spider.name},
                serializer=self.serializer,
            )
        except TypeError as e:
            raise ValueError("Failed to instantiate queue class ‘%s‘: %s",
                             self.queue_cls, e)

        try:
            self.df = load_object(self.dupefilter_cls)(
                server=self.server,
                key=self.dupefilter_key % {‘spider‘: spider.name},
                debug=spider.settings.getbool(‘DUPEFILTER_DEBUG‘),
            )
        except TypeError as e:
            raise ValueError("Failed to instantiate dupefilter class ‘%s‘: %s",
                             self.dupefilter_cls, e)

        if self.flush_on_start:
            self.flush()
        # notice if there are requests already in the queue to resume the crawl
        if len(self.queue):
            spider.log("Resuming crawl (%d requests scheduled)" % len(self.queue))

    def close(self, reason):
        if not self.persist:
            self.flush()

    def flush(self):
        self.df.clear()
        self.queue.clear()

    def enqueue_request(self, request):
        if not request.dont_filter and self.df.request_seen(request):
            self.df.log(request, self.spider)
            return False
        if self.stats:
            self.stats.inc_value(‘scheduler/enqueued/redis‘, spider=self.spider)
        self.queue.push(request)
        return True

    def next_request(self):
        block_pop_timeout = self.idle_before_close
        request = self.queue.pop(block_pop_timeout)
        if request and self.stats:
            self.stats.inc_value(‘scheduler/dequeued/redis‘, spider=self.spider)
        return request

    def has_pending_requests(self):
        return len(self) > 0

from scrapy import signals

from scrapy.exceptions import DontCloseSpider

from scrapy.spiders import Spider, CrawlSpider

 

from . import connection

 

 

# Default batch size matches default concurrent requests setting.

DEFAULT_START_URLS_BATCH_SIZE = 16

DEFAULT_START_URLS_KEY = ‘%(name)s:start_urls‘

 

 

class RedisMixin(object):

    """Mixin class to implement reading urls from a redis queue."""

    # Per spider redis key, default to DEFAULT_START_URLS_KEY.

    redis_key = None

    # Fetch this amount of start urls when idle. Default to DEFAULT_START_URLS_BATCH_SIZE.

    redis_batch_size = None

    # Redis client instance.

    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‘, DEFAULT_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:

            self.redis_batch_size = settings.getint(

                ‘REDIS_START_URLS_BATCH_SIZE‘, DEFAULT_START_URLS_BATCH_SIZE,

            )

 

        try:

            self.redis_batch_size = int(self.redis_batch_size)

        except (TypeError, ValueError):

            raise ValueError("redis_batch_size must be an integer")

 

        self.logger.info("Reading start URLs from redis key ‘%(redis_key)s‘ "

                         "(batch size: %(redis_batch_size)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‘)

        fetch_one = self.server.spop if use_set else self.server.lpop

        # XXX: Do we need to use a timeout here?

        found = 0

        while found < self.redis_batch_size:

            data = fetch_one(self.redis_key)

            if not data:

                # Queue empty.

                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):

        # By default, data is an URL.

        if ‘://‘ in data:

            return self.make_requests_from_url(data)

        else:

            self.logger.error("Unexpected URL from ‘%s‘: %r", self.redis_key, data)

 

    def schedule_next_requests(self):

        """Schedules a request if available"""

        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."""

 

    @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."""

 

    @classmethod

    def from_crawler(self, crawler, *args, **kwargs):

        obj = super(RedisCrawlSpider, self).from_crawler(crawler, *args, **kwargs)

        obj.setup_redis(crawler)

        return obj