Update command_handler.py

command_handler.py

@@ -1,153 +1,210 @@
-# command_handler.py
+# cwa_service.py
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+import requests
+import re
 import pandas as pd
-from linebot.v3.messaging import TextMessage, ImageMessage
-
-# 匯入服務函式
-from cwa_service import fetch_cwa_alarm_list, fetch_significant_earthquakes
-from usgs_service import fetch_global_last24h_text, fetch_taiwan_df_this_year
-from plotting_service import create_and_save_map
-from ai_service import generate_ai_text
-from config import CURRENT_YEAR, HF_SPACE_URL
-
-def get_help_message() -> TextMessage:
-    """回傳包含所有可用指令的說明訊息。"""
-    text = (
-        "📖 指令列表\n\n"
-        "【地震資訊】\n"
-        "• /latest - 查詢最新一筆顯著地震(含圖)\n"
-        "• /global - 查詢全球近24小時顯著地震\n"
-        "• /taiwan - 顯示今年台灣顯著地震列表\n"
-        "• /map - 取得台灣顯著地震地圖連結\n"
-        "• /alert - 獲取 CWA 最新地震預警\n"
-        "• /significant - CWA 最新顯著有感地震\n\n"
-        "【AI 與工具】\n"
-        "• /ai <問題> - 與 AI 助理對話\n\n"
-        "【基本指令】\n"
-        "• /info - 關於此機器人\n"
-        "• /help - 顯示此說明"
-    )
-    return TextMessage(text=text)
-
-def get_info_message() -> TextMessage:
-    """回傳機器人資訊。"""
-    text = (
-        "🤖 關於我\n\n"
-        "我是一個多功能助理機器人，提供地震查詢與 AI 對話功能。\n\n"
-        "• 版本: 2.5\n"
-        "• 資料來源: CWA, USGS, Hugging Face\n"
-        "• 開發者: dayichen"
-    )
-    return TextMessage(text=text)
-
-def get_taiwan_earthquake_list() -> TextMessage:
-    """回傳近期的台灣地震文字列表。"""
-    result = fetch_taiwan_df_this_year()
-    if isinstance(result, pd.DataFrame):
-        count = len(result)
-        lines = [f"🇹🇼 今年 ({CURRENT_YEAR} 年) 台灣區域顯著地震 (M≥5.0)，共 {count} 筆:", "-" * 20]
-        for _, row in result.head(15).iterrows():
-            t = row["time_utc"].strftime("%Y-%m-%d %H:%M")
-            lines.append(
-                # [修改] 將 "震級" 改為 "規模"
-                f"規模: {row['magnitude']:.1f} | 日期時間: {t} (UTC)\n"
-                f"地點: {row['place']}\n"
-                f"報告連結: {row.get('url', '無')}"
-            )
-        if count > 15:
-            lines.append(f"... (還有 {count - 15} 筆，請用 /map 查看全部)")
-        reply_text = "\n\n".join(lines)
-    else:
-        reply_text = result
-    return TextMessage(text=reply_text)
-
-def get_taiwan_earthquake_map(base_url: str) -> list:
-    """產生地震地圖，並回傳包含連結的文字訊息。"""
-    result = fetch_taiwan_df_this_year()
-    if isinstance(result, pd.DataFrame):
-        filename = create_and_save_map(result)
-        image_url = f"{(HF_SPACE_URL or base_url)}/static/{filename}"
-        reply_text = (
-            "🗺️ 已為您繪製地震分佈圖。\n\n"
-            "請點擊以下連結查看圖片：\n"
-            f"{image_url}"
-        )
-        return [TextMessage(text=reply_text)]
-    else:
-        return [TextMessage(text=result)]
+from datetime import datetime, timedelta, timezone
+from config import CWA_API_KEY, CWA_ALARM_API, CWA_SIGNIFICANT_API, CWA_LOCAL_EQ_API
 
-def process_message(user_message_raw: str, request_base_url: str) -> list:
-    """處理使用者的文字訊息並回傳一個包含回覆訊息的列表。"""
-    user_message = (user_message_raw or "").strip()
+TAIPEI_TZ = timezone(timedelta(hours=8))
+
+# --- Helper Functions ---
+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]:
+    if not s: return ("未知", "未知")
+    try:
+        dt = datetime.fromisoformat(s.replace("Z", "+00:00"))
+        tw = dt.astimezone(TAIPEI_TZ).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 _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
     
-    # 將舊指令對應到新指令，提供向下相容
-    if user_message.lower() in ["地震", "quake"]:
-        user_message = "/global"
-    elif user_message.lower() in ["台灣地震", "臺灣地震"]:
-        user_message = "/taiwan"
-    elif user_message.lower() in ["台灣地震畫圖", "臺灣地震畫圖"]:
-        user_message = "/map"
-    elif user_message.lower() == "地震預警":
-        user_message = "/alert"
-
-    if user_message.startswith('/'):
-        parts = user_message.split(' ', 1)
-        command = parts[0].lower()
-        arg = parts[1] if len(parts) > 1 else ""
-
-        if command == '/help':
-            return [get_help_message()]
-        elif command == '/info':
-            return [get_info_message()]
-        elif command == '/latest':
-            return get_latest_earthquake_reply()
-        elif command == '/global':
-            return [TextMessage(text=fetch_global_last24h_text())]
-        elif command == '/taiwan':
-            return [get_taiwan_earthquake_list()]
-        elif command == '/map':
-            return get_taiwan_earthquake_map(request_base_url)
-        elif command == '/alert':
-            return [TextMessage(text=fetch_cwa_alarm_list(limit=5))]
-        elif command == '/significant':
-            return [TextMessage(text=fetch_significant_earthquakes(limit=5))]
-        elif command == '/ai':
-            if not arg:
-                return [TextMessage(text="請輸入問題，例如：/ai 台灣最高的山是哪座？")]
-            return [TextMessage(text=generate_ai_text(arg))]
-
-    # 若非指令，則預設交給 AI 處理 (作為備援)
-    return [TextMessage(text=generate_ai_text(user_message))]
-
-# --- [新功能] 處理 /latest 指令的回覆 ---
-def get_latest_earthquake_reply() -> list:
-    """獲取最新地震資訊並組合成 LINE 訊息"""
+    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 地震預警並格式化輸出。"""
     try:
-        latest_eq = fetch_latest_significant_earthquake()
-        if not latest_eq:
-            return [TextMessage(text="✅ 近期無顯著有感地震報告。")]
+        r = requests.get(CWA_ALARM_API, timeout=10)
+        r.raise_for_status()
+        payload = r.json()
+    except Exception as e:
+        return f"❌ 地震預警查詢失敗：{e}"
 
-        # 組合文字訊息
-        mag_str = f"{latest_eq['Magnitude']:.1f}" if latest_eq.get('Magnitude') is not None else "—"
-        depth_str = f"{latest_eq['Depth']:.0f}" if latest_eq.get('Depth') is not None else "—"
-        
-        text_message_content = (
-            f"🚨 CWA 最新顯著有感地震\n"
-            f"----------------------------------\n"
-            f"時間: {latest_eq.get('TimeStr', '—')}\n"
-            f"地點: {latest_eq.get('Location', '—')}\n"
-            f"規模: M{mag_str} | 深度: {depth_str} km\n"
-            f"報告: {latest_eq.get('URL', '無')}"
+    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('}', '}}')
+        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"
+            f"時間: {tw_str}（台灣）\n"
+            f"預警地區: {areas}"
         )
-        reply_messages = [TextMessage(text=text_message_content)]
+    return "\n\n".join(lines).strip()
 
-        # 如果有圖片網址，則新增圖片訊息
-        if latest_eq.get("ImageURL"):
-            image_url = latest_eq["ImageURL"]
-            reply_messages.append(
-                ImageMessage(original_content_url=image_url, preview_image_url=image_url)
+# --- 顯著有感地震 (E-A0015-001) ---
+def _parse_significant_earthquakes(obj: dict) -> pd.DataFrame:
+    """最穩健的解析邏輯，供 /significant 和 /latest 共用"""
+    records = obj.get("records") or obj.get("Records") or {}
+    quakes = records.get("earthquake") or records.get("Earthquake") or []
+    rows = []
+    for q in quakes:
+        ei = q.get("EarthquakeInfo") or q.get("earthquakeInfo") or {}
+        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:
+        time_series = pd.to_datetime(df["Time"], errors="coerce")
+        if pd.api.types.is_datetime64_any_dtype(time_series):
+            df["Time"] = time_series.dt.tz_localize("UTC").dt.tz_convert(TAIPEI_TZ)
+    return df
+
+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 fetch_latest_significant_earthquake() -> dict | None:
+    """從 CWA 獲取最新一筆顯著地震，並重用現有的解析邏輯"""
+    if not CWA_API_KEY:
+        raise ValueError("錯誤：尚未設定 CWA_API_KEY Secret。")
+    
+    now = datetime.now(timezone.utc)
+    time_from = (now - timedelta(days=2)).strftime("%Y-%m-%dT%H:%M:%S")
+    params = {"Authorization": CWA_API_KEY, "format": "JSON", "limit": 1}
+    
+    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.sort_values(by="Time", ascending=False).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
+
+# --- 小區域有感地震 (E-A0016-001) ---
+def fetch_local_earthquakes(area_name: str = "", limit: int = 5) -> str:
+    """從 CWA 獲取小區域有感地震報告。如果 area_name 為空，則查詢全台灣。"""
+    if not CWA_API_KEY: return "❌ 查詢失敗：管理者尚未設定 CWA_API_KEY。"
+    
+    params = {"Authorization": CWA_API_KEY, "format": "JSON", "limit": limit}
+    title = "🇹🇼 台灣近期小區域有感地震:"
+
+    if area_name:
+        normalized_area = _normalize_cwa_area_name(area_name)
+        params["AreaName"] = normalized_area
+        title = f"🚨 「{normalized_area}」近期小區域有感地震:"
+    
+    try:
+        r = requests.get(CWA_LOCAL_EQ_API, params=params, timeout=15)
+        r.raise_for_status()
+        data = r.json()
         
-        return reply_messages
+        earthquakes = data.get("records", {}).get("Earthquake", [])
+        if not earthquakes:
+            msg = f"✅ 在「{area_name}」" if area_name else "✅ 台灣"
+            return f"{msg}近期沒有小區域有感地震報告。"
 
+        lines = [title, "-" * 20]
+        for eq in earthquakes:
+            info = eq.get("earthquakeInfo", {})
+            epi = info.get("epicenter", {})
+            mag = info.get("magnitude", {})
+            depth = info.get("depth", {})
+            
+            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"地點: {epi.get('location', '—')}\n"
+                f"規模: M{mag.get('magnitudeValue', '—')} | 深度: {depth.get('value', '—')} km\n"
+                f"主要影響區域: {intensity_str}"
+            )
+        return "\n\n".join(lines)
+        
     except Exception as e:
-        return [TextMessage(text=f"❌ 查詢最新地震失敗：{e}")]
+        return f"❌ 小區域地震查詢失敗：{e}"