LINEで送信予約はできないのですか？
企業向けのLINE公式アカウントでは可能ですが、個人でできなければ作ってしまいましょう！
LINEで送信予約を実装するツール
LINEで送信予約を実装するにはLINE NotifyというAPIサービスを利用します。
APIとは、Application Programming Interface（アプリケーション プログラミング インターフェース）といって、Webサービスやソフトウェアの機能を、別のWebサービスやソフトウェアから呼び出せるようにしたものです。LINE Notifyは、自分、もしくは自分が入っているグループラインにメッセージを送るために使用します。下記公式サイトから設定を行いましょう。
LINEアクセストークンの取得
LINE Notifyの登録には、LINEアカウントで連携しているメールアドレスとパスワードを使用します。QRコードからもログインできます。ログインができたら、WEBページ右上に登録したあなたの名前が表示されますので、名前の部分をクリックします。そうすると、「マイページ・登録サービス管理・ログアウト」と表示されますので、そのなかのマイページをクリックします。
すると次のような画面が出てきますので、「トークンを発行する」をクリックします。
次に出てくる画面で通知の際に表示されるトークン名を記入して、今回設定するグループラインを選択したら「発行する」をクリックします。発行されたトークンは後で使用するので、忘れないようにどこかにメモをしておきましょう。
また今回はグループラインを使用しますので、設定したグループラインにLINE Notifyを招待しておきます。これでLINEの方の準備は完了です。
アプリケーションの概要
今回のアプリケーションは、Excelのファイルに作成したスケジュール表をもとに、指定した時間になると指定した内容のメッセージがグループラインに送られるという仕様で作成していきます。
実行環境の整備
まず、必要なライブラリをインストールします。
pip install pandas requests apscheduler openpyxlPythonファイルを作成し、同じディレクトリpython-line内に今回のプログラムで使用するExcelファイルを配置します。Pythonファイルはline.py、Excelファイルはschedule.xlsxとします。
ライブラリのインポート
次にline.pyにプログラムを書いていきます。
まず、インストールしたライブラリをインポートします。
import pandas as pd
import requests
from apscheduler.schedulers.blocking import BlockingScheduler # タスクスケジュールの管理
from apscheduler.jobstores.base import JobLookupError
from datetime import datetime
import osここでの JobLookupError は新しくスケジュールを設定する際に、すでに設定されている古いスケジュールを削除しようとしたとき、そのスケジュールがすでに存在しない場合に発生するエラーでそれを対策するためにインポートしておきます。
各種変数の設定
次に取得したLINEアクセストークンを 変数 LINE_NOTIFY_TOKEN に代入します。
‘YOUR_LINE_NOTIFY_TOKEN’ の部分にあなたのグループラインのアクセストークンを間違いないように記述して下さい。
# LINE Notifyのアクセストークン
LINE_NOTIFY_TOKEN = 'YOUR_LINE_NOTIFY_TOKEN'使用するExcelファイルのパスを変数に代入します。
# Excelファイルのパス
excel_file_path = 'schedule.xlsx'あとで利用するスケジュールを管理するリストを作成します。
# スケジュールを管理するリスト
scheduled_jobs = []メッセージを送信するプログラム
メッセージを送信する関数 send_line_notify( ) を作成します。引数に message を受け取って各種パラメーターを設定します。データパラメーターには LINE NotifyのAPIエンドポイント、headerにLINEのアクセストークンを含め、引数 message で受け取ったメッセージを辞書オブジェクトに格納します。
# メッセージを送信する関数
def send_line_notify(message):
# 各種パラメーターを設定して送信処理 - APIエンドポイント,アクセストークン、メッセージ
response = requests.post(
'https://notify-api.line.me/api/notify',
headers={'Authorization': f'Bearer {LINE_NOTIFY_TOKEN}'},
data={'message': message}
)
# HTTPリクエストが失敗した場合のメッセージ表示
if response.status_code != 200:
print(f"Failed to send message: {response.status_code}, {response.text}")Excelファイルを読み込んでメッセージ送信スケジュールを設定する
次にExcelファイルを読み込んでメッセージを送信を管理する関数 read_excel_data( ) を作成します。引数に scheduler を受け取ります。これはプログラムの実行時にインスタンス化するスケジューラーのオブジェクトです。
pandasの read_excel( ) 関数でExcelファイルを読み込んだあとにfor文で必要な列のデータを抽出します。抽出したデータは文字列で入力されているのでdatetimeモジュールを使って日付や時間を扱うためのdatetimeオブジェクトに変換します。その後に必要なパラメーターをセットし、 scheduler.add_job( ) メソッドでタスク実行スケジュールを設定します。
また例外処理などは最小限にしています。ご了承下さい。
# Excelファイルを読み込んでスケジュールを設定する関数
def read_excel_data(scheduler):
global scheduled_jobs
# 残っている古いデータを削除する
for job_id in scheduled_jobs:
try:
scheduler.remove_job(job_id)
except JobLookupError:
print(f"Job ID {job_id} not found.")
scheduled_jobs.clear()
# Excelファイルの読み込み
df = pd.read_excel(excel_file_path)
# 必要なデータを抽出
for _, row in df.iterrows():
send_date = row['日付']
send_time = row['送信予定時刻']
message = row['メッセージ']
# datetimeオブジェクトに変換
schedule_time = datetime.strptime(f"{send_date} {send_time}", '%Y-%m-%d %H:%M')
print(f"Scheduling message at {schedule_time} with message: {message}")
# タスクをスケジュールに設定して管理リストに追加
job = scheduler.add_job(send_line_notify, 'date', run_date=schedule_time, args=[message])
scheduled_jobs.append(job.id)ここで先ほど作成した scheduled_jobs のリストにスケジュールの job.id を登録していきます。新しくデータを読み込む前に、順次 scheduler.remove_job() で古いデータを削除していくのですが、仮に削除する対象のデータがなくとも JobLookupError によって例外処理を行っているのでプログラムが停止することなく処理が続行されます。
ちなみに、scheduler.add_job( ) メソッドにセットするパラメーターは、実際にメッセージの送信を実行する関数 send_line_notify( ) と受け取る引数の message 、指定した日時に一度だけ実行するトリガーの ‘date’ と変換したdatetimeオブジェクトです。
Excelファイルの更新を監視
Excelファイルを読み込んでスケジュールを設定するところまではできましたが、このままではプログラムが実行されたあとにExcelファイルのスケジュールが更新されてもその内容が反映されません。
そのため、Excelファイルを監視してファイルの更新日時が変更されたときに再度Excelファイルを読み込むように処理を加えていきます。関数 monitor_file( ) を作成します。
# ファイルの更新を監視する関数
def monitor_file(scheduler):
global last_modified_time
# Excelファイルの更新日時を記録
current_modified_time = os.path.getmtime(excel_file_path)
# 更新日時が変わった場合、ファイルが変更されたと判断して処理を行う
if current_modified_time != last_modified_time:
print("File has been modified.")
#再度Excelファイルを読み込む
read_excel_data(scheduler)
# 最終更新日時を更新
last_modified_time = current_modified_time変数 current_modified_time に os.path.getmtime( ) 関数を使ってファイルの更新日時を記録します。グローバル変数に設定している last_modified_time は 最初にプログラムを実行したときに取得するExcelファイルの最終更新日時を記録する変数です。以後この変数の値同士を比較して相違があれば変更があったと判断して、再度ファイルの読み込みを行います。
プログラムの実行
今までに設定したプログラムを起動します。
# スケジューラーをインスタンス化
scheduler = BlockingScheduler()
# 初回読み込み
read_excel_data(scheduler)
# ファイルの最終更新日時を記録
last_modified_time = os.path.getmtime(excel_file_path)
# 10秒ごとにファイルの更新をチェック
scheduler.add_job(monitor_file, 'interval', seconds=10, args=[scheduler])
# スケジューラーを開始
try:
scheduler.start()
except (KeyboardInterrupt, SystemExit):
scheduler.shutdown() ‘interval’ は指定した間隔ごとにタスクを実行するトリガータイプです。seconds=10 で間隔を10秒に設定して スケジューラーオブジェクトを関数 monitor_file( ) に渡します。
最後に scheduler.start( ) によってスケジューラーを開始しますが、ユーザーによってCtrl+C が押されたときやシステム終了のシグナルを捕捉したときは scheduler.shutdown( ) によってスケジューラーのタスクをクリアして安全に終了します。
これでプログラムは完成です。
import pandas as pd
import requests
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.jobstores.base import JobLookupError
from datetime import datetime
import os
# LINE Notifyのアクセストークン
LINE_NOTIFY_TOKEN = 'YOUR_LINE_NOTIFY_TOKEN'
# Excelファイルのパス
excel_file_path = 'schedule.xlsx'
# スケジュールを管理するリスト
scheduled_jobs = []
# メッセージを送信する関数
def send_line_notify(message):
# 各種パラメーターを設定して送信処理
response = requests.post(
'https://notify-api.line.me/api/notify',
headers={'Authorization': f'Bearer {LINE_NOTIFY_TOKEN}'},
data={'message': message}
)
if response.status_code != 200:
print(f"Failed to send message: {response.status_code}, {response.text}")
# Excelファイルを読み込んでスケジュールを設定する関数
def read_excel_data(scheduler):
global scheduled_jobs
# 残っている古いデータを削除する
for job_id in scheduled_jobs:
try:
scheduler.remove_job(job_id)
except JobLookupError:
print(f"Job ID {job_id} not found.")
scheduled_jobs.clear()
# Excelファイルの読み込み
df = pd.read_excel(excel_file_path)
# 必要なデータを抽出
for _, row in df.iterrows():
send_date = row['日付']
send_time = row['送信予定時刻']
message = row['メッセージ']
# datetimeオブジェクトに変換
schedule_time = datetime.strptime(f"{send_date} {send_time}", '%Y-%m-%d %H:%M')
print(f"Scheduling message at {schedule_time} with message: {message}")
# タスクをスケジュールに設定して管理リストに追加
job = scheduler.add_job(send_line_notify, 'date', run_date=schedule_time, args=[message])
scheduled_jobs.append(job.id)
# ファイルの更新を監視する関数
def monitor_file(scheduler):
global last_modified_time
# Excelファイルの更新日時を記録
current_modified_time = os.path.getmtime(excel_file_path)
# 更新日時が変わった場合、ファイルが変更されたと判断して処理を行う
if current_modified_time != last_modified_time:
print("File has been modified.")
#再度Excelファイルを読み込む
read_excel_data(scheduler)
# 最終更新日時を更新
last_modified_time = current_modified_time
# スケジューラーをインスタンス化
scheduler = BlockingScheduler()
# 初回読み込み
read_excel_data(scheduler)
# ファイルの最終更新日時を記録
last_modified_time = os.path.getmtime(excel_file_path)
# 10秒ごとにファイルの更新をチェック
scheduler.add_job(monitor_file, 'interval', seconds=10, args=[scheduler])
# スケジューラーを開始
try:
scheduler.start()
except (KeyboardInterrupt, SystemExit):
scheduler.shutdown()
無事、メッセージが送れました!
おわりに
このアプリケーションは、常時起動していないとメッセージの送信ができないため、実装にはいくつかの工夫が必要です。例えば、Bashでのnohupコマンドを使用してバックグラウンドで動作させたり、Windowsのタスクスケジューラーを活用してパソコン起動時やスリープ復帰時にアプリケーションを自動的に起動させる方法があります。またクラウドサーバーを利用することで、常に稼働している環境でアプリケーションを動かし続けることも可能なのでいろいろとカスタマイズしてみて下さい。
最後までご覧頂きありがとうございました。
