Update cwa_service.py

cwa_service.py

@@ -1,147 +1,87 @@
-# cwa_service.py (Final Defensive Parsing Version)
-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))
+# cwa_service.py
 
-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
+import os
+import requests
+from datetime import datetime, timezone, timedelta
+from dotenv import load_dotenv
 
-def _parse_cwa_time(s: str) -> tuple[str, str]:
-    if not s: return ("未知", "未知")
-    dt_utc = None
-    try:
-        dt_utc = datetime.fromisoformat(s.replace("Z", "+00:00"))
-    except ValueError:
-        try:
-            dt_local = datetime.strptime(s, "%Y-%m-%d %H:%M:%S")
-            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, "未知")
+# 從 .env 文件加載環境變數
+load_dotenv()
 
-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", ""))
-        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_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"
-            f"時間: {tw_str}（台灣）\n"
-            f"地點: {areas}"
-        )
-    return "\n\n".join(lines).strip()
+# 從環境變數中獲取 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 _parse_significant_earthquakes(obj: dict) -> pd.DataFrame:
-    records = obj.get("records", {})
-    quakes = records.get("Earthquake", [])
-    rows = []
-    for q in quakes:
-        # [偵錯] 如果需要，可以取消下面這行的註解，它會在 Log 中印出最原始的資料
-        # print(f"原始地震資料: {q}") 
-        
-        ei = q.get("EarthquakeInfo", {})
-        
-        # [修正] 使用更穩健的方式取得所有資料，檢查所有已知的大小寫和備用名稱
-        epic = ei.get("Epicenter") or ei.get("epicenter") or {}
-        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), 
-            "Magnitude": _to_float(mag_raw),
-            "Location": epic.get("Location") or epic.get("location"), 
-            "URL": q.get("Web") or q.get("ReportURL"),
-        })
-        
-    df = pd.DataFrame(rows)
-    if not df.empty and "Time" in df.columns:
-        df["Time"] = pd.to_datetime(df["Time"], errors="coerce", utc=True).dt.tz_convert(TAIPEI_TZ)
-    return df
+# 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 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 get_latest_earthquake():
+    """
+    從 CWA API 獲取最新的有感地震資訊。
 
-def fetch_latest_significant_earthquake() -> dict | None:
+    Returns:
+        tuple: 包含地震詳細資訊的訊息和報告圖片的 URL。如果出錯則返回錯誤訊息和 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
+        response = requests.get(CWA_API_URL)
+        response.raise_for_status()  # 如果請求失敗 (狀態碼不是 2xx)，則引發異常
+        data = response.json()
 
-        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
+        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}"
+            )
+            
+            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
     except Exception as e:
-        raise e
+        print(f"An unexpected error occurred: {e}")
+        return f"發生未知錯誤: {e}", None
 
+# 你可以在本地端測試這個函式
+if __name__ == '__main__':
+    message, image_url = get_latest_earthquake()
+    print("--- 訊息 ---")
+    print(message)
+    print("\n--- 圖片 URL ---")
+    print(image_url)