氣象機器人 (1) - 雷達回波與地震資訊

透過之前的幾篇教學，應該已經熟悉一個 LINE BOT 的開發流程，接下來會銜接之前的範例，搭配氣象爬蟲的功能，開始實一個 LINE 的氣象機器人，這篇教學會先進行環境的設定，並串接雷達回波圖與地震資訊。

建立 LINE BOT Channel

參考「建立 LINE Channel」文章，進入 LINE Developer 的控制台，新增一個氣象機器人的 LINE Channel。

LINE BOT 教學 ( Python )- 開發氣象機器人 (1)

串接 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()

LINE BOT 教學 ( Python )- 開發氣象機器人 (1)

Webhook 完成後，回到 LINE 將這個氣象機器人加入好友，並傳送訊息，在 Colab 裡就會出現傳送的訊息內容。

LINE BOT 教學 ( Python )- 開發氣象機器人 (1)

串接雷達回波圖

參考「LINE Notify 傳送雷達回波圖」文章，由於雷達回波圖的圖片網址是固定的，所以只要修改原本的程式碼，當收到的訊息為「雷達回波圖」或「雷達回波」時，使用 reply message 的方法回傳圖片訊息即可 ( 圖片後方記得加上時間戳記，避免因為緩存的關係都顯示相同的圖片 )

雷達回波圖片網址：

較大範圍：https://cwbopendata.s3.ap-northeast-1.amazonaws.com/MSC/O-A0058-001.png

鄰近區域：https://cwbopendata.s3.ap-northeast-1.amazonaws.com/MSC/O-A0058-003.png

注意，2023 年 9 月份中央氣象局修改雷達回波圖網址，將 MSC 改為 Observation，修改後的圖片網址：https://cwbopendata.s3.ap-northeast-1.amazonaws.com/Observation/O-A0058-003.png。

下方的程式碼修改的重點如下：

建立了一個 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 BOT 教學 ( Python )- 開發氣象機器人 (1)

串接地震資訊

參考「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 裡面輸入「地震」或「地震資訊」，就會收到最新一筆的地震資料和地震圖。

LINE BOT 教學 ( Python )- 開發氣象機器人 (1)

小結

氣象機器人的功能不只如此，接下來的幾篇會繼續介紹如何串接即時氣象、天氣預報和空氣品質資訊。

繼續閱讀：

氣象機器人 (2) - 目前氣象資訊

意見回饋

如果有任何建議或問題，可傳送「意見表單」給我，謝謝～

Python 教學

氣象機器人 (1) - 雷達回波與地震資訊

建立 LINE BOT Channel

串接 Colab + ngrok 測試用的 Webhook

串接雷達回波圖

串接地震資訊

小結

意見回饋

Python 教學

基本介紹

資料型別

語法觀念

函式操作

內建函式＆方法

標準函式庫＆模組

網路爬蟲

網頁服務與應用

LINE BOT 教學

OpenCV 教學

AI 影像辨識教學

NumPy 教學

matplotlib 圖表

Tkinter 介面設計

PyQt5 介面設計

PyQt6 介面設計

影音處理範例

實際應用範例

基礎範例

數學範例

ZeroJudge 解答