爬虫最后一天,爬取到的数据存到mysql中，爬虫和下载中间件加代理cookieheaderselenium随机生成uersagent去重规则源码分析(布隆过滤器)scrapy-redis实现分布式爬虫

Posted 2023-04-09 人生苦短，我用python

爬到的数据存到mysql中

class FirstscrapyMySqlPipeline:
    def open_spider(self, spider):
        print(\'我开了\')
        self.conn = pymysql.connect(
            user=\'root\',
            password="",
            host=\'127.0.0.1\',
            database=\'cnblogs\',
            port=3306)
        self.cursor = self.conn.cursor()

    def close_spider(self, spider):
        print(\'我关了\')

        self.cursor.close()
        self.conn.close()

    # 这个很重要
    def process_item(self, item, spider):
        sql = \'\'\'INSERT INTO aritcle (title,author_img,author_name,`desc`,url,content) VALUES(%s,%s,%s,%s,%s,%s);\'\'\'
        print(len(item[\'content\']))  # 可能会超长
        # print(\'-----\')
        self.cursor.execute(sql,
                            args=[item[\'title\'], item[\'author_img\'], item[\'author_name\'], item[\'desc\'], item[\'url\'],item[\'content\']
                                  ])
        self.conn.commit()
        return item


cnblogs 中代码
import scrapy

from bs4 import BeautifulSoup
from firstscrapy.items import FirstscrapyItem
from scrapy.http.request import Request


class CnblogsSpider(scrapy.Spider):
    name = "cnblogs"
    allowed_domains = ["www.cnblogs.com"]
    start_urls = ["http://www.cnblogs.com/"]

    # def parse(self, response):
    # 爬完start_urls地址后会执行它，有个response 爬完后的响应对象，http响应的所有东西都在里面
    # print(\'爬取完了\')
    # from scrapy.http.response.html import HtmlResponse
    # print(type(response))
    # print(response.status)

    # 解析返回的数据
    # 方式一   bs4  解析
    #     print(response.text)
    # soup = BeautifulSoup(response.text,\'lxml\')

    #  方式二:  自带的解析
    #     response.css()
    #     response.xpath()

    # def parse(self, response):
    #     # 解析数据  css解析
    #     article_list = response.css(\'article.post-item\')
    #     for article in article_list:
    #         title = article.css(\'div.post-item-text>a::text\').extract_first()
    #         author_img = article.css(\'div.post-item-text img::attr(src)\').extract_first()
    #         author_name = article.css(\'footer span::text\').extract_first()
    #         desc_old = article.css(\'p.post-item-summary::text\').extract()
    #         desc = desc_old[0].replace(\'\\n\', \'\').replace(\' \', \'\')
    #         if not desc:
    #             desc = desc_old[1].replace(\'\\n\', \'\').replace(\' \', \'\')
    #             url = article.css(\'div.post-item-text>a::attr(href)\').extract_first()
    #         #     # 文章真正的内容，没拿到，它不在这个页面中，它在下一个页面中
    #         print(title)
    #         print(author_img)
    #         print(author_name)
    #         print(desc)
    #         print(url)
    # def parse(self, response):
    #     # 解析数据  xpath解析
    #     article_list = response.xpath(\'//*[@id="post_list"]/article\')
    #     # article_list = response.xpath(\'//article[contains(@class,"post-item")]\')
    #     for article in article_list:
    #         title = article.xpath(\'.//div/a/text()\').extract_first()
    #         author_img = article.xpath(\'.//div//img/@src\').extract_first()
    #         author_name = article.xpath(\'.//footer//span/text()\').extract_first()
    #         desc_old = article.xpath(\'.//p/text()\').extract()
    #         desc = desc_old[0].replace(\'\\n\', \'\').replace(\' \', \'\')
    #         if not desc:
    #             desc = desc_old[1].replace(\'\\n\', \'\').replace(\' \', \'\')
    #             url = article.xpath(\'.//div/a/@href\').extract_first()
    #         # 文章真正的内容，没拿到，它不在这个页面中，它在下一个页面中
    #         print(title)
    #         print(author_img)
    #         print(author_name)
    #         print(desc)
    #         print(url)
    def parse(self, response):
        # 解析数据  xpath解析
        article_list = response.xpath(\'//*[@id="post_list"]/article\')
        # article_list = response.xpath(\'//article[contains(@class,"post-item")]\')
        # l=[]
        for article in article_list:
            item = FirstscrapyItem()
            title = article.xpath(\'.//div/a/text()\').extract_first()
            item[\'title\'] = title
            author_img = article.xpath(\'.//div//img/@src\').extract_first()
            item[\'author_img\'] = author_img
            author_name = article.xpath(\'.//footer//span/text()\').extract_first()
            item[\'author_name\'] = author_name
            desc_old = article.xpath(\'.//p/text()\').extract()
            desc = desc_old[0].replace(\'\\n\', \'\').replace(\' \', \'\')
            if not desc:
                desc = desc_old[1].replace(\'\\n\', \'\').replace(\' \', \'\')
            item[\'desc\'] = desc
            url = article.xpath(\'.//div/a/@href\').extract_first()
            item[\'url\'] = url
            # yield item
            # print(title)

            # yield Request(url=url, callback=self.parser_detail, meta=\'item\': item)  # 爬完后执行的解析方法

            yield Request(url=url, callback=self.parser_detail, meta=\'item\': item)  # 爬完后执行的解析方法


        next = \'https://www.cnblogs.com\' + response.css(\'div.pager>a:last-child::attr(href)\').extract_first()
        print(next)  # 继续爬取
        yield Request(url=next, callback=self.parse)
        # yield Request(url=next)


    def parser_detail(self, response):
        # print(response.status)
        #  解析出文章内容
        content = response.css(\'#cnblogs_post_body\').extract_first()
        # print(str(content))
        # 怎么放到item中
        item = response.meta.get(\'item\')
        if content:
            item[\'content\'] = content
        else:
            item[\'content\'] = \'没查到\'

        yield item

settings中代码
ITEM_PIPELINES = 
   # "firstscrapy.pipelines.FirstscrapyFilePipeline": 300, # 数字越小 优先级越高
   "firstscrapy.pipelines.FirstscrapyMySqlPipeline": 301,

爬虫和下载中间件

# 爬虫中间件：处于爬虫和引擎之间的
# 下载中间件：处于引擎和下载器之间的

# 咱们主要用的是下载中间件，爬虫中间件，了解即可
     进来的时候是个Request对象
     出去的时候是个Response对象
# 爬虫中间件
class FirstscrapySpiderMiddleware:
    # 进入到爬虫时，触发它
    def process_spider_input(self, response, spider):
        return None
    # 出爬虫时，触发它
    def process_spider_output(self, response, result, spider):
        for i in result:
            yield i
    #出异常时，触发它
    def process_spider_exception(self, response, exception, spider):
        pass
    # 第一次爬取，
    def process_start_requests(self, start_requests, spider):
        for r in start_requests:
            yield r

    # 爬虫开启会触发
    def spider_opened(self, spider):
        spider.logger.info("Spider opened: %s" % spider.name)

# 下载中间件

class FirstscrapyDownloaderMiddleware:
   # 请求来了，从引擎进入到下载器会触发，这里有request对象
    def process_request(self, request, spider):
        # 必须返回以下情况:
        # - return None:继续下一个中间件
        # - return a Response object：结束掉，这个请求就不爬取了，回到引擎中
        # - return a Request object ：把请求对象给引擎，引擎把它再次放到调度器中，等待下一次调度
        # - raise IgnoreRequest: 就会执行process_exception() 
        return None

    def process_response(self, request, response, spider):
        # 请求下载完成，回来经过：request请求对象, response响应对象
        # - return a Response object：继续走下一个中间件的process_response
        # - return a Request object ：把请求对象给引擎，引擎把它再次放到调度器中，等待下一次调度
        # - raise IgnoreRequest：就会执行process_exception() 
        return response

    def process_exception(self, request, exception, spider):
        中间件中抛异常会走到
        pass

 # 下载中间件的process_request，因为有request对象，就是要爬取的对象
     修改请求头
     加cookie
     加代理
    --------
     集成selenium

加代理，cookie、header、加入selenium

加代理

在中间件里写东西 想要生效 先把配置文件的设置打开
DOWNLOADER_MIDDLEWARES = 
   "firstscrapy.middlewares.FirstscrapyDownloaderMiddleware": 543,

# 第一步：
    # 在下载中间件写process_request方法
    def get_proxy(self):
        import requests
        res = requests.get(\'http://127.0.0.1:5010/get/\').json()
        if res.get(\'https\'):
            return \'https://\' + res.get(\'proxy\')
        else:
            return \'http://\' + res.get(\'proxy\')
     def process_request(self, request, spider):
            request.meta[\'proxy\'] = self.get_proxy()
            return None
# 第二步：代理可能不能用，会触发process_exception，在里面写
    def process_exception(self, request, exception, spider):
        print(\'-----\',request.url)  # 这个地址没有爬
        return request

加cookie、修改请求头、随机生成UserAgent

###  加cookie
def process_request(self, request, spider):
        print(request.cookies)
        request.cookies[\'name\']=\'lqz\'
        return None
# 修改请求头
    def process_request(self, request, spider):
        print(request.headers)
        request.headers[\'referer\'] = \'http://www.lagou.com\'
        return None
# 动态生成User-agent使用
    def process_request(self, request, spider):
        # fake_useragent模块
        from fake_useragent import UserAgent
        ua = UserAgent()
        request.headers[\'User-Agent\']=str(ua.random)
        print(request.headers)

        return None

集成selenium

# 集成selenium因为有的页面，是执行完js后才渲染完，必须使用selenium去爬取数据才完整

# 第一步：在爬虫类中写
from selenium import webdriver
class CnblogsSpider(scrapy.Spider):
    bro = webdriver.Chrome(executable_path=\'./chromedriver.exe\')
    bro.implicitly_wait(10)
    def close(spider, reason):
        spider.bro.close() #浏览器关掉
        
# 第二步：在中间件中
    def process_request(self, request, spider):
        # 爬取下一页这种地址---》用selenium，但是文章详情，就用原来的
        if \'sitehome/p\' in request.url:
            spider.bro.get(request.url)
            from scrapy.http.response.html import HtmlResponse
            response = HtmlResponse(url=request.url, body=bytes(spider.bro.page_source, encoding=\'utf-8\'))
            return response
        else:
            return None

==去重规则源码分析(布隆过滤器)

# 调度器可以去重，研究一下，如何去重的
# 要爬取的Request对象，在进入到scheduler调度器排队之前，先执行enqueue_request，它如果return False，这个Request就丢弃掉，不爬了----》如何判断这个Request要不要丢弃掉，执行了self.df.request_seen(request)，它来决定的-----》RFPDupeFilter类中的方法----》request_seen---》会返回True或False----》如果这个request在集合中，说明爬过了，就return True，如果不在集合中，就加入到集合中，然后返回False

# 调度器源码
from scrapy.core.scheduler import Scheduler
     # 这个方法如果return True表示这个request要爬取，如果return False表示这个网址就不爬了(已经爬过了)
    def enqueue_request(self, request: Request) -> bool:
        # request当次要爬取的地址对象
        if self.df.request_seen(request):
            # 有的请情况，在爬虫中解析出来的网址，不想爬了，就就可以指定
            # yield Request(url=url, callback=self.detail_parse, meta=\'item\': item,dont_filter=True)
            # 如果符合这个条件，表示这个网址已经爬过了 
            return False
        return True
    
    
# self.df 去重类 是去重类的对象 RFPDupeFilter
    -在配置文件中如果配置了：DUPEFILTER_CLASS = \'scrapy.dupefilters.RFPDupeFilter\'表示，使用它作为去重类，按照它的规则做去重
     RFPDupeFilter的request_seen
       def request_seen(self, request: Request) -> bool:
        # request_fingerprint 生成指纹
        fp = self.request_fingerprint(request) #request当次要爬取的地址对象
        #判断 fp 在不在集合中，如果在，return True
        if fp in self.fingerprints:
            return True
        #如果不在，加入到集合，return False
        self.fingerprints.add(fp)
        return False

# 传进来是个request对象，生成的是指纹
     爬取的网址：https://www.cnblogs.com/teach/p/17238610.html?name=lqz&age=19
     和        https://www.cnblogs.com/teach/p/17238610.html?name=lqz&age=19
     它俩是一样的，返回的数据都是一样的，就应该是一条url，就只会爬取一次
     所以 request_fingerprint  就是来把它们做成一样的(核心原理是把查询条件排序，再拼接到后面)
    
   
     生成指纹，指纹是什么？ 生成的指纹放到集合中去重
         www.cnblogs.com?name=lqz&age=19
         www.cnblogs.com?age=19&name=lqz
         上面的两种地址生成的指纹是一样的
        # 测试指纹
        from scrapy.utils.request import RequestFingerprinter
        from scrapy import Request

        fingerprinter = RequestFingerprinter()
        request1 = Request(url=\'http://www.cnblogs.com?name=lqz&age=20\')
        request2 = Request(url=\'http://www.cnblogs.com?age=20&name=lqz\')

        res1 = fingerprinter.fingerprint(request1).hex()
        res2 = fingerprinter.fingerprint(request2).hex()
        print(res1)
        print(res2)

布隆过滤器

# 总结：scrapy的去重规则
     根据配置的去重类RFPDupeFilter的request_seen方法，如果返回True，就不爬了，如果返回False就爬
     后期咱们可以使用自己定义的去重类，实现去重
    
    
 # 更小内存实现去重
     如果是集合：存的数据库越多，占内存空间越大，如果数据量特别大，可以使用布隆过滤器实现去重

 # 布隆过滤器：https://zhuanlan.zhihu.com/p/94668361
    #bloomfilter：是一个通过多哈希函数映射到一张表的数据结构，能够快速的判断一个元素在一个集合内是否存在，具有很好的空间和时间效率。（典型例子，爬虫url去重）

    # 原理： BloomFilter 会开辟一个m位的bitArray(位数组)，开始所有数据全部置 0 。当一个元素（www.baidu.com）过来时，能过多个哈希函数（h1,h2,h3....）计算不同的在哈希值，并通过哈希值找到对应的bitArray下标处，将里面的值 0 置为 1 。

    
# Python中使用布隆过滤器
# 测试布隆过滤器
# 可以自动扩容指定错误率，底层数组如果大于了错误率会自动扩容
# from pybloom_live import ScalableBloomFilter
# bloom = ScalableBloomFilter(initial_capacity=100, error_rate=0.001, mode=ScalableBloomFilter.LARGE_SET_GROWTH)
# url = "www.cnblogs.com"
# url2 = "www.liuqingzheng.top"
# bloom.add(url)
# bloom.add(url2)
# print(url in bloom)
# print(url2 in bloom)

from pybloom_live import BloomFilter

bf = BloomFilter(capacity=10)
url = \'www.baidu.com\'
bf.add(url)
bf.add(\'aaaa\')
bf.add(\'ggg\')
bf.add(\'deww\')
bf.add(\'aerqaaa\')
bf.add(\'ae2rqaaa\')
bf.add(\'aerweqaaa\')
bf.add(\'aerwewqaaa\')
bf.add(\'aerereweqaaa\')
bf.add(\'we\')


print(url in bf)
print("wa" in bf)

# 如果有去重的情况，就可以使用集合---》但是集合占的内存空间大，如果到了亿级别的数据量，想一种更小内存占用，而去重的方案----》布隆过滤器
# 布隆过滤器：通过不同的hash函数，加底层数组实现的极小内存去重
# python中如何使用：pybloom_live  
     指定错误率
     指定大小


# 使用redis实现布隆过滤器
     编译redis---》把第三方扩展布隆过滤器编译进去，才有这个功能
     https://zhuanlan.zhihu.com/p/94668736
# 重写scrapy的过滤类

==scrapy-redis实现分布式爬虫

# 什么是分布式爬虫
     集群：一个项目，在多个机器上部署，每个机器完成完整的功能，称之为集群
    原来使用一台机器爬取cnblogs整站
     现在想使用3台机器爬取cnblogs整站
          每台机器爬取数据是不一样的
          最终组装成完整的数据
# 如果变成分布式，面临的问题
     1 去重集合，我们要使用同一个----》redis集合
     2 多台机器使用同一个调度器：Scheduler，排队爬取，使用同一个队列

# scrapy-redis 已经解决这个问题了，我只需要在我们单机基础上，改动一点，就变成了分布式爬虫
# 使用步骤
	第一步：安装scrapy-redis  ---》pip3 install scrapy-redis
    第二步：改造爬虫类
    from scrapy_redis.spiders import RedisSpider
    class CnblogSpider(RedisSpider):
        name = \'cnblog_redis\'
        allowed_domains = [\'cnblogs.com\']
        # 写一个key：redis列表的key，起始爬取的地址
        redis_key = \'myspider:start_urls\'
        
    第三步：配置文件配置
    # 分布式爬虫配置
    # 去重规则使用redis
    REDIS_HOST = \'localhost\'                            # 主机名
    REDIS_PORT = 6379                                   # 端口
    DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter" #看了源码
    SCHEDULER = "scrapy_redis.scheduler.Scheduler"   # 先进先出：队列，先进后出：栈
    # 持久化：文件，mysql，redis
    ITEM_PIPELINES = 
       \'cnblogs.pipelines.CnblogsFilePipeline\': 300,
       \'cnblogs.pipelines.CnblogsMysqlPipeline\': 100,
       \'scrapy_redis.pipelines.RedisPipeline\': 400,  #简单看
    
    第四步：在多台机器上启动scrapy项目，在一台机器起了多个scrapy爬虫进程，就相当于多台机器
    
    第五步：把起始爬取的地址放到redis的列表中
    lpush mycrawler:start_urls http://www.cnblogs.com/

拓展

# 原来的去重规则
class RFPDupeFilter(BaseDupeFilter):
   def request_seen(self, request):
        # request_fingerprint 生成指纹
        fp = self.request_fingerprint(request) #request当次要爬取的地址对象
        #判断 fp 在不在集合中，如果在，return True
        if fp in self.fingerprints:
            return True
        #如果不在，加入到集合，return False
        self.fingerprints.add(fp)
        return False


# scrapy-redis的去重规则
class RFPDupeFilter(BaseDupeFilter):
    def request_seen(self, request):
        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
    
    
    
    
# 持久化
class RedisPipeline(object):

    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

# 扩展
https://zhuanlan.zhihu.com/p/91643259
https://www.zhihu.com/column/c_1175438244715651072

爬虫数据存储至TXTCSVExcel.

数据存储

• TXT
• Excel
• CSV
• CSV文件互写

数据的存储是爬虫开发中一个很重要的环节，而存储的形式也分为很多种，大体来说分为两种。一种是将爬取到的数据储存到文件中，另一种就是将数据直接存储到数据库中。

本文举例子讲解如何将爬取到的数据存储到 TXT，Excel， CSV 文件中。其实这些方法大同小异，原理都差不多，只不过是分别运用了不同的库函数，使用的不同的方法而已。

TXT

存到 TXT 文件是最简单也是最好理解的，主要是用到了 open 函数。我们爬取 http://seputu.com/，获取每个标题下的章节内容、链接。

首先使用 Requests 访 问 http://seputu.com/， 获取 HTML 文 档内容。 接着分析 http://seputu.com/首页的 HTML 结构，确定要抽取标记的位置，分析如下：

标题和章节都被包围在 <div class="mulu">标记下，标题位于其中 的 <div class="mulu-title">下的<h2>中，章节位于其中的<div class="box">下的<a>中，代码如下所示:

我们将每一章节内容单独存入txt中，以章节名命名。

# 爬虫数据存储到txt中
import requests
from bs4 import BeautifulSoup

# 获取用户代理：在浏览器地址栏：about://version

# 1.获取请求
url = 'http://seputu.com'
headers = {'User_Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36'}

# 获取requet请求对象
req = requests.get(url, headers=headers)

# 2.解析数据
# 获取soup对象
soup = BeautifulSoup(req.text, 'html.parser')

# 获取所有的<div class='mulu'>
for div in soup.findAll(class_='mulu'):
    # div下的h2标签下的是标题
    h2 = div.find('h2')
    # 排除h2为空的情况
    if h2:
        # 获取标题
        h2_title = h2.string

        # 保存数据，以每个章节大标题命名
        with open('./txt_data/'+h2_title+'.txt', 'w', encoding='utf-8') as f:
            # 获取每一个标题下的章节
            '''
                <div class='mulu>
                    <div class='box'>
                        <ul><li><a>
            '''
            for a in div.find(class_='box').find_all('a'):
                a_href = a.get('href')
                a_title = a.get('title')
                a_txt = a.string
                # print(a_href, '---', '---', a_txt)
                # 将数据写入
                f.write(a_txt+'\\t'+a_href+'\\n')

Excel

将上述示例获取到的章节内容和链接数据保存到 Excel 中。

# 爬虫数据存储到txt中
import requests
from bs4 import BeautifulSoup
import openpyxl

# 获取用户代理：在浏览器地址栏：about://version

# 1.获取请求
url = 'http://seputu.com'
headers = {'User_Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36'}

# 2.解析请求
req = requests.get(url, headers=headers)

# 3.获取数据
soup = BeautifulSoup(req.text, 'html.parser')

# 4.保存数据
# 创建工作簿
wb = openpyxl.Workbook()

# 删除多余的Sheet
# ws = wb.active
# wb.remove_sheet('Sheet')

for div in soup.find_all(class_='mulu'):
    h2 = div.find('h2')
    if h2:
        # 获取标题
        h2_title = h2.string

        # 根据每个章节获取工作表
        ws = wb.create_sheet(h2_title)

        # 创建表头
        ws.append(['章节名称', '章节链接', '标题'])

        # 获取章节名称 标题 链接
        for a in div.find(class_='box').find_all('a'):
            a_href = a['href']
            a_title = a.get('title')
            a_txt = a.string
            ws.append([a_txt, a_href, a_title])

# 保存excel
wb.save('fiction.xlsx')

CSV

将列表数据写入CSV中

# 将一维数组写入CSV文件中

fo = open("price.csv", "w")  # 没有该文件则会创建

ls = [['郑州', '111', '123', '121.3'], ['北京', '452', '44a', 'sda']]

for data in ls:
    fo.write(','.join(data) + "\\n")  # 以','分割，末尾加\\n

fo.close()

CSV文件互写

# 将一个CSV文件写入另一个里面

# 读取 CSV
ls = []
with open("price2016.csv", 'r') as fr:
    for line in fr:  # 将CSV文件中二维数据读入到列表变量中
        line = line.replace("\\n", "")
        ls.append(line.split(","))
        
# 数据预处理
for i in range(len(ls)):  # 遍历列表变量计算百分数
    for j in range(len(ls[i])):
        if (ls[i][j]).replace(".", "").isnumeric():  # 用于判断一个字符串是否类似101.5小数构成，由于Python没有单个函数判断其余字符是否都是数字
            ls[i][j] = "{:.2}%".format(float(ls[i][j])/100)  # 将数据单位换为 %

with open("price.csv", "w") as fw:
    for row in ls:  # 将列表变量中的两位数数据输出到CSV文件中
        print(row)
        fw.write(",".join(row)+"\\n")

感谢！

加油!