Scrapy は、Webサイトをクロールし、ページから構造化データを抽出するために使用されるWebスクレイピングフレームワークです。Scrapyに関してはわかりやすい記事がたくさんあるので、ここでは実装サンプルを紹介しまくるスタンスにしようと思います。

インストール

pip install scrapy
# or
poetry add scrapy

チュートリアルを試す

こちらにしたがって、チュートリアルを試してみます。

scrapy startproject tutorial
or
poetry run scrapy startproject tutorial

を実行するとtutorialフォルダができてその下にテンプレートのソースコード一式が出力されます。

tutorial/spidersフォルダの下に以下の内容でquotes_spider.pyを作ります。

import scrapy
from typing import List

class QuotesSpider(scrapy.Spider):
    name: str = "quotes"

    def start_requests(self):
        urls: List[str] = [
            'http://quotes.toscrape.com/page/1/',
            'http://quotes.toscrape.com/page/2/',
        ]
        for url in urls:
            yield scrapy.Request(url=url, callback=self.parse)

    def parse(self, response):
        page: str = response.url.split("/")[-2]
        filename: str = 'quotes-%s.html' % page
        with open(filename, 'wb') as f:
            f.write(response.body)
        self.log('Saved file %s' % filename)

クローリングを実行します。

scrapy crawl quotes
or
poetry run scrapy crawl quotes

quotes-1.htmlとquotes-2.htmlが出力されていれば成功です。

Scrapyの構成要素

何も考えずチュートリアルを実行してみましたが、Scrapyにはいくつかの登場人物がいます。

Spiders

上のチュートリアルでも実装しましたが、どのようにスクレイピングするかを定義する中心的なクラスです。Spiderクラスだけでスクレイピングを完結させることもできます。

Spiderクラスでは、以下の流れで処理が実行されます

  1. start_requests()で最初のRequestオブジェクト（のリスト）を返却します
  2. Requestが実行され、callback関数であるparse()が呼び出される。引数にResponseオブジェクトが渡ってくる
  3. callback関数parse()では、Responseオブジェクトを元に処理を行い（例えばファイルを出力したりとか、DBに記録したりとか）、次のRequestオブジェクト（or リスト）かItemオブジェクトを返却する
  4. Requestオブジェクト（or リスト）が返却された場合は、Requestが実行され、またparse()が呼び出される
  5. Itemオブジェクトが返却された場合は、Pipeline定義にしたがってPipelineが実行される

Item Pipeline

Item Pipelineは、取得したResonseに対してなんらかのアクションを行いたい場合に利用するものです。利用しなくても全然構いません。
一般的な用途は、以下の通りとのことです。

  • HTMLデータのクレンジング
  • スクレイピングされたデータの検証（アイテムに特定のフィールドが含まれていることの確認）
  • 重複チェック（およびそれらのドロップ）
  • 削られたアイテムをデータベースに保存する

Pipelineクラスはprocess_item(self, item, spider)という関数を持つシンプルなクラスです。
process_item()では何らかのアクションを行った後に、Itemオブジェクト or Deferredを返却するか、DropItemエラーをraiseするか、のいずれかの処理を行う必要があります。

個人の見解ですが、Pipelineを利用する場合、Webリクエストに関してはSpiderがその責務を負い、結果のフィルターや加工に関してはPipelineが責務を負う、というような責任分担が重要だと思います。
なので、何らかのアクションをした結果で新しくRequestを投げたくなることがあるかもしれませんが、おそらくPipeline実装すべきではなく、Spiderで実装すべきだと思います。
下の方にこれに該当する実装サンプルがあるので見てみてください。

Items

ItemsはPipelineにデータを渡すための入れものです。dictionaries・Item objects・dataclass objects・attrs objectsなどいくつかの種類があるそうです。

  • dictionaries: ただのdict
  • Item objects: scrapy.item.Itemを継承したクラスのオブジェクト
  • dataclass objects: dataclassを使って定義したクラスのオブジェクト
  • attrs objects: attr.sを使って定義したクラスのオブジェクト

上のチュートリアルのサンプルコードにもItemは出てきてないのですが、Item Pipelineを使わないのであればItemの定義は特に必要ありません。

実装サンプル

pythonスクリプトから呼び出す

scrapyはコマンド実行するのが基本のようですが、pythonスクリプトから実行することもできます。
チュートリアルコードを少し変更して、指定したURLのHTMLを抽出してファイルに出力するサンプルを書いてみようと思います。

main.pyを作って、その中でMySpiderクラスを作成し、CrawlerProcessでSpiderクラスを実行する処理を書きました。

import scrapy
from scrapy.http import Response
from scrapy.crawler import CrawlerProcess
from typing import List, Dict, Any

class MySpider(scrapy.Spider):
    name = 'my_spider'

    def __init__(self, urls: List[str], *args, **kwargs):
        # Request対象のURLを指定
        self.start_urls = urls
        super(MySpider, self).__init__(*args, **kwargs)

    def parse(self, response: Response):
        page: str = response.url.split("/")[-2]
        filename: str = 'quotes-%s.html' % page
        with open(filename, 'wb') as f:
            f.write(response.body)
        self.log('Saved file %s' % filename)

def main():
    # スクレイピング設定 see: https://docs.scrapy.org/en/latest/topics/settings.html
    settings: Dict[str, Any] = {
        'DOWNLOAD_DELAY': 3,
        'TELNETCONSOLE_ENABLED': False,
    }

    # クローリング実行
    process: CrawlerProcess = CrawlerProcess(settings=settings)
    process.crawl(MySpider, ['http://quotes.toscrape.com/page/1/', 'http://quotes.toscrape.com/page/2/'])
    process.start()  # the script will block here until the crawling is finished

if __name__ == "__main__":
    main()

ポイントは以下の通りです。

  • MySpiderのコンストラクタ（__init__()）にurlsという引数を追加しており、このurlsをインスタンス変数start_urlsに設定することで、リクエスト対象のURLとして設定している
  • parse()の中身はチュートリアルコードと全く同じ
  • main()関数の中でまずスクレイピング設定をDictで定義している。これはテンプレートだとsettings.pyで定義している内容を移植したもの
    • 設定が固定であればsettings.pyをそのまま使ってもOKだが、Dictであれば動的に変更することもできるので、こちらの方が優秀
  • クローリングの実行はCrawlerProcessもしくはCrawlerRunnerで実行できるが今回はCrawlerProcessを使っている（正直使い分けがよくわからない）
  • process.crawl()の第二引数以降がSpiderクラスのコンストラクタに引数として渡される
    • 複数引数をとっても問題ない。またキーワード引数もOK（例：urls=['http://hoge', 'http://fuga']）

あとは以下のように実行します。

python main.py

チュートリアルコードと同様にquotes-1.htmlとquotes-2.htmlが出力されると思います。

このように、簡単にpythonスクリプトからScrapyを実行することができますし、設定もDictで渡せるし、対象URLも引数で渡せるので、動的にスクレイピング設定・スクレイピング対象を変更することも可能です。

個人的にはコマンドライン実行よりこちらの方が柔軟性が高く好きです。なので、これ以降の実装サンプルはCrawlerProcessを使ってSpiderを実行するスタンスで書いていきたいと思います。

Item Pipelineを使う

ResponseオブジェクトのステータスとBODYサイズでフィルターして、OKだったらファイルに出力するというPipelineを作ってみました。

まずは、items.pyを作ってMyItemクラスを定義します。今回はdataclassを使って定義してみました。

from dataclasses import dataclass

@dataclass
class MyItem:
    url: str
    status: int
    title: str
    body: str

次に、pipelines.pyを作って、MyItemクラスを受け取ってフィルターして、ファイルを出力するPipelineを定義します。
Pipelineクラスはprocess_timeという名前の関数が必須で、第二引数にItemオブジェクト、第三引数にSpiderオブジェクトを受け取り、Itemオブジェクト or defer を返却するか、DropItemをraiseする必要があります。

from scrapy.exceptions import DropItem
from scrapy import Spider
from items import MyItem

class StatusFilterPipeline:
    """ステータスでフィルターするパイプライン"""
    def process_item(self, item: MyItem, spider: Spider) -> MyItem:
        if item.status != 200:
            raise DropItem(f'Status is not 200. status: {item.status}')
        return item

class BodyLengthFilterPipeline:
    """BODYサイズでフィルターするパイプライン"""
    def process_item(self, item: MyItem, spider: Spider) -> MyItem:
        if len(item.body) < 11000:
            raise DropItem(f'Body length less than 11000. body_length: {len(item.body)}')
        return item

class OutputFilePipeline:
    """ファイル出力するパイプライン"""
    def process_item(self, item: MyItem, spider: Spider):
        filename: str = f'quotes-{item.url.split("/")[-2]}.html'
        with open(filename, 'wb') as f:
            f.write(item.body)

後は、main.pyを作ってMySpiderクラスを作り、CrawlerProcessで実行します。

from scrapy import Spider
from scrapy.http import Response
from scrapy.crawler import CrawlerProcess
from typing import List, Dict, Any, Iterator
from items import MyItem

class MySpider(Spider):
    name = 'my_spider'

    def __init__(self, urls: List[str], *args, **kwargs):
        self.start_urls = urls
        super(MySpider, self).__init__(*args, **kwargs)

    def parse(self, response: Response) -> Iterator[MyItem]:
        yield MyItem(
            url=response.url,
            status=response.status,
            title=response.xpath('//title/text()').extract_first(),
            body=response.body,
        )

def main():
    # スクレイピング設定 see: https://docs.scrapy.org/en/latest/topics/settings.html
    settings: Dict[str, Any] = {
        'DOWNLOAD_DELAY': 3,
        'TELNETCONSOLE_ENABLED': False,
        'ITEM_PIPELINES': {
            'pipelines.StatusFilterPipeline': 100,
            'pipelines.BodyLengthFilterPipeline': 200,
            'pipelines.OutputFilePipeline': 300,
        },
    }

    # クローリング実行
    process: CrawlerProcess = CrawlerProcess(settings=settings)
    process.crawl(MySpider, ['http://quotes.toscrape.com/page/1/', 'http://quotes.toscrape.com/page/2/', 'http://quotes.toscrape.com/page/3/'])
    process.start()  # the script will block here until the crawling is finished

if __name__ == "__main__":
    main()

ポイントは以下です。

  • parse()関数は、ジェネレーター関数になっていて、MyItemオブジェクトをyieldしている
  • settingsのITEM_PIPELINESでPipelineの順番を定義している。ここにはPipelineクラスのパスを指定する必要がある。また数字（0-1000）の順にパイプラインは実行される

あとは以下のように実行します。

python main.py

BODYサイズのフィルターの結果、3つのURLのうちquotes-1.htmlとquotes-2.htmlの二つのファイルが出力されていれば成功です。
ちなみに、フィルターでDropItemをraiseして除外したItemについては、以下のようにWARNINGログが出力されています。

2020-08-13 19:11:28 [scrapy.core.scraper] WARNING: Dropped: Body length less than 11000. body_length: 10018

Item Pipelineはこんな感じで利用します。
Pipelineによって責務が明確になるのが良さそうですが、理解すべき事項が増えてしまうので、個人的には特別な知識がなくてもコードを見ればわかるように、ただの関数として定義してしまう方が好みですね。

Item Pipelineを使うケースでparse時にItemと次のRequestのyieldを両方実行する

あるページをスクレイピングしてItemを取得しつつ、ページ内のリンクを取得して次のリクエストを行うようなことをやりたいケースはあると思います。そのようなケースではSpiderクラスのparse()の中でItemとRequestの両方をyieldすれば良いはずです。

Item-Pipelineを使うのケースでResponseをparseする際に、NEXTボタンのリンクURLを取得して次のRequestを作りつつ、該当ページのResponse自体はMyItemに詰めてPipelineに流すような感じにしようと思います。

実装は、Item-Pipelineを使うのmain.pyのparse()に3行追加して、main()でprocess.crawl()する際に渡すURLを1ページ目だけに絞っただけです。

class MySpider(Spider):
    ### (省略) ###

    def parse(self, response: Response) -> Iterator[Union[MyItem, Request]]:
        yield MyItem(
            url=response.url,
            status=response.status,
            title=response.xpath('//title/text()').extract_first(),
            body=response.body,
        )
        if len(response.css('li.next > a')) > 0:  # <= これ以下3行を追加
            next_url: str = response.urljoin(response.css('li.next > a').attrib['href'])
            yield Request(url=next_url, callback=self.parse)

def main():
    ### (省略) ###

    # クローリング実行
    process: CrawlerProcess = CrawlerProcess(settings=settings)
    process.crawl(MySpider, ['http://quotes.toscrape.com/page/1/'])  # <= 開始URLをpage=1だけに変更
    process.start()  # the script will block here until the crawling is finished

実行すると、quotes-1.html・quotes-2.html・quotes-8.html・quotes-9.htmlの4ページ分のHTMLが出力されて、その他のページはDropItemされた旨のログが出ていたので、正常にSpiderのparse()処理で、Itemと次のRequestの両方をyieldすることができていたようです。

LinkExtractorを利用する

ScrapyにはリンクをたどってURLを抽出することに特化した、LinkExtractorクラスが用意されています。これを使って、リンクURLをログに出力するだけのサンプルを書いてみようと思います。

main.pyを作って、MyLinkExtractSpiderクラスを実装し、RuleにLinkExtractorを指定しただけのサンプルを実装しました。

from scrapy.http import Response
from scrapy.crawler import CrawlerProcess
from scrapy.spiders import CrawlSpider, Rule
from scrapy.linkextractors import LinkExtractor
from typing import Dict, Any

class MyLinkExtractSpider(CrawlSpider):
    name = 'my_link_extract_spider'
    start_urls = ['http://quotes.toscrape.com/']
    rules = (
        Rule(
            LinkExtractor(
                allow=r'http://quotes.toscrape.com/page/\d+/',  # 許可するURLのパターンを正規表現で
                unique=True,  # URLをユニークにするかどうか
                tags=['a'],  #  対象とするタグ
            ),
            follow=True,  # Responseオブジェクトのリンクをたどるかどうか
            callback='log_url'  # 各リンクに対するResponseを受け取った際に呼び出されるコールバック関数
        ),
    )

    def log_url(self, response: Response):
        print(f'response.url: {response.url}')

def main():
    settings: Dict[str, Any] = {
        'DOWNLOAD_DELAY': 3,
        'TELNETCONSOLE_ENABLED': False,
    }

    process: CrawlerProcess = CrawlerProcess(settings=settings)
    process.crawl(MyLinkExtractSpider)
    process.start()  # the script will block here until the crawling is finished

if __name__ == "__main__":
    main()
  • LinkExtractorのコンストラクタ引数で色々と指定できます
    • 詳細はこちらを参照
    • 今回は実装してないですが、process_valueに関数を指定できますので、例えばhrefにjavascriptの処理が入っていたとしても関数でパースしてURL部分を抜き出すことも可能です
  • Ruleの引数はこちらを参照ください

これもpython main.pyで実行できます。

response.url: http://quotes.toscrape.com/page/1/
response.url: http://quotes.toscrape.com/page/2/
response.url: http://quotes.toscrape.com/page/3/
: (省略)
response.url: http://quotes.toscrape.com/page/10/

実行すると以下のように無事リンクURLがログに出力されました。

SitemapSpiderを利用する

サイトマップファイル（sitemap.xmlやrobots.txt）からSitemapSpiderを使ってサイト内のURL一覧を取得することもできます。
こちらにサンプルがあるので、それを参考に、雑ですがQiitaのURL一覧をログに出力するだけのSpiderを作ってみました。

from scrapy.spiders import SitemapSpider

class MySitemapSpider(SitemapSpider):
    name = 'my_sitemap_spider'
    sitemap_urls = ['https://qiita.com/robots.txt']  # sitemap.xmlやrobots.txtを指定する

    def parse(self, response: Response):
        print(f'response.url: {response.url}')

実際に動かしてみると、以下のような感じで大量にURLを抽出することができました。

response.url: https://qiita.com/tags/comprehension
response.url: https://qiita.com/tags/%23rute53
response.url: https://qiita.com/tags/%23dns
response.url: https://qiita.com/tags/%EF%BC%83lightsail
:

サイトマップから抽出するのであれば、相当速いのかと期待してたのですが、全てのURLにRequestを投げるので、普通にLinkExtractorと同じレベルで時間がかかります。また、サイトマップが最新に更新されているという保証もないため、これだけに頼るのは危険そうです。

その他

scrapy shellは便利

scrapy shell 対象URL

のように実行すると、REPLが立ち上がって、コード断片を実行することができるようになります。

たとえば、scrapy shell http://quotes.toscrape.com/を実行した場合は以下のオブジェクトが利用できます。

[s] Available Scrapy objects:
[s]   scrapy     scrapy module (contains scrapy.Request, scrapy.Selector, etc)
[s]   crawler    <scrapy.crawler.Crawler object at 0x102d21610>
[s]   item       {}
[s]   request    <GET http://quotes.toscrape.com/>
[s]   response   <200 http://quotes.toscrape.com/>
[s]   settings   <scrapy.settings.Settings object at 0x102d219a0>
[s]   spider     <DefaultSpider 'default' at 0x1030efb80>
[s] Useful shortcuts:
[s]   fetch(url[, redirect=True]) Fetch URL and update local objects (by default, redirects are followed)
[s]   fetch(req)                  Fetch a scrapy.Request and update local objects 
[s]   shelp()           Shell help (print this help)
[s]   view(response)    View response in a browser

なので、これを見ながら以下のような感じで気軽に試せるのが良い感じです。

>>> response.css('li.next > a')
[<Selector xpath="descendant-or-self::li[@class and contains(concat(' ', normalize-space(@class), ' '), ' next ')]/a" data='<a href="/page/2/">Next <span aria-hi...'>]
>>> response.css('li.next > a').attrib['href']
'/page/2/'

ヘッドレスブラウザ対応

試してないのですが、scrapy-splashを利用すると、ヘッドレスブラウザ上で対象ページを実行することができるようです。
https://github.com/scrapy-plugins/scrapy-splash

READMEを軽く読んだだけですが、scrapy.Requestの代わりにSplashRequest をyieldすると、ヘッドレスブラウザsplashで該当ページが読み込まれ、SplashRequestに渡したJavaScriptコードが実行され、その中でreturnされた値がHTTP Responseとして返却されるようです。

JavaScriptコード内では当然ながらdocument.querySelector()などは実行可能なので、特定のエレメントを取得して属性を取得するようなことは可能っぽいですが、画面遷移などに関しては未知数です。

複数の画面をまたいで画面内のエレメントをクリックしながら画面遷移するようなことはおそらくできないので、過度な期待はできないかな、、という印象です。ヘッドレスブラウザが必要であれば自分なら無難にpyppeteerを利用すると思います。

CrawlerProcess(CrawlerRunner)を同一プロセス内で二回実行する

CrawlerProcess(CrawlerRunner)を使ってクローリングする場合、twisted.internet.reactorを使って処理が完了を検知します。しかし、このtwisted.internet.reactorが曲者で、同一プロセス上で二度開始することができません。

以下のようなサンプルプログラムを作って、二回クローリングを実行すると

    process: CrawlerProcess = CrawlerProcess(settings=settings)

    # 1回目実行
    process.crawl(MySpider, ['http://quotes.toscrape.com/page/1/', 'http://quotes.toscrape.com/page/2/'])
    process.start()  # the script will block here until the crawling is finished
    # 2回目実行
    process.crawl(MySpider, ['http://quotes.toscrape.com/page/1/', 'http://quotes.toscrape.com/page/2/'])
    process.start()  # the script will block here until the crawling is finished

以下のようにtwisted.internet.error.ReactorNotRestartableエラーが発生します。

Traceback (most recent call last):
  File "main.py", line 41, in <module>
    main()
  File "main.py", line 37, in main
    process.start()  # the script will block here until the crawling is finished
  File "~/python_scraping_sample/.venv/lib/python3.8/site-packages/scrapy/crawler.py", line 327, in start
    reactor.run(installSignalHandlers=False)  # blocking call
  File "~/python_scraping_sample/.venv/lib/python3.8/site-packages/twisted/internet/base.py", line 1282, in run
    self.startRunning(installSignalHandlers=installSignalHandlers)
  File "~/python_scraping_sample/.venv/lib/python3.8/site-packages/twisted/internet/base.py", line 1262, in startRunning
    ReactorBase.startRunning(self)
  File "~/python_scraping_sample/.venv/lib/python3.8/site-packages/twisted/internet/base.py", line 765, in startRunning
    raise error.ReactorNotRestartable()
twisted.internet.error.ReactorNotRestartable

結論から言うと、これを回避する方法は、別プロセスで実行する方法しかうまくいきませんでした。 （process._stop_reactor()は特に意味がなく、その他色々試しましたが全部ダメでした）

concurrent.futures.ProcessPoolExecutorやmultiprocessing.Processを使って以下のような感じで実装すると問題なく動きます。

from multiprocessing import Process

def start_crawl(settings: Dict[str, Any], urls: List[str]):
    process: CrawlerProcess = CrawlerProcess(settings=settings)
    process.crawl(MySpider, urls)  #MySpiderの実装は省略
    process.start()  # the script will block here until the crawling is finished

def main():
    settings: Dict[str, Any] = {
        'DOWNLOAD_DELAY': 3,
        'TELNETCONSOLE_ENABLED': False,
    }

    # クローリングを別プロセスで二回実行
    Process(target=start_crawl, args=(settings, ['http://quotes.toscrape.com/page/1/'])).start()
    Process(target=start_crawl, args=(settings, ['http://quotes.toscrape.com/page/2/', 'http://quotes.toscrape.com/page/3/'])).start()

if __name__ == "__main__":
    main()

ちなみに、自分はPySparkのworkerノード上でクローリング処理を実行した際に、この問題に遭遇しました。workerノード上で動くTaskExecutorは同一プロセスで要求されたタスクを順次処理するため、クローリング処理を含むタスクを2回目に実行した際にこのエラーが発生しました。。。

リポジトリ

今回検証したソースコードは全てgithubにあげてありますので、必要があればこちらを参照ください。

さいごに

ドキュメントを見ながら色々と試してみましたが、実装自体は結構簡単でした。
設定だけで振る舞いを変えることができるのも、便利な点と言えると思います。
一方で、ドキュメントを見ずにソースコードだけを見て、どう動くのかを理解するのは難しく学習コストが高いライブラリという印象でした。

個人的には、requests+BeautifulSoupやrequests_htmlのようなコードを見れば分かるライブラリとScrapyのどちらが採用する？聞かれると、正直かなり迷うと思います。たぶん、LinkExtractorやSitemapSpiderのように目的にばっちり合致するケースではScrapyを利用し、それ以外は他のライブラリを選択する気がします。。

関連記事
  1. pythonでスクレイピングする際に利用するライブラリ比較
  2. 標準モジュールunittestでpythonのテストを書く
  3. Pythonにおけるマルチプロセスでの状態管理
  4. pyppeteerの使い方
  5. Pythonでasyncioを使った非同期処理