Update app.py

app.py

@@ -1,5 +1,5 @@
 import os
-# ✅ 避免 Matplotlib 在唯讀路徑寫入設定
+# ✅ Matplotlib config/cache 放入可寫路徑
 os.environ["MPLCONFIGDIR"] = "/tmp/matplotlib"
 
 import uuid
@@ -68,7 +68,30 @@ app = Flask(__name__)
 configuration = Configuration(access_token=CHANNEL_ACCESS_TOKEN)
 handler = WebhookHandler(CHANNEL_SECRET)
 
-# --- Routes ---
+# -------------------- 共用小工具 --------------------
+def _iso(dt: datetime) -> str:
+    return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
+
+def _base_url_for_images() -> str:
+    if HF_SPACE_URL:
+        return HF_SPACE_URL.rstrip("/")
+    return request.url_root.rstrip("/")
+
+def _fmt_dt_utc(ts_ms: int | float | None) -> str:
+    """將毫秒或秒級 timestamp 轉成 YYYY-MM-DD HH:MM (UTC)，未知回傳 '未知'"""
+    if ts_ms is None:
+        return "未知"
+    try:
+        # 多數 API 為毫秒；若是秒也能容錯
+        if ts_ms > 1e12:  # 毫秒
+            dt = datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc)
+        else:
+            dt = datetime.fromtimestamp(ts_ms, tz=timezone.utc)
+        return dt.strftime("%Y-%m-%d %H:%M")
+    except Exception:
+        return "未知"
+
+# -------------------- 路由 --------------------
 @app.route("/", methods=["GET"])
 def home():
     return """
@@ -101,12 +124,9 @@ def healthz():
 def serve_static(filename):
     return send_from_directory(STATIC_DIR, filename)
 
-# --- Earthquake Data Functions ---
+# -------------------- 資料存取：USGS --------------------
 USGS_API_BASE_URL = "https://earthquake.usgs.gov/fdsnws/event/1/query"
 
-def _iso(dt: datetime) -> str:
-    return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
-
 def fetch_global_last24h_text(min_mag=5.0, limit=10) -> str:
     """顯示近 24 小時全球 M≥min_mag 地震，含日期。"""
     now_utc = datetime.now(timezone.utc)
@@ -130,7 +150,6 @@ def fetch_global_last24h_text(min_mag=5.0, limit=10) -> str:
         for f in features:
             p = f["properties"]
             t_utc = datetime.fromtimestamp(p["time"] / 1000, tz=timezone.utc)
-            # ✅ 改用含日期格式
             lines.append(
                 f"震級: {p['mag']:.1f} | 日期時間: {t_utc.strftime('%Y-%m-%d %H:%M')} (UTC)\n地點: {p.get('place','')}"
             )
@@ -177,7 +196,101 @@ def fetch_taiwan_df_this_year(min_mag=5.0) -> pd.DataFrame | str:
     except Exception as e:
         return f"❌ 查詢失敗: {e}"
 
-# --- Map Creation ---
+# -------------------- 資料存取：CWA 地震預警 --------------------
+CWA_ALARM_API = "https://app-2.cwa.gov.tw/api/v1/earthquake/alarm/list"
+
+def fetch_cwa_alarm_list(limit: int = 5) -> str:
+    """
+    取得交通部中央氣象署「地震預警」清單並格式化為文字。
+    盡量容錯不同欄位名稱；最多輸出 limit 筆。
+    """
+    try:
+        r = requests.get(CWA_ALARM_API, timeout=10)
+        r.raise_for_status()
+        data = r.json()
+    except Exception as e:
+        return f"❌ 地震預警查詢失敗：{e}"
+
+    # 嘗試從常見包裝鍵取出陣列
+    items = None
+    for key in ("data", "records", "result", "list", "items"):
+        if isinstance(data, dict) and isinstance(data.get(key), list):
+            items = data.get(key)
+            break
+    if items is None:
+        # 也可能根本就是陣列
+        if isinstance(data, list):
+            items = data
+        else:
+            return "⚠️ 無法解析地震預警資料（未知格式）。"
+
+    if not items:
+        return "✅ 目前沒有地震預警。"
+
+    lines = ["🚨 地震預警（最新）:", "-" * 20]
+    count = 0
+    for it in items:
+        if count >= limit:
+            break
+        # 嘗試各種常見欄位名稱
+        event_id = it.get("eventId") or it.get("eventid") or it.get("id") or "—"
+        # 可能是 ISO 字串、毫秒或秒
+        origin = it.get("originTime") or it.get("time") or it.get("origin_time") or it.get("earthquake_time")
+        dt_str = "未知"
+        if isinstance(origin, (int, float)):
+            dt_str = _fmt_dt_utc(origin)
+        elif isinstance(origin, str):
+            # 直接顯示（若是 ISO 字串）
+            try:
+                # 嘗試 parse 並以 UTC 格式化
+                dt = datetime.fromisoformat(origin.replace("Z", "+00:00"))
+                dt_str = dt.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M")
+            except Exception:
+                dt_str = origin  # 無法解析就原樣輸出
+
+        mag = it.get("magnitude") or it.get("mag") or it.get("ml") or it.get("mw") or "—"
+        depth = it.get("depth") or "—"
+        lat = it.get("latitude") or it.get("lat") or "—"
+        lon = it.get("longitude") or it.get("lon") or "—"
+        place = it.get("location") or it.get("place") or it.get("region") or "—"
+
+        # 預警區域/縣市（可能是陣列或字串）
+        areas = it.get("areas") or it.get("alertAreas") or it.get("counties") or it.get("cities")
+        if isinstance(areas, list):
+            # 可能是字串陣列或物件陣列
+            try:
+                area_names = []
+                for a in areas:
+                    if isinstance(a, str):
+                        area_names.append(a)
+                    elif isinstance(a, dict):
+                        name = a.get("name") or a.get("area") or a.get("city") or a.get("county")
+                        if name:
+                            area_names.append(str(name))
+                areas_txt = "、".join(area_names) if area_names else "—"
+            except Exception:
+                areas_txt = "—"
+        elif isinstance(areas, str):
+            areas_txt = areas
+        else:
+            areas_txt = "—"
+
+        lines.append(
+            f"事件: {event_id}\n"
+            f"日期時間: {dt_str} (UTC)\n"
+            f"震級/深度: M{mag} / {depth} km\n"
+            f"位置: {place}（lat {lat}, lon {lon}）\n"
+            f"預警區域: {areas_txt}"
+        )
+        lines.append("")  # 空行
+        count += 1
+
+    if len(items) > limit:
+        lines.append(f"... 另有 {len(items) - limit} 筆。")
+
+    return "\n".join(lines).strip()
+
+# -------------------- 繪圖 --------------------
 def create_and_save_map(df: pd.DataFrame) -> str:
     fig, ax = plt.subplots(figsize=(9, 6), dpi=150)
     lon_min, lon_max = 118.5, 123.5
@@ -215,13 +328,7 @@ def create_and_save_map(df: pd.DataFrame) -> str:
     plt.close(fig)
     return filename
 
-def _base_url_for_images() -> str:
-    # 盡量用公開的 HF Space URL，確保 LINE 能抓到圖
-    if HF_SPACE_URL:
-        return HF_SPACE_URL.rstrip("/")
-    return request.url_root.rstrip("/")
-
-# --- LINE Webhook ---
+# -------------------- LINE Webhook --------------------
 @app.route("/callback", methods=["POST"])
 def callback():
     signature = request.headers.get("X-Line-Signature")
@@ -232,7 +339,7 @@ def callback():
         abort(400)
     return "OK"
 
-# --- Message Handler ---
+# -------------------- Message Handler --------------------
 @handler.add(MessageEvent, message=TextMessageContent)
 def handle_message(event):
     user_message = (event.message.text or "").strip().lower()
@@ -240,7 +347,15 @@ def handle_message(event):
     with ApiClient(configuration) as api_client:
         line_bot_api = MessagingApi(api_client)
 
-        # 繪圖
+        # ➊ 地震預警（新指令）
+        if "地震預警" in user_message:
+            reply_text = fetch_cwa_alarm_list(limit=5)
+            line_bot_api.reply_message_with_http_info(
+                ReplyMessageRequest(reply_token=event.reply_token, messages=[TextMessage(text=reply_text)])
+            )
+            return
+
+        # ➋ 臺灣地震畫圖
         if ("臺灣地震畫圖" in user_message) or ("台灣地震畫圖" in user_message):
             result = fetch_taiwan_df_this_year()
             if isinstance(result, pd.DataFrame):
@@ -264,7 +379,7 @@ def handle_message(event):
             line_bot_api.reply_message_with_http_info(reply)
             return
 
-        # 說明
+        # ➌ 說明
         if user_message == "/help":
             text = (
                 "📖 地震預警 dayichen 指令說明\n\n"
@@ -272,6 +387,7 @@ def handle_message(event):
                 "➡️ 地震 / quake\n   說明：查詢全球最近 24 小時內，M≥5.0 的顯著地震（含日期時間）。\n\n"
                 "➡️ 臺灣地震 / 台灣地震\n   說明：查詢今年以來台灣區域 (21–26°N, 119–123°E) M≥5.0 地震（含日期時間）。\n\n"
                 "➡️ 臺灣地震畫圖 / 台灣地震畫圖\n   說明：繪製今年台灣區域 M≥5.0 地震分佈圖並回傳圖片。\n\n"
+                "➡️ 地震預警\n   說明：取得中央氣象署地震預警清單（最新 5 筆）。\n\n"
                 "➡️ 你好\n   說明：顯示歡迎訊息。"
             )
             line_bot_api.reply_message_with_http_info(
@@ -279,7 +395,7 @@ def handle_message(event):
             )
             return
 
-        # 台灣清單（已含日期）
+        # ➍ 台灣清單
         if ("臺灣地震" in user_message) or ("台灣地震" in user_message):
             result = fetch_taiwan_df_this_year()
             if isinstance(result, pd.DataFrame):
@@ -298,7 +414,7 @@ def handle_message(event):
             )
             return
 
-        # 全球 24 小時（已含日期）
+        # ➎ 全球 24 小時
         if ("地震" in user_message) or ("quake" in user_message):
             reply_text = fetch_global_last24h_text()
             line_bot_api.reply_message_with_http_info(
@@ -306,7 +422,7 @@ def handle_message(event):
             )
             return
 
-        # 打招呼
+        # ➏ 打招呼
         if ("你好" in user_message) or ("hi" in user_message):
             line_bot_api.reply_message_with_http_info(
                 ReplyMessageRequest(