Update cwa_service.py

cwa_service.py

@@ -1,9 +1,29 @@
 # cwa_service.py
+import requests
+import re
+import pandas as pd
+from datetime import datetime, timedelta, timezone
+from config import CWA_API_KEY, CWA_ALARM_API, CWA_SIGNIFICANT_API
 
-# (檔案中其他的程式碼保持不變...)
+TAIPEI_TZ = timezone(timedelta(hours=8))
+
+def _to_float(x):
+    if x is None: return None
+    s = str(x).strip()
+    m = re.search(r"[-+]?\d+(?:\.\d+)?", s)
+    return float(m.group()) if m else None
+
+def _parse_cwa_time(s: str) -> tuple[str, str]:
+    if not s: return ("未知", "未知")
+    try:
+        dt = datetime.fromisoformat(s.replace("Z", "+00:00"))
+        tw = dt.astimezone(TAIPEI_TZ).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()
@@ -12,14 +32,11 @@ def fetch_cwa_alarm_list(limit: int = 5) -> str:
         return f"❌ 地震預警查詢失敗：{e}"
 
     items = payload.get("data", [])
-    if not items:
-        return "✅ 目前沒有地震預警。"
+    if not items: return "✅ 目前沒有地震預警。"
 
     def _key(it):
-        try:
-            return datetime.fromisoformat(it.get("originTime", "").replace("Z", "+00:00"))
-        except:
-            return datetime.min.replace(tzinfo=timezone.utc)
+        try: return datetime.fromisoformat(it.get("originTime", "").replace("Z", "+00:00"))
+        except: return datetime.min.replace(tzinfo=timezone.utc)
     
     items = sorted(items, key=_key, reverse=True)
     lines = ["🚨 地震預警（最新）:", "-" * 20]
@@ -31,19 +48,9 @@ def fetch_cwa_alarm_list(limit: int = 5) -> str:
         msg_type = str(it.get('msgType', '—')).replace('{', '{{').replace('}', '}}')
         msg_no = str(it.get('msgNo', '—')).replace('{', '{{').replace('}', '}}')
         
-        # --- 修改後的邏輯 ---
-        # 專門處理 alertAreas 列表
         alert_areas_list = it.get('alertAreas')
-        if isinstance(alert_areas_list, list) and alert_areas_list:
-            # 如果 alertAreas 是一個非空列表，將地區名稱用逗號串接起來
-            areas_str = ", ".join(str(area) for area in alert_areas_list)
-        else:
-            # 否則，維持顯示為 '—'
-            areas_str = "—"
-        
-        # 最後再處理 f-string 可能的逸出字元
+        areas_str = ", ".join(str(area) for area in alert_areas_list) if isinstance(alert_areas_list, list) and alert_areas_list else "—"
         areas = areas_str.replace('{', '{{').replace('}', '}}')
-        # --- 修改結束 ---
         
         mag_str = f"{mag:.1f}" if mag is not None else "—"
         depth_str = f"{depth:.0f}" if depth is not None else "—"
@@ -55,4 +62,70 @@ def fetch_cwa_alarm_list(limit: int = 5) -> str:
         )
     return "\n\n".join(lines).strip()
 
-# (檔案中其他的程式碼保持不變...)
+def _parse_significant_earthquakes(obj: dict) -> pd.DataFrame:
+    records = obj.get("records", {})
+    quakes = records.get("Earthquake", [])
+    rows = []
+    for q in quakes:
+        ei = q.get("EarthquakeInfo", {})
+        epic = ei.get("Epicenter", {})
+        mag_info = ei.get("Magnitude", {})
+        depth_raw = ei.get("FocalDepth")
+        mag_raw = mag_info.get("MagnitudeValue")
+        rows.append({
+            "ID": q.get("EarthquakeNo"), "Time": ei.get("OriginTime"),
+            "Lat": _to_float(epic.get("EpicenterLatitude")),
+            "Lon": _to_float(epic.get("EpicenterLongitude")),
+            "Depth": _to_float(depth_raw), "Magnitude": _to_float(mag_raw),
+            "Location": epic.get("Location"), "URL": q.get("Web"),
+        })
+    df = pd.DataFrame(rows)
+    if not df.empty and "Time" in df.columns:
+        df["Time"] = pd.to_datetime(df["Time"], errors="coerce").dt.tz_convert(TAIPEI_TZ)
+    return df
+
+def fetch_significant_earthquakes(days: int = 7, limit: int = 5) -> str:
+    if not CWA_API_KEY: return "❌ 顯著地震查詢失敗：管理者尚未設定 CWA_API_KEY。"
+    now = datetime.now(timezone.utc)
+    time_from = (now - timedelta(days=days)).strftime("%Y-%m-%d")
+    params = {"Authorization": CWA_API_KEY, "format": "JSON", "timeFrom": time_from}
+    try:
+        r = requests.get(CWA_SIGNIFICANT_API, params=params, timeout=15)
+        r.raise_for_status()
+        data = r.json()
+        df = _parse_significant_earthquakes(data)
+        if df.empty: return f"✅ 過去 {days} 天內沒有顯著有感地震報告。"
+        df = df.sort_values(by="Time", ascending=False).head(limit)
+        lines = [f"🚨 CWA 最新顯著有感地震 (近{days}天內):", "-" * 20]
+        for _, row in df.iterrows():
+            mag_str = f"{row['Magnitude']:.1f}" if pd.notna(row['Magnitude']) else "—"
+            depth_str = f"{row['Depth']:.0f}" if pd.notna(row['Depth']) else "—"
+            lines.append(
+                f"時間: {row['Time'].strftime('%Y-%m-%d %H:%M') if pd.notna(row['Time']) else '—'}\n"
+                f"地點: {row['Location'] or '—'}\n"
+                f"規模: M{mag_str} | 深度: {depth_str} km\n"
+                f"報告: {row['URL'] or '無'}"
+            )
+        return "\n\n".join(lines)
+    except Exception as e:
+        return f"❌ 顯著地震查詢失敗：{e}"
+
+def fetch_latest_significant_earthquake() -> dict | None:
+    if not CWA_API_KEY: raise ValueError("錯誤：尚未設定 CWA_API_KEY Secret。")
+    params = {"Authorization": CWA_API_KEY, "format": "JSON", "limit": 1, "orderby": "OriginTime desc"}
+    r = requests.get(CWA_SIGNIFICANT_API, params=params, timeout=15)
+    r.raise_for_status()
+    data = r.json()
+    df = _parse_significant_earthquakes(data)
+    if df.empty: return None
+
+    latest_eq_data = df.iloc[0].to_dict()
+    
+    quakes = data.get("records", {}).get("Earthquake", [])
+    if quakes:
+        latest_eq_data["ImageURL"] = quakes[0].get("ReportImageURI")
+
+    if pd.notna(latest_eq_data.get("Time")):
+        latest_eq_data["TimeStr"] = latest_eq_data["Time"].strftime('%Y-%m-%d %H:%M')
+
+    return latest_eq_data