作ったもの

以前作った写真に任意の日付を描画するラインボットを改良し、データベースに名前と生年月日を保存、撮影日から何歳頃の写真かを計算して画像へ書き込むラインボットを作成しました。
githubはこちら

・成長記録アシスタント君(ID:@033cynwe)

環境

macOS
Python3.7.1
flask1.1.1
line-bot-sdk1.14.0
flask-sqlalchemy2.4.1
PostgreSQL

この記事で書かないこと

LineBotのチャンネル作成
herokuへのデプロイ

全文

main.py

main.py

from flask import Flask, request, abort
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import (FollowEvent, PostbackEvent, TemplateSendMessage, MessageAction,\
                                            ButtonsTemplate, DatetimePickerTemplateAction, ImageMessage, \
                                            ImageSendMessage, MessageEvent, TextMessage, TextSendMessage)
from pathlib import Path
from PIL import Image, ImageDraw, ImageFont, ImageFilter
import datetime
import os

import database

app = Flask(__name__)
app.debug = False

#環境変数取得
YOUR_CHANNEL_ACCESS_TOKEN = os.environ["YOUR_CHANNEL_ACCESS_TOKEN"]
YOUR_CHANNEL_SECRET = os.environ["YOUR_CHANNEL_SECRET"]

line_bot_api = LineBotApi(YOUR_CHANNEL_ACCESS_TOKEN)
handler = WebhookHandler(YOUR_CHANNEL_SECRET)

#画像の参照元パス
SRC_IMAGE_PATH = "static/images/{}.jpg"
MAIN_IMAGE_PATH = "static/images/{}_main.jpg"
PREVIEW_IMAGE_PATH = "static/images/{}_preview.jpg"

@app.route("/callback", methods=['POST'])
def callback():
    signature = request.headers['X-Line-Signature']
    body = request.get_data(as_text=True)
    app.logger.info("Request body: " + body)

    try:
        handler.handle(body, signature)
    except InvalidSignatureError:
        abort(400)

    return 'OK'

#フォローイベント
@handler.add(FollowEvent)
def handle_follow(event):
    line_bot_api.reply_message(
        event.reply_token,
        TextSendMessage(text=
        "友達登録ありがとう。写っている人が何歳ごろかを画像に書き込むよ。画像を送ってみてね。"))

#画像の受け取り
@handler.add(MessageEvent, message=ImageMessage)
def handle_image(event):
    global message_id, user_id, name_list, user_dict, num, src_image_path
    
    #ユーザー情報を格納する為のリスト
    name_list = []
    day_list = []
    user_dict = {}
    #message_idを取得
    message_id = event.message.id
    #user_idを取得
    user_id = event.source.user_id
    
    #ファイル名をmessage_idに変換したパス
    src_image_path = Path(SRC_IMAGE_PATH.format(message_id)).absolute()

    # 画像をHerokuへ一時保存
    save_image(message_id, src_image_path)
    
    #ユーザー情報を確認し登録がない場合はパスする
    try:
        name_list, day_list = database.serch_data(user_id)
    except TypeError:
        pass
        
    #登録数
    num = len(name_list)
    
    #登録がない場合、名前を確認する
    if num == 0:
        line_bot_api.reply_message(
            event.reply_token,
            TextSendMessage(text="写真に写っている人の名前は？"))
        #user_idを追加
        database.add_data(user_id)
    #登録がある場合、誰が写ってるか確認
    elif num != 0:
        #nameとdayで辞書を作成
        user_dict = dict(zip(name_list, day_list))
        
        #1人登録の場合
        if num == 1:
            name_1 = name_list[0]
            buttons_template = TemplateSendMessage(
                alt_text="誰が写ってる？",
                template=ButtonsTemplate(
                    text="誰が写ってる？", actions=[
                        MessageAction(label=name_1, text=name_1),
                        MessageAction(label="その他", text="その他")
                    ]
                )
            )
        
        #2人登録の場合
        elif num == 2:
            name_1 = name_list[0]
            name_2 = name_list[1]
            buttons_template = TemplateSendMessage(
                alt_text="誰が写ってる？",
                template=ButtonsTemplate(
                    text="誰が写ってる？", actions=[
                        MessageAction(label=name_1, text=name_1),
                        MessageAction(label=name_2, text=name_2),
                        MessageAction(label="その他", text="その他")
                    ]
                )
            )
        
        #３人登録の場合
        elif num == 3:
            name_1 = name_list[0]
            name_2 = name_list[1]
            name_3 = name_list[2]
            buttons_template = TemplateSendMessage(
                alt_text="誰が写ってる？",
                template=ButtonsTemplate(
                    text="誰が写ってる？", actions=[
                        MessageAction(label=name_1, text=name_1),
                        MessageAction(label=name_2, text=name_2),
                        MessageAction(label=name_3, text=name_3),
                        MessageAction(label="その他", text="その他")
                    ]
                )
            )
        
        line_bot_api.reply_message(event.reply_token, buttons_template)

#テキストの受け取り
@handler.add(MessageEvent, message=TextMessage)
def handle_text(event):
    global text_name, birthday
    
    #登録が無い場合、生年月日の確認
    if num == 0:
        text_name = event.message.text
        select_day(event)
    #その他が選択された場合、名前を確認
    elif event.message.text == "その他":
        line_bot_api.reply_message(
            event.reply_token,
            TextSendMessage(text="写真に写っている人の名前は？"))
    else:
        text_name = event.message.text
            
        #名前が登録済みの場合、生年月日を取得
        if text_name in name_list:
            birthday = user_dict[text_name]
        
        #ifの場合撮影日、それ以外の場合生年月日の選択            
        select_day(event)
    
#画像を処理して送信
@handler.add(PostbackEvent)
def handle_postback(event):
    global birthday
    
    #ファイル名をmessage_idに変換したパス
    main_image_path = MAIN_IMAGE_PATH.format(message_id)
    preview_image_path = PREVIEW_IMAGE_PATH.format(message_id)
    
    #birthdayが未定義の場合
    if not "birthday" in globals():
        #日付選択アクションの結果ををbirthdayに代入
        birthday = event.postback.params["date"]
        #nameとdayを更新
        database.update_data(user_id, num, text_name, birthday)
        
        #撮影日の選択    
        select_day(event)
    #birthdayが定義済みの場合
    elif "birthday" in globals():
        #画像処理
        date_the_image(src_image_path, Path(main_image_path).absolute(), event)
        date_the_image(src_image_path, Path(preview_image_path).absolute(), event)

        # 画像を指定
        image_message = ImageSendMessage(
                original_content_url=f"https://<herokuアプリ名>.herokuapp.com/{main_image_path}",
                preview_image_url=f"https://<herokuアプリ名>.herokuapp.com/{preview_image_path}"
        )
    
        #ログの取得
        app.logger.info(f"https://<herokuアプリ名>.herokuapp.com/{main_image_path}")
        
        #画像の送信
        line_bot_api.reply_message(event.reply_token, image_message)
        
        #変数birthdayを削除
        del birthday
        
        #データベースの変更を保存し、接続を切る
        database.close_db()

#画像保存関数
def save_image(message_id: str, save_path: str) -> None:
    # message_idから画像のバイナリデータを取得
    message_content = line_bot_api.get_message_content(message_id)
    with open(save_path, "wb") as f:
        # 取得したバイナリデータを書き込む
        for chunk in message_content.iter_content():
            f.write(chunk)

#日付選択関数
def select_day(event):
    #birthdayが定義済みの場合
    if "birthday" in globals():
        message = "撮影日を選択してね"
    #birthdayが未定義の場合
    elif not "birthday" in globals():
        message = "生年月日を選択してね"
    
    #日付選択アクション    
    date_picker = TemplateSendMessage(
        alt_text=message,
        template=ButtonsTemplate(
            text=message,
            actions=[
                DatetimePickerTemplateAction(
                    label="選択",
                    data="action=buy&itemid=1",
                    mode="date",
                    initial=str(datetime.date.today()),
                    max=str(datetime.date.today())
                )
            ]
        )
    )
    
    line_bot_api.reply_message(event.reply_token, date_picker)

#画像処理関数
def date_the_image(src: str, desc: str, event) -> None:
    im = Image.open(src)
    draw = ImageDraw.Draw(im)
    font = ImageFont.truetype("./fonts/AquaKana.ttc", 50)
    
    #撮影日を取得
    date = event.postback.params["date"]
    #撮影日から生年月日を引いて、生後日数を計算
    how_old = datetime.datetime.strptime(date, "%Y-%m-%d") - datetime.datetime.strptime(str(birthday), "%Y-%m-%d")
    #生後日数と365(日)で商と余りを計算
    years, days = divmod(how_old.days, 365)
    #余りと30(日)で商を計算
    month = days // 30
    text = text_name + f"({years}才{month}ヶ月)"
    
    #テキストのサイズ
    text_width = draw.textsize(text, font=font)[0]
    text_height = draw.textsize(text, font=font)[1]
    margin = 10
    x = im.width - text_width
    y = im.height - text_height
    #描画する矩形のサイズ
    rect_size = ((text_width + margin * 6), (text_height + margin * 2))
    #矩形の描画
    rect = Image.new("RGB", rect_size, (0, 0, 0))
    #矩形を透明にする為のマスク
    mask = Image.new("L", rect_size, 128)
    
    #画像に矩形とマスクを貼り付け
    im.paste(rect, (x - margin * 6, y - margin * 3), mask)
    #テキストの書き込み
    draw.text((x - margin * 3, y - margin * 2), text, fill=(255, 255, 255), font=font)
    im.save(desc)

if __name__ == "__main__":
    #app.run()
    port = int(os.getenv("PORT", 5000))
    app.run(host="0.0.0.0", port=port)

database.py

database.py

from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import create_engine, Column, String, Date
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session
from sqlalchemy.orm.exc import NoResultFound
import datetime
import os

#環境変数からデータベースのURLを取得
DATABASE_URL = os.environ["DATABASE_URL"]

engine = create_engine(DATABASE_URL)
Base = declarative_base()

#テーブル定義
class User(Base):
	__tablename__ = "user_list"
	user_id = Column("user_id", String(50), primary_key=True)
	name1 = Column("name1", String(10))
	day1 = Column("day1", Date)
	name2 = Column("name2", String(10))
	day2 = Column("day2", Date)
	name3 = Column("name3", String(10))
	day3 = Column("day3", Date)

Base.metadata.create_all(engine)
session = Session(bind=engine)

#user_idの検索
def serch_data(user_id):
	try:
		#user_idで検索
		res = session.query(User.name1, User.day1, User.name2, User.day2, User.name3, User.day3).filter(User.user_id==f"{user_id}").one()
		
		#nameとdayをそれぞれリストへ挿入
		name_list = [n for n in res if type(n) is str]
		day_list = [str(d) for d in res if type(d) is datetime.date]
		return name_list, day_list
	#user_idの登録が無い場合パスする
	except NoResultFound:
		pass

#user_idの登録
def add_data(user_id):
	session.add(User(user_id=f"{user_id}"))

#ユーザー情報の更新
def update_data(user_id, num, text_name, birthday):
	user_data = session.query(User).filter(User.user_id==f"{user_id}").one()
	
	#登録数に応じてnameとdayを保存
	if num == 0:
		user_data.name1 = text_name
		user_data.day1 = birthday
	elif num == 1:
		user_data.name2 = text_name
		user_data.day2 = birthday
	elif num == 2:
		user_data.name3 = text_name
		user_data.day3 = birthday
    
#データベースの変更を保存し、接続を切る
def close_db():
	session.commit()
	session.close()

解説

main.py

準備

main.py

from flask import Flask, request, abort
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import (FollowEvent, PostbackEvent, TemplateSendMessage, MessageAction,\
                                            ButtonsTemplate, DatetimePickerTemplateAction, ImageMessage, \
                                            ImageSendMessage, MessageEvent, TextMessage, TextSendMessage)
from pathlib import Path
from PIL import Image, ImageDraw, ImageFont, ImageFilter
import datetime
import os

import database

app = Flask(__name__)
app.debug = False

#環境変数取得
YOUR_CHANNEL_ACCESS_TOKEN = os.environ["YOUR_CHANNEL_ACCESS_TOKEN"]
YOUR_CHANNEL_SECRET = os.environ["YOUR_CHANNEL_SECRET"]

line_bot_api = LineBotApi(YOUR_CHANNEL_ACCESS_TOKEN)
handler = WebhookHandler(YOUR_CHANNEL_SECRET)

#画像の参照元パス
SRC_IMAGE_PATH = "static/images/{}.jpg"
MAIN_IMAGE_PATH = "static/images/{}_main.jpg"
PREVIEW_IMAGE_PATH = "static/images/{}_preview.jpg"

@app.route("/callback", methods=['POST'])
def callback():
    signature = request.headers['X-Line-Signature']

    body = request.get_data(as_text=True)
    app.logger.info("Request body: " + body)

    try:
        handler.handle(body, signature)
    except InvalidSignatureError:
        abort(400)

    return 'OK'

モジュールのインポートと事前に設定しておいた環境変数を取得しますが、詳しい役割は適宜調べて頂ければと思います。
後述する画像参照元のパスを設定し、{}の部分を画像を受け取った時のmessage_idに置き変えます。

フォロー時にテキストを送信

main.py

#フォローイベント
@handler.add(FollowEvent)
def handle_follow(event):
    line_bot_api.reply_message(
        event.reply_token,
        TextSendMessage(text=
        "友達登録ありがとう。写っている人が何歳ごろかを画像に書き込むよ。画像を送ってみてね。"))

ユーザーが友達追加時にメッセージを送ります。

画像の受け取り

main.py

#画像の受け取り
@handler.add(MessageEvent, message=ImageMessage)
def handle_image(event):
    global message_id, user_id, name_list, user_dict, num, src_image_path
    
    #ユーザー情報を格納する為のリスト
    name_list = []
    day_list = []
    user_dict = {}
    #message_idを取得
    message_id = event.message.id
    #user_idを取得
    user_id = event.source.user_id
    
    #ファイル名をmessage_idに変換したパス
    src_image_path = Path(SRC_IMAGE_PATH.format(message_id)).absolute()

    # 画像をHerokuへ一時保存
    save_image(message_id, src_image_path)
    
    #ユーザー情報を確認し登録がない場合はパスする
    try:
        name_list, day_list = database.serch_data(user_id)
    except TypeError:
        pass
        
    #登録数
    num = len(name_list)
    
    #登録がない場合、名前を確認する
    if num == 0:
        line_bot_api.reply_message(
            event.reply_token,
            TextSendMessage(text="写真に写っている人の名前は？"))
        #user_idを追加
        database.add_data(user_id)
    #登録がある場合、誰が写ってるか確認
    elif num != 0:
        #nameとdayで辞書を作成
        user_dict = dict(zip(name_list, day_list))
        
        #1人登録の場合
        if num == 1:
            name_1 = name_list[0]
            buttons_template = TemplateSendMessage(
                alt_text="誰が写ってる？",
                template=ButtonsTemplate(
                    text="誰が写ってる？", actions=[
                        MessageAction(label=name_1, text=name_1),
                        MessageAction(label="その他", text="その他")
                    ]
                )
            )
        
        #2人登録の場合
        elif num == 2:
            name_1 = name_list[0]
            name_2 = name_list[1]
            buttons_template = TemplateSendMessage(
                alt_text="誰が写ってる？",
                template=ButtonsTemplate(
                    text="誰が写ってる？", actions=[
                        MessageAction(label=name_1, text=name_1),
                        MessageAction(label=name_2, text=name_2),
                        MessageAction(label="その他", text="その他")
                    ]
                )
            )
        
        #３人登録の場合
        elif num == 3:
            name_1 = name_list[0]
            name_2 = name_list[1]
            name_3 = name_list[2]
            buttons_template = TemplateSendMessage(
                alt_text="誰が写ってる？",
                template=ButtonsTemplate(
                    text="誰が写ってる？", actions=[
                        MessageAction(label=name_1, text=name_1),
                        MessageAction(label=name_2, text=name_2),
                        MessageAction(label=name_3, text=name_3),
                        MessageAction(label="その他", text="その他")
                    ]
                )
            )
        
        line_bot_api.reply_message(event.reply_token, buttons_template)

MessageEvent以外でも使いたい変数をglobalに指定し、参照できるようにしておきます。

try文のname_list, day_list = database.serch_data(user_id)で、名前を格納するname_list、生年月日を格納するday_listを、databaseファイルのserch_data関数で取得したリスト内容に更新します。
この時、user_idの登録が無い場合何も値を返さず更新できないので、exceptでTypeErrorを指定しておきます。

データを参照した結果、登録が無い場合とある場合で処理を分けます。
無い場合は名前の確認を行いuser_idを追加します。
ある場合はuser_dict = dict(zip(name_list, day_list))で、名前をキーに生年月日を参照する為の辞書を作成します。
その後、登録数が1~3人で更に処理を分け、写っている人の確認をTemplateSendMessageで行います。

テキストの受け取り

main.py

#テキストの受け取り
@handler.add(MessageEvent, message=TextMessage)
def handle_text(event):
    global text_name, birthday
    
    #登録が無い場合、生年月日の確認
    if num == 0:
        text_name = event.message.text
        select_day(event)
    #その他が選択された場合、名前を確認
    elif event.message.text == "その他":
        line_bot_api.reply_message(
            event.reply_token,
            TextSendMessage(text="写真に写っている人の名前は？"))
    else:
        text_name = event.message.text
            
        #名前が登録済みの場合、生年月日を取得
        if text_name in name_list:
            birthday = user_dict[text_name]
        
        #ifの場合撮影日、それ以外の場合生年月日の選択            
        select_day(event)

if num == 0:は、登録が無い場合の名前の確認を行った返事を受けた時に実行され、名前を取得した後生年月日の確認に進みます。

elif event.message.text == "その他":は、登録があった場合のTemplateSendMessageでその他が選択された時に実行され、名前の確認を行います。

else:は上記以外のTemplateSendMessageで登録済みの名前が選択された時と、その他で名前の確認を行った返事を受けた場合に実行されます。
登録済みの場合はbirthday = user_dict[text_name]で名前をキーに生年月日を取得します。
その他の返事の場合はそのまま日付選択に進み、生年月日を選択してもらいます。

日付選択アクションの処理

main.py

#画像を処理して送信
@handler.add(PostbackEvent)
def handle_postback(event):
    global birthday
    
    #ファイル名をmessage_idに変換したパス
    main_image_path = MAIN_IMAGE_PATH.format(message_id)
    preview_image_path = PREVIEW_IMAGE_PATH.format(message_id)
    
    #birthdayが未定義の場合
    if not "birthday" in globals():
        #日付選択アクションの結果ををbirthdayに代入
        birthday = event.postback.params["date"]
        #nameとdayを更新
        database.update_data(user_id, num, text_name, birthday)
        
        #撮影日の選択    
        select_day(event)
    #birthdayが定義済みの場合
    elif "birthday" in globals():
        #画像処理
        date_the_image(src_image_path, Path(main_image_path).absolute(), event)
        date_the_image(src_image_path, Path(preview_image_path).absolute(), event)

        # 画像を指定
        image_message = ImageSendMessage(
                original_content_url=f"https://<herokuのアプリ名>.herokuapp.com/{main_image_path}",
                preview_image_url=f"https://<herokuのアプリ名>.herokuapp.com/{preview_image_path}"
        )
    
        #ログの取得
        app.logger.info(f"https://<herokuのアプリ名>.herokuapp.com/{main_image_path}")
        
        #画像の送信
        line_bot_api.reply_message(event.reply_token, image_message)
        
        #変数birthdayを削除
        del birthday
        
        #データベースの変更を保存し、接続を切る
        database.close_db()

if not "birthday" in globals():でbirthdayが未定義の（日付選択アクションで生年月日の選択が行われた）時の処理を行います。
birthday = event.postback.params["date"]で選択された日付を取得し、database.update_data(user_id, num, text_name, birthday)で、データベースの名前と生年月日の更新をし、撮影日の選択をしてもらいます。

elif "birthday" in globals():でbirthdayが定義済みの（日付選択アクションで撮影日の選択が行われた）時の処理を行います。
image_message = ImageSendMessage( original_content_url=f"https://<herokuのアプリ名>.herokuapp.com/{main_image_path}", preview_image_url=f"https://<herokuのアプリ名>.herokuapp.com/{preview_image_path}" )で処理済みの画像を指定し送信します。

del birthdayで変数birthdayを消去し、連続で画像が送られた時でも正常に処理が行えるようにします。
最後にdatabaseファイルのclose_db()でデータベースの変更を保存し、接続を切って終了です。

画像保存関数

main.py

#画像保存関数
def save_image(message_id: str, save_path: str) -> None:
    # message_idから画像のバイナリデータを取得
    message_content = line_bot_api.get_message_content(message_id)
    with open(save_path, "wb") as f:
        # 取得したバイナリデータを書き込む
        for chunk in message_content.iter_content():
            f.write(chunk)

本当は画像のExif情報を取得し自動で撮影日を入れたかったのですが、この方法だと取得できなかったので苦肉の策として上記の日付選択アクションという形をとりました。
もしやり方を知っている人がいたら教えてください。

日付選択関数

main.py

#日付選択関数
def select_day(event):
    #birthdayが定義済みの場合
    if "birthday" in globals():
        message = "撮影日を選択してね"
    #birthdayが未定義の場合
    elif not "birthday" in globals():
        message = "生年月日を選択してね"
    
    #日付選択アクション    
    date_picker = TemplateSendMessage(
        alt_text=message,
        template=ButtonsTemplate(
            text=message,
            actions=[
                DatetimePickerTemplateAction(
                    label="選択",
                    data="action=buy&itemid=1",
                    mode="date",
                    initial=str(datetime.date.today()),
                    max=str(datetime.date.today())
                )
            ]
        )
    )
    
    line_bot_api.reply_message(event.reply_token, date_picker)

生年月日が登録済みの場合は1回、未登録の場合は合計2回この関数が実行されます。
その為、birthdayが定義済みの場合撮影日に、未定義の場合生年月日にメッセージを変更します。

画像処理関数

main.py

#画像処理関数
def date_the_image(src: str, desc: str, event) -> None:
    im = Image.open(src)
    draw = ImageDraw.Draw(im)
    font = ImageFont.truetype("./fonts/AquaKana.ttc", 50)
    
    #撮影日を取得
    date = event.postback.params["date"]
    #撮影日から生年月日を引いて、生後日数を計算
    how_old = datetime.datetime.strptime(date, "%Y-%m-%d") - datetime.datetime.strptime(str(birthday), "%Y-%m-%d")
    #生後日数と365(日)で商と余りを計算
    years, days = divmod(how_old.days, 365)
    #余りと30(日)で商を計算
    month = days // 30
    text = text_name + f"({years}才{month}ヶ月)"
    
    #テキストのサイズ
    text_width = draw.textsize(text, font=font)[0]
    text_height = draw.textsize(text, font=font)[1]
    margin = 10
    x = im.width - text_width
    y = im.height - text_height
    #描画する矩形のサイズ
    rect_size = ((text_width + margin * 6), (text_height + margin * 2))
    #矩形の描画
    rect = Image.new("RGB", rect_size, (0, 0, 0))
    #矩形を透明にする為のマスク
    mask = Image.new("L", rect_size, 128)
    
    #画像に矩形とマスクを貼り付け
    im.paste(rect, (x - margin * 6, y - margin * 3), mask)
    #テキストの書き込み
    draw.text((x - margin * 3, y - margin * 2), text, fill=(255, 255, 255), font=font)
    im.save(desc)

font = ImageFont.truetype("./fonts/AquaKana.ttc", 50)で描画するフォントとサイズを指定します。
日本語を描画する場合は対応していないフォントもあるので、プレビューなどで日本語が入っているものを選択しましょう。

date = event.postback.params["date"]で撮影日を取得し、how_old = datetime.datetime.strptime(date, "%Y-%m-%d") - datetime.datetime.strptime(str(birthday), "%Y-%m-%d")で撮影日から生年月日を引いた日数を取得します。
years, days = divmod(how_old.days, 365)で日数を365(日)で割った商と余りを計算し、yearsに年齢、daysに余りの日数を代入します。
month = days // 12で余りの日数を月に変換し、text = text_name + f"({years}才{month}ヶ月)"で書き込む内容を定義します。

rect_size = ((text_width + margin * 6), (text_height + margin * 2))で矩形とマスクのサイズを、テキストの左右30px上下10pxずつ余白をとります。
最後に、im.paste(rect, (x - margin * 6, y - margin * 3), mask)で矩形とマスクの位置を指定し貼り付け、その中心になるようにdraw.text((x - margin * 3, y - margin * 2), text, fill=(255, 255, 255), font=font)でテキストを書き込み完成です。

実行

main.py

if __name__ == "__main__":
    #app.run()
    port = int(os.getenv("PORT", 5000))
    app.run(host="0.0.0.0", port=port)

database.py

準備

database.py

from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import create_engine, Column, String, Date
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session
from sqlalchemy.orm.exc import NoResultFound
import datetime
import os

#環境変数からデータベースのURLを取得
DATABASE_URL = os.environ["DATABASE_URL"]

engine = create_engine(DATABASE_URL)
Base = declarative_base()

まず、databaseファイルの必要なモジュールをインポートします。
DATABASE_URLはheroku configで設定してあるデータベースのURLを指定します。
engine = create_engine(DATABASE_URL)、Base = declarative_base()は次に行うテーブル作成に使用します。

テーブル作成

database.py

#テーブル定義
class User(Base):
	__tablename__ = "user_list"
	user_id = Column("user_id", String(50), primary_key=True)
	name1 = Column("name1", String(10))
	day1 = Column("day1", Date)
	name2 = Column("name2", String(10))
	day2 = Column("day2", Date)
	name3 = Column("name3", String(10))
	day3 = Column("day3", Date)

Base.metadata.create_all(engine)
session = Session(bind=engine)

class User(Base):で、作成時みのテーブル名とカラム内容を指定します。
この時、どこかのカラムにprimary_keyを指定する必要があるので、重複が起こらないカラムに指定します。
Base.metadata.create_all(engine)、session = Session(bind=engine)でデータベースに接続する準備を行います。

user_idの検索

database.py

#user_idの検索
def serch_data(user_id):
	try:
		#user_idで検索
		res = session.query(User.name1, User.day1, User.name2, User.day2, User.name3, User.day3).filter(User.user_id==f"{user_id}").one()
		
		#nameとdayをそれぞれリストへ挿入
		name_list = [n for n in res if type(n) is str]
		day_list = [str(d) for d in res if type(d) is datetime.date]
		return name_list, day_list
	#user_idの登録が無い場合パスする
	except NoResultFound:
		pass

res = session.query(User.name1, User.day1, User.name2, User.day2, User.name3, User.day3).filter(User.user_id==f"{user_id}").one()でuser_idが一致する行のuser_id以外の情報を取得します。
user_idは重複することがないので、one()で1つのみ指定します。

name_list = [n for n in res if type(n) is str]でstr型(名前)をname_listへ、day_list = [str(d) for d in res if type(d) is datetime.date]でdatetime.date型(生年月日)をday_listへ文字列として挿入します。
resで情報を取得した時、空のカラムはNoneが取得されますが上記のどちらの型にも当てはまらないので、if ~ is not Noneなどは必要ありません。
return name_list, day_listで2つのリストを返し、main.pyのリストを更新します。

resでuser_idの登録がなかった場合、NoResultFoundとなるのでexceptで指定し回避します。

ユーザー情報の変更

database.py

#user_idの登録
def add_data(user_id):
	session.add(User(user_id=f"{user_id}"))

#ユーザー情報の更新
def update_data(user_id, num, text_name, birthday):
	user_data = session.query(User).filter(User.user_id==f"{user_id}").one()
	
	#登録数に応じてnameとdayを保存
	if num == 0:
		user_data.name1 = text_name
		user_data.day1 = birthday
	elif num == 1:
		user_data.name2 = text_name
		user_data.day2 = birthday
	elif num == 2:
		user_data.name3 = text_name
		user_data.day3 = birthday

session.add(User(user_id=f"{user_id}"))でuser_idの登録が無かった場合追加します。

user_data = session.query(User).filter(User.user_id==f"{user_id}").one()でユーザー情報を取得し、ifで分岐させ適当なカラムを更新します。
この時、必要なカラムのみを取得し更新しようとしましたがエラーが出たので全ての情報を取得した方が無難なようです。

データベースの接続を切る

database.py

#データベースの変更を保存し、接続を切る
def close_db():
	session.commit()
	session.close()

session.commit()でデータベースの変更を保存し、session.close()で接続を切って終了です。

ファイル構成

app
- main.py
- database.py
- Procfile
- requirements
- runtime
- static
  - images
    - hoge
- font
  - AquaKana.ttf

まとめ

前回の完成から約2ヶ月で、初めて扱うデータベースを何とか目標としていた所まで組み込むことができました。
途中でMySQLからPostgreSQLへの変更や、当初使う予定のなかったsqlalchemyの使用など新しいことに挑戦しましたが前回と比べ比較的スムーズに実装することができ、僅かながら成長を感じることができました。

このボットについては一旦完成として、時間があるときに不便な点などを改良したアプリなどを作成してみたいです。

参考