專案介紹

使用Scrapy框架進行爬取伯樂線上的所有技術文章

所用知識點

• Scrapy專案的建立
• Scrapy框架Shell命令的使用
• Scrapy自帶的圖片下載管道
• Scrapy自定義圖片下載管道(繼承自帶的管道)
• Scrapy框架ItemLoader的使用
• Scrapy自定義ItemLoader
• Scrapy中同步將Item儲存入Mysq資料庫
• Scrapy中非同步將Item儲存入Mysq資料庫

喜歡的話關注收藏評論轉發比心麼麼噠！Python學習交流群548377875內有大量的專案開發和新手教學視訊五千人大群等著你來加入

專案初始

建立新專案

scrapy startproject bole
 

建立爬蟲

scrapy genspider jobbole blog.jobbole.com

爬蟲除錯

為了方便對爬蟲進行除錯，在專案目錄中建立一個main.py檔案

from scrapy.cmdline import execute
import sys,os
	
# 將專案目錄動態設定到環境變數中
# os.path.abspath(__file__) 獲取main.py的路徑
# os.path.dirname(os.path.abspath(__file__) 獲取main.py所處目錄的上一級目錄
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
execute(['scrapy','crawl','jobbole'])
 

在爬蟲開始執行時，建議修改專案中的配置檔案,找到ROBOTSTXT_OBEY將其改為False,如果不修改的話，Scrapy會自動的查詢網站的ROBOTS協議，會過濾不符合協議的URL 

在windows環境下可能會出現No moudle named 'win32api',因此需要執行pip install pypiwin32 

如果下載速度過慢可使用豆瓣源進行安裝

前置知識

XPath語法簡介

CSS常用選擇器

Scrapy shell模式

在解析頁面的時候如果要檢視執行結果則必須要執行Scrapy爬蟲發起一個請求，而Scrapy提供了一種方便的除錯方法可以只請求一次。

scrpay shell http://blog.jobbole.com/111144/

文章解析

文章詳情頁

Xpath的解析方式

def parse_detail(self, response):
    # xpath方式進行解析
    # 文章標題
    title = response.xpath('//div[@class="entry-header"]/h1/text()').extract_first()
    # 釋出時間
    create_time = response.xpath('//p[@class="entry-meta-hide-on-mobile"]/text()').extract_first().replace('·','').strip()
    # 點贊數
    # contains函式是找到class中存在vote-post-up這個類
    up_num = response.xpath('//span[contains(@class,"vote-post-up")]/h10/text()').extract_first()
    # 收藏數
    fav_num = response.xpath('//span[contains(@class,"bookmark-btn")]/text()').extract_first()
    match_re = re.match('.*?(\d+).*',fav_num)
    if match_re:
        fav_num = match_re.group(1)
    else:
        fav_num = 0
    # 評論數
    comment_num = response.xpath('//a[@href="#article-comment"]/span/text()').extract_first()
    match_re = re.match('.*?(\d+).*', comment_num)
    if match_re:
        comment_num = match_re.group(1)
    else:
        comment_num = 0
    # 文章正文
    content = response.xpath('//div[@class="entry"]').extract_first()
    # 獲取標籤
    tags_list = response.xpath('//p[@class="entry-meta-hide-on-mobile"]/a/text()').extract()
    tags_list = [element for element in tags_list if not element.strip().endswith('評論')]
    tags = ",".join(tags_list)

CSS解析方式

def parse_detail(self, response):
    # CSS方式進行解析
    # 文章標題
    title = response.css('div.entry-header h1::text').extract_first()
    # 釋出時間
    create_time = response.css('p.entry-meta-hide-on-mobile::text').extract_first().replace('·','').strip()
    # 點贊數
    up_num = response.css('span.vote-post-up h10::text').extract_first()
    # 收藏數
    fav_num = response.css('span.bookmark-btn::text').extract_first()
    match_re = re.match('.*?(\d+).*',fav_num)
    if match_re:
        fav_num = match_re.group(1)
    else:
        fav_num = 0
    # 評論數
    comment_num = response.css('a[href="#article-comment"] span::text').extract_first()
    match_re = re.match('.*?(\d+).*', comment_num)
    if match_re:
        comment_num = match_re.group(1)
    else:
        comment_num = 0
    # 文章正文
    content = response.css('div.entry').extract_first()
    # 獲取標籤
    tags_list = response.css('p.entry-meta-hide-on-mobile a::text').extract()
    tags_list = [element for element in tags_list if not element.strip().endswith('評論')]
    tags = ",".join(tags_list)

列表頁

def parse(self, response):

    # 獲取文章列表中的每一篇文章的url交給Scrapy下載並解析
    article_nodes = response.css('div#archive .floated-thumb .post-thumb a')
    for article_node in article_nodes:
        # 解析每個文章的封面圖
        font_image_url = article_node.css('img::attr(src)').extract_first("")
        # 解析每個文章的url
        article_url = article_node.css('::attr(href)').extract_first("")
        # 智慧對url進行拼接，如果url中不帶有域名則會自動新增域名
        # 通過在Request中設定meta資訊來進行資料的傳遞
        yield Request(url=parse.urljoin(response.url, article_url),meta={'font_image_url':parse.urljoin(response.url, font_image_url)}, callback=self.parse_detail)

    # 獲取文章的下一頁url地址，並交給自身解析
    next_url = response.css('a.next.page-numbers::attr(href)').extract_first('')
    if next_url:
        yield Request(url=parse.urljoin(response.url, next_url), callback=self.parse)
        
 def parse_detail(self, response):
        article_item = JobBoleArticleItem()
        # 從response中獲取資料
        # 文章封面圖
        font_image_url = response.meta.get('font_image_url', '')
        # CSS方式進行解析
        # 文章標題
        title = response.css('div.entry-header h1::text').extract_first()
        # 釋出時間
        create_time = response.css('p.entry-meta-hide-on-mobile::text').extract_first().replace('·','').strip()
        # 點贊數
        up_num = response.css('span.vote-post-up h10::text').extract_first()
        # 收藏數
        fav_num = response.css('span.bookmark-btn::text').extract_first()
        match_re = re.match('.*?(\d+).*',fav_num)
        if match_re:
            fav_num = match_re.group(1)
        else:
            fav_num = 0
        # 評論數
        comment_num = response.css('a[href="#article-comment"] span::text').extract_first()
        match_re = re.match('.*?(\d+).*', comment_num)
        if match_re:
            comment_num = match_re.group(1)
        else:
            comment_num = 0
        # 文章正文
        content = response.css('div.entry').extract_first()
        # 獲取標籤
        tags_list = response.css('p.entry-meta-hide-on-mobile a::text').extract()
        tags_list = [element for element in tags_list if not element.strip().endswith('評論')]
        tags = ",".join(tags_list)

        article_item["title"] = title
        article_item["create_time"] = create_time
        article_item["url"] = response.url
        article_item["font_image_url"] = [font_image_url]
        article_item["up_num"] = up_num
        article_item["fav_num"] = fav_num
        article_item["comment_num"] = comment_num
        article_item["content"] = content
        article_item["tags"] = tags

        yield article_item

定義Items

class JobBoleArticleItem(scrapy.Item):
    title = scrapy.Field()
    create_time = scrapy.Field()
    url = scrapy.Field()
    url_object_id = scrapy.Field()
    font_image_url = scrapy.Field()
    font_image_path = scrapy.Field()
    up_num = scrapy.Field()
    fav_num = scrapy.Field()
    comment_num = scrapy.Field()
    tags = scrapy.Field()
    content = scrapy.Field()

pipeline管道的使用

Scrapy自帶的圖片下載管道

在settings.py中的pipeline處新增 scrapy.pipeline.images.ImagesPipeline

ITEM_PIPELINES = {
   'bole.pipelines.BolePipeline': 300,
    'scrapy.pipeline.images.ImagesPipeline' : 200
}
# 設定圖片url的欄位，scraoy將從item中找出此欄位進行圖片下載
IMAGES_URLS_FIELD = "font_image_url"
# 設定圖片下載儲存的目錄
project_path = os.path.abspath(os.path.dirname(__file__))
IMAGES_STORE = os.path.join(project_path, "images")
# 表示只下載大於100x100的圖片
IMAGES_MIN_HEIGHT = 100
IMAGES_MIN_WIDTH = 100

之後執行專案可能包PIL未找到，因此需要pip install pillow 

此外scrapy的圖片下載預設是接受一個數組，因此在賦值的時候需要article_item["font_image_url"] = [font_image_url]

自定義圖片下載管道

雖然Scrapy自帶的下載中介軟體很好用，但是如果我要獲取圖片下載後儲存的路徑則官方自帶就不能滿足需求，因此需要我們自定義管道

# 自定義圖片下載處理的中介軟體
class ArticleImagePipeline(ImagesPipeline):
    # 過載函式，改寫item處理完成的函式
    def item_completed(self, results, item, info):
        for key, value in results:
            font_image_path = value["path"]
        item["font_image_path"] = font_image_path
        return item

使用Scrapy自帶的管道將Item匯出成Json檔案

from scrapy.exporters import JsonItemExporter



# 使用Scrapy自帶的JsonExporter將item匯出為json

class JsonExportPipeline(object):

    # 呼叫scrapy提供的JsonExporter匯出json檔案

    def __init__(self):

        self.file = open('article_export.json', 'wb')

        self.exporter = JsonItemExporter(self.file, encoding="utf-8", ensure_ascii=False)

        self.exporter.start_exporting()



    # 重寫Item處理

    def process_item(self, item, spider):

        self.exporter.export_item(item)

        return item



    def spider_closed(self, spider):

        self.exporter.finish_exporting()

        self.file.close()

自定義管道將Item儲存為Json檔案

import codecs,json

# 自定義將Item匯出為Json的管道
class ArticleWithJsonPipeline(object):
    # 爬蟲初始化時呼叫
    def __init__(self):
        # 開啟json檔案
        # 使用codecs能夠解決編碼方面的問題
        self.file = codecs.open('article.json','w',encoding="utf-8")

    # 重寫Item處理
    def process_item(self, item, spider):
        # 需要關閉ensure_ascii，不然中文字元會顯示不正確
        lines = json.dump(dict(item), ensure_ascii=False)+'\n'
        # 將一行資料寫入
        self.file.write(lines)
        return item

    # 爬蟲結束時呼叫
    def spider_closed(self, spider):
        # 關閉檔案控制代碼
        self.file.close()

同步化將Item儲存入資料庫

pip install mysqlclient 安裝Mysql客戶端庫

import MySQLdb

# 同步機制寫入資料庫
class ArticleWithMysqlPipeline(object):
    def __init__(self):
        self.conn = MySQLdb.connect('127.0.0.1', 'root', 'root', 'scrapy', charset="utf8", use_unicode=True)
        self.cursor = self.conn.cursor()

    def process_item(self, item, spider):
        insert_sql = '''
            INSERT INTO 
            jobbole_article (title, create_time, url, url_object_id, font_image_url, comment_num, up_num, fav_num, tags, content)
            VALUES 
            (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
        '''
        self.cursor.execute(insert_sql, (item["title"], item["create_time"], item["url"], item["url_object_id"], item["font_image_url"][0],
                                    item["comment_num"], item["up_num"], item["fav_num"], item["tags"], item["content"]))
        self.conn.commit()

    def spider_closed(self, spider):
        self.conn.close()

非同步化將Item儲存入資料庫

因為Scrapy的解析速度非常快，加上文章的內容較大，因此會出現資料庫的操作速度趕不上解析速度會產生阻塞，因此採用非同步化的方式來進行資料的插入

import MySQLdb.cursors
from twisted.enterprise import adbapi

# 非同步操作寫入資料庫
class ArticleTwiterMysqlPipeline(object):
    # scrapy會自動執行此方法，將setting檔案中的配置讀入
    @classmethod
    def from_settings(cls, settings):
        param = dict(
            host = settings["MYSQL_HOST"],
            db = settings["MYSQL_DBNAME"],
            user = settings["MYSQL_USERNAME"],
            passwd = settings["MYSQL_PASSWORD"],
            charset = "utf8",
            cursorclass = MySQLdb.cursors.DictCursor,
            use_unicode = True
        )
        #需要使用連線模組的模組名
        dbpool = adbapi.ConnectionPool("MySQLdb", **param)
        return cls(dbpool)

    def __init__(self, dbpool):
        self.dbpool = dbpool

    # 使用twisted非同步將資料插入到資料庫中
    def process_item(self, item, spider):
        query = self.dbpool.runInteraction(self.do_insert, item)
        query.addErrback(self.handle_error, item, spider)

    # 自定義錯誤處理
    def handle_error(self, failure, item, spider):
        print(failure)
        print(item)

    def do_insert(self, cursor, item):
        insert_sql = '''
            INSERT INTO 
            jobbole_article (title, create_time, url, url_object_id, font_image_url, font_image_path, comment_num, up_num, fav_num, tags, content)
            VALUES 
            (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
        '''
        cursor.execute(insert_sql, (item["title"], item["create_time"], item["url"], item["url_object_id"], item["font_image_url"][0],
                                    item["font_image_path"], item["comment_num"], item["up_num"], item["fav_num"], item["tags"], item["content"]))

專案改進

前面使用了最基本的方式來解析的文章詳情頁，這樣使得spider的程式碼十分長，不容易維護，因此可以採用自定義ItemLoder的方式方便對規則的管理

spider檔案的修改

class JobboleSpider(scrapy.Spider):
    # 爬蟲的名稱 後續啟動爬蟲是採用此名稱
    name = "jobbole"
    # 爬取允許的域名
    allowed_domains = ["blog.jobbole.com"]
    # 起始url列表 ， 其中的每個URL會進入下面的parse函式進行解析
    start_urls = ['http://blog.jobbole.com/all-posts/']

    # 列表頁面的解析
    def parse(self, response):
        # 獲取文章列表中的每一篇文章的url交給Scrapy下載並解析
        article_nodes = response.css('div#archive .floated-thumb .post-thumb a')
        for article_node in article_nodes:
            # 解析每個文章的封面圖
            font_image_url = article_node.css('img::attr(src)').extract_first("")
            # 解析每個文章的url
            article_url = article_node.css('::attr(href)').extract_first("")
            # 智慧對url進行拼接，如果url中不帶有域名則會自動新增域名
            # 通過在Request中設定meta資訊來進行資料的傳遞
            yield Request(url=parse.urljoin(response.url, article_url),meta={'font_image_url':parse.urljoin(response.url, font_image_url)}, callback=self.parse_detail)

        # 獲取文章的下一頁url地址，並交給自身解析
        next_url = response.css('a.next.page-numbers::attr(href)').extract_first('')
        if next_url:
            yield Request(url=parse.urljoin(response.url, next_url), callback=self.parse)

    # 詳情頁面的解析
    def parse_detail(self, response):
        article_item = JobBoleArticleItem()
        # 從response中獲取文章封面圖
        font_image_url = response.meta.get('font_image_url', '')
        item_loader = JobBoleArticleItemLoader(item=JobBoleArticleItem(),response=response)
        item_loader.add_css('title', 'div.entry-header h1::text')
        item_loader.add_css('create_time', 'p.entry-meta-hide-on-mobile::text')
        item_loader.add_value('url', response.url)
        item_loader.add_value('url_object_id', get_md5(response.url))
        item_loader.add_value('font_image_url', [font_image_url])
        item_loader.add_css('comment_num', 'a[href="#article-comment"] span::text')
        item_loader.add_css('content', 'div.entry')
        item_loader.add_css('tags', 'p.entry-meta-hide-on-mobile a::text')
        item_loader.add_css('up_num', '.vote-post-up h10')
        item_loader.add_css('fav_num', 'div.post-adds > span.btn-bluet-bigger.href-style.bookmark-btn.register-user-only::text')

        article_item = item_loader.load_item()
        yield article_item

自定義的ItemLoader

import datetime
import re

import scrapy
from scrapy.loader import ItemLoader
from scrapy.loader.processors import MapCompose, TakeFirst, Join


# 去除文字中的點
def remove_dote(value):
    return value.replace('·','').strip()


# 時間轉換處理
def date_convert(value):
    try:
        create_time = datetime.datetime.strptime(value, "%Y/%m/%d").date()
    except Exception as e:
        create_time = datetime.datetime.now().date()
    return create_time


# 獲得數字
def get_num(value):
    match_re = re.match('.*?(\d+).*', value)
    if match_re:
        num = match_re.group(1)
    else:
        num = 0
    return int(num)


# 獲取點贊數
def get_up_num(value):
    match_re = re.match('<h10 id=".*?">(\d+)</h10>', value)
    if match_re:
        num = match_re.group(1)
    else:
        num = 0
    return int(num)


# 去掉tag中的評論
def remove_comment_tag(value):
    if "評論" in value:
       return ""
    return value


# 預設返回
def return_value(value):
    return value


# 自定義ITemLoader
class JobBoleArticleItemLoader(ItemLoader):
    # 改寫預設的output_processor
    default_output_processor = TakeFirst()


# 伯樂線上Item
class JobBoleArticleItem(scrapy.Item):
    title = scrapy.Field()
    create_time = scrapy.Field(
        # 該傳入的欄位值要批量處理的函式
        input_processor=MapCompose(remove_dote, date_convert),
    )
    url = scrapy.Field()
    url_object_id = scrapy.Field()
    font_image_url = scrapy.Field(
        output_processor = MapCompose(return_value)
    )
    font_image_path = scrapy.Field()
    up_num = scrapy.Field(
        input_processor = MapCompose(get_up_num)
    )
    fav_num = scrapy.Field(
        input_processor=MapCompose(get_num),
    )
    comment_num = scrapy.Field(
        input_processor=MapCompose(get_num),
    )
    tags = scrapy.Field(
        input_processor=MapCompose(remove_comment_tag),
        output_processor = Join(',')
    )
    content = scrapy.Field()
```