Notifier LINE des informations sur l'exploitation du train

Quel genre d'application?

Gratter les informations de fonctionnement de Yahoo! Route Information et notifier LINE.

Source (GitHub) LINE Notify

Grattage

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


class NotFoundElementError(Exception):
    """Erreur lorsque l'élément n'existe pas"""


class Collecter:
    """Classe de collection"""
    def __init__(self):
        """Obtenir des variables d'environnement"""
        try:
            self.urls = list(os.environ['TRAIN_URLS'].split())
        except KeyError:
            raise NotFoundElementError('url get failed!')

    def format_train_info(self, info, err_trains):
        """Mise en forme des informations de fonctionnement

        @param:
          info [ <str:Route>, <str:Détails> ]
          err_trains [ <str:URL d'échec des informations d'opération> ]
        @return:
          train_info <str:Informations sur le fonctionnement>
        """
        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!'
        else:
            train_info += 'This is Error url!'
            for url in err_trains:
                train_info += '\n' + url
        return train_info

    def get_train_info(self):
        """Collecte d'informations sur les opérations

        @return:
          format_train_info(train_info, err_trains) <str:Informations sur le fonctionnement>
        """
        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)

J'ai essayé d'implémenter le traitement de tâches parallèles par multi-thread. À la suite des requêtes GET pour 6 URL en parallèle, il est devenu environ 1 seconde plus rapide.

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

Il peut ne pas être possible d'augmenter la vitesse en raison de la surcharge, il est donc nécessaire de déterminer l'utilisation.

Notifier LINE

import os
import requests


class Line:
    """Classe de notification LINE"""
    def __init__(self):
        """Obtenir des variables d'environnement"""
        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):
        """Collection réussie

        @param:
          info <str:Informations sur le fonctionnement>
        """
        requests.post(self.url,
                      headers=self.headers,
                      params={'message': info})

    def send_error(self, err_msg):
        """Échec de la collecte

        @param:
          err_msg <str:Message d'erreur>
        """
        requests.post(self.url,
                      headers=self.headers,
                      params={'message': err_msg})

finalement

Pour Selenium, vous devez télécharger WebDrive, mais Beautiful Soup est facile à utiliser simplement en installant la bibliothèque. Il est plus rapide que Selenium car il ne fait pas fonctionner le navigateur. Selenium est plus pratique pour les sites dynamiques utilisant JavaScript. Fondamentalement, Beautiful Soup est utilisé, et le sélénium semble être utilisé partiellement.

référence

concurrent.futures - Exécution de tâches parallèles