[PYTHON] Benachrichtigen Sie LINE über Informationen zum Zugbetrieb

Was für eine App?

Löschen Sie die Betriebsinformationen aus Yahoo! Route Information und benachrichtigen Sie LINE.

Quelle (GitHub) LINE Notify

Schaben

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


class NotFoundElementError(Exception):
    """Fehler, wenn das Element nicht vorhanden ist"""


class Collecter:
    """Sammlungsklasse"""
    def __init__(self):
        """Umgebungsvariablen abrufen"""
        try:
            self.urls = list(os.environ['TRAIN_URLS'].split())
        except KeyError:
            raise NotFoundElementError('url get failed!')

    def format_train_info(self, info, err_trains):
        """Formung von Betriebsinformationen

        @param:
          info [ <str:Route>, <str:Einzelheiten> ]
          err_trains [ <str:URL für Fehler bei Betriebsinformationen> ]
        @return:
          train_info <str:Betriebsinformationen>
        """
        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):
        """Sammlung von Betriebsinformationen

        @return:
          format_train_info(train_info, err_trains) <str:Betriebsinformationen>
        """
        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)

Ich habe versucht, die parallele Aufgabenverarbeitung per Multithread zu implementieren. Infolge von GET-Anfragen für 6 URLs parallel wurde es ungefähr 1 Sekunde schneller.

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

Aufgrund des Overheads ist es möglicherweise nicht möglich, die Geschwindigkeit zu erhöhen. Daher muss die Verwendung ermittelt werden.

LINE benachrichtigen

import os
import requests


class Line:
    """LINE-Benachrichtigungsklasse"""
    def __init__(self):
        """Umgebungsvariablen abrufen"""
        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):
        """Erfolgreiche Sammlung

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

    def send_error(self, err_msg):
        """Sammlungsfehler

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

Schließlich

Für Selen müssen Sie WebDrive herunterladen, aber Beautiful Soup ist einfach zu verwenden, indem Sie einfach die Bibliothek installieren. Es ist schneller als Selenium, da es den Browser nicht bedient. Selen ist für dynamische Websites mit JavaScript bequemer. Grundsätzlich ist es besser, Beautiful Soup und teilweise Selen zu verwenden.

Referenz

concurrent.futures - Parallele Taskausführung