TL;DR

AWS IoTボタンからTeamsの自動投稿を行いたいが、そこに至るまでピタゴラスイッチ的パイプライン構築を行う
前半として、AWS IoTボタン登録からLambda実行、RSSフィード生成までを記載する

背景・モチベーション

自社のテレワークの前提として出社、休憩、退社時にTeamsでその旨を投稿する
わざわざそのために、自社PCにログインし、ワンタッチパスワードを発行し、仮想OSにログインし、Teamsを開く必要がある
このためパイプラインを構築し、その部分を自動化したかった

構成

最終的に下記の構成になった

AWS IoTボタン → Lambda → (自宅Webアプリ) → S3(RSSフィードxml配置) ← PowerAutomate → Teams

Lambaの作成

後に登録するAWS IoTボタンと同じリージョンで下記のようなLambaを登録する
今回は、実際の処理部分は自宅のWebAPIサーバで行っており、Lamba側はリクエストを転送するのみにしている
(のちほどロジック部分のLambdaに移植する予定)

handler.py

# coding: utf-8

import json

from external_api import request_home_api
 
def lambda_handler(event, context):
    clicktype = event['deviceEvent']['buttonClicked']['clickType']
    print(clicktype)

    if clicktype == "SINGLE":
        post_data("work")
        message = "ボタンが1回押されました" 
    elif clicktype == "DOUBLE":
        post_data("break")
        message = "ボタンが2回押されました"
    elif clicktype == "LONG":
        post_data("work")
        message = "ボタンが長押しされました"
    else:
        message = "clickTypeを正常に取得できませんでした"

def post_data(data_type):
    url_path = ['post_chat', data_type]
    res = request_home_api(*url_path ,method='POST', body={'from': 'lambda'})
    print(res)
    print(res.text)

AWS IoTボタン登録

スマホアプリAWS IoT 1-ClickでAWS IoTボタンを追加する
(結構簡単にできる)

アプリTopのデバイスを登録から「デバイスIDで登録」をタップ
箱側面のバーコードをスキャン
ボタンを長押しし、ペアリングを行う
Wifi設定などを行う
ボタンに紐づくプロジェクトを登録する
プレイスメント登録を行う
プロジェクト登録にて、作成済みLambdaとプロジェクトを紐づける

投稿メッセージを生成するロジックの実装

下記のようなロジックを実装し、投稿するメッセージを生成する
(今回はPython DjangoでRESTfulgAPIサーバを実装し、勤務ステータスをDBで保持している)

url.py

from django.conf.urls import include, url
from . import views


urlpatterns = [
    url(r'^post_chat/(work|break)$', views.post_chat, name='post_chat')
]

view.py

from __future__ import unicode_literals

import json
import logging
from django.http.response import JsonResponse
from django.shortcuts import get_object_or_404

from .models import At_home_status
from .helpers get_worktime_message

logger = logging.getLogger('views')

def post_chat(request, kubun):
    member = get_object_or_404(At_home_status, name="hoge")
    if kubun == "work":
        message_num = 2 if member.at_work_flg else 1
        message = get_worktime_message(message_num)
        member.update(at_work=not member.at_work_flg, message=message)
    elif kubun == "break":
        if not member.at_work_flg:
            return JsonResponse({"message": ""})
        in_break = json.loads(member.memo).get('in_break', False)
        message_num = 4 if in_break else 3
        message = get_worktime_message(message_num)
        member.update(in_break=not in_break, message=message)
    
    push_message("chat post: {}".format(message), debug=True)
    
    return JsonResponse({"message": message_num})

helpers.py

from __future__ import unicode_literals

import datetime

def get_worktime_message(message_num):
    message_dict = {1: "おはようございます。本日の業務を開始します。", 2: "本日の業務を終了します。", 3: "お昼休憩に入ります。{}再開予定です。", 4: "業務を再開します。"}
    message = message_dict.get(message_num)
    if message_num == 3:
        now = datetime.datetime.now()
        delta_min = -1 * (now.minute % 5) if now.minute % 5 < 3 else 5 - (now.minute % 5)
        break_end = now + datetime.timedelta(hours=1) + datetime.timedelta(minutes=delta_min)
        message = message.format(break_end.strftime('%H:%M'))
    return message

RRS フィードの発行

今回の記事のキモはこちらになる
後続のTeams自動投稿をPowerAutomateで実装することになるが、驚くべきことにライセンスの問題でInternetFacingなエンドポイントを作成してのWebhook作成できない。
代替手段を探していたところRSSフィードのポーリングによるトリガー実行が可能だった
このため、上記処理で生成したメッセージをもとにRSSfeed用のxmlを生成し、S3にpushすることでRSSフィード更新を行うことで、PowerAutomateにステータス更新を伝える
下記 publish_rssfile() にメッセージを渡すことでS3(CloudFront)にRSSfeedの更新を行っている

publish_rss.py

import json
import os
import datetime
import codecs
from homegateway.libs.boto_helpers import s3_put_object


message_template = '''<?xml version='1.0' encoding='UTF-8'?>
<feed xmlns='http://www.w3.org/2005/Atom' xml:lang='ja'>
<entry>
<title type="text">{message}</title>
<content type="html">{message}</content>
<published>{datetime}</published>
<updated>{datetime}</updated>
</entry>
</feed>'''

def publish_rssfile(message=""):
    now_str = datetime.datetime.now().strftime("%Y-%m-%dT%H:%M:%S+09:00")
    file_str = message_template.format(message=message, datetime=now_str)
    with codecs.open("/tmp/rss.xml", "w", "utf-8") as rss_file:
        rss_file.write(file_str)

    s3_put_object("lytitong-rssfeedtest", "/tmp/rss.xml")
    os.remove("/tmp/rss.xml")

    return None

IoTボタンからTeamsへの自動投稿を行う(前半)