Scrapy を使ってクローリング・スクレイピングしてみる

Python 製のスクレイピング・ライブラリ Scrapy を使ってみる。

目次

Scrapy プロジェクトを作成する

作業ディレクトリを作り、pipenv を使って環境構築し、Scrapy をインストールしていく。

# Pipfile を初期生成する
$ pipenv --python 3.7
# Pipfile、Pipfile.lock、.venv/ が生成される

# Scrapy をインストールする
$ pipenv install scrapy

# 「pipenv run scrapy」と打たずに「scrapy」とコマンドが実行できるようシェルを読み込む
$ pipenv shell

# Scrapy プロジェクトを作成する
$ scrapy startproject my_scrapy .
# scrapy.cfg ファイル、my_scrapy/ ディレクトリが生成される

scrapy startproject 【プロジェクト名】 【作成するパス】 というコマンドで、Scrapy 用のプロジェクトを作る。カレントディレクトリに作るようにすると、以下のようなファイルが自動生成されるはずだ。

【作業ディレクトリ】/
├ scrapy.cfg
└ 【プロジェクト名】/ … 上の例では「my_scrapy/」
   ├ __init__.py
   ├ items.py
   ├ middlewares.py
   ├ pipelines.py
   ├ settings.py
   └ spiders/
      └ __init__.py

それぞれのファイルの意味は順を追って理解していくとする。

クローリングの設定を変更する

まずは settings.py を開き、クローリングの設定を変更していく。やっておきたい設定は以下のとおり。

スパイダーを作成する

スクレイピング関連の一般的な用語整理。

Scrapy では、これらの処理をまるっと「Spider」と呼ぶクラスが担う形になる。Scrapy でいうところの「スパイダー」を作ってみよう。

# 作成した Scrapy プロジェクトに移動する
$ cd my_scrapy/

# 以下のコマンドで Spider を作成する
$ scrapy genspider my_example example.com

scrapy genspider 【Spider 名】 【取得するサイトのドメイン】 と記す。ドメイン部分は、https:// などのプロトコルを付けず、ドメイン配下を付けてはならない。

1つの Spider につき、1つのサイトのドメインを扱う形になる。

このようにコマンドを実行すると、./【Scrapy プロジェクト名】/spiders/【指定した Spider 名】.py というファイルができているはずだ。

# -*- coding: utf-8 -*-

import scrapy

class MyExampleSpider(scrapy.Spider):
  name = 'my_example'
  allowed_domains = ['example.com']
  start_urls = ['http://example.com/']
  
  def parse(self, response):
    pass

コレで Spider の雛形が出来たことになる。

start_urls 部分でクロール対象の URL を指定する。一般的な用語でいう「スパイダー」の役割は、この start_urls に URL を列挙することで担う。クロール対象 URL 部分を動的に処理して収集していくこともできるが、今回は割愛。予めクロールしたい URL を配列で列挙しておくこととする。

「クロール」処理は、Scrapy が自動的に行い、レスポンスを parse() 関数に渡してくれる。開発者は parse() 関数内で「スクレピング」処理を実装していけば良い、というワケだ。

以降、もう少し詳しく説明していこう。

Item を定義する

スクレイピングしたデータは、最終的に CSV や JSON の形式で出力できる。こうした構造化データを表現するために、Scrapy では Item と呼ばれるクラスを作成しておくことになる。

items.py というファイルがあるので、コレを開き、次のように実装しておこう。

# -*- coding: utf-8 -*-

from scrapy import Field, Item

# Item を定義する
class MyExampleItem(Item):
  # 適当にフィールドを定義する
  page_url      = Field()
  page_title    = Field()
  page_headline = Field()

任意のクラス名で MyExampleItem を作った。class のカッコ部分で Item を渡していて、コレにより scrapy.Item クラスを継承している。

page_url や page_title といったフィールド名は任意。取得したいモノに合わせてフィールドを定義しよう。そこに scrapy.Field() を代入して、Scrapy 用のフィールドを宣言してあげている。

この書き方はお決まりなので、深く考えずこのように実装する。

ページ遷移を伴う場合や、取得するデータの構造が異なる場合は、複数の Item クラスを作成して良い。Spider のクラスでは、この items.py から、使いたい Item クラスを import して使う形になる。

スクレイピング処理を実装する

スクレイプした結果を詰める Item クラスを作成したので、いよいよスクレイピング処理を実装してみよう。今回は上で見せた Spider クラスのとおり、example.com にアクセスして、そのページのデータを抜き取ってみる。

# -*- coding: utf-8 -*-

import scrapy

# Item を import しておく
from my_scrapy.items import MyExampleItem

class MyExampleSpider(scrapy.Spider):
  name = 'my_example'
  allowed_domains = ['example.com']
  
  # クロール対象の URL を指定する
  start_urls = ['https://example.com/']
  
  def parse(self, response):
    # 結果を詰める Item を作成する
    item = MyExampleItem()
    # フィールドごとにデータを抽出して詰める
    item['page_title']    = response.css('title::text').extract_first()
    item['page_url']      = response.url
    item['page_headline'] = response.css('h1::text').extract_first()
    # Item を出力する
    yield item

こんな感じ。

基本は、parse() 関数が自動的に渡してくれる response オブジェクトから、上手くデータを抽出していくだけ。

example.com にアクセスすると分かるが、ページには h1 要素があるので、この要素を特定して、中身のテキストを拾っているのが item['page_headline'] の代入部分。

とにかくひたすらページの構造に合わせて、このように取得を繰り返していくだけなので、後は愚直に作業していく…。

作成した Spider を実行する

Spider が実装できたら、実際に実行してみよう。作成した Scrapy プロジェクトに移動して、次のようにコマンドを実行する。

$ scrapy crawl my_example

すると、コンソールにデバッグログなどが出力され、最終的に yield Item で出力させた Item 情報が確認できるかと思う。

分かりにくければ、次のように JSON 形式で結果だけ表示させてみよう。

# JSON 形式でコンソール出力する
$ scrapy crawl my_example -t json -o stdout: --nolog

結果をファイルに出力することもできる。-o オプションで任意のファイル名を指定すると、その拡張子に応じて JSON や CSV 等の形式で Item を出力してくれる。

# JSON ファイルに書き出す
$ scrapy crawl my_example -o my_result.json

コレが基本的な使い方となる。

ページ遷移を伴う・複数 Item を扱う場合

次は少し高度で、実用的な例を紹介する。ココまでだと、単一の URL にアクセスして、そのページ内の情報を引っこ抜くだけだった。だが実際は、エントリポイントとなるページから、リンク先のページに遷移して、そのページの情報を取得したかったりする。今回はそんな例を作成してみよう。

題材

題材は僕のサイト Neo's World にアクセスし、グローバルナビゲーションメニューの項目一つひとつのリンクを踏んで、遷移先のページにある h1 要素のテキストを引っこ抜いてみよう。

トップページにあるグローバルナビゲーションメニューは、次のように実装されていることとする。

<nav id="nav">
  <ul>
    <li><a href="/about/index.html">About</a></li>
    <li><a href="/music/index.html">Music</a></li>
    <li><a href="/games/index.html">Games</a></li>
    <li><a href="/gallery/index.html">Gallery</a></li>
    <li><a href="/etc/index.html">Etc.</a></li>
  </ul>
</nav>

遷移先ページは5ページあるワケだ。

Item を作成する

今回は「トップページの情報」と「遷移先ページの情報」とをまとめて、1つの Item に出力しようと思う。JSON でいうとこんな感じだ。

[
  {
    "index_page": {
      "title": "トップページのタイトル",
      "url"  : "トップページの URL"
    },
    "child_page": {
      "title"   : "遷移先ページのタイトル",
      "url"     : "遷移先ページの URL",
      "headline": "遷移先ページの見出しテキスト"
    }
  },
  // 同じ構成で、あと4つ、合計5行分のデータ
]

先程 HTML で見せたナビゲーションメニューは5項目あったので、最終的に JSON 配列で5つの要素が取れれば OK だ。index_page プロパティに含めるデータは、5つとも同じ情報が入ることになる。index_page.child_page.headline というように、子プロパティに遷移先ページのデータを持たせていくことも不可能ではないが、Item の取り回しが面倒臭くなるので今回は避ける。

こうしたデータに相当する Item を実装しておく。

# -*- coding: utf-8 -*-

from scrapy import Field, Item

# トップページと遷移先ページの両データを持つ、出力用の Item
class NeosWorldItem(Item):
  index_page = Field()
  child_page = Field()

# トップページのデータを格納する Item
class IndexPageItem(Item):
  title = Field()
  url   = Field()

# 遷移先ページのデータを格納する Item
class ChildPageItem(Item):
  title    = Field()
  url      = Field()
  headline = Field()

このように、3つの Item クラスを作成する。IndexPageItem と ChildPageItem のインスタンスを、NeosWorldItem の各フィールドに持たせて、yield NeosWorldItem と出力する構成だ。

Spider を作成する

新たに Spider を作成する。

$ scrapy genspider neos_world neo.s21.xrea.com

生成された Spider ファイルを開いて以下のように実装する。

# -*- coding: utf-8 -*-

import scrapy
from my_scrapy.items import NeosWorldItem, IndexPageItem, ChildPageItem

class NeosWorldSpider(scrapy.Spider):
  name = 'neos_world'
  allowed_domains = ['neo.s21.xrea.com']
  # クロール対象の URL を指定する
  start_urls = ['http://neo.s21.xrea.com/']
  
  # トップページ (start_urls) のスクレイピング処理
  def parse(self, response):
    # トップページ用 Item を作成する
    index_page_item = IndexPageItem()
    index_page_item['title'] = response.css('title::text').extract_first()
    index_page_item['url']   = response.url
    
    # ナビゲーションメニューのリンク要素を取得する
    nav_link_elems = response.css('#nav > ul > li > a')
    
    # 万が一、ナビゲーションメニューが1つも存在しない場合は、異常終了とする
    if(not nav_link_elems):
      # 異常時は IndexPageItem だけを格納した結果出力用 Item を出力して終了する
      neos_world_item = NeosWorldItem()
      neos_world_item['index_page_item'] = index_page_item
      yield neos_world_item
      return
    
    # (リンク要素が正常に見つかれば) リンク要素を1つずつ処理する
    for nav_link_elem in nav_link_elems:
      # 1つのリンクの href 属性値を取得する
      nav_href = nav_link_elem.css('a::attr(href)').extract_first()
      # 遷移元 URL と href 属性値をかけあわせて、遷移先 URL のフルパスを構成する
      next_url = response.urljoin(nav_href)
      
      # 遷移先ページにアクセスし、parse_child() 関数に後続の処理を行わせる
      yield scrapy.Request(next_url, callback = self.parse_child, meta = {
        'index_page_item': index_page_item
      })
  
  # 遷移先ページのスクレイピング処理
  def parse_child(self, response):
    # meta で送信されたトップページ用 Item を引き出しておく
    index_page_item = response.meta['index_page_item']
    
    # 遷移先ページ用 Item を作成する
    child_page_item = ChildPageItem()
    child_page_item['title']    = response.css('title::text').extract_first()
    child_page_item['url']      = response.url
    child_page_item['headline'] = response.css('h1::text').extract_first()
    
    # 結果出力用 Item を作成する
    neos_world_item = NeosWorldItem()
    neos_world_item['index_page_item'] = index_page_item
    neos_world_item['child_page_item'] = child_page_item
    # Item を出力する
    yield neos_world_item

少し長くなったがこんな感じ。いくつかポイントがあるので押さえておこう。

parse() 以降で呼ぶ yield が分かりづらいかもしれないが、

という動きをするので、この2つの動きをベースに、最終的にどんな Item を出力したいか、というところをイメージしながら、コールバック関数を作っていくことになる。

入れ子の Item を出力しようとしたら…

少し前に index_page.child_page.headline というような入れ子関係を作るのが大変、といったが、コレをやろうとすると、こんな実装になるだろう。

from scrapy import Field, Item

# トップページと、遷移先ページのデータを格納する Item
class IndexPageItem(Item):
  title = Field()
  url   = Field()
  child_pages = Field()  # 配列で ChildPageItem を格納していく

# 遷移先ページのデータを格納する Item
class ChildPageItem(Item):
  title    = Field()
  url      = Field()
  headline = Field()

# -*- coding: utf-8 -*-

import scrapy
from my_scrapy.items import IndexPageItem, ChildPageItem

class NeosWorldSpider(scrapy.Spider):
  name = 'neos_world'
  allowed_domains = ['neo.s21.xrea.com']
  start_urls = ['http://neo.s21.xrea.com/']
  
  def parse(self, response):
    # トップページ用 Item を作成する (同様の処理なので省略)
    index_page_item = IndexPageItem()
    # 遷移先ページの ChildPageItem を持たせるプロパティを初期化しておく
    index_page_item['child_pages'] = []
    
    # 遷移先 URL の配列を作る
    child_urls = []
    for nav_link_elem in response.css('#nav > ul > li > a'):
      child_url = response.urljoin(nav_link_elem.css('a::attr(href)').extract_first())
      child_urls.append(child_url)
    
    # 遷移先 URL を配列から1つ取り出し (配列からは取り出した要素が消える)、スクレイピングを行う
    next_url = child_urls.pop()
    yield scrapy.Request(next_url, callback = self.parse_child, meta = {
      'index_page_item': index_page_item,
      'child_urls'     : child_urls
    })
  
  # 遷移先ページのスクレイピング処理
  def parse_child(self, response):
    # meta で送信されたトップページ用 Item を引き出しておく
    index_page_item = response.meta['index_page_item']
    # 遷移先ページの URL 配列も引き出しておく
    child_urls = response.meta['child_urls']
    
    # 遷移先ページ用 Item を作成する (同様の処理なので省略)
    child_page_item = ChildPageItem()
    # スクレイピングしたデータを IndexPageItem に追加する
    index_page_item['child_pages'].append(child_page_item)
    
    # 遷移先ページの URL 配列が空になっていたら、IndexPageItem を出力して終了する
    if not child_urls:
      yield index_page_item
      return
    
    # 遷移先 URL を配列から1つ取り出し (配列からは取り出した要素が消える)、スクレイピングを行う
    next_url = child_urls.pop()
    yield scrapy.Request(next_url, callback = self.parse_child, meta = {
      'index_page_item': index_page_item,
      'child_urls'     : child_urls
    })

一気に関数の実行順序が複雑になったのが分かるだろうか。

なかなか理解するのが難しいと思うので、やってみたい方は print デバッグしながら動きを追ってみると良いだろう。

child_page よりさらに先のページにも遷移して、3階層の入れ子を作りたい、と考え始めるともっとしんどくなるので、Item の入れ子はしない方が良いだろう。

以上

今回はココまで。pipelines.py を実装すれば、収集した Item を DB 投入するところまで一気に実行できたりとか、Scrapy には他にも色々な機能がたくさんあるので、ぜひ活用していってもらいたい。