气象机器人 (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 import Flask, request
from pyngrok import ngrok # Colab 環境需要，本機環境不需要
# 載入 LINE Message API 相關函式庫
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import MessageEvent, TextMessage, TextSendMessage, StickerSendMessage, ImageSendMessage, LocationSendMessage
import json
app = Flask(__name__)
# Colab 環境需要下面這三行，本機環境不需要
port = "5000"
public_url = ngrok.connect(port).public_url
print(f" * ngrok tunnel \"{public_url}\" -> \"http://127.0.0.1:{port}\" ")
@app.route("/", methods=['POST'])
def linebot():
body = request.get_data(as_text=True) # 取得收到的訊息內容
try:
line_bot_api = LineBotApi('你的 Access Token') # 確認 token 是否正確
handler = WebhookHandler('你的 Channel Serect') # 確認 secret 是否正確
signature = request.headers['X-Line-Signature'] # 加入回傳的 headers
handler.handle(body, signature) # 綁定訊息回傳的相關資訊
json_data = json.loads(body) # 轉換內容為 json 格式
print(json_data) # 印出內容
except:
print(body) # 如果發生錯誤，印出收到的內容
return 'OK' # 驗證 Webhook 使用，不能省略
if __name__ == "__main__":
app.run()
Webhook 完成后，回到 LINE 将这个气象机器人加入好友，并传送消息，在 Colab 里就会出现传送的消息内容。
串接雷达回波图
参考“LINE Notify 传送雷达回波图”文章，由于雷达回波图的图片网址是固定的，所以只要修改原本的程序码，当收到的消息为“雷达回波图”或“雷达回波”时，使用 reply message 的方法回传图片消息即可 ( 图片后方记得加上时间戳记，避免因为缓存的关系都显示相同的图片 )
雷达回波图片网址：https://cwaopendata.s3.ap-northeast-1.amazonaws.com/Observation/O-A0058-001.png
下方的程序码修改的重点如下：
- 将 access token 和 channel_secret 独立为变数。
- 使用 if 判断消息是否出现“雷达回波图”或“雷达回波”。
- 使用标准函数库 time 来增加图片的时间戳记。
from flask import Flask, request
from pyngrok import ngrok # Colab 環境需要，本機環境不需要
import time
# 載入 LINE Message API 相關函式庫
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import MessageEvent, TextMessage, TextSendMessage, StickerSendMessage, ImageSendMessage, LocationSendMessage
import json
app = Flask(__name__)
# Colab 環境需要下面這三行，本機環境不需要
port = "5000"
public_url = ngrok.connect(port).public_url
print(f" * ngrok tunnel \"{public_url}\" -> \"http://127.0.0.1:{port}\" ")
access_token = '你的 Access Token'
channel_secret = '你的 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) # 印出內容
type = json_data['events'][0]['message']['type']
if type == 'text':
text = json_data['events'][0]['message']['text']
if text == '雷達回波圖' or text == '雷達回波':
line_bot_api.push_message(user_id, TextSendMessage(text='馬上找給你！抓取資料中....')) # 一開始先發送訊息
img_url = f'https://cwaopendata.s3.ap-northeast-1.amazonaws.com/Observation/O-A0058-001.png?{time.time_ns()}'
img_message = ImageSendMessage(original_content_url=img_url, preview_image_url=img_url)
line_bot_api.reply_message(reply_token, img_message) # 回傳訊息
else:
text_message = TextSendMessage(text=text)
line_bot_api.reply_message(reply_token,text_message)
except Exception as e:
print(e) # 發生錯誤就印出完整錯誤內容
return 'OK' # 驗證 Webhook 使用，不能省略
if __name__ == "__main__":
app.run()
串接地震信息
首先执行下方函数，观察是否能够正确抓到地震信息，由于气象局提供的地震信息有两组，因此两组都要抓取数据，再透过判断时间的先后顺序，取出最新的地震数据。
import requests
def earth_quake():
result = []
try:
code = '你的氣象 token'
# 小區域 https://opendata.cwa.gov.tw/dataset/earthquake/E-A0016-001
url = f'https://opendata.cwa.gov.tw/api/v1/rest/datastore/E-A0016-001?Authorization={code}'
req1 = requests.get(url) # 爬取資料
data1 = req1.json() # 轉換成 json
eq1 = data1['records']['Earthquake'][0] # 取得第一筆地震資訊
t1 = data1['records']['Earthquake'][0]['EarthquakeInfo']['OriginTime']
# 顯著有感 https://opendata.cwa.gov.tw/dataset/all/E-A0015-001
url2 = f'https://opendata.cwa.gov.tw/api/v1/rest/datastore/E-A0015-001?Authorization={code}'
req2 = requests.get(url2) # 爬取資料
data2 = req2.json() # 轉換成 json
eq2 = data2['records']['Earthquake'][0] # 取得第一筆地震資訊
t2 = data2['records']['Earthquake'][0]['EarthquakeInfo']['OriginTime']
result = [eq1['ReportContent'], eq1['ReportImageURI']] # 先使用小區域地震
if t2>t1:
result = [eq2['ReportContent'], eq2['ReportImageURI']] # 如果顯著有感地震時間較近，就用顯著有感地震
except Exception as e:
print(e)
result = ['抓取失敗...','']
return result
earth_quake()
将爬取地震信息的程序和 LINE BOT 的程序组合，就能透过发送“地震”两个字来触发抓取地震信息的函数。
from flask import Flask, request
from pyngrok import ngrok # Colab 環境需要，本機環境不需要
import json, time, requests
# 載入 LINE Message API 相關函式庫
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import MessageEvent, TextMessage, TextSendMessage, StickerSendMessage, ImageSendMessage, LocationSendMessage
app = Flask(__name__)
# Colab 環境需要下面這三行，本機環境不需要
port = "5000"
public_url = ngrok.connect(port).public_url
print(f" * ngrok tunnel \"{public_url}\" -> \"http://127.0.0.1:{port}\" ")
def earth_quake():
result = []
code = '你的 token'
try:
# 小區域 https://opendata.cwa.gov.tw/dataset/earthquake/E-A0016-001
url = f'https://opendata.cwa.gov.tw/api/v1/rest/datastore/E-A0016-001?Authorization={code}'
req1 = requests.get(url) # 爬取資料
data1 = req1.json() # 轉換成 json
eq1 = data1['records']['Earthquake'][0] # 取得第一筆地震資訊
t1 = data1['records']['Earthquake'][0]['EarthquakeInfo']['OriginTime']
# 顯著有感 https://opendata.cwa.gov.tw/dataset/all/E-A0015-001
url2 = f'https://opendata.cwa.gov.tw/api/v1/rest/datastore/E-A0015-001?Authorization={code}'
req2 = requests.get(url2) # 爬取資料
data2 = req2.json() # 轉換成 json
eq2 = data2['records']['Earthquake'][0] # 取得第一筆地震資訊
t2 = data2['records']['Earthquake'][0]['EarthquakeInfo']['OriginTime']
result = [eq1['ReportContent'], eq1['ReportImageURI']] # 先使用小區域地震
if t2>t1:
result = [eq2['ReportContent'], eq2['ReportImageURI']] # 如果顯著有感地震時間較近，就用顯著有感地震
except Exception as e:
print(e)
result = ['抓取失敗...','']
return result
access_token = '你的 Access Token'
channel_secret = '你的 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) # 印出內容
type = json_data['events'][0]['message']['type']
if type == 'text':
text = json_data['events'][0]['message']['text']
if text == '雷達回波圖' or text == '雷達回波':
line_bot_api.push_message(user_id, TextSendMessage(text='馬上找給你！抓取資料中....'))
img_url = f'https://cwaopendata.s3.ap-northeast-1.amazonaws.com/Observation/O-A0058-001.png?{time.time_ns()}'
img_message = ImageSendMessage(original_content_url=img_url, preview_image_url=img_url)
line_bot_api.reply_message(reply_token,img_message)
elif text == '地震':
line_bot_api.push_message(user_id, TextSendMessage(text='馬上找給你！抓取資料中....'))
reply = earth_quake() # 執行函式，讀取數值
text_message = TextSendMessage(text=reply[0]) # 取得文字內容
line_bot_api.reply_message(reply_token,text_message) # 傳送文字
line_bot_api.push_message(user_id, ImageSendMessage(original_content_url=reply[1], preview_image_url=reply[1])) # 傳送圖片
else:
text_message = TextSendMessage(text=text)
line_bot_api.reply_message(reply_token,text_message)
except Exception as e:
print(e) # 發生錯誤就印出完整錯誤內容
return 'OK' # 驗證 Webhook 使用，不能省略
if __name__ == "__main__":
app.run()
完成后重新执行，设定新的 ngrok 网址为 Webhook，验证成功后，在 LINE 里面输入“地震”，就会收到最新一笔的地震数据和地震图。
小结
气象机器人的功能不只如此，接下来的几篇会继续介绍如何串接即时气象、天气预报和空气质量信息。
继续阅读：
微信扫码关注
抖音扫码关注