Update cwa_service.py

cwa_service.py

@@ -1,84 +1,62 @@
-# cwa_service.py (修正與優化版本)
-
+# cwa_service.py (Final Defensive Parsing Version)
 import requests
 import re
-import pandas as pd  # 請確保 'pandas' 已加入 requirements.txt
+import pandas as pd
 from datetime import datetime, timedelta, timezone
 from config import CWA_API_KEY, CWA_ALARM_API, CWA_SIGNIFICANT_API
 
-# 定義台北時區 (UTC+8)
 TAIPEI_TZ = timezone(timedelta(hours=8))
 
-def _to_float(x) -> float | None:
-    """將輸入值穩健地轉換為浮點數，從字串中提取第一個數字。"""
+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]:
-    """
-    穩健地解析 CWA API 可能回傳的兩種時間格式 (ISO 格式或本地時間格式)。
-    返回 (台北時間字串, UTC時間字串)。
-    """
     if not s: return ("未知", "未知")
     dt_utc = None
     try:
-        # 嘗試解析 ISO 8601 格式 (例如 "2025-08-22T14:06:15Z")
         dt_utc = datetime.fromisoformat(s.replace("Z", "+00:00"))
     except ValueError:
-        # 如果失敗，嘗試解析本地時間格式 (例如 "2025-08-22 14:06:15")
         try:
             dt_local = datetime.strptime(s, "%Y-%m-%d %H:%M:%S")
-            # 假設此格式為台北時間，並轉換為標準 UTC 時間
             dt_local = dt_local.replace(tzinfo=TAIPEI_TZ)
             dt_utc = dt_local.astimezone(timezone.utc)
         except Exception:
-            # 如果兩種格式都失敗，直接返回原始字串
             return (s, "未知")
-    
     if dt_utc:
         tw_str = dt_utc.astimezone(TAIPEI_TZ).strftime("%Y-%m-%d %H:%M")
         utc_str = dt_utc.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M")
         return (tw_str, utc_str)
-    
     return (s, "未知")
 
 def fetch_cwa_alarm_list(limit: int = 5) -> str:
-    """獲取最新的地震預警列表。"""
     try:
         r = requests.get(CWA_ALARM_API, timeout=10)
         r.raise_for_status()
         payload = r.json()
     except Exception as e:
         return f"❌ 地震預警查詢失敗：{e}"
-
     items = payload.get("data", [])
     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)
     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", ""))
-        
-        # [修正] 移除不必要的 .replace('{', '{{')，直接使用 str() 即可
-        identifier = str(it.get('identifier', '—'))
-        msg_type = str(it.get('msgType', '—'))
-        msg_no = str(it.get('msgNo', '—'))
-        
+        identifier = str(it.get('identifier', '—')).replace('{', '{{').replace('}', '}}')
+        msg_type = str(it.get('msgType', '—')).replace('{', '{{').replace('}', '}}')
+        msg_no = str(it.get('msgNo', '—')).replace('{', '{{').replace('}', '}}')
         location_desc_list = it.get('locationDesc')
-        areas = ", ".join(str(area) for area in location_desc_list) if isinstance(location_desc_list, list) and location_desc_list else "—"
-
+        areas_str = ", ".join(str(area) for area in location_desc_list) if isinstance(location_desc_list, list) and location_desc_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 "—"
-        
         lines.append(
             f"事件: {identifier} | 類型: {msg_type}#{msg_no}\n"
             f"規模/深度: M{mag_str} / {depth_str} km\n"
@@ -88,55 +66,48 @@ def fetch_cwa_alarm_list(limit: int = 5) -> str:
     return "\n\n".join(lines).strip()
 
 def _parse_significant_earthquakes(obj: dict) -> pd.DataFrame:
-    """從 CWA API 的 JSON 回應中解析出顯著地震資料，並轉換為 Pandas DataFrame。"""
     records = obj.get("records", {})
     quakes = records.get("Earthquake", [])
     rows = []
     for q in quakes:
+        # [偵錯] 如果需要，可以取消下面這行的註解，它會在 Log 中印出最原始的資料
+        # print(f"原始地震資料: {q}") 
+        
         ei = q.get("EarthquakeInfo", {})
         
-        # 使用非常防禦性的方式取得資料，以應對 API 欄位名稱大小寫不一致或變動的問題
+        # [修正] 使用更穩健的方式取得所有資料，檢查所有已知的大小寫和備用名稱
         epic = ei.get("Epicenter") or ei.get("epicenter") or {}
-        mag_info = ei.get("Magnitude") or ei.get("magnitude") or {}
-        depth_raw = ei.get("FocalDepth") or ei.get("depth")
-        mag_raw = mag_info.get("MagnitudeValue") or mag_info.get("magnitudeValue")
+        mag_info = ei.get("Magnitude") or ei.get("magnitude") or ei.get("EarthquakeMagnitude") or {}
+        depth_raw = ei.get("FocalDepth") or ei.get("depth") or ei.get("Depth")
+        mag_raw = mag_info.get("MagnitudeValue") or mag_info.get("magnitudeValue") or mag_info.get("Value") or mag_info.get("value")
         
         rows.append({
             "ID": q.get("EarthquakeNo"), "Time": ei.get("OriginTime"),
             "Lat": _to_float(epic.get("EpicenterLatitude") or epic.get("epicenterLatitude")),
             "Lon": _to_float(epic.get("EpicenterLongitude") or epic.get("epicenterLongitude")),
-            "Depth": _to_float(depth_raw),
+            "Depth": _to_float(depth_raw), 
             "Magnitude": _to_float(mag_raw),
-            "Location": epic.get("Location") or epic.get("location"),
+            "Location": epic.get("Location") or epic.get("location"), 
             "URL": q.get("Web") or q.get("ReportURL"),
-            "ImageURL": q.get("ReportImageURI"), # 也順便解析圖片 URL
         })
         
     df = pd.DataFrame(rows)
     if not df.empty and "Time" in df.columns:
-        # 使用 pandas 內建功能將時間欄位轉換為帶有時區的 datetime 物件
         df["Time"] = pd.to_datetime(df["Time"], errors="coerce", utc=True).dt.tz_convert(TAIPEI_TZ)
     return df
 
 def fetch_significant_earthquakes(days: int = 7, limit: int = 5) -> str:
-    """獲取過去 N 天內的顯著地震列表。"""
     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} 天內沒有顯著有感地震報告。"
-        
-        # 依時間排序並取前 N 筆
         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 "—"
@@ -152,28 +123,25 @@ def fetch_significant_earthquakes(days: int = 7, limit: int = 5) -> str:
         return f"❌ 顯著地震查詢失敗：{e}"
 
 def fetch_latest_significant_earthquake() -> dict | None:
-    """獲取最新的一筆顯著地震的詳細資料 (包含圖片)。"""
     try:
-        if not CWA_API_KEY: 
-            raise ValueError("錯誤：尚未設定 CWA_API_KEY Secret。")
-            
+        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
+        if df.empty: return None
 
         latest_eq_data = df.iloc[0].to_dict()
         
-        # 將 Time object 轉換為格式化的字串，方便後續使用
+        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
     except Exception as e:
-        # [優化] 發生錯誤時，印出日誌並回傳 None，避免程式崩潰
-        print(f"[錯誤] 獲取最新顯著地震失敗: {e}")
-        return None
+        raise e
+