電車の運行情報をLINEに通知する

14349 ワード
Python LineNotify BeautifulSoup スクレイピング Python テキストリンク

どんなアプリ？

Yahoo!路線情報から運行情報をスクレイピングし、LINEに通知する。

ソース(GitHub)
LINE Notify

スクレイピングする


import os
from concurrent.futures import ThreadPoolExecutor
import requests
from bs4 import BeautifulSoup


class NotFoundElementError(Exception):
    """要素が存在しない時のエラー"""


class Collecter:
    """収集クラス"""
    def __init__(self):
        try:
            with open('train_urls.txt', 'r') as f:
                self.urls = f.read().splitlines()
        except FileNotFoundError:
            raise NotFoundElementError('url get failed!')

    def format_train_info(self, info, err_trains):
        """運行情報整形

        @param:
          info [ <str: 路線>, <str: 詳細> ]
          err_trains [ <str: 運行情報失敗URL> ]
        @return:
          train_info <str: 運行情報>
        """
        train_info = '\n'
        for i in info:
            try:
                lead, _, detail = i[1].strip('\n').split('\n')
                train_info += '{0}\n{1}\n{2}\n\n'.format(i[0], lead, detail)
            except ValueError:
                raise ValueError('format failed!')

        if not err_trains:
            train_info += 'Collect Complete!'
            return train_info

        train_info += 'This is Error url!'
        for url in err_trains:
            train_info += '\n' + url
        return train_info

    def get_train_info(self):
        """運行情報収集

        @return:
          format_train_info(train_info, err_trains) <str: 運行情報>
        """
        pool = ThreadPoolExecutor()
        res_list = pool.map(requests.get, self.urls)

        train_info, err_trains = [[], []]
        for res in res_list:
            try:
                res.raise_for_status()
            except requests.exceptions.RequestException:
                err_trains.append(res.url)
                continue

            bs_obj = BeautifulSoup(res.text, 'lxml')
            try:
                route = bs_obj.h1.text
                detail = bs_obj.find(id='mdServiceStatus').text
            except AttributeError:
                err_trains.append(res.url)
            else:
                train_info.append([route, detail])

        if not train_info:
            raise NotFoundElementError('collect failed!')
        return self.format_train_info(train_info, err_trains)

マルチスレッドによる並列タスク処理を実装してみた。
6つのURLを並列でGETリクエストした結果、約1秒速くなった。

pool = ThreadPoolExecutor()
res_list = pool.map(requests.get, self.urls)

オーバーヘッドによる高速化が見込めない場合もあるので、使い所を見極める必要がある。

LINEに通知する

import os
import requests


class Line:
    """LINE通知クラス"""
    def __init__(self):
        """環境変数取得"""
        try:
            self.url = os.environ['LINE_API_URL']
            self.token = os.environ['LINE_API_TOKEN']
            self.headers = {'Authorization': 'Bearer ' + self.token}
        except KeyError as err:
            raise KeyError(err)

    def send_success(self, info):
        """収集成功

        @param:
          info <str: 運行情報>
        """
        requests.post(self.url,
                      headers=self.headers,
                      params={'message': info})

    def send_error(self, err_msg):
        """収集失敗

        @param:
          err_msg <str: エラーメッセージ>
        """
        requests.post(self.url,
                      headers=self.headers,
                      params={'message': err_msg})

最後に

Seleniumの場合、WebDriveのダウンロードが必要だが、Beautiful Soupはライブラリのインストールだけで手軽に使用できる。
ブラウザ操作も行わないため、Seleniumより動作が速い。
JavaScriptによる動的サイトでは、Seleniumの方が便利。
基本はBeautiful Soupを使用し、部分的にSeleniumを使用するのが良さそう。

参考

concurrent.futures -- 並列タスク実行