アプリケーションのスケルトンを説明します。merge_face.py、box_upload.pyアプリケーションについて、部位ごとに掲載していきます。各API等の仕様については、リンクしていきますので、そちらをご参照ください。
Develop編①
宣言
# Pythonで必要なライブラリ
import os
from io import BytesIO
import json
import pprint
import requests
from pathlib import Path
from flask import Flask, request, abort
import base64
import random
from datetime import datetime
from PIL import Image, ImageDraw, ImageFont
# LINE Botで必要なライブラリ
from linebot import (
LineBotApi, WebhookHandler
)
from linebot.exceptions import (
InvalidSignatureError
)
from linebot.models import (
FollowEvent, MessageEvent, TextMessage, TextSendMessage,
ImageMessage,ImageSendMessage,StickerMessage, StickerSendMessage
)
# A3RT APIで必要なライブラリ
import pya3rt
# Face++_APIで必要なライブラリ
import face_detect as fd
import face_detect_azure as afd
import merge_face as mf
# BOX_APIで必要なライブラリ
import box_upload as bu
環境変数など
# LINE Bot
YOUR_CHANNEL_ACCESS_TOKEN = os.environ['YOUR_CHANNEL_ACCESS_TOKEN']
YOUR_CHANNEL_SECRET = os.environ['YOUR_CHANNEL_SECRET']
# Talk_API
A3RT_TALK_APIKEY = os.environ['A3RT_TALK_APIKEY']
# Face_Detect,Merge_Face
FACE_API_KEY = os.environ['FACE_API_KEY']
FACE_API_SECRET = os.environ['FACE_API_SECRET']
# Face_API
ASURE_API_KEY = os.environ['ASURE_API_KEY']
ASURE_API_SECRET = os.environ['ASURE_API_SECRET']
# Box_API
BOX_FOLDER_ID = os.environ["BOX_FOLDER_ID"]
BOX_CLIENT_ID = os.environ["BOX_CLIENT_ID"]
BOX_CLIENT_SECRET = os.environ["BOX_CLIENT_SECRET"]
BOX_ENTERPRISE_ID = os.environ["BOX_ENTERPRISE_ID"]
BOX_JWT_KEY_ID = os.environ["BOX_JWT_KEY_ID"]
BOX_RSA_PRIVATE_KEY_FILEPATH = os.environ["BOX_RSA_PRIVATE_KEY_FILEPATH"]
BOX_RSA_PRIVATE_KEY_PASSPHRASE = os.environ["BOX_RSA_PRIVATE_KEY_PASSPHRASE"]
# FlaskのInstance作成
app = Flask(__name__)
# LINE BotのInstance作成
line_bot_api = LineBotApi(YOUR_CHANNEL_ACCESS_TOKEN)
handler = WebhookHandler(YOUR_CHANNEL_SECRET)
Flask_app.runの定義
# このプログラムがメインだった場合にどう稼働するか定義
if __name__ == "__main__":
port = int(os.getenv("PORT", 5000))
app.run(host="0.0.0.0", port=port)
LINE からCallBackにPostしたときのURLと関数の定義
# LINE から、CallBackにPostしたときのURLと関数の定義
# callback
@app.route("/callback", methods=['POST'])
def callback():
# get X-Line-Signature header value
signature = request.headers['X-Line-Signature']
# get request body as text UTF8 Only
body = request.get_data(as_text=True)
# handle webhook body
try:
handler.handle(body, signature)
except InvalidSignatureError:
print("Invalid signature. Please check your channel access token/channel secret.")
abort(400)
# レスポンスにOKと設定
return 'OK'
LINEのhandlerに、EventDrivenの関数の定義を追加
# フォローイベント時 ----------------------------------------------
@handler.add(FollowEvent)
def handle_follow(event):
# LINEから、テスト接続する際に、reply_tokenが'0...0'となってしまう。
# テストする際にline_bot_api内でerrorとなってしまうのを回避
if event.reply_token == "00000000000000000000000000000000":
return
else:
# LINE Botへのレスポンスメッセージを記載
line_bot_api.reply_message(
event.reply_token,
TextSendMessage(text='初めましてbotです。')
)
# スタンプイベント時 ----------------------------------------------
@handler.add(MessageEvent, message=StickerMessage)
def handle_sticker_message(event):
# LINEから、テスト接続する際に、reply_tokenが'f...f'となってしまう。
# テストする際にline_bot_api内でerrorとなってしまうのを回避
if event.reply_token == "ffffffffffffffffffffffffffffffff":
return
else:
# LINE Botへのレスポンススタンプを記載
# https://devdocs.line.me/files/sticker_list.pdf
# のスタンプidをランダムにて設定する
line_bot_api.reply_message(
event.reply_token,
StickerSendMessage(
package_id='2',
sticker_id=str(random.randrange(29, 47, 1))
)
)
return
トークイベント時：Talk_API
@handler.add(MessageEvent, message=TextMessage)
def handle_message(event):
# LINEから、テスト接続する際に、reply_tokenが'0...0'となってしまう。
# テストする際にline_bot_api内でerrorとなってしまうのを回避
if event.reply_token == "00000000000000000000000000000000":
return
else:
# TalkAPIに、イベントmessageをString形式で受け渡し、
# String形式のchatを返却値として受け取る。
reply_text = talkapi_response(event.message.text)
# LINE Botへのレスポンスメッセージを記載
line_bot_api.reply_message(
event.reply_token,
TextSendMessage(text=reply_text)
)
return
# A3RTのTalkAPIの呼び出し関数
def talkapi_response(text):
# Talk_APIのInstance作成
apikey = A3RT_TALK_APIKEY
client = pya3rt.TalkClient(apikey)
# Talk_APIのInstance作成
response = client.talk(text)
# Talk_APIのレスポンスステータスのチェック
if response['status'] == 0:
# normalなら、chatメッセージを返却
ret = response['results'][0]['reply']
else:
# errorなら、無言メッセージを返却
ret = '・・・・・・・・・'
return ret
イメージイベント時：
@handler.add(MessageEvent, message=(ImageMessage, AudioMessage))
def handle_image_audio_message(event):
# イベントイメージIDを取得
push_img_id = event.message.id
# イベントイメージをlineサーバより取得
message_content = line_bot_api.get_message_content(push_img_id)
# イベントイメージをiter_contentで、少しずつpush_imgにバイナリデータとして順次代入
push_img = b""
for chunk in message_content.iter_content():
push_img += chunk
# イメージデータの画素数と形式のJPEG化 ※ご参考-------------------
# メモリに保持してディレクトリ偽装みたいなことをして、PILで読み込む
im = Image.open(BytesIO(push_img))
# PILで1600 x Height の比率にして、JPGバイナリデータに戻す。
if im.width > 1600:
push_img = change_image(im,1600)
# イメージデータの画素数と形式のJPEG化---------------------------
# 3種類の回答をさせるため、時刻1分刻みに呼び出し関数を変更するように制御
dt_now = datetime.now()
dt_chk = str(dt_now.minute)[-1]
chk1={"1":"1","4":"4","7":"7"}
chk2={"2":"2","5":"5","8":"8"}
chk3={"3":"3","6":"6","9":"9","0":"0"}
# イメージデータの画素数と形式のJPEG化。※ご参考
def change_image(im,size) :
# size x Height の比率にする
if im.width > size:
proportion = size / im.width
im = im.resize((int(im.width * proportion), int(im.height * proportion)),Image.NEAREST)
# 空のインスタンス（メモリに保持してディレクトリ偽装みたいなことをして）を作り、保存する。
img = io.BytesIO()
im.save(img,"JPEG")
# バイナリデータを取得する（open().read()状態）
return img.getvalue()
イメージイベント時：Face_Detect部
if dt_chk in chk1 :
# Face_Detect_APIに、イベントイメージをString形式で受け渡し、
# String形式の判別結果を返却値として受け取る。
msg=call_image_apis1(push_img)
# LINE Botへのレスポンスメッセージを記載
line_bot_api.reply_message(
event.reply_token,
TextSendMessage(text=msg)
)
return
# Face_Detect_APIの呼び出し関数
def call_image_apis1(push_img):
msg = ''
# イメージは、base64エンコードを実施する
push_img64 = base64.b64encode(push_img)
# Face_Detect_API.pyの呼び出し。イメージと環境変数をパラメータとする。
msg = fd.face_detect(push_img64,FACE_API_KEY,FACE_API_SECRET)
return msg
イメージイベント時：Face_API部
elif dt_chk in chk2 :
# Face_APIに、イベントイメージをString形式で受け渡し、
# String形式の判別結果を返却値として受け取る。
msg=call_image_apis2(push_img)
# LINE Botへのレスポンスメッセージを記載
line_bot_api.reply_message(
event.reply_token,
TextSendMessage(text=msg)
)
return
# Face_APIの呼び出し関数
def call_image_apis2(push_img):
msg = ''
# Face_API.pyの呼び出し。イメージと環境変数をパラメータとする。
msg_list = afd.face_detect_azure(push_img,ASURE_API_KEY,ASURE_API_SECRET)
# Face_APIは、イメージ内の画像に対して、リスト形式でメッセージが返却されるため、ここでメッセージを連結
for m in msg_list:
msg += str(m['msg'])+'\n'
return msg
イメージイベント時：Merge_Face部
else:
# Merge_Face_APIに、イベントイメージをString形式で受け渡し、
# String形式の判別結果を返却値として受け取る。
# Boxストレージ上に一時的に保存する名称もここで設定する。
msg_url=call_image_apis3(push_img,f"{push_img_id}.jpg")
# LINE BotへのレスポンスメッセージとBOXストレージの共有URLを記載。
line_bot_api.reply_message(
event.reply_token,
TextSendMessage(text="セキュアなBOXストレージに一時的に合成写真を格納しました。\n" + msg_url)
)
return
# Merge_Faceの呼び出し関数
def call_image_apis3(push_img,push_img_name):
## Merge Imageの顔位置をDetect APIで取得する。-----------------------------------
push_img_rectangle = ''
# 今回はイメージの中で、一番右側の人
msg_list = afd.face_detect_azure(push_img,ASURE_API_KEY,ASURE_API_SECRET)
for m in msg_list:
push_img_rectangle = str(m['faceRectangle'])
# イメージは、base64エンコードを実施する
push_img64 = base64.b64encode(push_img)
## Merge Imageの顔位置をDetect APIで取得する。-----------------------------------
## Template Imageの顔位置をDetect APIで取得する。-----------------------------------
tmp_img = ""
tmp_img_rectangle = ""
# Template ImageをHeroku上より取得
with open('./static/images/merge_face_templete.jpg', 'rb') as f:
tmp_img = f.read()
# 今回はイメージの中で、一番右側の人
msg_list = afd.face_detect_azure(tmp_img,ASURE_API_KEY,ASURE_API_SECRET)
for m in msg_list:
tmp_img_rectangle = str(m['faceRectangle'])
# イメージは、base64エンコードを実施する
tmp_img64 = base64.b64encode(tmp_img)
## Template Imageの顔位置をDetect APIで取得する。-----------------------------------
# Face_API.pyの呼び出し。イメージ、イメージの顔位置、環境変数をパラメータとする。
merge_img64 = mf.merge_face(tmp_img64,tmp_img_rectangle,push_img64,push_img_rectangle,FACE_API_KEY,FACE_API_SECRET)
# base64エンコードイメージのデコードを実施する
merge_img = base64.b64decode(merge_img64)
# BOX_API.pyの呼び出し。保存ファイル名、合成イメージ、環境変数をパラメータとする。返却値はBOXストレージの共有URLとなる。
msg_url = bu.box_upload(push_img_name,merge_img,BOX_FOLDER_ID,BOX_CLIENT_ID,BOX_CLIENT_SECRET,BOX_ENTERPRISE_ID,BOX_JWT_KEY_ID,'./static/images/private.pem',BOX_RSA_PRIVATE_KEY_PASSPHRASE)
return msg_url
# 本当はiImageSendMessageで、URLをレスポンスしたかったが、短縮URLには未対応のようだ。
# line_bot_api.reply_message(
# event.reply_token,
# ImageSendMessage(
# original_content_url=msg_url,
# preview_image_url=msg_url
# )
# )
UptimeRobotから、GETしたときのURLと関数の定義
# box_remove
@app.route("/box_remove", methods=['GET'])
def box_remove():
try:
# BOX_API.pyの呼び出し。BOX_FOLDER_IDと環境変数をパラメータとする。
# BOXストレージ上のファイル削除をするため、返却値なし
bu.box_remove(BOX_FOLDER_ID,BOX_CLIENT_ID,BOX_CLIENT_SECRET,BOX_ENTERPRISE_ID,BOX_JWT_KEY_ID,'./static/images/private.pem',BOX_RSA_PRIVATE_KEY_PASSPHRASE)
except Exception as e:
print(e.message)
return 'OK'
