Scrapy のクロール結果からファイルを Cloud Storage に保存したい

Scrapy のドキュメントにはパイプラインでファイルを保存するという項目があります。

Downloading and processing files and images

Scrapy は S3 と Google Cloud Storage 向けのクラスをありがたいことに用意してくれているので基本は設定で各種パラメタを定義して機能を有効にするだけでOKです。
ただしこの機能は Client の "デフォルトのアカウント"を使用するため、"サービスアカウントのJSON"は指定できません。

そこで、"サービスアカウントのJSON"を指定できるようにいくつかオーバーライドが必要でした。

対象環境

Python : Anaconda + 3.7

今回も引き続き Jupyter notebook から動かしてます。

準備

以下のコマンドで google-cloud-storage の Python API をインストールします。

noteboook

!pip install --upgrade google-cloud-storage

パイプラインの実装

以下のような手順でパイプラインを実装します。

1. GCSFilesStore のオーバーライド

GCSFilesStoreを継承したMyGCSFilesStore を作成してコンストラクタでjsonを渡せるようにしておきます。

class MyGCSFilesStore(GCSFilesStore):
    def __init__(self, uri, cred_file):
        from google.cloud import storage
        client = storage.Client.from_service_account_json(cred_file)
        bucket, prefix = uri[5:].split('/', 1)
        self.bucket = client.bucket(bucket)
        self.prefix = prefix

コンストラクタの cred_file に json ファイルのパスが渡ってくる予定です。
Client.from_service_account_json(cred_file) でサービスアカウントのJSONを指定したインスタンス生成が行われます。

2. ImagesPipeline のオーバーライド

今回は画像ファイルをターゲットとして、ImagesPipeline をオーバーライドしてみます。

2-1. コンストラクタ

今回はコンストラクタで settings.py の設定から jsonファイルのパスを取得するようにしてみました。

class GCSFilePipeline(ImagesPipeline):
    def __init__(self, store_uri, download_func=None, settings=None):
        self.cred_file = settings.get('GOOGLE_ACCOUNT_JSON')
        super(GCSFilePipeline, self).__init__(store_uri,download_func,settings)

2-2. FileStoreの生成

Scrapy では保存先のスキーマに応じて FileStore を切り替える作りですが、今回は決め打ちでOK のため先の手順で作った MyGCSFilesStore をダイレクトに生成します。

    def _get_store(self, uri):
        return MyGCSFilesStore(uri, self.cred_file)

MyGCSFilesStore 生成時に self.cred_file の jsonファイルのパスを渡しています。

2-3. フィールド名の変更

Scrapy の実装では image_urls や file_urls が画像URLで保存後のURLは images や files となっていますが、これを変更するために get_media_requests メソッドと item_completed メソッドをオーバーライドします。

    def get_media_requests(self, item, info):
        for file_url in item['sample_image']:
            yield scrapy.Request(file_url)

    def item_completed(self, results, item, info):
        image_paths = [x['path'] for ok, x in results if ok]
        if not image_paths:
            raise DropItem("Item contains no images")
        item['sample_image'] = image_paths
        return item

今回は sample_image というフィールドに、元のURLが入っており、画像をGCSに保存後はバケットのパスに差し替える、という処理をしています。

結果のクラス

全体的には以下のようなファイルになりました。

gscfilepipeline.py

import scrapy
from scrapy.pipelines.files import GCSFilesStore
from scrapy.pipelines.images import ImagesPipeline
from scrapy.exceptions import DropItem

class MyGCSFilesStore(GCSFilesStore):
    def __init__(self, uri, cred_file):
        from google.cloud import storage
        client = storage.Client.from_service_account_json(cred_file)
        bucket, prefix = uri[5:].split('/', 1)
        self.bucket = client.bucket(bucket)
        self.prefix = prefix

class GCSFilePipeline(ImagesPipeline):
    def __init__(self, store_uri, download_func=None, settings=None):
        self.cred_file = settings.get('GOOGLE_ACCOUNT_JSON')
        super(GCSFilePipeline, self).__init__(store_uri,download_func,settings)

    def _get_store(self, uri):
        return MyGCSFilesStore(uri, self.cred_file)

    def get_media_requests(self, item, info):
        for file_url in item['sample_image']:
            yield scrapy.Request(file_url)

    def item_completed(self, results, item, info):
        image_paths = [x['path'] for ok, x in results if ok]
        if not image_paths:
            raise DropItem("Item contains no images")
        item['sample_image'] = image_paths
        return item

spider の方でサイトからリンクを取得して画像のURLを item["sample_image"] に入れておくようにしてください。

設定(settings.py)の修正

設定にて上記のパイプラインを有効にするとともに、GCSFileStorage 用に設定を追加します。
先日のCloudFireStorePipelineより優先度を上げて、ファイルの取得が終わってから(保存先の画像パスに差し替えた後で) FireStore に書き込まれるようにします。

ITEM_PIPELINES = {
    'myProject.gcsfilepipeline.GCSFilePipeline': 200,
    'myProject.pipelines.CloudFireStorePipeline': 300,
}

GCS_PROJECT_ID = 'my-gcp-project'
FILES_STORE = 'gs://my-backet-path/sample_images/'
IMAGES_STORE = 'gs://my-backet-path/sample_images/'

GOOGLE_ACCOUNT_JSON = './cred/scrapy_service_account.json'

IMAGES_STORE_GCS_ACL = 'publicRead'

GCS_PROJECT_ID はそのままプロジェクト名を入れてください。

GSCの保存先のバケットとパスの設定ですが、FILES_STORE と IMAGES_STORE のどちらも定義が必要です。。。

GOOGLE_ACCOUNT_JSON にはサービスアカウントJSONのパスを指定します。

IMAGES_STORE_GCS_ACLは保存したファイルのパーミッションです。今回は外部からアクセスできるように publicRead としています。

Scrapy では設定で画像のリサイズなどにも対応しているようです。とりあえず今回はそのままのサイズで行ってみたいとおもいます。

参考サイト

継承などで処理を横取りする場合は結局、本体側のソースコードの理解がマストです。
ドキュメントからソースコードへのリンクがあればもうちょいやりやすいんですけどね。。。

以上です。

【2019年5月】Scrapy の自作パイプラインで Cloud Storage に画像を叩き込む