pyppeteer + HeadlessChromeで社用ツールを自動化する #Python

この記事は、LIFULLその2 Advent Calendar 2018の6日目の記事です。

※注意: 例示するツールとしてチャットワークを使っていますが、チャットワーク公式のAPIが公開されているのでわざわざクローリングする必要はありません。

勤怠管理や稟議など、けっこう面倒なシステムって多いですよね？特に社用ツールはやや古いものが多く、（リプレースの計画も進んでいるようなのですが）触っていて楽しいものではなく、私の場合はどうしても後回しにして、楽しいPython実装を優先してしまうことが多いです。同じように「好きなものは先に食べてしまう」タイプの人って多いですよね？

以前の上司もそういうタイプだったので、「俺もよく忘れるんだよｗけど俺よりひどいから気をつけて」って感じで私も気にしてなかったのですが、最近上司が変わって（当然）注意されてしまいました。

↑怒られたときの例。その節は申し訳ありませんでした。

というわけで今回はpyppeteerという、PythonからHeadlessChromeを呼び出せるツールを使えば、簡単に自動化できるものもあるんじゃないか？という実験をシェアします。スクリプトから自動化できれば、自動でjsonやyamlファイルを生成して入力することができ、実装作業に近づくためモチベーションも上がるはずです。

サンプル実装

サンプルとして次のような、チャットワークにメッセージを送信するコードを作ってみました。

チャットワークにログインする
メッセージを送信する

async/await使っているのは、pyppeteerの元になったライブラリがnode.jsのもので、pyppeteerの関数もそちらのAPIを再現する形で実装されており、そちらに合わせるほうが楽だったからです。

main.py

import asyncio
from src import crawler

async def main():
    async with crawler.ChatworkSession("your_email", "your_password") as s:
        await s.post_message("こんにちは", room_id="********")

if __name__ == "__main__":
    asyncio.get_event_loop().run_until_complete(main())

こちらのようにPageのtypeメソッドで入力し、clickメソッドでボタンを押して送信するというものが多いです。非同期的に行ってしまうため、「待つ」動作が必要になることもあります。

ちなみに、HeadlessChromeはライブラリの初回起動時に自動でインストールされます。こうしたツールを作るときに非常に楽ですね。

他の社用ツールもほぼこの流れで対策できるはずです。

src/crawler.py

from pyppeteer import launch
import asyncio


class ChatworkSession(object):
    def __init__(self, login_email, login_password):
        self._login_email = login_email
        self._login_password = login_password

    async def __aenter__(self):
        self._browser = await launch(headless=True)
        self._page = await self._browser.newPage()
        await self._page.goto('https://www.chatwork.com/login.php')
        await self._page.type('input[name=email]', self._login_email)
        await self._page.type('input[name=password]', self._login_password)
        await self._page.click('input[name=login]')
        await asyncio.wait([self._page.waitForNavigation()]) # 読み込むまで待つ
        return self

    async def post_message(self, message, room_id):
        await self._page.goto(f"https://www.chatwork.com/#!rid{room_id}")
        await self._page.type('textarea[id=_chatText]', message)
        await self._page.click("div[id=_sendButton]")

    async def __aexit__(self, exc_type, exc, tb):
        await self._browser.close()

懸念や難しい点

async/awaitの実装にまだ慣れません…。また、きちんとテストされたコードではないため、awaitの実行順によってはエラーが起きると思います。同期的なメソッドでラップしてしまうのもありかもしれません。

参考

asyncio公式ドキュメント
pyppeteer公式ドキュメント
puppeteer （node.jsからHeadlessChromeを扱うライブラリ。同じことができるはずです）