Pythonでいくつかの通知処理サンプルを作ってみる

処理で判定をした後のアラートや処理終了時の通知などで
メールや特定のサービスに通知したいケースがちょいちょいあり、
そのたびに考えるのがめんどくさくなったので、まとめてみた。

環境

google colaboratoryにて実施。
Pythonのバージョンは以下。

Mail

まずはベタにメール通知。

SMTPサーバはGmailを想定
※Gmailを使って送信する際には通常のパスワードではなく、アプリパスワードを発行すること。
通常のパスワードを使用すると[SMTPAuthenticationError]が発生すると思います。
参考：https://www.gocca.work/python-mailerror/

import smtplib
from email.mime.text import MIMEText
from email.utils import formatdate

# 送信元のメアドと名前、パスワード（Gmailの場合はアプリパスワードを発行し使用する）
FROM_ADDRESS = '[e-mail]'
FROM_NAME = 'XXXXXX'
MY_PASSWORD = 'xxxxx'

# 送信先はリスト型にしておく
TO_ADDRESS = ['[e-mail]', '[e-mail]']

from_mail_addr = FROM_ADDRESS
_from_name = FROM_NAME
_from_addr = "{} <{}>".format(_from_name, from_mail_addr)


# 送信するメッセージを作成
## https://docs.python.org/ja/3.5/library/email-examples.html
def create_message(list_to_addr, subject, body):
    msg = MIMEText(body)
    msg['Subject'] = subject
    # msg['From'] = _from_addr
    msg['From'] = _from_name
    msg['To'] = ','.join(list_to_addr)
    # よくわからないがBCCに入れてもBCCという欄に表示されるだけなので止める
    # msg['Bcc'] = ','.join(list_bcc_addrs)
    msg['Date'] = formatdate()
    return msg

# GmaiのSMTPサーバを使って送信
def send_gmail(list_to_addr, msg):
    smtpobj = smtplib.SMTP('smtp.gmail.com', 587)
    smtpobj.ehlo()
    smtpobj.starttls()
    smtpobj.ehlo()
    smtpobj.login(FROM_ADDRESS, MY_PASSWORD)
    smtpobj.sendmail(from_mail_addr, list_to_addr, msg.as_string())
    smtpobj.close()


# 実際に送信する処理
## メッセージ作成
list_to_addr = TO_ADDRESS
subject = "＜通知サンプルですよ＞"
import datetime
dt_now = datetime.datetime.now()
# 改行は\nで設定すればよい
body = "送信内容ですよ\n改行は反映されるのでしょうか？\n{}"
body = body.format(dt_now.strftime('%Y年%m月%d日 %H:%M:%S'))

msg = create_message(list_to_addr, subject, body)

## 送信処理
send_gmail(list_to_addr, msg)

LINE

続いてLINE。
LINEに通知したい場合は大きく分けて二つ。
自分に通知するだけでいいのか、BOTとして特定のユーザに通知するか？
※通知先としては人気であるが、後者の特定のユーザに通知しようと思うと結構めんどくさい。

自分に通知

以下でルームに対してトークンを発行する
https://notify-bot.line.me/ja/

※自分のみに通知する以外に特定のグループに通知する場合もこれでOK

import requests
line_notify_api = 'https://notify-api.line.me/api/notify'

payload = {'message': send_message}
headers = {'Authorization': 'Bearer ' + line_notify_token}
line_notify = requests.post(line_notify_api, data=payload, headers=headers)

BOTで通知

LINE Bot APIを使って、フレンドの特定のユーザへプッシュすることも可能
※ただし対象ユーザのユーザIDが分かっていることが前提

こっちは以下からBOTを作成する
https://developers.line.biz/console

import json
import requests

# HTTPヘッダを設定
header = {
  'Content-Type': 'application/json',
  'Authorization': 'Bearer ' + _access_token,
}

# 送信データを設定
send_data = {
  'to': to_user_id,
  'messages': [
      {
          'type': 'text',
          'text': send_message
      }
  ]
}

# 実行
url = 'https://api.line.me/v2/bot/message/push'
res = requests.post(url, headers=header, data=json.dumps(send_data))
if(res.status_code != 200):
  print("error : [{}]".format(res.text))

Slack

チャットツールに通知をしたいというのは最近の流行りかと。

incoming-webhookとして対象チャネルにURLを作成して、POSTで投げ込めばOK

import requests
import json
requests.post(slack_url, data=json.dumps({'text': send_message}))

簡単。

Teams

基本的にはSlackと一緒。
incoming-webhookとして対象チャネルにURLを作成して、POSTで投げ込めばOK。
Slackと違うTeamsの特徴なだけですが、タイトルがつけられます

HTMLで出力されるため、\nでは改行されないのでほかの通知と一緒にする場合は置換処理を入れるといいと思います。

import requests
import json
send_message = send_message.replace('\n', '<BR>')
requests.post(teams_webhook_url, data=json.dumps({'title': subject, 'text': send_message}))

IFTTT

最後は何にでも飛ばせるようにIFTTTのWebhookに飛ばしてみる
※最終的な出力はLINEとしておく

ちょいとややこしい手順を簡単にしか記載しないので
IFTTTでWebhookを作ったことない人はあきらめて他のページを参照した方がいい。

１．New AppletでトリガーにWebhooksを指定。

２．WebhooksにはReceive a web requestしかないので、それを選択。
３．イベント名をつけてあげる。（この名前がWebhookのURLにも入るので注意）
４．アクションは好きなものを選ぶ
　とりあえずLINEという想定として、出力内容も設定する
５．アプレット名を決めてFInishを押下。

ここからがちとややこしい
６．右上のMy servicesを選択し、webhooksサービスを表示させる
※このURLに直でもいいかもしれない[https://ifttt.com/maker_webhooks]
７．Settingsを押下
※このURLに直でもいいかもしれない[https://ifttt.com/maker_webhooks/settings]

８．URLが記載されているので、そのURLをブラウザに入力。
　この画面がIFTTTのWebhookの設定画面となります。

　この設定画面に表示されているkeyとさきほどのイベント名によってIFTTTのWebhookのURLは決定されます。

　https://maker.ifttt.com/trigger/<イベント名>/with/key/

ここまでくればあとはSlackやTEAMSのWebhookと同じようにURLにPOSTで投げ込んであげればOK
パラメータはvalue1,value2,value3の３つになります

import requests
requests.post(url_ifttt, data={'value1': subject, 'value2': send_message, 'value3': ''})

IFTTTを経由することができれば、送り先はどうとでもなると思うので、好きな通知先に通知することができると思います。

まとめ

全部をまとめて、一括で通知するようにするとこんな感じになる。

import smtplib
from email.mime.text import MIMEText
from email.utils import formatdate
import requests
import json

# 送信元のメアドと名前、パスワード（Gmailの場合はアプリパスワードを発行し使用する）
FROM_ADDRESS = '[e-mail]'
FROM_NAME = 'XXXXXX'
MY_PASSWORD = 'xxxxx'

# 送信先はリスト型にしておく
TO_ADDRESS = ['[e-mail]', '[e-mail]']

def create_message(list_to_addr, subject, body):
    msg = MIMEText(body)
    msg['Subject'] = subject
    msg['From'] = FROM_NAME 
    msg['To'] = ','.join(list_to_addr)
    # よくわからないがBCCに入れてもBCCという欄に表示されるだけなので止める
    # msg['Bcc'] = ','.join(list_bcc_addrs)
    msg['Date'] = formatdate()
    return msg

def send_gmail(list_to_addr, msg):
    smtpobj = smtplib.SMTP('smtp.gmail.com', 587)
    smtpobj.ehlo()
    smtpobj.starttls()
    smtpobj.ehlo()
    smtpobj.login(FROM_ADDRESS, MY_PASSWORD)
    smtpobj.sendmail(FROM_ADDRESS, list_to_addr, msg.as_string())
    smtpobj.close()

def Notice_Mail(subject, send_message):
  list_to_addr = TO_ADDRESS
  msg = create_message(list_to_addr, subject, send_message)
  send_gmail(list_to_addr, msg)

line_notify_token = 'xxxxxxxxxxx'
def Notice_LINE_Self(send_message):
  line_notify_api = 'https://notify-api.line.me/api/notify'

  payload = {'message': send_message}
  headers = {'Authorization': 'Bearer ' + line_notify_token}
  line_notify = requests.post(line_notify_api, data=payload, headers=headers)

_access_token ='xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
def Notice_LINE_push(to_user_id, send_message):

  # HTTPヘッダを設定
  header = {
      'Content-Type': 'application/json',
      'Authorization': 'Bearer ' + _access_token,
  }

  # 送信データを設定
  send_data = {
      'to': to_user_id,
      'messages': [
          {
              'type': 'text',
              'text': send_message
          }
      ]
  }

  # 実行
  url = 'https://api.line.me/v2/bot/message/push'
  res = requests.post(url, headers=header, data=json.dumps(send_data))
  if(res.status_code != 200):
    print("error : [{}]".format(res.text))

slack_url = 'https://hooks.slack.com/services/xxxxxxxx/xxxxxxxx/xxxxxx'
def Notice_Slack(send_message):
  requests.post(slack_url, data=json.dumps({'text': send_message}))

teams_webhook_url = 'https://outlook.office.com/webhook/xxxxxxx/IncomingWebhook/xxxxxxx/xxxxxx'
def Notice_TEAMS(subject, send_message):
  send_message = send_message.replace('\n', '<BR>')
  requests.post(teams_webhook_url, data=json.dumps({'title': subject, 'text': send_message}))

url_ifttt = 'https://maker.ifttt.com/trigger/eventid/with/key/xxxxx'
def Notice_IFTTT(subject, send_message):
  requests.post(url_ifttt, data={'value1': subject, 'value2': send_message, 'value3': ''})

def Notice(subject, body):
  # 通知処理：メール
  Notice_Mail(subject, body)

  # 通知処理：LINE_Notify
  Notice_LINE_Self(body)

  # 通知処理：LINE_push
  userid = 'xxxxxxxxxx'
  Notice_LINE_push(userid, body)

  # 通知処理：Slack
  Notice_Slack(body)

  # 通知処理：TEAMS
  Notice_TEAMS(subject, body)

  # 通知処理：IFTTT（経由でLINEへ）
  Notice_IFTTT(subject, body)


import datetime
dt_now = datetime.datetime.now()

# 改行は\nで設定すればよい
body = "送信内容ですよ\n改行は反映されるのでしょうか？\n{}"
body = body.format(dt_now.strftime('%Y年%m月%d日 %H:%M:%S'))

subject = "＜通知サンプルですよ : {}＞".format(dt_now.strftime('%Y%m%d %H%M%S'))

# 通知
Notice(subject, body)