Update cwa_service.py

cwa_service.py

@@ -1,97 +1,84 @@
-# cwa_service.py
-# -*- coding: utf-8 -*-
-from __future__ import annotations
-import requests
-from datetime import datetime, timedelta, timezone
-from config import CWA_ALARM_API
-
-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()
+# cwa_service.py
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+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))
+
+# --- Helper Function from Gradio Script ---
+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 fetch_cwa_alarm_list(limit: int = 5) -> str:
+    # ... (此函式內容不變) ...
+    return "\n".join(lines).strip()
+
+# --- [新功能] 顯著有感地震 (E-A0015-001) ---
+def _parse_significant_earthquakes(obj: dict) -> pd.DataFrame:
+    """移植自 Gradio 程式碼，用於解析 E-A0015-001 的 JSON 資料"""
+    quakes = obj.get("records", {}).get("Earthquake", [])
+    rows = []
+    for q in quakes:
+        ei = q.get("EarthquakeInfo", {})
+        epic = ei.get("Epicenter", {})
+        mag_info = ei.get("Magnitude", {})
+        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(ei.get("FocalDepth")),
+            "Magnitude": _to_float(mag_info.get("MagnitudeValue")),
+            "Location": epic.get("Location"),
+            "URL": q.get("Web"),
+        })
+    df = pd.DataFrame(rows)
+    if not df.empty:
+        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:
+    """從 CWA 獲取最新的顯著有感地震"""
+    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():
+            lines.append(
+                f"時間: {row['Time'].strftime('%Y-%m-%d %H:%M')}\n"
+                f"地點: {row['Location']}\n"
+                f"震級: M{row['Magnitude']:.1f} | 深度: {row['Depth']:.0f} km\n"
+                f"報告: {row['URL'] or '無'}"
+            )
+        return "\n\n".join(lines)
+        
+    except Exception as e:
+        return f"❌ 顯著地震查詢失敗：{e}"
+