氣象機器人 (1) - 雷達回波與地震資訊
透過之前的幾篇教學,應該已經熟悉一個 LINE BOT 的開發流程,接下來會銜接之前的範例,搭配氣象爬蟲的功能,開始實一個 LINE 的氣象機器人,這篇教學會先進行環境的設定,並串接雷達回波圖與地震資訊。
建立 LINE BOT Channel
參考「建立 LINE Channel」文章,進入 LINE Developer 的控制台,新增一個氣象機器人的 LINE Channel。
串接 Colab + ngrok 測試用的 Webhook
參考「使用 ngrok 服務」、「使用 Google Cloud Functions」和「建立並串接 Webhook」三篇教學,安裝 ngrok 和 line-bot-sdk,執行下方的程式碼產生 Webhook,回到 LINE Channel,確認 Webhook 可以正常運作 ( 驗證 Verify 後出現 success 表示 Webhook 沒有問題 )。
注意!如果使用 ngrok 搭配 Colab 做開發測試,每次執行後產生的網址都不同,需要重複更新 LINE 的 Webhook。
from flask_ngrok import run_with_ngrok
from flask import Flask, request
# 載入 LINE Message API 相關函式庫
from linebot import LineBotApi, WebhookHandler
from linebot.models import MessageEvent, TextMessage, TextSendMessage
# 載入 json 標準函式庫,處理回傳的資料格式
import json
app = Flask(__name__)
@app.route("/", methods=['POST'])
def linebot():
body = request.get_data(as_text=True) # 取得收到的訊息內容
try:
line_bot_api = LineBotApi('你的 LINE Channel access token') # 確認 token 是否正確
handler = WebhookHandler('你的 LINE Channel secret') # 確認 secret 是否正確
signature = request.headers['X-Line-Signature'] # 加入回傳的 headers
handler.handle(body, signature) # 綁定訊息回傳的相關資訊
json_data = json.loads(body) # 轉換內容為 json 格式
print(json_data) # 印出內容
except:
print('error') # 如果發生錯誤,印出 error
return 'OK' # 驗證 Webhook 使用,不能省略
if __name__ == "__main__":
run_with_ngrok(app) # 串連 ngrok 服務
app.run()
Webhook 完成後,回到 LINE 將這個氣象機器人加入好友,並傳送訊息,在 Colab 裡就會出現傳送的訊息內容。
串接雷達回波圖
參考「LINE Notify 傳送雷達回波圖」文章,由於雷達回波圖的圖片網址是固定的,所以只要修改原本的程式碼,當收到的訊息為「雷達回波圖」或「雷達回波」時,使用 reply message 的方法回傳圖片訊息即可 ( 圖片後方記得加上時間戳記,避免因為緩存的關係都顯示相同的圖片 )
雷達回波圖片網址:
下方的程式碼修改的重點如下:
- 建立了一個 reply_image 函式,使用 requests 的 GET 方法傳送圖片。
- 因為傳送訊息時需要 access token 和 reply token,所以將這兩個值變成 reply_message 函式的參數。
- 將 access token 和 channel_secret 獨立為變數。
- 使用 if 判斷訊息是否出現「雷達回波圖」或「雷達回波」。
- 使用標準函式庫 time 來增加圖片的時間戳記。
from flask_ngrok import run_with_ngrok
from flask import Flask, request
# 載入 LINE Message API 相關函式庫
from linebot import LineBotApi, WebhookHandler
from linebot.models import MessageEvent, TextMessage, TextSendMessage
# 載入 json 標準函式庫,處理回傳的資料格式
import requests, json, time
app = Flask(__name__)
access_token = '你的 LINE Channel access token'
channel_secret = '你的 LINE Channel secret'
@app.route("/", methods=['POST'])
def linebot():
body = request.get_data(as_text=True) # 取得收到的訊息內容
try:
line_bot_api = LineBotApi(access_token) # 確認 token 是否正確
handler = WebhookHandler(channel_secret) # 確認 secret 是否正確
signature = request.headers['X-Line-Signature'] # 加入回傳的 headers
handler.handle(body, signature) # 綁定訊息回傳的相關資訊
json_data = json.loads(body) # 轉換內容為 json 格式
reply_token = json_data['events'][0]['replyToken'] # 取得回傳訊息的 Token ( reply message 使用 )
user_id = json_data['events'][0]['source']['userId'] # 取得使用者 ID ( push message 使用 )
print(json_data) # 印出內容
if 'message' in json_data['events'][0]: # 如果傳送的是 message
if json_data['events'][0]['message']['type'] == 'text': # 如果 message 的類型是文字 text
text = json_data['events'][0]['message']['text'] # 取出文字
if text == '雷達回波圖' or text == '雷達回波': # 如果是雷達回波圖相關的文字
# 傳送雷達回波圖 ( 加上時間戳記 )
reply_image(f'https://cwbopendata.s3.ap-northeast-1.amazonaws.com/MSC/O-A0058-003.png?{time.time_ns()}', reply_token, access_token)
else:
reply_message(text, reply_token, access_token) # 如果是一般文字,直接回覆同樣的文字
except:
print('error') # 如果發生錯誤,印出 error
return 'OK' # 驗證 Webhook 使用,不能省略
if __name__ == "__main__":
run_with_ngrok(app) # 串連 ngrok 服務
app.run()
# LINE 回傳圖片函式
def reply_image(msg, rk, token):
headers = {'Authorization':f'Bearer {token}','Content-Type':'application/json'}
body = {
'replyToken':rk,
'messages':[{
'type': 'image',
'originalContentUrl': msg,
'previewImageUrl': msg
}]
}
req = requests.request('POST', 'https://api.line.me/v2/bot/message/reply', headers=headers,data=json.dumps(body).encode('utf-8'))
print(req.text)
完成後重新執行,設定新的 ngrok 網址為 Webhook,驗證成功後 ( 有時因為快取的緣故,需要重新產生兩次才會正常運作 ),LINE 裡面輸入「雷達回波圖」或「雷達回波」,就會收到雷達回波圖的圖片。
串接地震資訊
參考「LINE Notify 傳送地震資訊」文章,先前往「氣象資料開放平臺」申請個人個授權碼,將上方的程式碼加入抓取地震資訊的函式,修改的重點如下:
- 建立一個 reply_message 函式,使用 requests 的 GET 方法傳送訊息。
- 建立一個 push_message 函式,使用 requests 的 GET 方法傳送訊息 ( 因為 reply 的方法只能傳送一次,如果要傳送兩次就需要使用 push 的方法 )。
- 建立一個 earth_quake 函式,負責爬取地震資訊,組合成串列,串列的第一個項目為文字說明,第二個項目為地震圖,爬取資訊後,回傳第一筆資料。
- 使用 if 判斷訊息是否出現「地震」或「地震資訊」。
from flask_ngrok import run_with_ngrok
from flask import Flask, request
from linebot import LineBotApi, WebhookHandler
from linebot.models import MessageEvent, TextMessage, TextSendMessage
import requests, json, time
app = Flask(__name__)
access_token = '你的 LINE Channel access token'
channel_secret = '你的 LINE Channel secret'
@app.route("/", methods=['POST'])
def linebot():
body = request.get_data(as_text=True)
try:
line_bot_api = LineBotApi(access_token)
handler = WebhookHandler(channel_secret)
signature = request.headers['X-Line-Signature']
handler.handle(body, signature)
json_data = json.loads(body)
reply_token = json_data['events'][0]['replyToken']
user_id = json_data['events'][0]['source']['userId']
print(json_data)
if 'message' in json_data['events'][0]:
if json_data['events'][0]['message']['type'] == 'text':
text = json_data['events'][0]['message']['text']
if text == '雷達回波圖' or text == '雷達回波':
reply_image(f'https://cwbopendata.s3.ap-northeast-1.amazonaws.com/MSC/O-A0058-003.png?{time.time_ns()}', reply_token, access_token)
elif text == '地震資訊' or text == '地震': # 如果是地震相關的文字
msg = earth_quake() # 爬取地震資訊
push_message(msg[0], user_id, access_token) # 傳送地震資訊 ( 用 push 方法,因為 reply 只能用一次 )
reply_image(msg[1], reply_token, access_token) # 傳送地震圖片 ( 用 reply 方法 )
else:
reply_message(text, reply_token, access_token) # 如果是一般文字,直接回覆同樣的文字
except:
print('error')
return 'OK'
if __name__ == "__main__":
run_with_ngrok(app)
app.run()
# 地震資訊函式
def earth_quake():
msg = ['找不到地震資訊','https://example.com/demo.jpg'] # 預設回傳的訊息
try:
code = '你的氣象資料授權碼'
url = f'https://opendata.cwb.gov.tw/api/v1/rest/datastore/E-A0016-001?Authorization={code}'
e_data = requests.get(url) # 爬取地震資訊網址
e_data_json = e_data.json() # json 格式化訊息內容
eq = e_data_json['records']['earthquake'] # 取出地震資訊
for i in eq:
loc = i['earthquakeInfo']['epiCenter']['location'] # 地震地點
val = i['earthquakeInfo']['magnitude']['magnitudeValue'] # 地震規模
dep = i['earthquakeInfo']['depth']['value'] # 地震深度
eq_time = i['earthquakeInfo']['originTime'] # 地震時間
img = i['reportImageURI'] # 地震圖
msg = [f'{loc},芮氏規模 {val} 級,深度 {dep} 公里,發生時間 {eq_time}。', img]
break # 取出第一筆資料後就 break
return msg # 回傳 msg
except:
return msg # 如果取資料有發生錯誤,直接回傳 msg
# LINE push 訊息函式
def push_message(msg, uid, token):
headers = {'Authorization':f'Bearer {token}','Content-Type':'application/json'}
body = {
'to':uid,
'messages':[{
"type": "text",
"text": msg
}]
}
req = requests.request('POST', 'https://api.line.me/v2/bot/message/push', headers=headers,data=json.dumps(body).encode('utf-8'))
print(req.text)
# LINE 回傳訊息函式
def reply_message(msg, rk, token):
headers = {'Authorization':f'Bearer {token}','Content-Type':'application/json'}
body = {
'replyToken':rk,
'messages':[{
"type": "text",
"text": msg
}]
}
req = requests.request('POST', 'https://api.line.me/v2/bot/message/reply', headers=headers,data=json.dumps(body).encode('utf-8'))
print(req.text)
# LINE 回傳圖片函式
def reply_image(msg, rk, token):
headers = {'Authorization':f'Bearer {token}','Content-Type':'application/json'}
body = {
'replyToken':rk,
'messages':[{
'type': 'image',
'originalContentUrl': msg,
'previewImageUrl': msg
}]
}
req = requests.request('POST', 'https://api.line.me/v2/bot/message/reply', headers=headers,data=json.dumps(body).encode('utf-8'))
print(req.text)
完成後重新執行,設定新的 ngrok 網址為 Webhook,驗證成功後 ( 有時因為快取的緣故,需要重新產生兩次才會正常運作 ),LINE 裡面輸入「地震」或「地震資訊」,就會收到最新一筆的地震資料和地震圖。
小結
氣象機器人的功能不只如此,接下來的幾篇會繼續介紹如何串接即時氣象、天氣預報和空氣品質資訊。
繼續閱讀:
意見回饋
如果有任何建議或問題,可傳送「意見表單」給我,謝謝~