Create fetch_cwa_alarm_list.py

fetch_cwa_alarm_list.py

@@ -0,0 +1,118 @@
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+import requests
+from datetime import datetime, timedelta, timezone
+
+CWA_ALARM_API = "https://app-2.cwa.gov.tw/api/v1/earthquake/alarm/list"
+
+def _parse_cwa_time(s: str) -> tuple[str, str]:
+    """
+    解析 CWA 的 originTime。
+    - 若帶有時區或 'Z'，尊重其時區；輸出台灣時間與 UTC。
+    - 若無時區（常見 'YYYY-MM-DD HH:MM:SS'），視為台灣時間 (UTC+8)。
+    回傳: (tw_str, utc_str) -> "YYYY-MM-DD HH:MM"
+    """
+    if not s:
+        return ("未知", "未知")
+    try:
+        if "T" in s or s.endswith("Z") or "+" in s:
+            dt = datetime.fromisoformat(s.replace("Z", "+00:00"))
+        else:
+            dt = datetime.strptime(s, "%Y-%m-%d %H:%M:%S")
+            dt = dt.replace(tzinfo=timezone(timedelta(hours=8)))
+        tw = dt.astimezone(timezone(timedelta(hours=8))).strftime("%Y-%m-%d %H:%M")
+        utc = dt.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M")
+        return (tw, utc)
+    except Exception:
+        return (s, "未知")
+
+def fetch_cwa_alarm_list(limit: int = 5) -> str:
+    """
+    取得中央氣象署「地震預警」清單並格式化為文字（抓常見欄位；最多 limit 筆）。
+    顯示：事件ID、狀態/訊息型別#序號、震級、深度、震中座標、台灣時間與UTC、預警地區。
+    """
+    try:
+        r = requests.get(CWA_ALARM_API, timeout=10)
+        r.raise_for_status()
+        payload = r.json()
+    except Exception as e:
+        return f"❌ 地震預警查詢失敗：{e}"
+
+    # 取出清單
+    items = None
+    if isinstance(payload, dict):
+        items = payload.get("data") or payload.get("records") or payload.get("list") or payload.get("items")
+    if items is None and isinstance(payload, list):
+        items = payload
+    if not items:
+        return "✅ 目前沒有地震預警。"
+
+    # 依 originTime 由新到舊排序
+    def _key(it):
+        s = it.get("originTime") or ""
+        try:
+            if "T" in s or s.endswith("Z") or "+" in s:
+                dt = datetime.fromisoformat(s.replace("Z", "+00:00"))
+            else:
+                dt = datetime.strptime(s, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone(timedelta(hours=8)))
+            return dt.astimezone(timezone.utc)
+        except Exception:
+            return datetime.min.replace(tzinfo=timezone.utc)
+
+    try:
+        items = sorted(items, key=_key, reverse=True)
+    except Exception:
+        pass
+
+    lines = ["🚨 地震預警（最新）:", "-" * 20]
+    count = 0
+    for it in items:
+        if count >= limit:
+            break
+
+        identifier = it.get("identifier") or it.get("eventId") or it.get("id") or "—"
+        status = it.get("status") or "—"
+        msg_type = it.get("msgType") or "—"
+        msg_no = it.get("msgNo") or it.get("msgSeq") or "—"
+
+        def _num(x):
+            xs = str(x)
+            ok = xs.replace(".", "", 1).replace("-", "", 1).isdigit()
+            return float(xs) if ok else None
+
+        mag = _num(it.get("magnitudeValue") or it.get("magnitude") or it.get("ml") or it.get("mw"))
+        mag_str = f"{mag:.1f}" if mag is not None else "—"
+
+        depth = _num(it.get("depth"))
+        depth_str = f"{depth:.0f}" if depth is not None else "—"
+
+        lat = _num(it.get("epicenterLat") or it.get("latitude") or it.get("lat"))
+        lon = _num(it.get("epicenterLon") or it.get("longitude") or it.get("lon"))
+        lat_str = f"{lat:.2f}" if lat is not None else "—"
+        lon_str = f"{lon:.2f}" if lon is not None else "—"
+
+        origin = it.get("originTime") or ""
+        tw_str, utc_str = _parse_cwa_time(origin)
+
+        areas = it.get("locationDesc") or it.get("areas") or it.get("alertAreas")
+        if isinstance(areas, list):
+            areas_txt = "、".join(str(a) for a in areas if a)
+        elif isinstance(areas, str):
+            areas_txt = areas
+        else:
+            areas_txt = "—"
+
+        lines.append(
+            f"事件: {identifier} | 狀態: {status} | 類型: {msg_type}#{msg_no}\n"
+            f"震級/深度: M{mag_str} / {depth_str} km\n"
+            f"震中: lat {lat_str}, lon {lon_str}\n"
+            f"時間: {tw_str}（台灣） / {utc_str}（UTC）\n"
+            f"預警地區: {areas_txt}"
+        )
+        lines.append("")
+        count += 1
+
+    if len(items) > limit:
+        lines.append(f"... 另有 {len(items) - limit} 筆。")
+
+    return "\n".join(lines).strip()