はじめに

つい最近、台湾で大きな地震がありましたね。
一人でも多くの無事と一日も早い復興を願ってます。

さて、今回は台湾の最新の地震情報を取得できるAPIについてご紹介したいと思います。
台湾には中央気象署という気象業務を行う公的機関があり、日本でいう気象庁のような役割を果たしています。

中央気象署は、台湾国内の最新の気象情報を入手できるオープンデータを提供しており、会員登録をすれば誰でも無料で利用することができます。

まずは会員登録を済ませましょう。

中央気象署会員になる

1. 上記のサイトにジャンプし、利用規約を確認したら「同意」をクリックします。

2. 登録情報を入力します。 Google翻訳を掛ければ読めると思いますが、下記に解説します。

電子郵件 - メールアドレス
密碼 - パスワード（6文字以上、12文字以下）（スペース、「+」「%」「&」は使用不可）
再次輸入密碼 - パスワードの再入力
訂閱 - 受け取りたいメールの種類をチェック
- 每日氣象電子報 - 天気日報
- 天氣警、特報 - 気象警報
- 地震報告 - 地震情報

そして最後の「驗證碼」が確認コードです。
全て入力し終わったら「送出」を押しましょう。

パスワードを自動生成に頼っていて、パスワード欄をコピー&ペーストで入力した場合は認証に失敗する場合があります。
最後まで入力したのに「驗證碼失效」が出る場合は、パスワードを手入力してみてください。または、別のパスワードも試してみてください。

3. この段階では仮会員です。
入力したメールアドレス宛に確認メールが届くので、URLをクリックします。

これで本登録は完了です！

アクセスキーの発行

APIにアクセスするには、まずアクセスキーを入手する必要があります。

1. 上記のオープンデータ用HPにアクセスし、右上の「登入/註冊」をクリックします。
2. 「氣象會員登入」を選択し、メールアドレスとパスワードを入力、最後にCAPTCHAチェックしてログインします。
3. 自動でアクセスキーを入手するページに飛ぶので、「取得授權碼」をクリックして、アクセスキーを発行します。右に出てくる「CWA-」で始まる文字がアクセスキーです。

発行したアクセスキーは忘れずにメモしておきましょう。

もしアクセスキーを紛失した場合は、「更新授權碼」で再発行できます。
しかし発行済みのアクセスキーは無効となる上、7日間は再発行できなくなるので、注意しましょう。

APIにアクセスする

1. 上記リンクをクリックすると、地震津波情報のオープンデータ一覧が確認できます。
今回は震度3級以上の地震情報を取得したいので、「顯著有感地震報告」を選択します。
2. 「資料擷取API服務說明網址」項目の「API」をクリックします。

3. Swaggerの選択した項目にジャンプしました。右に「Try it out」をクリックします。
4. 各パラメータを入力します。一番上の「Authorization」が必須項目で、先ほど入手したアクセスキーをコピペしておきます。
5. 最後に「Execute」をクリックすると結果が返されます。あとは煮るなり焼くなり好きにしましょう。

LINEに通知する

LINE Notifyを使って通知します。

ソースコード

pip install requests
pip install googletrans==3.1.0a0

import requests
import time
import datetime
from googletrans import Translator
import re
import os
import urllib.request


def cst_to_jst(dt):  # 台湾時間から日本時間への変換
    cst = datetime.datetime.strptime(dt, "%Y-%m-%d %H:%M:%S")
    jst = cst + datetime.timedelta(hours=1)
    return jst


def translate(str_zh):  # 翻訳
    translator = Translator()
    trans_jp = translator.translate(str_zh, dest="ja")
    return trans_jp.text


def notifysend(message, img):  # 通知系APIへのPOST
    notify_url = "https://notify-api.line.me/api/notify"
    if img:
        with open(img, "rb") as f:
            r = requests.post(notify_url,
                              headers={"Authorization": "Bearer " + token},
                              data={"message": message},
                              files={"imageFile": f.read()},
                              params={"notificationDisabled": True}
                              )
    else:
        r = requests.post(notify_url,
                          headers={"Authorization": "Bearer " + token},
                          data={"message": message},
                          params={"notificationDisabled": True}
                          )
    return r


def notify_postlimit(p):  # APIレート制限確認用
    ratelimit = p.headers.get("X-RateLimit-Limit")
    ratelimit_re = p.headers.get("X-RateLimit-Remaining")
    imgratelimit = p.headers.get("X-RateLimit-ImageLimit")
    imgratelimit_re = p.headers.get("X-RateLimit-ImageRemaining")
    ratereset = p.headers.get("X-RateLimit-Reset")
    reset_time = int(ratereset) - int(time.time())
    m, s = divmod(reset_time, 60)

    print(f"LINE Notify送信完了。 {datetime.datetime.now()}")
    print(f"残りAPIコール回数：{ratelimit_re} / {ratelimit}")
    print(f"残りImage Upload回数；{imgratelimit_re} / {imgratelimit}")
    print(f"上限リセットまで、あと{m}m{s}s\n")

token = input("LINE Notifyのトークンを入力")
checkNum = 0

url = "https://opendata.cwa.gov.tw/api/v1/rest/datastore/E-A0015-001?Authorization={Authkey}&limit=1&format=JSON&AreaName="
jsonData = requests.get(url).json()

eqNo1 = jsonData["records"]["Earthquake"][0]["EarthquakeNo"]

while True:
    time.sleep(10)
    jsonData = requests.get(url).json()
    eqNo2 = jsonData["records"]["Earthquake"][0]["EarthquakeNo"]

    if eqNo2 != eqNo1:
        eqNo1 = eqNo2
        checkNum += 1
        print(f"EarthQuakeNoの変化を検知。メッセージを送信します。 日時：{datetime.datetime.now()} 回数：{checkNum}\n")

        reportType = jsonData["records"]["Earthquake"][0]["ReportType"]
        reportColor = jsonData["records"]["Earthquake"][0]["ReportColor"]  # 程度の強さ
        # 色の区分は仮のもの。エラーが出たら疑うべき
        if reportColor == "綠色":  # 下の基準に満たない
            eqDeal = ""  # 表現の度合い
            expression = "。"
        elif reportColor == "黃色":  # M5.5かつ震度4級以上
            eqDeal = "やや強い"
            expression = "。"
        elif reportColor == "橙色":  # M6.0かつ震度5弱以上
            eqDeal = "強い"
            expression = "！"
        elif reportColor == "紅色":  # M6.5かつ震度6弱以上
            eqDeal = "非常に強い"
            expression = "！！"
        else:
            raise ValueError("想定外のreportColor")

        riURI = jsonData["records"]["Earthquake"][0]["ReportImageURI"]  # 震度地図
        imgName = riURI.split("/")[-1]
        os.makedirs("eqmap", exist_ok=True)
        dl_path = f"./eqmap/{imgName}"
        urllib.request.urlretrieve(riURI, dl_path)

        eqInfo = jsonData["records"]["Earthquake"][0]["EarthquakeInfo"]

        originTime = eqInfo["OriginTime"]  # 発生時刻
        otFormat = f"{cst_to_jst(originTime):%d日 %H時%M分頃、}"  # フォーマット済み発生時刻（JST）

        source = eqInfo["Source"]  # 情報元

        epiCenter = eqInfo["Epicenter"]["Location"]  # 震源地
        epiCenter = translate(epiCenter)

        depth = str(eqInfo["FocalDepth"]) + "km"  # 深さ

        mType = eqInfo["EarthquakeMagnitude"]["MagnitudeType"]  # マグニチュード
        mValue = eqInfo["EarthquakeMagnitude"]["MagnitudeValue"]
        if mType == "芮氏規模":  # ローカルマグニチュード
            mUnit = "ML"
        else:
            mUnit = "M"

        shakingArea = jsonData["records"]["Earthquake"][0]["Intensity"]["ShakingArea"]
        shakingArea = sorted(shakingArea, key=lambda a: a["AreaIntensity"], reverse=True)
        maxInt = shakingArea[0]["AreaIntensity"]
        skaList = []

        for ska in shakingArea:
            if re.findall("最大震度.*地區", ska["AreaDesc"]):
                areaInt = ska["AreaIntensity"]
                county = translate(ska["CountyName"])
                res = f"【震度{areaInt}】\n{county}"
                skaList.append(res)

        mainHead = f"\n【🇹🇼CWA{reportType}】\n{otFormat}台湾で{eqDeal}地震があったピヨ{expression}"
        mainBody = f"震源地：{epiCenter}\n震源の深さ：{depth}\n最大震度：{maxInt}\n推定マグニチュード：{mUnit}{mValue}"
        mainFoot = f"ソース：{source}"

        mainAns = f"{mainHead}\n\n{mainBody}\n\n{mainFoot}"
        notifysend(mainAns, dl_path)

        skaHead = "\n各地の震度は次の通りピヨ🐤"
        skaBody = "\n".join(skaList)

        skaAns = f"{skaHead}\n\n{skaBody}"
        post = notifysend(skaAns, None)
        notify_postlimit(post)

    else:
        checkNum += 1
        print(f"更新完了。 更新日時：{datetime.datetime.now()}　回数：{checkNum}\n")

最後に

気象庁の防災情報もJSON化してほしい（切実）

出典