ScrapyのItem Pipelineで複数Spiderの完了をSlackに通知する

### できたもの

![スクリーンショット 2022-01-24 0.47.12.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/119757/b31a93f1-ddd1-9cd2-ee40-091da3f5b0e3.png)
![スクリーンショット 2022-01-24 0.45.03.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/119757/43624bb1-8392-aa61-40fd-f2b2ba4dd2df.png)


### 環境
* Python: 3.7.10
* Scrapy: 2.5.1


### 背景

個人的にいくつかのクローラーをScrapyで運用していて、クロールが終わったりエラーになったら通知してほしいなと思いました。
ぱっと思いついたやり方は以下の通りでした。

* （クロール結果がS3に出力されるので）Lambdaで出力を検知して通知する
    * エラーでS3に何も出力されなかったら検知できない
* Scrapyの各Spider（`spiders/*.py`）内に通知処理を書く
    * [Signals](https://docs.scrapy.org/en/latest/topics/signals.html)という機能を使えばできそう
    * エラー検知用のメソッドがある
    * 同じScrapyプロジェクト内に複数のSpiderを動かしていて、同じ処理を複数の場所で書きたくない
* Scrapyの[Item Pipeline](https://docs.scrapy.org/en/latest/topics/item-pipeline.html)を使う
    * ここに書けばすべてのSpiderに適用される
    * エラー検知用のメソッドはないが、自分で書けばよい

ちゃんとエラー検知したいなら専用のメソッドがある`Signals`がよいと思いますが、何ヶ月か運用してエラーは起きなかったので、今回は比較的簡単な`Item Pipeline`を使いました。

なお、[公式リファレンス](https://docs.scrapy.org/en/latest/topics/item-pipeline.html)にもあるように、本来はアイテムごとのバリデーションやDBへの保存処理を行うものです。

また、PythonからSlackに通知するやり方はいろんな記事があるので割愛しますが、このあたりを参考にしました。

* [SlackのIncoming Webhooksを使い倒す - Qiita](https://qiita.com/ik-fib/items/b4a502d173a22b3947a0)
* [Creating rich message layouts - Slack API](https://api.slack.com/messaging/composing/layouts)
* [Message Formatting
 | Slack](https://api.slack.com/docs/messages/builder)


### 実装方法

#### pipelines.py に処理を記述

Scrapyのプロジェクト内に`pipelines.py`があると思うので、そこに通知処理を書きます。

```python:pipelines.py
from itemadapter import ItemAdapter
import json
import requests
import traceback


def post_slack(message_json):
    webhook_url = "SLACK_WEBHOOK_URL"
    requests.post(webhook_url, json.dumps(message_json))

class [ProjectName]Pipeline:
    def process_item(self, item, spider):
        return item

    def close_spider(self, spider):
        ```spiderが閉じたら起動する処理```

        # クロールが完了したらSlackに通知する
        try:
            stats = spider.crawler.stats.get_stats()
            message_json = {
                "text": f"{spider.name}のクロールが完了しました！",
                "attachments": [
                    {
                        "text": f"レコード数：{stats['item_scraped_count']} 件"
                    }
                ]
            }

        # エラーなら内容を通知する
        except:
            message_json = {
                "text": f"*{spider.name}のクロールでエラーが発生しました！*",
                "attachments": [
                    {
                        "color": "danger",
                        "text": traceback.format_exc()
                    }
                ]
            }
        post_slack(message_json)
```

* `close_spider`はデフォルトで用意されているメソッドで、ここに処理を書くとSpiderが閉じたときに実行されます
* `spider.crawler.stats.get_stats()`はクローラー内のいろんな統計情報が保存されている辞書です
    * 自分でキーを追加することもできるようです
    * 参考：[Scrapyで統計情報を記録する - nullpo.io](https://www.nullpo.io/2020/05/18/python-scrapy-stats/)
* `traceback.format_exc()`はエラー内容を表示してくれる関数です
* **注意：上の書き方だと、最初のアイテムのスクレイピングが成功して途中で失敗するケースは拾えません**
    * 主なエラーケースは`KeyError: 'item_scraped_count'`になります
    * `item_scraped_count`は最初のアイテムの処理が終わった段階でキーが作成されるので、最初が正常に処理できたら`KeyError`になりません
    * 今回は、サイト構造が全体的に変わって何も取得できないケースが拾えればよかったので、上のような実装にしました


#### settings.pyへの追記

上の処理を反映するために、クラス名を下のように`settings.py`に追記します（デフォルトだとコメントアウトされているはず）。
`300`などの数字は優先度を表していて、0-1000の数字が小さい順に適用されるようです。
参考：[Settings - Scrapy](https://docs.scrapy.org/en/latest/topics/settings.html#item-pipelines)

```python:settings.py
ITEM_PIPELINES = {
   "[project_name].pipelines.[ProjectName]Pipeline": 300,
}
```

### 感想

これで自分用のSlackに完了通知が来るようになり、ちゃんと動いてくれてるなとクローラーを労う気持ちが生まれました。