fetch_cwa_alarm_list.py

fetch_cwa_alarm_list.py

@@ -1,104 +0,0 @@
-# -*- 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]:
-    """回傳 (台灣時間, UTC)；若字串無時區，預設視為台灣時間。"""
-    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:
-    """抓 CWA 地震預警並格式化輸出。"""
-    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 "✅ 目前沒有地震預警。"
-
-    # 由新到舊
-    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
-
-    def _num(x):
-        xs = str(x)
-        ok = xs.replace(".", "", 1).replace("-", "", 1).isdigit()
-        return float(xs) if ok else None
-
-    lines = ["🚨 地震預警（最新）:", "-" * 20]
-    for idx, it in enumerate(items[:limit]):
-        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 "—"
-
-        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("")
-
-    if len(items) > limit:
-        lines.append(f"... 另有 {len(items) - limit} 筆。")
-
-    return "\n".join(lines).strip()