Scrapy的Item_loader机制详解

Posted 2020-11-28 zhangyafei

一、ItemLoader与Item的区别

• ItemLoader是负责数据的收集、处理、填充,item仅仅是承载了数据本身
• 数据的收集、处理、填充归功于item loader中两个重要组件:
  
  • 输入处理input processors
  • 输出处理output processors

二、ItemLoader的使用

• 1、创建一个项目并创建一个爬虫

• 2、在item.py中使用

import redis
import scrapy
from scrapy.loader import ItemLoader
from scrapy.loader.processors import MapCompose, TakeFirst, Join
from w3lib.html import remove_tags

from utils.common import extract_num


def add_jobbole(value):
    return value + ‘zhangyafei‘


def date_convert(value):
    try:
        value = value.strip().replace(‘·‘, ‘‘).strip()
        create_date = datetime.datetime.strptime(value, "%Y/%m/%d").date()
    except Exception as e:
        create_date = datetime.datetime.now().date
    return create_date


def get_nums(value):
    try:
        if re.match(‘.*?(d+).*‘, value).group(1):
            nums = int(re.match(‘.*?(d+).*‘, value).group(1))
        else:
            nums = 0
    except:
        nums = 0
    return nums


def remove_comment_tags(value):
    if "评论" in value:
        return ‘‘
    return value


def return_value(value):
    return value


def gen_suggests(index, info_tuple):
    # 根据字符串生成搜索建议数组
    used_words = set()
    suggests = []
    for text, weight in info_tuple:
        if text:
            # 调用es的analyze接口分析字符串
            words = es.indices.analyze(index=index, analyzer="ik_max_word", params={‘filter‘: ["lowercase"]}, body=text)
            anylyzed_words = set([r["token"] for r in words["tokens"] if len(r["token"]) > 1])
            new_words = anylyzed_words - used_words
        else:
            new_words = set()

        if new_words:
            suggests.append({"input": list(new_words), "weight": weight})

    return suggests


class ArticleItemLoader(ItemLoader):
    # 自定义itemloader
    default_output_processor = TakeFirst()


class JobboleArticleItem(scrapy.Item):
    title = scrapy.Field()
    create_date = scrapy.Field(
        input_processor=MapCompose(date_convert),
    )
    url = scrapy.Field()
    url_object_id = scrapy.Field()
    front_image_url = scrapy.Field(
        output_processor=MapCompose(return_value)
    )
    front_image_path = scrapy.Field()
    praise_nums = scrapy.Field(
        input_processor=MapCompose(get_nums)
    )
    comment_nums = scrapy.Field(
        input_processor=MapCompose(get_nums)
    )
    fav_nums = scrapy.Field(
        input_processor=MapCompose(get_nums)
    )
    tags = scrapy.Field(
        input_processor=MapCompose(remove_comment_tags),
        output_processor=Join(",")
    )
    content = scrapy.Field()

    def get_insert_sql(self):
        insert_sql = """
            insert into jobbole_article(title, url, create_date, fav_nums)
            VALUES (%s, %s, %s, %s) ON DUPLICATE KEY UPDATE content=VALUES(fav_nums)
        """
        params = (self["title"], self["url"], self["create_date"], self["fav_nums"])

        return insert_sql, params

    def save_to_es(self):
        article = ArticleType()
        article.title = self[‘title‘]
        article.create_date = self["create_date"]
        article.content = remove_tags(self["content"])
        article.front_image_url = self["front_image_url"]
        if "front_image_path" in self:
            article.front_image_path = self["front_image_path"]
        article.praise_nums = self["praise_nums"]
        article.fav_nums = self["fav_nums"]
        article.comment_nums = self["comment_nums"]
        article.url = self["url"]
        article.tags = self["tags"]
        article.meta.id = self["url_object_id"]

        article.suggest = gen_suggests(ArticleType._doc_type.index, ((article.title, 10), (article.tags, 7)))

        article.save()

        redis_cli.incr("jobbole_count")

        return

　spider中的使用

 def parse(self, response):
        """
        1.获取文章列表页的文章url交给scrapy下载后并进行解析
        2.获取下一页的url交给scrapy进行下载，下载完成后交给parse解析
        """
        """
        解析文章列表页中的所有文章url交给scrapy下载并进行解析
        """
        if response.status == 404:
            self.fail_urls.append(response.url)
            self.crawler.stats.inc_value("failed_urls")

        post_nodes = response.css(‘#archive .post-thumb a‘)
        for post_node in post_nodes:
            img_url = post_node.css(‘img::attr(src)‘).extract_first()
            # img_url = [img_url if ‘http:‘ in img_url else (‘http:‘ + img_url)]
            post_url = post_node.css(‘::attr(href)‘).extract_first()
            yield scrapy.Request(url=parse.urljoin(response.url, post_url), meta={‘img_url‘: img_url},
                                 callback=self.parse_detail)
        next_url = response.css(‘.next.page-numbers::attr(href)‘).extract_first()
        # 获取下一页的url交给scrapy下载并进行解析
        if next_url:
            yield scrapy.Request(url=next_url, callback=self.parse)

    def parse_detail(self, response):
        # 通过item loader加载item
        front_image_url = response.meta.get("front_image_url", "")  # 文章封面图
        item_loader = ArticleItemLoader(item=JobboleArticleItem(), response=response)
        item_loader.add_css("title", ".entry-header h1::text")
        item_loader.add_value("url", response.url)
        item_loader.add_value("url_object_id", get_md5(response.url))
        item_loader.add_css("create_date", "p.entry-meta-hide-on-mobile::text")
        item_loader.add_value("front_image_url", [front_image_url])
        item_loader.add_css("praise_nums", ".vote-post-up h10::text")
        item_loader.add_css("comment_nums", "a[href=‘#article-comment‘] span::text")
        item_loader.add_css("fav_nums", ".bookmark-btn::text")
        item_loader.add_css("tags", "p.entry-meta-hide-on-mobile a::text")
        item_loader.add_css("content", "div.entry")

        article_item = item_loader.load_item()

        yield article_item

三、常见的内置处理器

• 1、Identity

  不对数据进行处理,直接返回原来的数据

• 2、TakeFirst

  返回第一个非空值,常用于单值字段的输出处理

• 3、Join

  相当于把列表中的元素拼接起来

• 4、MapCompose把几个方法组合起来

四、数据清洗方法详解

processor

scrapy提供了一个processors类，里面有下列几种方法:Join,TakeFirst,MapCompose,Compose,Identity,SelectJmes

对这几种方法的用法简单介绍一下：

from scrapy.loader.processors import Join,TakeFirst,MapCompose,Compose,Identity,SelectJmes

#以特定字符连接，示例以空连接，对字符串也能操作
c = Join(‘‘)
c([‘a‘,‘b‘])
>>>‘ab‘
#********************

#传入函数的列表的每一个元素都会经过第一个函数，
#得到值在经过第二个函数，如果有返回值为None的，则抛弃，
#最后返回一个列表
c=MapCompose(str.strip,str.upper)
c([‘  a   ‘,‘b‘])
>>>[‘A‘, ‘B‘]
#********************

#如果传入一个列表时则会报下面这个错误
#descriptor ‘strip‘ requires a ‘str‘ object but received a ‘list‘
#但如果Compose的第一个函数是取列表的第一个元素，不会报错
#即Compose是处理单一数据，MapCompose是批量处理
c=Compose(str.strip,str.upper)
c(‘  ac   ‘)
>>>‘AC‘
#********************

#拿到JSON格式数据时会有作用
proc = SelectJmes(‘a‘) 
proc({‘a‘:‘b‘,‘c‘:‘d‘})
>>>‘b‘

 input--output

Item Loader 为每个 Item Field 单独提供了一个 Input processor 和一个 Output processor；

Input processor 一旦它通过 add_xpath()，add_css()，add_value() 方法收到提取到的数据便会执行，执行以后所得到的数据将仍然保存在 ItemLoader 实例中；当数据收集完成以后，ItemLoader 通过 load_item() 方法来进行填充并返回已填充的 Item 实例。

即input_processor是在收集数据的过程中所做的处理，output_processor是数据yield之后进行的处理，通过下面这个例子会更加理解：

#type字段取出来时是‘type‘: [‘2室2厅‘, ‘中楼层/共6层‘]

#定义一个在第一个元素后面加a的函数
def adda(value):
    return value[0]+‘a‘

type = scrapy.Field(output_processor = Compose(adda))
>>>‘type‘: ‘2室2厅a‘

type = scrapy.Field(input_processor = Compose(adda))
>>>‘type‘: [‘2室2厅a‘, ‘中楼层/共6层a‘]
#如果使用MapCompose的话，两个结果会一样，这也是Compose和MapCompose的区别

当指定了取列表的第一个元素后，有些信息想保留整个列表便可以使用name_out，Identity()是取自身的函数。

class TeItem(ItemLoader):
    default_out_processor = TakeFirst()
    name_out = Identity()

也可以在基于scrapy.Item的item中定义一些规则：

class Scrapy1Item(scrapy.Item):
    name = scrapy.Field(output_processor=Identity())

优先级

scrapy提供了很多种方式去自定义输入输出的内容，具有一定的优先级，优先级最高的是name_out这种，其次是在scrapy.Field()中定义的output_processor和input_processor，最后是default_out_processor = TakeFirst()这种。