「python で作る対話システム」の天気予報ボットを参考に公式ラインに天気予報させてみた。

pythonで作る対話システムに書かれた内容と変えた点

・XMLではなくSQLite3を使って状態遷移をさせた

ポイント

・本のコードをそのまま公式ラインに使用しようとすると遷移情報の保持が出来なかったため、SQLite3を使用した。

次に目指すもの

「大阪じゃなくて」のように入力を取り消すことが出来るようにする。
pythonで作る対話システムの非タスク編へ

プログラムの構造

下記：天気予報公式ラインのコード(訂正版）

w_system.py

import sys
from PySide2 import QtCore, QtScxml
import requests
import json
from datetime import datetime, timedelta, time
import sqlite3

DATES = ['今日','明日']

TYPES = ['天気','気温']

# 都道府県名から緯度と経度を取得するための辞書
PLACES = {'北海道': (43.06, 141.35), '青森': (40.82, 140.74), '岩手': (39.7, 141.15), '宮城': (38.27, 140.87),
        '秋田': (39.72, 140.1), '山形': (38.24, 140.36), '福島': (37.75, 140.47), '茨城': (36.34, 140.45),
        '栃木': (36.57, 139.88), '群馬': (36.39, 139.06), '埼玉': (35.86, 139.65), '千葉': (35.61, 140.12),
        '東京': (35.69, 139.69), '神奈川': (35.45, 139.64), '新潟': (37.9, 139.02), '富山': (36.7, 137.21),
        '石川': (36.59, 136.63), '福井': (36.07, 136.22), '山梨': (35.66, 138.57), '長野': (36.65, 138.18),
        '岐阜': (35.39, 136.72), '静岡': (34.98, 138.38), '愛知': (35.18, 136.91), '三重': (34.73, 136.51),
        '滋賀': (35.0, 135.87), '京都': (35.02, 135.76), '大阪': (34.69, 135.52), '兵庫': (34.69, 135.18),
        '奈良': (34.69, 135.83), '和歌山': (34.23, 135.17), '鳥取': (35.5, 134.24), '島根': (35.47, 133.05),
        '岡山': (34.66, 133.93), '広島': (34.4, 132.46), '山口': (34.19, 131.47), '徳島': (34.07, 134.56),
        '香川': (34.34, 134.04), '愛媛': (33.84, 132.77), '高知': (33.56, 133.53), '福岡': (33.61, 130.42),
        '佐賀': (33.25, 130.3), '長崎': (32.74, 129.87), '熊本': (32.79, 130.74), '大分': (33.24, 131.61),
        '宮崎': (31.91, 131.42), '鹿児島': (31.56, 130.56), '沖縄': (26.21, 127.68)}


current_weather_url = 'http://api.openweathermap.org/data/2.5/weather'
forecast_url = 'http://api.openweathermap.org/data/2.5/forecast'
appid = '' # 自身のAPPIDを入れてください  


def get_current_weather(lat,lon):
# 天気情報を取得    
    response = requests.get("{}?lat={}&lon={}&lang=ja&units=metric&APPID={}".format(self.current_weather_url,lat,lon,self.appid))
    return response.json()

def get_tomorrow_weather(lat,lon):
    # 今日の時間を取得
    today = datetime.today()
    # 明日の時間を取得
    tomorrow = today + timedelta(days=1)
    # 明日の正午の時間を取得
    tomorrow_noon = datetime.combine(tomorrow, time(12,0))
    # UNIX時間に変換
    timestamp = tomorrow_noon.timestamp()
    # 天気情報を取得
    response = requests.get("{}?lat={}&lon={}&lang=ja&units=metric&APPID={}".format(forecast_url,lat,lon,appid))
    dic = response.json()
    # 3時間おきの天気情報についてループ
    for i in range(len(dic["list"])):
        # i番目の天気情報（UNIX時間）
        dt = float(dic["list"][i]["dt"])
        # 明日の正午以降のデータになった時点でその天気情報を返す
        if dt >= timestamp:
            return dic["list"][i]
    return ""

def create_db(cur):
    cur.execute('CREATE TABLE test(place STRING,date STRING, type STRING)')

def update_db(cur, text):
    for place in PLACES:
        if place in text:
            cur.execute(f'INSERT INTO test values("{place}","","")')
    for date in DATES:
        if date in text:
            cur.execute(f'INSERT INTO test values("", "{date}", "")')
    for type_ in TYPES:
        if type_ in text:
            cur.execute(f'INSERT INTO test values("", "", "{type_}")')

def read_db(cur):
    cur.execute('SELECT * FROM test')
    data = cur.fetchall()
    place, date, type_ = "","",""
    for p, d, t in data:
        place = p or place
        date = d or date
        type_ = t or type_
    print(place,date,type_)
    return place,date,type_

def clear_db():
    dbname = "weather.db"
    conn = sqlite3.connect(dbname)
    cur = conn.cursor()
    cur.execute('DROP TABLE test')

def reply_000(place,date,type_):
    return 'ご用件は？'

def reply_001(place,date,type_):
    return '都道府県といつか（今日・明日）を教えてください'

def reply_011(place,date,type_):
    return '都道府県を教えてください'

def reply_100(place,date,type_):
    return 'いつか（今日・明日）と何（天気・気温）を知りたいか教えてください'

def reply_110(place,date,type_):
    return '何（天気・気温）を知りたいですか'

def reply_101(place,date,type_):
    return 'いつか（今日・明日）を教えてください'

def reply_010(place,date,type_):
    return '都道府県と何（天気・気温）を知りたいか教えてください'

def reply_111(place,date,type_):
    print("placeは",place)
    lat = PLACES[place][0] # placeから緯度を取得
    lon = PLACES[place][1] # placeから経度を取得       
    print("lat=",lat,"lon=",lon)
    #テーブルの消去
    clear_db()
    if date == "今日":
        print("今日の天気予報")
        cw = get_current_weather(lat,lon)
        if type_ == "天気":
            return(cw["weather"][0]["description"]+"です")
        elif type_ == "気温":
            return(str(cw["main"]["temp"])+"度です")
    elif date == "明日":
        tw = get_tomorrow_weather(lat,lon)
        if type_ == "天気":
            return(tw["weather"][0]["description"]+"です")
        elif type_ == "気温":
            return (str(tw["main"]["temp"])+"度です")

def start(text):
    dbname = "weather.db"
    conn = sqlite3.connect(dbname)
    cur = conn.cursor()
    if text == "start":
        create_db(cur)
    else:
        update_db(cur,text)
    conn.commit()
    place,date,type_ = read_db(cur)
    print(place,"だお")
    return place,date,type_

PATTERN = {
    (False,False,False):reply_000,
    (False,False,True):reply_001,
    (False,True,True):reply_011,
    (True,False,False):reply_100,
    (True,True,False):reply_110,
    (True,False,True):reply_101,
    (False,True,False):reply_010,
    (True,True,True):reply_111,
}

def apply(text):
    place, date, type_ = start(text)
    return PATTERN[place!="",date!="",type_!=""](place,date,type_)

main.py

from flask import Flask,request,abort
from linebot import LineBotApi,WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import MessageEvent,TextMessage,TextSendMessage
import os
import requests
import pprint
import w_system

app=Flask(__name__)
#環境変数の取得
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)


@app.route("/callback",methods=["POST"])
def callback():
    print("コールバック1")
    signature=request.headers["X-Line-Signature"]
    print(signature)
    body=request.get_data(as_text=True)
    app.logger.info("Request body"+body)

    try:
        handler.handle(body,signature)
    except InvalidSignatureError:
        abort(400)
    print('コールバック')
    return "OK"


@handler.add(MessageEvent,message=TextMessage)
def handle_message(event):
    print('ハンドルメッセージ')
    #入力された文字列を格納
    push_text = event.message.text
    reply_text = w_system.apply(push_text)

    #リプライ部分の記述
    line_bot_api.reply_message(event.reply_token,TextSendMessage(text=reply_text))


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