Python&LINE Notifyでサイトの更新をLINEへ自動通知する

環境

  • Windows 10
  • Python 3.9.4
  • VSCode

使用するソースコード

以下の公開リポジトリに置いています

GitHub - masayan1126/tao-py-py: 作業自動化用 Python ライブラリ
作業自動化用 Python ライブラリ. Contribute to masayan1126/tao-py-py development by creating an account on GitHub.

LINE Notifyトークンの取得

LINE Notifyにログインし、マイページでアクセストークンを発行します

LINE Notify
LINE NotifyはGitHub,IFTTT,MackerelなどのWebサービスからの通知を、LINEで受信することが出来る便利なサービスです。

トークン名を入力後、通知先を選択し、発行

次画面でトークンが表示されるので控える

通知用のサンプルプログラムの作成

ここまでで固定のメッセージがプログラムからLINEへ送信できるかチェックしてみてください

import requests
from packages.my_rss.config import CONFIG

def send_line_notify(notification_message):
    # 先ほど取得したトークンをconfig.pyから読み込み
    LINE_NOTIFY_TOKEN = CONFIG["LINE_NOTIFY_TOKEN"]
    LUNE_NOTIFY_URL = "https://notify-api.line.me/api/notify"
    headers = {"Authorization": f"Bearer {LINE_NOTIFY_TOKEN}"}
    data = {"message": f"{notification_message}"}
    requests.post(LUNE_NOTIFY_URL, headers=headers, data=data)

send_line_notify("サイトが更新されました。")

サイトの更新を検知して通知できるようにする

プログラムの作成

cronで定期実行し、実行時に前回のサイトの情報と今回のサイトの情報を比較し、変換があった場合に更新があった旨をLINEへ通知します。

なお、今回はテキストファイルに、前回のサイト情報を保持する方法をとりたいと思います。

やることは以下の3ステップです。

  1. BeautifulSoupもしくは、Selenium等でサイト情報(例えば、新着情報箇所のhtml)を取得
  2. テキストから、前回のサイト情報を取得
  3. 1.と2.を比較して、一致していなければ更新があったことになるので、その場合はテキストを更新して、LINE通知する
masayan
masayan

本記事に関係のある3.についてのみ詳細に記載します。1.や2.については以下の記事を参照ください。また、Line Notifyで使用するトークンや対象のサイトurlなどはすべてconfig.pyやenv.pyに記載するようにし、.gitignoreに追加することでgit管理対象外としています。

【Python】スクレイピング: SeleniumとBeautiful Soupの主なメソッド一覧
本記事では、Pythonでスクレイピングをする際によく使用するSeleniumとBeautiful Soupの主なメソッド一覧について紹介しています
PythonのSeleniumでWebサイトのログイン操作を自動化する
本記事では、PythonのSeleniumでWebサイトのログイン操作を自動化するための方法について紹介しています。効率化につながりとても便利なのでぜひ活用してみてください

ユーケースの呼び出し

packages\my_rss\main.py

from packages.my_rss.Application.rss_notification_usecase import RssNotificationUsecase
from shared.Domain.Notification.notification import Notification
from packages.my_rss.env import ENV
from packages.my_rss.config import CONFIG

LINE_NOTIFY_URL = ENV["LINE_NOTIFY_URL"]
site_url = ENV["SITE_URL"]
message = "サイトが更新されました。" "\n" f"{site_url}"
LINE_NOTIFY_TOKEN = CONFIG["LINE_NOTIFY_TOKEN"]

notification = Notification(LINE_NOTIFY_URL, message, LINE_NOTIFY_TOKEN)

RssNotificationUsecase(
    notification=Notification(LINE_NOTIFY_URL, message, LINE_NOTIFY_TOKEN)
).notify_to_line()
ユースケースからラインへの通知用のサービスを呼び出す

packages\my_rss\Application\rss_notification_usecase.py

from time import sleep
from shared.Domain.Notification.line_notification_service import LineNotificationService
from shared.Domain.Notification.notification import Notification
from shared.Domain.Scraping.i_html_analyzer import IHtmlAnalyzer
from shared.Domain.Scraping.soup_factory import SoupFactory
from shared.Domain.Text.text_file_service import TextFileService
from shared.Domain.FileSystem.x_file_system_path import XFileSystemPath
from shared.Domain.String.xstr import XStr
from shared.Domain.Text.x_text import XText
from shared.Domain.Url.x_url import XUrl
from shared.di_container import DiContainer
from shared.i_factory import IFactory
from shared.Domain.Log.x_logger import XLogger
from selenium.common.exceptions import SessionNotCreatedException
from packages.my_rss.env import ENV

class RssNotificationUsecase:
    def __init__(self, notification: Notification):
        self.notification = notification

    def notify_to_line(self) -> int:
        # 1.BeautifulSoupもしくは、Selenium等でサイト情報(例えば、新着情報箇所のhtml)を取得
        # 2.テキストから、前回のサイト情報を取得
        
        # 3.1.と2.を比較して、一致していなければ更新があったことになるので、その場合はテキストを更新して、LINE通知する
        if prev != current:
            content = TextFileService(x_text=XText(filepath)).write(
                content=[current],
                is_overwrite=True,
                encoding="UTF-8",
                needs_indention=False,
            )

            return LineNotificationService(self.notification).send()
通知用のサービスが通知用のクラスを受け取って通知する

shared\Domain\Notification\line_notification_service.py

import requests
from shared.Domain.Notification.notification import Notification

class LineNotificationService:
    def __init__(self, notification: Notification):
        self.notification = notification

    def send(self) -> int:
        headers = {"Authorization": f"Bearer {self.notification.token()}"}
        data = {"message": f"{self.notification.message()}"}
        res = requests.post(
            self.notification.destination_url(), headers=headers, data=data
        )
        return res.status_code

通知クラス

shared\Domain\Notification\notification.py

class Notification:
    def __init__(self, destination_url: str, message: str, token: str = ""):
        self._destination_url = destination_url
        self._message = message
        self._token = token

    def destination_url(self) -> str:
        return self._destination_url

    def message(self) -> str:
        return self._message

    def token(self) -> str:
        return self._token

Cronに設定して定期実行

あとは、cronに上記プログラムを定期実行できるように設定するだけです。私は、Xserverを個人的に使用しているのでこちらを使用します。

Xserverのcron設定については以下の記事で詳細に説明していますので、割愛します。

XserverでCronを使用してPythonプログラムを定期実行する
本記事では、XserverでCronを使用してPythonプログラムを定期実行するについて紹介しています。

Python学習におすすめの書籍

独習Python/山田祥寛【3000円以上送料無料】
bookfan 1号店 楽天市場店
¥ 3,300(2023/01/31 18:22時点)
タイトルとURLをコピーしました