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

環境
使用するソースコード
LINE Notifyトークンの取得
通知用のサンプルプログラムの作成
サイトの更新を検知して通知できるようにする
1. プログラムの作成
2. Cronに設定して定期実行
Python学習におすすめの書籍

環境

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ステップです。

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

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