天天看點
傳回

Python爬蟲之Scrapy架構的使用（二）一：建立爬蟲二：settings配置檔案三：建立啟動檔案四：爬蟲檔案内容五：item檔案六：pipelines檔案七：使用scrapy中的ItemLoader提取資料

Python爬蟲之Scrapy架構的使用（二）

  • 一：建立爬蟲
  • 二：settings配置檔案
  • 三：建立啟動檔案
  • 四：爬蟲檔案内容
  • 五：item檔案
  • 六：pipelines檔案
  • 七：使用scrapy中的ItemLoader提取資料
    • 7.1 爬蟲檔案内容
    • 7.2 input_processor和output_processor
    • 7.3 重用和擴充ItemLoaders

一：建立爬蟲 

scrapy startproject ArticleSpider
cd ArticleSpider
scrapy genspider cnblogs cnblogs.com

二：settings配置檔案 

ROBOTSTXT_OBEY = False
DOWNLOAD_DELAY = 1
DEFAULT_REQUEST_HEADERS = {
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
    'Accept-Language': 'en',
    'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_7_0) AppleWebKit/535.11 (KHTML, like Gecko) Chrome/17.0.963.56 Safari/535.11'
}
LOG_LEVEL = "WARNING"		# 增加此行配置

三：建立啟動檔案 

from scrapy.cmdline import execute
import sys, os

sys.path.append(os.path.dirname(os.path.abspath(__file__)))
execute(['scrapy', 'crawl', 'cnblogs'])

四：爬蟲檔案内容 

import scrapy
from urllib import parse
import json
from ArticleSpider.items import ArticlespiderItem
from ArticleSpider.utils.common import get_url_md5

class CnblogsSpider(scrapy.Spider):
    name = 'cnblogs_bak'
    allowed_domains = ['cnblogs.com']
    start_urls = ['https://news.cnblogs.com/']

    def parse(self, response):
        # news_list = response.xpath('//div[@id="news_list"]//div[@class="news_block"]')
        news_list = response.css('div#news_list div.news_block')
        for div in news_list:
            detail_url = div.xpath('.//h2/a/@href').get()
            thumbnail_url_data = div.xpath('.//img[@class="topic_img"]/@src').get()
            if not thumbnail_url_data:
                thumbnail_url = []
            elif thumbnail_url_data.startswith('/'):
                thumbnail_url = ['https:' + thumbnail_url_data]
            else:
                thumbnail_url = [thumbnail_url_data]
            yield scrapy.Request(url=parse.urljoin(response.url, detail_url), callback=self.parse_detail,meta={'thumbnail_url': thumbnail_url})
        # 提取下一頁
        next_url = response.xpath('//div[@class="pager"]/a[contains(text(), "Next")]/@href').get()
        yield scrapy.Request(url=parse.urljoin(response.url, next_url), callback=self.parse)

    def parse_detail(self, response):
        thumbnail_url = response.meta.get('thumbnail_url', '')
        title = response.xpath('//div[@id="news_title"]/a/text()').get()
        author = response.xpath('//div[@id="news_info"]/span[@class="news_poster"]/a/text()').get()
        pubdate = response.xpath('//div[@id="news_info"]/span[@class="time"]/text()').get()
        pubdate = pubdate.replace('釋出于', '').lstrip()
        content = ''.join(response.xpath('//div[@id="news_body"]//*').getall())
        tag_list = response.xpath('//div[@class="news_tags"]//a/text()').getall()
        tags = ','.join(tag_list)
        origin_url = response.url
        url_hash_id = get_url_md5(origin_url)
        content_id = response.xpath('.//input[@id="lbContentID"]/@value').get()
        item = ArticlespiderItem(title=title, author=author, pubdate=pubdate, content=content, tags=tags,thumbnail_url=thumbnail_url, origin_url=origin_url, url_hash_id=url_hash_id)
        # 不建議将再有的url請求放在該方法中，建議将url請求 yield 出去
        yield scrapy.Request(parse.urljoin(response.url, '/NewsAjax/GetAjaxNewsInfo?contentId={}'.format(content_id)),callback=self.parse_json_data, meta={'item': item})

    def parse_json_data(self, response):
        item = response.meta.get('item')
        json_data = json.loads(response.text)
        item['dig_count'] = json_data.get('DiggCount')
        item['comment_count'] = json_data.get('CommentCount')
        item['view_count'] = json_data.get('TotalView')
        yield item

五：item檔案 

class ArticlespiderItem(scrapy.Item):
    title = scrapy.Field()
    author = scrapy.Field()
    pubdate = scrapy.Field()
    content = scrapy.Field()
    tags = scrapy.Field()
    dig_count = scrapy.Field()
    comment_count = scrapy.Field()
    view_count = scrapy.Field()
    thumbnail_url = scrapy.Field()
    thumbnail_path = scrapy.Field()
    origin_url = scrapy.Field()
    url_hash_id = scrapy.Field()

六：pipelines檔案 

import codecs
import json
import pymysql
from scrapy.exporters import JsonItemExporter
from scrapy.pipelines.images import ImagesPipeline
from twisted.enterprise import adbapi

class ArticlespiderPipeline:
    def process_item(self, item, spider):
        return item

class ArticleImagesPipeline(ImagesPipeline):
	# 圖檔下載下傳的pipeline
    def item_completed(self, results, item, info):
        if 'thumbnail_url' in item:
            thumbnail_path = ''
            for ok, value in results:
                thumbnail_path = value['path'] or ''
            item['thumbnail_path'] = thumbnail_path
            # thumbnail_path儲存格式為：full/xxxx.jpg
        return item

class JsonWithEncodingPipeline(object):
    '''自定義導出為json檔案pipeline，一行一條資料'''
    def __init__(self):
        self.file = codecs.open('article.json', 'w', encoding='utf-8')
    def process_item(self, item, spider):
        lines = json.dumps(dict(item), ensure_ascii=False) + '\n'
        self.file.write(lines)
        return item
    def spider_close(self, spider):
        self.file.close()

class JsonExporterPipeline:
    '''
    自定義導出為json檔案pipeline，儲存後的資料為一個清單
    該類中預設會執行三個方法：open_spider, close_spider，process_item。
    '''
    def __init__(self):
        self.file = open('article_export.json', 'wb')
        self.exporter = JsonItemExporter(self.file, ensure_ascii=False, encoding='utf-8')
        self.exporter.start_exporting()
    def open_spider(self, spider):
        print('start...')
    def process_item(self, item, spider):
        self.exporter.export_item(item)
        return item
    def close_spider(self, spider):
        self.exporter.finish_exporting()
        self.file.close()
        print('finish...')

class MysqlPipeline:
    '''儲存資料到mysql資料庫pipeline'''
    def __init__(self):
        self.conn = pymysql.connect(host='172.17.2.36', port=3306, user='root', password='xxxxxx',database='spider', charset='utf8')
        self.cursor = self.conn.cursor()
    def open_spider(self, spider):
        print('start...')
    def process_item(self, item, spider):
        insert_sql = 'insert into cnblogs values(%s, %s, %s,%s,%s,%s,%s,%s,%s,%s,%s,%s)'
        self.cursor.execute(insert_sql, (
            item.get('title', ''),
            item.get('author', ''),
            item.get('pubdate', '1970-07-01'),
            item.get('content', ''),
            item.get('origin_url', ''),
            item.get('url_hash_id', ''),
            item.get('tags', ''),
            item.get('dig_count', 0),
            item.get('comment_count', 0),
            item.get('view_count', 0),
            ''.join(item.get('thumbnail_url', [])),
            item.get('thumbnail_path', '')
        ))
        self.conn.commit()
        return item
    def close_spider(self, spider):
        self.cursor.close()
        self.conn.close()
        print('finish...')

class MysqlTwistedPipeline:
    # 自定義異步儲存資料到mysql資料庫類
    def __init__(self, dbpool):
        self.dbpool = dbpool
    @classmethod
    def from_settings(cls, settings):
        # 從settings配置檔案讀取參數
        db_params = dict(
            host=settings['MYSQL_HOST'],
            port=settings['MYSQL_PORT'],
            database=settings['MYSQL_DATABASE'],
            user=settings['MYSQL_USER'],
            password=settings['MYSQL_PASSWORD'],
            charset='utf8',
            cursorclass=pymysql.cursors.DictCursor
        )
        db_pool = adbapi.ConnectionPool('pymysql', **db_params)
        return cls(db_pool)
    def process_item(self, item, spider):
        query = self.dbpool.runInteraction(self.insert_item, item)
        query.addErrback(self.handle_error, item, spider)
        return item
    def insert_item(self, cursor, item):
        insert_sql = 'insert into cnblogs values(%s, %s, %s,%s,%s,%s,%s,%s,%s,%s,%s,%s) on duplicate key update view_count=values(view_count)'
        cursor.execute(insert_sql, (
            item.get('title', ''),
            item.get('author', ''),
            item.get('pubdate', '1970-07-01'),
            item.get('content', ''),
            item.get('origin_url', ''),
            item.get('url_hash_id', ''),
            item.get('tags', ''),
            item.get('dig_count', 0),
            item.get('comment_count', 0),
            item.get('view_count', 0),
            ''.join(item.get('thumbnail_url', [])),
            item.get('thumbnail_path', '')
        ))
    def handle_error(self, error, item, spider):
        print('=' * 10 + 'error' + '=' * 10)
        print(error)

settings檔案配置：

ITEM_PIPELINES = {
    'ArticleSpider.pipelines.ArticlespiderPipeline': 300,
    # 'scrapy.pipelines.images.ImagesPipeline': 1,
    'ArticleSpider.pipelines.ArticleImagesPipeline': 1,
    # 'ArticleSpider.pipelines.JsonWithEncodingPipeline': 2,
    # 'ArticleSpider.pipelines.JsonExporterPipeline': 3,
    # 'ArticleSpider.pipelines.MysqlPipeline': 4,
    'ArticleSpider.pipelines.MysqlTwistedPipeline': 5,
}
# 圖檔下載下傳pipeline相關配置
IMAGES_URLS_FIELD = 'thumbnail_url'
IMAGES_STORE = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'images')
# mysql資料庫資訊配置
MYSQL_HOST = '172.17.2.36'
MYSQL_PORT = 3306
MYSQL_DATABASE = 'spider'
MYSQL_USER = 'root'
MYSQL_PASSWORD = 'xxxxxx'

到此，運作爬蟲啟動檔案即可爬取資料。

七：使用scrapy中的ItemLoader提取資料

當項目很大，提取的字段數以百計，那麼各種提取規則會越來越多，按照這種方式來做，維護的工作将會是一場噩夢！是以scrapy就提供了ItemLoader這樣一個容器，在這個容器裡面可以配置item中各個字段的提取規則。可以通過函數分析原始資料，并對Item字段進行指派，非常的便捷。

可以這麼來看 Item 和 Itemloader：Item提供儲存抓取到資料的容器，而 Itemloader提供的是填充容器的機制。

Itemloader提供的是一種靈活，高效的機制，可以更友善的被spider或source format （HTML, XML, etc）擴充并重寫，更易于維護，尤其是分析規則特别複雜繁多的時候。

7.1 爬蟲檔案内容 

import scrapy
from urllib import parse
import json
from scrapy.loader import ItemLoader
from ArticleSpider.items import ArticlespiderItem
from ArticleSpider.utils.common import get_url_md5

class CnblogsSpider(scrapy.Spider):
    name = 'cnblogs'
    allowed_domains = ['cnblogs.com']
    start_urls = ['https://news.cnblogs.com/']
    
    def parse(self, response):
        # news_list = response.xpath('//div[@id="news_list"]//div[@class="news_block"]')
        news_list = response.css('div#news_list div.news_block')
        for div in news_list:
            detail_url = div.xpath('.//h2/a/@href').get()
            # 因為ImagesPipeline要求圖檔url是一個清單，是以上面的案例對該字段進行了處理，但使用itemload，這裡傳遞一個空字元串，itemload不會報錯會轉換成一個空清單
            thumbnail_url = div.xpath('.//img[@class="topic_img"]/@src').get()
            yield scrapy.Request(url=parse.urljoin(response.url, detail_url), callback=self.parse_detail,meta={'thumbnail_url': thumbnail_url})
        # 提取下一頁
        # next_url = response.xpath('//div[@class="pager"]/a[contains(text(), "Next")]/@href').get()
        # yield scrapy.Request(url=parse.urljoin(response.url, next_url), callback=self.parse)

    def parse_detail(self, response):
        # 使用item_loader提取資料。下面使用css選擇器提取資料，當然也可以使用
        item_loader = ItemLoader(item = ArticlespiderItem(), response = response)
        # 使用item_loader提取的資料都是list類型
        item_loader.add_css('title', '#news_title a::text')
        item_loader.add_css('author', '.news_poster a::text')
        item_loader.add_css('content', '#news_content')
        item_loader.add_css('tags', '.news_tags a::text')
        item_loader.add_css('pubdate', '#news_info .time::text')
        item_loader.add_value('origin_url', response.url)
        item_loader.add_value('thumbnail_url', response.meta.get('thumbnail_url', ''))
        item_loader.add_value('url_hash_id', get_url_md5(response.url))
        # 當所有資料被收集起來之後，還需要調用 ItemLoader.load_item() 方法， 實際上填充并且傳回了之前通過調用 add_xpath()，add_css()，and add_value() 所提取和收集到的資料。
        item = item_loader.load_item()
        content_id = response.xpath('.//input[@id="lbContentID"]/@value').get()
        # 不建議将url請求放在該方法中，建議将url請求 yield 出去
        yield scrapy.Request(parse.urljoin(response.url, '/NewsAjax/GetAjaxNewsInfo?contentId={}'.format(content_id)),callback=self.parse_json_data, meta={'item': item})

    def parse_json_data(self, response):
        item = response.meta.get('item')
        json_data = json.loads(response.text)
        item['dig_count'] = json_data.get('DiggCount')
        item['comment_count'] = json_data.get('CommentCount')
        item['view_count'] = json_data.get('TotalView')
        yield item

7.2 input_processor和output_processor

使用輸入處理器input_processor和輸出處理器output_processor對提取到的資料進行加工處理。

從上面的示例中，可以看到，存在兩個問題：

第一，提取的資料，填充進去的對象都是List類型。而我們大部分的需求是要取第一個數值，取List中的第一個非空元素，那麼如何實作取第一個呢？

第二，在做item字段解析時，經常需要再進一步解析，過濾出我們想要的數值，例如用正規表達式将 $10 price中的數字10提取出來。那麼又如何對字段加一些處理函數呢？

input_processor and output_processor在item檔案定義字段時對資料進行處理，下面舉例說明用法：

from scrapy.loader.processors import MapCompose, TakeFirst
def date_convert(value):
    # 這裡的value傳遞過來的就是清單中的元素
    if value:
        value = re.sub('[\u4e00-\u9fa5]', '', value).strip()  # 去除字元串中的中文字元
        try:
            create_date = datetime.datetime.strptime(value, "%Y/%m/%d").date()
        except Exception as e:
            create_date = datetime.datetime.now().date()
        return create_date
class ArticlespiderItem(scrapy.Item):
    # define the fields for your item here like:
    # name = scrapy.Field()
    title = scrapy.Field()
    author = scrapy.Field()
    pubdate = scrapy.Field(input_processor=MapCompose(date_convert), output_processor=TakeFirst())
    # 可以定義多個函數，按順序寫到MapCompose中
    ......................

TakeFirst：傳回第一個非空（non-null/ non-empty）值，常用于單值字段的輸出處理器，無參數。

MapCompose：與Compose處理器類似，差別在于各個函數結果在内部傳遞的方式（會涉及到list對象解包的步驟）：

  1. 輸入值是被疊代的處理的，List對象中的每一個元素被單獨傳入，第一個函數進行處理，然後處理的結果被連接配接起來形成一個新的疊代器，并被傳入第二個函數，以此類推，直到最後一個函數。最後一個函數的輸出被連接配接起來形成處理器的輸出。
  2. 每個函數能傳回一個值或者一個值清單，也能傳回None（會被下一個函數所忽略）
  3. 這個處理器提供了很友善的方式來組合多個處理單值的函數。是以它常用于輸入處理器，因為傳遞過來的是一個List對象。

Compose：用給定的多個函數的組合，來構造的處理器。list對象（注意不是指list中的元素），依次被傳遞到第一個函數，然後輸出，再傳遞到第二個函數，一個接着一個，直到最後一個函數傳回整個處理器的輸出。預設情況下，當遇到None值（list中有None值）的時候停止處理。可以通過傳遞參數stop_on_none = False改變這種行為。

Identity：最簡單的處理器，不進行任何處理，直接傳回原來的資料。無參數。

Join：傳回用分隔符連接配接後的值。分隔符預設為空格。不接受Loader contexts。當使用預設分隔符的時候，這個處理器等同于：u’ '.join

7.3 重用和擴充ItemLoaders

從上面的資訊來看，ItemLoaders是非常靈活的，但是假設有個需求，所有的字段，我們都要去取第一個，那麼如果有300個字段，我們就要添加300次，每個都要寫，就會覺得很麻煩。那麼有沒有辦法統一設定呢，答案是有的，如下：

如果想要實作每個字段都隻取第一個，那麼可以定義一個自己的ItemLoader類：ArticleItemLoader（繼承自ItemLoader類）

我們首先可以看一下原始的 ItemLoader 的定義：

class ItemLoader:
	# 可以看到是有預設的輸入/輸出處理器的，而且預設是什麼都不做
    default_item_class = Item
    default_input_processor = Identity()
    default_output_processor = Identity()
    default_selector_class = Selector

在item檔案中自定義一個ItemLoader類：

def date_convert(value):
    # 這裡的value傳遞過來的就是清單中的元素
    if value:
        value = re.sub('[\u4e00-\u9fa5]', '', value).strip()  # 去除字元串中的中文字元
        try:
            pubdate = datetime.datetime.strptime(value, "%Y/%m/%d").date()
        except Exception as e:
            pubdate = datetime.datetime.now().date()
        return pubdate
class ArticleItemLoad(ItemLoader):
    default_output_processor = TakeFirst()
class ArticlespiderItem(scrapy.Item):
    # define the fields for your item here like:
    # name = scrapy.Field()
    title = scrapy.Field()
    author = scrapy.Field()
    pubdate = scrapy.Field(input_processor=MapCompose(date_convert))
    content = scrapy.Field()
    tags = scrapy.Field(output_processor=Join(separator=','))
    dig_count = scrapy.Field()
    comment_count = scrapy.Field()
    view_count = scrapy.Field()
    thumbnail_url = scrapy.Field(output_processor=Identity())
    thumbnail_path = scrapy.Field()
    origin_url = scrapy.Field()
    url_hash_id = scrapy.Field()

下面把爬蟲檔案修改全部使用自定義的ItemLoader類：

class CnblogsSpider(scrapy.Spider):
    name = 'cnblogs'
    allowed_domains = ['cnblogs.com']
    start_urls = ['https://news.cnblogs.com/']

    def parse(self, response):
        # news_list = response.xpath('//div[@id="news_list"]//div[@class="news_block"]')
        news_list = response.css('div#news_list div.news_block')
        for div in news_list:
            detail_url = div.xpath('.//h2/a/@href').get()
            thumbnail_url_data = div.xpath('.//img[@class="topic_img"]/@src').get()
            if not thumbnail_url_data:
                thumbnail_url = []
            elif thumbnail_url_data.startswith('/'):
                thumbnail_url = ['https:' + thumbnail_url_data]
            else:
                thumbnail_url = [thumbnail_url_data]
            yield scrapy.Request(url=parse.urljoin(response.url, detail_url), callback=self.parse_detail,meta={'thumbnail_url': thumbnail_url})
        # 提取下一頁
        # next_url = response.xpath('//div[@class="pager"]/a[contains(text(), "Next")]/@href').get()
        # yield scrapy.Request(url=parse.urljoin(response.url, next_url), callback=self.parse)

    def parse_detail(self, response):
        # 使用item_loader提取資料
        item_loader = ArticleItemLoad(item=ArticlespiderItem(), response=response)
        item_loader.add_css('title', '#news_title a::text')
        item_loader.add_css('author', '.news_poster a::text')
        item_loader.add_css('content', '#news_content')
        item_loader.add_css('tags', '.news_tags a::text')
        item_loader.add_css('pubdate', '#news_info .time::text')
        item_loader.add_value('origin_url', response.url)
        item_loader.add_value('thumbnail_url', response.meta.get('thumbnail_url', ''))
        item_loader.add_value('url_hash_id', get_url_md5(response.url))
        # 下面的parse_json_data方法可以不使用ItemLoader
        # 如果需要使用ItemLoader那麼延緩使用item_loader.load_item()方法
        # item = item_loader.load_item()
        content_id = response.xpath('.//input[@id="lbContentID"]/@value').get()
        # 不建議将url請求放在該方法中，建議将url請求 yield 出去
        yield scrapy.Request(parse.urljoin(response.url, '/NewsAjax/GetAjaxNewsInfo?contentId={}'.format(content_id)), callback=self.parse_json_data, meta={'item_loader': item_loader})

    def parse_json_data(self, response):
        item_loader = response.meta.get('item_loader')
        json_data = json.loads(response.text)
        item_loader.add_value('dig_count', json_data.get('DiggCount'))
        item_loader.add_value('comment_count', json_data.get('CommentCount'))
        item_loader.add_value('view_count', json_data.get('TotalView'))
        item = item_loader.load_item()
        yield item
Python爬蟲 scrapy架構 ItemLoad
上一篇: 11.Scrapy架構基礎-使用Scrapy抓取資料并儲存到mongodb一、Scrapy安裝二、使用scrapy爬取資料三、儲存資料到mongodb四、再多學一點五、作業（這是一個考驗）
下一篇: 圖像視訊基礎知識

繼續閱讀