クローラーの作り方 - Advanced 編

LAPRAS アウトプットリレー7日目の記事です！
こんにちは！LAPRAS クローラーエンジニアの @Chanmoro です！

先日クローラーの作り方 - Basic 編という記事を書かせて頂きましたが、今回の記事では「クローラーの作り方 - Advanced 編」と題しまして、本格的にクローラーを開発する際にどういう問題に直面するかと、クローラーをどのような設計にしておくとこれらの問題に対処しやすいメンテナブルなクローラーにできるかというのを簡単に紹介したいと思います。

クローラー運用でよくある問題の例

さて、クローラーの作り方 - Basic 編で紹介したような方法でミニマムなクローラーを実装することができますが、そこから更に定期的にクロールを繰り返してデータを更新する運用を始めると、大抵の場合で以下のような問題に直面することになります。

• クロール先の HTML 構造が変化する
• クロール先の障害や一時的なエラー
• id が変化する

クローラー開発者にとってはどれも「あるあるー」という感じの問題なのですが、クローラーを運用し続けるには少なくともこれらの課題をクリアする必要があります。

「クロール先の HTML 構造が変化する」問題

クロール先のサービスでは絶えず変更が続けられているので、ある日突然 HTML 構造が変化して目的のデータが取得できなくなる可能性があります。
設計によっては、おかしな値が取得されてデータベースを上書きされてこれまでに蓄積したデータが全て壊れる可能性も十分あります。

僕がこれまでにこの課題に対して対応したことがあるのは以下のような方法です。

• クロールして取得される値に対してバリデーションを入れる
• クロール先サービスを定点観測するテストを定期実行する

クロールして取得される値に対してバリデーションを入れる

これはもしかすると最初に思いつくかもしれませんが、クロール先から取得されるデータに対するバリデーションを導入する方法です。
フォーマットや型を限定することができるデータに対してはバリデーションのルールを実装することでおかしなデータがデータベースに入ることを防ぐことができます。
しかし、この方法の弱点は特定のフォーマットが決められないデータや、必須項目ではないデータに対してはバリデーションのルールを定義することができないので、一部の項目にしか適用できないデメリットがあります。

クロール先サービスを定点観測するテストを定期実行する

この方法は、定点観測 = クロール先サービスに実際にアクセスして取得されるデータが想定の値になっているか、のテストを定期的に実行することで HTML 構造の変化を検知する方法です。
特定のアカウントやページに対するテストを書き、それを定期実行することで実装できます。

scrapy には contract という機能があり、指定した URL に対して実際にアクセスをし取得されたレスポンスに対してテストを書くことができるのですが、そこからアイディアを得ました。

ただしこの方法は、自分で定点観測用のアカウントを作成したりデータを用意できる場合は有用ですが、そうではない場合は対象のアカウントやページが更新される度にテストが失敗することになります。(テストの詳細に依存はしますが)
更に、例えば為替のデータのような高頻度で内容が変化するようなデータに対しては厳密なテストを書くことができません。

クロール先の HTML 構造が変化する問題への対象として2つの方法を紹介しましたが、もちろんこれらが全てではありません。

「クロール先の障害や一時的なエラー」問題

クロール先サービス側で障害が発生しているときに、一時的にエラーのレスポンスが返されることも有り得ます。
レスポンスのステータスコードを見てクローラーの処理を分岐しているような場合にこれが問題となることがあります。

例えばレスポンスステータスが 404 だった場合に該当のデータが削除されていたと判断し、該当のデータを削除する処理を実装していたとします。
この時クロール先サービス側では仕様のバグや何らかのエラーにより一時的に 404 が返されていたとすると、実際には該当のデータは削除されていないが、クローラー側では削除されたと誤って判断してしまう問題が発生します。

この問題の対処については、しばらく時間をあけてリトライすることで一時的なレスポンスなのかを見分ける方法が有効です。

「id が変化する」問題

URL にクロール対象の ID が含まれる設計になっているサービスで、後から ID を変更できる仕様になっているサービスをクロールする場合、変更後の id も変更前のデータと同じとして扱いたい場合があります。
例えばユーザーの ID を変更できたり、一度投稿されたデータの URL を変更できる仕様になっているようなサービスのクロールです。

サービスによっては 301 リダイレクトしてくれるので、この場合は旧 URL と新 URL を比べることで id の対応が分かります。
この場合の対処は比較的簡単で、301 リダイレクトされたあとの URL に含まれる id を取得してデータを更新することで追従できます。
注意点としては、クロール先データの id は可変なのでクローラーのシステム上では id としては扱ってはいけないということになります。

また、ものによっては旧 URL が 404 になり、新しい URL への対応が分からない場合があるので、旧 URL のデータを削除して新 URL のデータが新規に追加されるのを待つしかない場合というのもあり得ます。

メンテナブルなコードでクローラーを表現する

さて、これまでに紹介した 3 つの問題は大抵のクローラーが直面することになるのではないかと想像しています。

紹介したコード はかなりベタベタな実装だったので、今回紹介したような問題に対処する実装をパッと見どこに入れたらいいかが分かりません。

そこで、例えば以下のようにクローラーの処理をいくつかのレイヤーに分割してみます。

• parser
  • HTML をパースしてデータを抽出する層
• crawler
  • LAPRAS NOTE へのリクエストを送り parser を呼び出す層
• usecase
  • crawler と、取得されたデータの永続化処理を呼び出す層

import json
import time
import dataclasses
from typing import List, Optional

import requests
from bs4 import BeautifulSoup


@dataclasses.dataclass(frozen=True)
class ArticleListPageParser:
    @dataclasses.dataclass(frozen=True)
    class ArticleListData:
        """
        記事一覧ページから取得されるデータを表すクラス
        """
        article_url_list: List[str]
        next_page_link: Optional[str]

    @classmethod
    def parse(self, html: str) -> ArticleListData:
        soup = BeautifulSoup(html, 'html.parser')
        next_page_link = soup.select_one("nav.navigation.pagination a.next.page-numbers")

        return self.ArticleListData(
            article_url_list=[a["href"] for a in soup.select("#main div.post-item h2 > a")],
            next_page_link=next_page_link["href"] if next_page_link else None
        )


@dataclasses.dataclass(frozen=True)
class ArticleDetailPageParser:
    @dataclasses.dataclass(frozen=True)
    class ArticleDetailData:
        """
        記事詳細ページから取得されるデータを表すクラス
        """
        title: str
        publish_date: str
        category: str
        content: str

    def parse(self, html: str) -> ArticleDetailData:
        soup = BeautifulSoup(html, 'html.parser')
        return self.ArticleDetailData(
            title=soup.select_one("h1").get_text(),
            publish_date=soup.select_one("article header div.entry-meta").find(text=True, recursive=False).replace("｜", ""),
            category=soup.select_one("article header div.entry-meta a").get_text(),
            content=soup.select_one("article div.entry-content").get_text(strip=True)
        )


@dataclasses.dataclass(frozen=True)
class LaprasNoteCrawler:
    INDEX_PAGE_URL = "https://note.lapras.com/"
    article_list_page_parser: ArticleListPageParser
    article_detail_page_parser: ArticleDetailPageParser

    def crawl_lapras_note_articles(self) -> List[ArticleDetailPageParser.ArticleDetailData]:
        """
        LAPRAS NOTE をクロールして記事のデータを全て取得する
        """
        return [self.crawl_article_detail_page(u) for u in self.crawl_article_list_page(self.INDEX_PAGE_URL)]

    def crawl_article_list_page(self, start_url: str) -> List[str]:
        """
        記事一覧ページをクロールして記事詳細の URL を全て取得する
        """
        print(f"Accessing to {start_url}...")
        # https://note.lapras.com/ へアクセスする
        response = requests.get(start_url)
        response.raise_for_status()
        time.sleep(10)

        # レスポンス HTML から記事詳細の URL を取得する
        page_data = self.article_list_page_parser.parse(response.text)
        article_url_list = page_data.article_url_list

        # 次ページのリンクがあれば取得する
        while page_data.next_page_link:
            print(f'Accessing to {page_data.next_page_link}...')
            response = requests.get(page_data.next_page_link)
            time.sleep(10)
            page_data = self.article_list_page_parser.parse(response.text)
            article_url_list += page_data.article_url_list

        return article_url_list

    def crawl_article_detail_page(self, url: str) -> ArticleDetailPageParser.ArticleDetailData:
        """
        記事詳細ページをクロールして記事のデータを取得する
        """
        # 記事詳細へアクセスする
        print(f"Accessing to {url}...")
        response = requests.get(url)
        response.raise_for_status()

        time.sleep(10)
        # レスポンス HTML から記事の情報を取得する
        return self.article_detail_page_parser.parse(response.text)


def collect_lapras_note_articles_usecase(crawler: LaprasNoteCrawler):
    """
    LAPRAS NOTE の記事のデータを全て取得してファイルに保存する
    """
    print("Start crawl LAPRAS NOTE.")
    article_list = crawler.crawl_lapras_note_articles()

    output_json_path = "./articles.json"
    with open(output_json_path, mode="w") as f:
        print(f"Start output to file. path: {output_json_path}")
        article_data = [dataclasses.asdict(d) for d in article_list]
        json.dump(article_data, f)
        print("Done output.")

    print("Done crawl LAPRAS NOTE.")


if __name__ == '__main__':
    collect_lapras_note_articles_usecase(LaprasNoteCrawler(
        article_list_page_parser=ArticleListPageParser(),
        article_detail_page_parser=ArticleDetailPageParser(),
    ))

コードはこちらに置いています。
https://github.com/Chanmoro/lapras-note-crawler/blob/master/advanced/crawler.py

このように parser, crawler, usecase の3つの層に分離したことで、先に紹介した問題に対応するために例えば以下のように変更を入れるというのがより明確になります。

• クロール先の HTML 構造が変化する
  • parser のレイヤーでバリデーションを入れる、もしくは定点観測のテストを書く
• クロール先の障害や一時的なエラー
  • crawler のレイヤーで状況に応じた例外や戻り値を返す
  • usecase のレイヤーでリトライやフローの分岐をする
• id が変化する
  • crawler からの戻り値に id が変化したことが分かる情報を付加する
  • usecase のレイヤーで取り込み済みのデータの突合と更新ロジックを処理する

まとめ

さて、今回の記事では「クローラーの作り方 - Advanced 編」と題して、クローラーを継続的に運用する際によく発生する問題と、それらに対応するためにどんなクローラーの設計にしておくと開発がしやすいかという内容について書きました。

今回紹介したのは分かりやすさのためのほんの一例なので、実際にはクロール先サービスの特性や、クロールしたデータを使いたいサービスの要件を満たすために設計を更に工夫していく必要があると思います。

しかしこのような設計はクローラーに限ったものではなく、外部サービスや API、ライブラリとのデータ連携があるシステムの一般的なデータモデリングの範囲の話がほとんどなので、クローラーを開発すればするほど、実はクローラー特有の設計というのはあまりないんじゃないか？という気がしてきています。

前回の クローラーの作り方 - Basic 編 から続いて2回に渡って、これまで僕がクローラーを開発してきた経験をベースとして、クローラー開発の実際について紹介してきました。

いまクローラー開発に困っている方、これからクローラーの開発をしてみたい方のお役にたてると嬉しいです！

それでは皆さんもよいクローラー開発ライフをお楽しみください！