Update cwa_service.py

cwa_service.py

@@ -5,7 +5,7 @@ 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
+from config import CWA_API_KEY, CWA_ALARM_API, CWA_SIGNIFICANT_API, CWA_LOCAL_EQ_API
 
 TAIPEI_TZ = timezone(timedelta(hours=8))
 
@@ -26,6 +26,18 @@ def _parse_cwa_time(s: str) -> tuple[str, str]:
     except Exception:
         return (s, "未知")
 
+def _normalize_cwa_area_name(area_name: str) -> str:
+    """自動校正縣市名稱，以符合 CWA API 查詢需求"""
+    area_name = area_name.replace("台", "臺")
+    if area_name.endswith("市") or area_name.endswith("縣"):
+        return area_name
+    
+    major_cities = "臺北,新北,基隆,桃園,新竹,臺中,嘉義,臺南,高雄".split(',')
+    if any(city in area_name for city in major_cities):
+         return f"{area_name}市"
+    else:
+         return f"{area_name}縣"
+
 # --- 地震預警 (CWA_ALARM_API) ---
 def fetch_cwa_alarm_list(limit: int = 5) -> str:
     """抓 CWA 地震預警並格式化輸出。"""
@@ -47,21 +59,17 @@ def fetch_cwa_alarm_list(limit: int = 5) -> str:
             return datetime.min.replace(tzinfo=timezone.utc)
     
     items = sorted(items, key=_key, reverse=True)
-
     lines = ["🚨 地震預警（最新）:", "-" * 20]
     for it in items[:limit]:
         mag = _to_float(it.get("magnitudeValue"))
         depth = _to_float(it.get("depth"))
         tw_str, _ = _parse_cwa_time(it.get("originTime", ""))
-        
         identifier = str(it.get('identifier', '—')).replace('{', '{{').replace('}', '}}')
         msg_type = str(it.get('msgType', '—')).replace('{', '{{').replace('}', '}}')
         msg_no = str(it.get('msgNo', '—')).replace('{', '{{').replace('}', '}}')
         areas = str(it.get('alertAreas') or '—').replace('{', '{{').replace('}', '}}')
-        
         mag_str = f"{mag:.1f}" if mag is not None else "—"
         depth_str = f"{depth:.0f}" if depth is not None else "—"
-
         lines.append(
             f"事件: {identifier} | 類型: {msg_type}#{msg_no}\n"
             f"震級/深度: M{mag_str} / {depth_str} km\n"
@@ -79,14 +87,10 @@ def _parse_significant_earthquakes(obj: dict) -> pd.DataFrame:
         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"),
+            "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 and "Time" in df.columns:
@@ -96,28 +100,21 @@ def _parse_significant_earthquakes(obj: dict) -> pd.DataFrame:
     return df
 
 def fetch_significant_earthquakes(days: int = 7, limit: int = 5) -> str:
-    if not CWA_API_KEY:
-        return "❌ 顯著地震查詢失敗：管理者尚未設定 CWA_API_KEY。"
-    
+    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} 天內沒有顯著有感地震報告。"
-
+        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"
@@ -125,6 +122,33 @@ def fetch_significant_earthquakes(days: int = 7, limit: int = 5) -> str:
                 f"報告: {row['URL'] or '無'}"
             )
         return "\n\n".join(lines)
-        
     except Exception as e:
         return f"❌ 顯著地震查詢失敗：{e}"
+
+# --- 小區域有感地震 (E-A0016-001) ---
+def fetch_local_earthquakes(area_name: str, limit: int = 3) -> str:
+    """從 CWA 獲取指定地區的小區域有感地震報告"""
+    if not CWA_API_KEY: return "❌ 查詢失敗：管理者尚未設定 CWA_API_KEY。"
+    normalized_area = _normalize_cwa_area_name(area_name)
+    params = {"Authorization": CWA_API_KEY, "format": "JSON", "limit": limit, "AreaName": normalized_area}
+    try:
+        r = requests.get(CWA_LOCAL_EQ_API, params=params, timeout=15)
+        r.raise_for_status()
+        data = r.json()
+        earthquakes = data.get("records", {}).get("Earthquake", [])
+        if not earthquakes: return f"✅ 在「{normalized_area}」近期沒有小區域有感地震報告。"
+        lines = [f"🚨 「{normalized_area}」近期小區域有感地震:", "-" * 20]
+        for eq in earthquakes:
+            info = eq.get("earthquakeInfo", {})
+            intensity_areas = eq.get("intensity", {}).get("shakingArea", [])
+            area_strs = [f"{area.get('areaDesc')} {area.get('areaIntensity')}級" for area in intensity_areas if area.get("areaIntensity") and float(area.get("areaIntensity", 0)) > 0]
+            intensity_str = "、".join(area_strs) if area_strs else "無具體震度回報"
+            lines.append(
+                f"報告: {eq.get('reportContent', '—')}\n"
+                f"時間: {info.get('originTime', '—')}\n"
+                f"規模: M{info.get('magnitude', {}).get('magnitudeValue', '—')} | 深度: {info.get('depth', {}).get('value', '—')} km\n"
+                f"主要影響區域: {intensity_str}"
+            )
+        return "\n\n".join(lines)
+    except Exception as e:
+        return f"❌ 小區域地震查詢失敗：{e}"