爬虫(三)存储数据

r囧r小猫 2022-05-23 05:09 204阅读 0赞

#### 存储数据 ####

##### 数据缓存 #####

通过前面的学习, 我们已经知道了如何从指定的页面中抓取数据, 以及如何保存抓取的结果, 但是我们没有考虑过这么一种情况, 就是我们可能需要从已经抓取过的页面提取出更多数据, 重新去下载这些页面对于规模不大的网站倒是没有问题, 但是如果能够把这些页面缓存起来, 对应用的性能会有明显的改善.

##### 使用NoSQL #####

Redis简介  
Redis是REmoteDIctionary Server的缩写, 它是一个用ANSI C编写的高性能的key-value存储系统, 与其他的key-value存储系统相对, Redis有以下一些特点(也是优点):

*  Redis的读写性能极高, 并且有丰富的特性(发布/订阅, 事务, 通知等).
 *  Redis支持数据的持久化(RDB和AOF两种方式), 可以将内存中的数据保存你在磁盘中, 重启的时候可以再次加载进行使用.
 *  Redis不仅仅支持简单的key-value类型的数据, 同时还hash, list, set, zset, hyperloglog, geo等数据类型.
 *  Redis支持主从复制(实现读写分离)以及哨兵模式(监控master是否宕机并调整配置).  
    Redis 的安装和配置  
    可以使用Linux系统的包管理工具(如CentOS里面的yum或者Ubuntu里面的apt)来安装Redis, 可以通过在Redis 的官方网站下载Redis 的源代码解压缩解归档之后进行构件安装.
    
        # wget http://download.redis.io/releases/redis-3.2.11.tar.gz
        # gunzip redis-3.2.11.tar.gz
        # tar -xvf redis-3.2.11.tar
        # cd redis-3.2.11
        # make && make install
    
    接下来我们将redis-3.2.11目录下的redis.conf配置文件赋值到用户主目录下并修改配置文件(如果你对配置文件不是很有把握就不要直接修改而是赋值一份再修改这个副本)
    
        # cd ..
        # cp redis-3.2.11/redis.conf redis.conf
        # vim redis.conf
    
    配置将Redis服务绑定到指定的IP地址和端口
    
    > 具体修改步骤已经在[redis读写分离][redis]里面讲到了.

##### Redis 的服务器和客户端 #####

接下来启动Redis服务器, 可以将服务器放在后台去运行.

# redis-server redis.conf &

接下来, 我们尝试用Redis客户端去连接服务器.

# redis-cli -h 172.18.61.250 -p 6379
    172.18.61.250:6379> auth 1qaz2wsx
    OK
    172.18.61.250:6379> ping
    PONG
    172.18.61.250:6379>

Redis有着非常丰富的数据类型, 也有很多的命令来操作这些数据,具体的内容可以查看[Redis命令参考][Redis], 在这个网站上, 除了Redis的命令参考, 还有Redis 的详细文档, 其中包括了通知, 事务, 主从复制, 持久化, 哨兵, 集群等等内容.

172.18.61.250:6379> set username admin
    OK
    172.18.61.250:6379> get username
    "admin"
    172.18.61.250:6379> hset student1 name hao
    (integer) 0
    172.18.61.250:6379> hset student1 age 38
    (integer) 1
    172.18.61.250:6379> hset student1 gender male
    (integer) 1
    172.18.61.250:6379> hgetall student1
    1) "name"
    2) "hao"
    3) "age"
    4) "38"
    5) "gender"
    6) "male"
    172.18.61.250:6379> lpush num 1 2 3 4 5
    (integer) 5
    172.18.61.250:6379> lrange num 0 -1
    1) "5"
    2) "4"
    3) "3"
    4) "2"
    5) "1"
    172.18.61.250:6379> sadd fruits apple banana orange apple grape grape
    (integer) 4
    172.18.61.250:6379> scard fruits
    (integer) 4
    172.18.61.250:6379> smembers fruits
    1) "grape"
    2) "orange"
    3) "banana"
    4) "apple"
    172.18.61.250:6379> zadd scores 90 zhao 78 qian 66 sun 95 lee
    (integer) 4
    172.18.61.250:6379> zrange scores 0 -1
    1) "sun"
    2) "qian"
    3) "zhao"
    4) "lee"
    172.18.61.250:6379> zrevrange scores 0 -1
    1) "lee"
    2) "zhao"
    3) "qian"
    4) "sun"

##### 在Python程序中使用Redis #####

可以使用pip安装redis模块, redis模块的核心是名为Redis 的类, 该类的对象代表一个Redis客户端, 通过该客户端可以向Redis服务器发送命令并获取执行的结果, 上面我们在Redis客户端使用的命令基本上就是Redis对象可以接受的消息, 如果了解了Redis 的命令就可以在Python中玩转Redis.

$ pip3 install redis
    $ python3
    
    >>> import redis
    >>> client = redis.Redis(host='1.2.3.4', port=6379, password='1qaz2wsx')
    >>> client.set('username', 'admin')
    True
    >>> client.hset('student', 'name', 'hao')
    1
    >>> client.hset('student', 'age', 38)
    1
    >>> client.keys('*')
    [b'username', b'student']
    >>> client.get('username')
    b'admin'
    >>> client.hgetall('student')
    {b'name': b'hao', b'age': b'38'}

##### MongoDB简介 #####

MongoDB是2009年问世的一个面向文档的数据库管理系统, 由C++语言编写, 宗旨在为Web应用提供可扩展的高性能数据存储解决方案. 虽然在划分类别的时候, MongoDB被认为是NoSQL的产品, 但是它更像一个介于关系数据库和非关系数据库之间的产品, 在非关系数据库中它的功能最丰富, 最像关系型数据库.

##### MongoDB 的安装和配置 #####

###### 使用MongoDB实现CRUD操作 ######

###### 示例 - 缓存知乎发现上的链接和页面代码 ######

from hashlib import sha1
    from urllib.parse import urljoin
    
    import pickle
    import re
    import requests
    import zlib
    
    from bs4 import BeautifulSoup
    from redis import Redis
    
    
    def main():
        # 指定种子页面
        base_url = 'https://www.zhihu.com/'
        seed_url = urljoin(base_url, 'explore')
        # 创建Redis客户端
        client = Redis(host='1.2.3.4', port=6379, password='1qaz2wsx')
        # 设置用户代理(否则访问会被拒绝)
        headers = {'user-agent': 'Baiduspider'}
        # 通过requests模块发送GET请求并指定用户代理
        resp = requests.get(seed_url, headers=headers)
        # 创建BeautifulSoup对象并指定使用lxml作为解析器
        soup = BeautifulSoup(resp.text, 'lxml')
        href_regex = re.compile(r'^/question')
        # 将URL处理成SHA1摘要(长度固定更简短)
        hasher_proto = sha1()
        # 查找所有href属性以/question打头的a标签
        for a_tag in soup.find_all('a', {'href': href_regex}):
            # 获取a标签的href属性值并组装完整的URL
            href = a_tag.attrs['href']
            full_url = urljoin(base_url, href)
            # 传入URL生成SHA1摘要
            hasher = hasher_proto.copy()
            hasher.update(full_url.encode('utf-8'))
            field_key = hasher.hexdigest()
            # 如果Redis的键'zhihu'对应的hash数据类型中没有URL的摘要就访问页面并缓存
            if not client.hexists('zhihu', field_key):
                html_page = requests.get(full_url, headers=headers).text
                # 对页面进行序列化和压缩操作
                zipped_page = zlib.compress(pickle.dumps(html_page))
                # 使用hash数据类型保存URL摘要及其对应的页面代码
                client.hset('zhihu', field_key, zipped_page)
        # 显示总共缓存了多少个页面
        print('Total %d question pages found.' % client.hlen('zhihu'))
    
    
    if __name__ == '__main__':
        main()

#### 面向对象多线程爬虫 ####

import logging
    from random import random
    from enum import Enum, unique
    from queue import Queue
    from threading import Thread, current_thread
    from time import sleep
    from urllib.parse import urlparse
    
    import requests
    from bs4 import BeautifulSoup
    
    
    @unique
    class SpiderStatus(Enum):
        """ 枚举类:用于定义符号常量, unique代表值唯一 """
        IDEL = 0
        WORIKING = 1
    
    # 解析页面, 如果解析不到返回None, 它会从我的字符集挨个进行解码重试, 如果都不成功则返回None, 捕获异常以后会进行下一次循环. 
    def decode_page(page_bytes, charsets=('utf-8',)):
        page_html = None
        for charset in charsets:
            try:
                page_html = page_bytes.decode(charset)
                break
            except Exception as e:
                pass
                # logging.error(e)
        return page_html
    
    
    class Retry(object):
        """ 包装类(重试爬取) """
        def __init__(self, *, retry_times=3, wait_secs=5, errors=(Exception,)):
            self.retry_times = retry_times # 重试次数 默认3次
            self.wait_secs = wait_secs # 重试等待时间 5s
            self.errors = errors  # 捕获的异常
    
        def __call__(self, fn):  # 包装类回调函数, 下面就是正常的包装器写法了
    
            def wrapper(*args, **kwargs):
    
                try:
                    return fn(*args, **kwargs)
                except self.errors as e:
                    logging.error(e)
                    sleep((random() + 1) * self.wait_secs)
    
            return wrapper
    
    
    class Spider(object):
        """ 爬虫类 """
        def __init__(self):
            # 初始化状态为IDEL(空闲)
            self.status = SpiderStatus.IDEL
    
        @Retry() # 包装类和包装器用法差不多, 多了一个括号, 下面方法用于抓取页面
        def fetch(self, current_url, *, user_agent=None, proxies=None, charsets=('utf-8',)):
            pass
    
        # 解析页面
        def parse(self, html_page, *, domain='m.sohu.com'):
            pass
        # 保存数据 
        def extract(self, html_page):
            pass
        # 抽取需要的数据
        def store(self, data_dict):
            pass
    
    
    class SpiderThread(Thread):
        """ 爬虫线程类 """
        def __init__(self, spider, tasks_queue, name):
            super().__init__(daemon=True)
            self.spider = spider #传入爬虫
            self.tasks_queue = tasks_queue # 传入任务队列
            self.name = name # 线程名称
    
        # 钩子函数, 启动线程(线程.start)以后会自动执行这个方法
        def run(self):
            while True:
                current_url = self.tasks_queue.get()
                visited_urls.add(current_url)
                self.spider.status = SpiderStatus.WORIKING
                html_page = self.spider.fetch(current_url)
                if html_page not in [None, '']:
                    url_links = self.spider.parse(html_page)
                    for url_link in url_links:
                        self.tasks_queue.put(url_link)
                self.spider.status = SpiderStatus.IDEL
    
    # 判断爬虫线程的状态 
    # any([])函数里面的列表元素如果全都未false则返回false, 否则又一个为True则都返回True
    # all([])函数里面的列表元素如果有一个为false则返回false, 全为True才返回True
    def is_any_alive(spider_threads):
        return any([spider_thread.spider.status == SpiderStatus.WORIKING for spider_thread in spider_threads])
    
    # 已经访问爬取过的网页集合, 用set是为了去除重复的网页
    visited_urls = set()
    
    
    def main():
        # 创建任务队列
        task_queue = Queue()
        # 在队列中放入初始种子页面
        task_queue.put('http://m.sohu.com/')
        # 列表生成式, 用于启动10个爬虫线程来爬取页面, 传入的爬虫姓名为t0-9
        spider_threads = [SpiderThread(Spider(), task_queue, name='t%d' % i) for i in range(10)]
        # 遍历该爬虫线程列表, 启动爬虫线程
        for spider_thread in spider_threads:
            spider_thread.start()
        # 如果任务列表不为空或者爬虫不在空闲状态则主线程会一直在此处循环下去 ,直到任务队列为空并且爬虫全部为空闲状态才会停止循环, 主线程结束, 因为设置所有的子线程爬虫都为守护线程, 那在主线程结束以后,所有的子线程也就结束了!
        while not task_queue.empty() or is_any_alive(spider_threads):
             pass
    
        print('Over!')
    
    
    if __name__ == '__main__':
        main()

#### 面向对象多进程爬虫(redis+mongo) ####

import logging
    
    from _sha1 import sha1
    from random import random
    from enum import Enum, unique
    from threading import Thread, current_thread
    from time import sleep
    from urllib.parse import urlparse
    
    import zlib
    
    import pickle
    from bs4 import BeautifulSoup
    
    import pymongo
    import redis
    import requests
    from bson import Binary
    
    
    @unique
    class SpiderStatus(Enum):
        """ 枚举类:用于定义符号常量, unique代表值唯一 """
        IDEL = 0
        WORIKING = 1
    
    
    # 用于解码
    def decode_page(page_bytes, charsets=('utf-8',)):
        page_html = None
        for charset in charsets:
            try:
                page_html = page_bytes.decode(charset)
                break
            except Exception as e:
                pass
                # logging.error(e)
        return page_html
    
    
    # 包装类, 用于设置重试次数, 等待事件, 捕获异常
    class Retry(object):
    
        def __init__(self, *, retry_times=3, wait_secs=5, errors=(Exception,)):
            self.retry_times = retry_times
            self.wait_secs = wait_secs
            self.errors = errors
    
        def __call__(self, fn):
    
            def wrapper(*args, **kwargs):
    
                try:
                    return fn(*args, **kwargs)
                except self.errors as e:
                    logging.error(e)
                    sleep((random() + 1) * self.wait_secs)
    
            return wrapper
    
    
    # 爬虫类
    class Spider(object):
    
        def __init__(self):
            self.status = SpiderStatus.IDEL
    
        @Retry()
        def fetch(self, current_url, *, user_agent=None, proxies=None, charsets=('utf-8',)):
            thread_name = current_thread().name
            print(f'[{thread_name}:{current_url}]')
            headers = {
       'user-agent':user_agent} if user_agent else {}
            resp = requests.get(current_url, headers=headers, proxies=proxies)
            return decode_page(resp.content, charsets=charsets) if resp.status_code == 200 else None
    
    
        def parse(self, html_page, *, domain='m.sohu.com'):
            soup = BeautifulSoup(html_page, 'lxml')
            for a_tag in soup.body.select('a[href]'):
    
                parser = urlparse(a_tag.attrs['href'])
                netloc = parser.netloc or 'm.sohu.com'
                scheme = parser.scheme or 'http'
                if netloc == domain and scheme != 'javascript':
                    path = parser.path
                    query = '?' + parser.query if parser.query else ''
                    full_url = f'{scheme}://{netloc}{path}{query}'
                    if not redis_client.sismember('visited_url', full_url):
                        redis_client.rpush('sohu_url',full_url)
    
    
    
        def extract(self, html_page):
            pass
    
        def store(self, data_dict):
            pass
    
    
    # 爬虫线程类, 继承自Thread类, 重构run方法,
    class SpiderThread(Thread):
    
        def __init__(self, spider, name):
            super().__init__(daemon=True)
            self.spider = spider
            self.name = name
    
    
        def run(self):
            while True:
                current_url = redis_client.lpop('sohu_url')
                if current_url:
                    current_url = current_url.decode('utf-8')
                    hasher = hasher_proto.copy()
                    hasher.update(current_url.encode('utf-8'))
                    doc_id = hasher.hexdigest()
                    self.spider.status = SpiderStatus.WORIKING
                    redis_client.sadd('visited_url', current_url)
                    html_page = self.spider.fetch(current_url)
                    if html_page not in [None, '']:
                        # 创建连接mongoDB的客户端
                        mongo_client = pymongo.MongoClient(host='116.196.107.20', port=27017)
                        # 在mongoDB的sohu数据库的webpages集合里面查找是否存在_id是doc_id的数据,如果没有则插入,
                        if not mongo_client.sohu.webpages.find_one({
       '_id':doc_id}):
                            mongo_client.sohu.webpages.insert_one({
       '_id':doc_id, 'url':current_url, 'html_page':zlib.compress(pickle.dumps(html_page))})
                            # Binary(zlib.compress(pickle.dumps(html_page))) # 以二进制方式存储对象
                        self.spider.parse(html_page)
                    self.spider.status = SpiderStatus.IDEL
    
    # 判断线程是否空闲
    def is_any_alive(spider_threads):
        return any([spider_thread.spider.status == SpiderStatus.WORIKING for spider_thread in spider_threads])
    
    # 创建sha1类的对象,用于生成sha1格式的摘要
    hasher_proto = sha1()
    
    # 创建连接redis的客户端对象
    redis_client = redis.Redis(host='116.196.107.20', port=63791, password='2905058')
    
    
    
    
    def main():
        # 如果redis数据库里面没有sohu_url的key值,则创建该列表,从右边插入数据
        if not redis_client.exists('sohu_url'):
            redis_client.rpush('sohu_url', 'http://m.sohu.com/')
        spider_threads = [SpiderThread(Spider(), name='t%d' % i) for i in range(10)]
        for spider_thread in spider_threads:
            spider_thread.start()
    
        while not redis_client.llen('sohu_url') or is_any_alive(spider_threads):
             pass
    
        print('Over!')
    
    
    if __name__ == '__main__':
        main()

[redis]: https://blog.csdn.net/qq_41637554/article/details/80514613
[Redis]: http://redisdoc.com/