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. IMG_CBDABF68DF8D-1.jpeg

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

Recommended Posts

Notifier LINE des informations sur l'exploitation du train
Notifier LINE des informations d'apparence TV de votre artiste préféré
J'ai essayé de notifier les informations de retard de train avec LINE Notify
Notifier LINE des informations de localisation (Google map) avec GPS Multi-Unit SORACOM Edition
[Python] Opération d'énumération
Fonctionnement de base des pandas
Fonctionnement de base des Pandas
Exemple d'implémentation du serveur LINE BOT pour un fonctionnement réel
Je veux obtenir des informations sur le fonctionnement de Yahoo Route
Fonctionnement du filtre (Aucun, liste)
Notifier Slack de l'envoi de GitHub
Essayez LINE Notify avec Python
Extraction de dessin au trait de l'illustration
Acquisition des informations météorologiques (DarkSky)
Maîtriser OpenWRT (LEDE): LINE Notify
[Python] Notification LINE des dernières informations à l'aide de la recherche automatique Twitter
Notifier LINE de la température corporelle du thermomètre BLE avec la tarte à la râpe # 1
Avertir LINE de la température corporelle du thermomètre BLE avec la tarte à la râpe n ° 2
Bases de la théorie de l'information quantique: opération logique par code de surface (Brading)