Update cwa_service.py

cwa_service.py

@@ -1,87 +1,179 @@
-# cwa_service.py
+# cwa_service.py (修正與優化版本)
 
-import os
 import requests
-from datetime import datetime, timezone, timedelta
-from dotenv import load_dotenv
+import re
+import pandas as pd  # 請確保 'pandas' 已加入 requirements.txt
+from datetime import datetime, timedelta, timezone
+from config import CWA_API_KEY, CWA_ALARM_API, CWA_SIGNIFICANT_API
 
-# 從 .env 文件加載環境變數
-load_dotenv()
+# 定義台北時區 (UTC+8)
+TAIPEI_TZ = timezone(timedelta(hours=8))
 
-# 從環境變數中獲取 CWA 授權碼
-CWA_API_KEY = os.getenv('CWA_API_KEY')
-if not CWA_API_KEY:
-    raise ValueError("CWA_API_KEY not found in environment variables.")
+def _to_float(x) -> float | None:
+    """將輸入值穩健地轉換為浮點數，從字串中提取第一個數字。"""
+    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
 
-# CWA API 端點
-CWA_API_URL = f"https://opendata.cwa.gov.tw/api/v1/rest/datastore/E-A0015-001?Authorization={CWA_API_KEY}&format=JSON"
-
-def get_latest_earthquake():
+def _parse_cwa_time(s: str) -> tuple[str, str]:
     """
-    從 CWA API 獲取最新的有感地震資訊。
-
-    Returns:
-        tuple: 包含地震詳細資訊的訊息和報告圖片的 URL。如果出錯則返回錯誤訊息和 None。
+    穩健地解析 CWA API 可能回傳的兩種時間格式 (ISO 格式或本地時間格式)。
+    返回 (台北時間字串, UTC時間字串)。
     """
+    if not s: return ("未知", "未知")
+    dt_utc = None
     try:
-        response = requests.get(CWA_API_URL)
-        response.raise_for_status()  # 如果請求失敗 (狀態碼不是 2xx)，則引發異常
-        data = response.json()
+        # 嘗試解析 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, "未知")
 
-        if data.get('success') == 'true' and data['records']['Earthquake']:
-            latest_earthquake = data['records']['Earthquake'][0]
-            
-            # --- 時間處理 ---
-            # 1. 取得 API 回傳的 UTC 時間字串 (格式: YYYY-MM-DDTHH:MM:SSZ)
-            origin_time_utc_str = latest_earthquake['EarthquakeInfo']['OriginTime']
-            
-            # 2. 將 UTC 字串轉換為帶有時區資訊的 datetime 物件
-            #    .replace(tzinfo=timezone.utc) 是為了明確告知這個時間是 UTC
-            utc_time = datetime.strptime(origin_time_utc_str, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
-            
-            # 3. 定義台北時區 (UTC+8)
-            taipei_tz = timezone(timedelta(hours=8))
-            
-            # 4. 將 UTC 時間轉換為台北時間
-            taipei_time = utc_time.astimezone(taipei_tz)
-            
-            # 5. 格式化為易於閱讀的字串
-            formatted_time = taipei_time.strftime('%Y-%m-%d %H:%M:%S')
-            
-            # --- 其他資訊處理 ---
-            location = latest_earthquake['EarthquakeInfo']['Epicenter']['Location']
-            magnitude_value = latest_earthquake['EarthquakeInfo']['Magnitude']['MagnitudeValue']
-            depth_value = latest_earthquake['EarthquakeInfo']['FocalDepth']
-            report_url = latest_earthquake.get('Web', '#') # 如果 Web 欄位不存在，給予一個預設值
-            report_image_url = latest_earthquake.get('ReportImageURI', None)
-            
-            # 組合訊息
-            message = (
-                f"🚨 CWA 最新顯著有感地震\n\n"
-                f"時間: {formatted_time}\n"
-                f"地點: {location}\n"
-                f"規模: M{magnitude_value} | 深度: {depth_value} km\n"
-                f"報告: {report_url}"
+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', '—'))
+        
+        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 "—"
+
+        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"
+            f"時間: {tw_str}（台灣）\n"
+            f"地點: {areas}"
+        )
+    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:
+        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")
+        
+        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),
+            "Magnitude": _to_float(mag_raw),
+            "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 "—"
+            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 message, report_image_url
-        else:
-            return "目前無法獲取最新的地震資訊。", None
-            
-    except requests.exceptions.RequestException as e:
-        print(f"Error fetching data from CWA API: {e}")
-        return f"連接中央氣象署 API 時發生錯誤: {e}", None
-    except (KeyError, IndexError) as e:
-        print(f"Error parsing CWA API response: {e}")
-        return "解析地震資料時發生錯誤，可能是資料格式有變。", None
+        return "\n\n".join(lines)
     except Exception as e:
-        print(f"An unexpected error occurred: {e}")
-        return f"發生未知錯誤: {e}", None
+        return f"❌ 顯著地震查詢失敗：{e}"
 
-# 你可以在本地端測試這個函式
-if __name__ == '__main__':
-    message, image_url = get_latest_earthquake()
-    print("--- 訊息 ---")
-    print(message)
-    print("\n--- 圖片 URL ---")
-    print(image_url)
+def fetch_latest_significant_earthquake() -> dict | None:
+    """獲取最新的一筆顯著地震的詳細資料 (包含圖片)。"""
+    try:
+        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()
+        
+        # 將 Time object 轉換為格式化的字串，方便後續使用
+        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