Upload 15 files

config/config.py

@@ -0,0 +1,63 @@
+# config.py (Gemini 最終版)
+import os
+import tempfile
+from datetime import datetime
+
+# ==============================================================================
+# 1. 執行環境設定 (適用於 Hugging Face Spaces)
+# ==============================================================================
+
+# 設定一個暫存目錄給 Matplotlib 快取字體 (若未來有繪圖功能)
+os.environ.setdefault("MPLCONFIGDIR", "/tmp/matplotlib")
+
+# 定義一個暫存目錄來存放生成的靜態檔案
+STATIC_DIR = os.getenv("STATIC_DIR", os.path.join(tempfile.gettempdir(), "static"))
+os.makedirs(STATIC_DIR, exist_ok=True)
+
+
+# ==============================================================================
+# 2. 憑證與金鑰 (從 Secret Variables 讀取)
+# ==============================================================================
+
+# LINE Bot 憑證
+CHANNEL_ACCESS_TOKEN = os.getenv("CHANNEL_ACCESS_TOKEN")
+CHANNEL_SECRET = os.getenv("CHANNEL_SECRET")
+
+# CWA (中央氣象署) API 金鑰
+CWA_API_KEY = os.getenv("CWA_API_KEY")
+
+# Google Gemini API 金鑰
+GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
+
+
+# ==============================================================================
+# 3. API 端點與 URL
+# ==============================================================================
+
+# CWA API 端點
+CWA_ALARM_API = "https://app-2.cwa.gov.tw/api/v1/earthquake/alarm/list"
+CWA_SIGNIFICANT_API = "https://opendata.cwa.gov.tw/api/v1/rest/datastore/E-A0015-001"
+
+# USGS API 端點
+USGS_API_BASE_URL = "https://earthquake.usgs.gov/fdsnws/event/1/query"
+
+# MCP 伺服器 (Gradio App) URL
+MCP_SERVER_URL = "https://cwadayi-mcp-2.hf.space"# MCP 伺服器 (Gradio App) URL
+
+# [新增] PWS 伺服器 API 端點
+PWS_API_URL = "https://cwadayi-mcp-pws.hf.space/gradio_api/mcp/sse"
+
+
+# CWA PWS 地震訊息 API 端點
+CWA_PWS_EARTHQUAKE_API = "https://cwadayi-app-show-pws.hf.space/cwa-earthquakes"
+
+# [新增] PTS 新聞 API 端點
+PTS_NEWS_API = "https://cwadayi-app-show-news.hf.space/pts-news"
+
+
+# ==============================================================================
+# 4. 一般應用程式設定
+# ==============================================================================
+
+# 顯示用的當年年份
+CURRENT_YEAR = datetime.now().year

config/data.py

@@ -1,7 +1,7 @@
 # config/data.py
 import pandas as pd
 
-# 課程進度表 (維持不變)
+# 課程進度表
 schedule_data = {
     "週次": ["第一週 (9/10)", "第二週 (9/17)", "第三週 (9/24)", "第四週 (10/1)", "第五週 (10/8)", "第六週 (10/15)", "第七週 (10/22)", "第八週 (10/29)", "第九週 (11/5)", "第十週 (11/12)", "第十一週 (11/19)", "第十二週 (11/26)", "第十三週 (12/3)", "第十四週 (12/10)", "第十五週 (12/17)", "第十六週", "十七週", "第十八週"],
     "地球物理知識": ["地球物理概要", "折射探勘學", "校園野外實驗 (折射探勘)", "地球物理專題演講", "折射探勘學", "重力探勘學", "重力探勘學", "期中考", "板塊構造與地震", "板塊構造與地震", "板塊構造與地磁", "板塊構造與地磁", "板塊構造與地熱", "板塊構造與地熱", "期末專題發表", "行憲紀念日 (放假)", "元旦 (放假)", "參訪大屯火山觀測中心 (暫定)"],
@@ -9,7 +9,7 @@ schedule_data = {
 }
 schedule_df = pd.DataFrame(schedule_data)
 
-# 知識庫 (擴充答案內容)
+# AI 助教知識庫
 KNOWLEDGE_BASE = {
     "course": [
         {"keywords": ["成績", "評分", "分數", "grading"], "answer": "本課程的評分標準經過精心設計，旨在全面評估您的學習成效。其中 **作業佔比 50%**，是最大的一部分，因為我們深信「做中學」的價值，這包括所有程式碼實作、數據分析報告與 GitHub 更新紀錄，用以衡量您的實踐能力。**期中考佔 40%**，主要檢核您對前半學期地球物理核心理論與程式工具的理解程度。最後 **10% 為平時成績**，我們鼓勵您積極參與課堂討論、勇於提問，因為主動的學習態度是深化知識的關鍵。"},
@@ -27,7 +27,7 @@ KNOWLEDGE_BASE = {
         {"keywords": ["板塊", "tectonics"], "answer": "板塊構造學說（Plate Tectonics）是現代地球科學的基石理論，它完美地解釋了全球的地震、火山活動與造山運動。該學說認為，地球最外層的岩石圈並非完整一塊，而是由十幾個巨大的、剛性的板塊所拼合而成。這些板塊漂浮在底下較具塑性的軟流圈之上，並以每年數公分的速度持續移動。在板塊的邊界處，由於板塊間的相互作用（碰撞、分離或錯動），累積了巨大的應力，最終導致了地震的發生，並形成了火山、海溝與山脈等壯麗的地質景觀。"},
         {"keywords": ["地磁", "古地磁", "磁場", "geomagnetism", "paleomagnetism"], "answer": "地球本身就像一個巨大的棒狀磁鐵，擁有一個從南極指向北極的磁場，這就是「地磁」。而「古地磁」則是指被岩石記錄下來的古代地球磁場。當火山噴發的熾熱岩漿冷卻時，其中的磁性礦物（如磁鐵礦）就像成千上萬個微小的指南針，它們會將當時地球磁場的方向和強度「鎖定」在岩石中。科學家發現，海底中洋脊兩側的玄武岩，記錄了地球磁場在數百萬年間多次反轉的歷史，並且呈現完美對稱的條帶狀分佈。這成為了海底擴張與板塊構造學說最經典、最無可辯駁的證據之一。"},
         {"keywords": ["地熱", "能源", "geothermal"], "answer": "地熱（Geothermal Energy）是源自地球內部的一種穩定且潔淨的可再生能源。其能量主要來自地球深處放射性元素衰變所產生的熱。我們主要透過鑽井到地下的「熱儲層」，將高溫的蒸汽或熱水引導至地表，用以推動渦輪機發電。地熱發電最大的優點是**穩定**，能 24 小時不間斷供電，不像太陽能或風能受天氣影響；同時**碳排放極低**。地熱資源通常集中在板塊邊界，因為這裡地殼較薄、岩漿活動頻繁。台灣正好位於歐亞板塊與菲律賓海板塊的碰撞帶上，因此擁有發展地熱的絕佳地理條件。"},
-        {"keywords": ["地震儀", "地震學", "seismograph", "seismology", "差別"], "answer": "這兩個詞彙的關係，就像是天文學家與望遠鏡的關係。**地震學（Seismology）** 是一門**科學**，是地球物理學的一個重要分支，它專門研究地震的成因、地震波的傳播、地球的內部構造以及如何減輕地震災害。而**地震儀（Seismograph）** 則是一種**儀器**，是地震學家進行研究時所使用的核心工具。它能精密地偵測並記錄地面的任何微小振動，並將這些振動轉換成一張稱為「地震圖」(Seismogram) 的紀錄。簡單來說：**地震學家使用地震儀來進行地震學研究。**"}
+        {"keywords": ["地震儀", "地震學", "seismograph", "seismology", "差別"], "answer": "這兩個詞彙的關係，就像是天文學家與望遠-鏡的關係。**地震學（Seismology）** 是一門**科學**，是地球物理學的一個重要分支，它專門研究地震的成因、地震波的傳播、地球的內部構造以及如何減輕地震災害。而**地震儀（Seismograph）** 則是一種**儀器**，是地震學家進行��究時所使用的核心工具。它能精密地偵測並記錄地面的任何微小振動，並將這些振動轉換成一張稱為「地震圖」(Seismogram) 的紀錄。簡單來說：**地震學家使用地震儀來進行地震學研究。**"}
     ],
 
     "tools": [
@@ -44,8 +44,8 @@ KNOWLEDGE_BASE = {
     ],
 
     "general": [
-        {"keywords": ["你好", "哈囉", "hello", "hi"], "answer": f"你好！我是本課程的 AI 助教，很高興能為您服務。今天是 {pd.Timestamp.now(tz='Asia/Taipei').strftime('%Y年%m月%d日')}，有任何關於課程、地球物理或程式工具的問題，都隨時可以問我喔！"},
-        {"keywords": ["你是誰", "功能", "幹嘛", "who are you"], "answer": "我是本課程專屬的 AI 助教。我的核心任務是根據內建的課程知識庫，即時回答您關於**地球物理核心概念**、**Python 程式工具**（如 PyGMT、ObsPy）以及**課程安排**（如評分標準、進度）的各種問題。當您在「互動體驗區」遇到程式碼問題時，我也能提供初步的除錯協助。您可以試著問我「如何安裝 Anaconda？」或「給我一些期末專題的靈感」。"},
+        {"keywords": ["你好", "哈囉", "hello", "hi"], "answer": f"你好！我是課程 AI 助教，很高興能為您服務。今天是 {pd.Timestamp.now(tz='Asia/Taipei').strftime('%Y年%m月%d日')}，有任何關於課程、地球物理或程式工具的問題，都隨時可以問我喔！"},
+        {"keywords": ["你是誰", "功能", "幹嘛", "who are you"], "answer": "我是本課程專屬的 AI 助教。我的核心任務是根據內建的課程知識庫，回答您關於**地球物理、程式工具**與**課程安排**的靜態問題。\n\n**更厲害的是，我現在還能查詢即時資訊！** 試著問我看看：\n- `「最新的 CWA 地震報告」`\n- `「今天有什麼新聞？」`\n- `「全球最近有什麼大地震？」`\n- `「現在有地震預警嗎？」`"},
         {"keywords": ["謝謝", "感謝", "thank"], "answer": "不客氣！很高興我的回答能對您有幫助。學習的路上有任何問題，都歡迎隨時再來找我討論。祝您學習順利，一起探索地球的奧秘！"},
         {"keywords": ["笑話", "好玩", "有趣"], "answer": "好的，這是一個我很喜歡的地質學冷笑話：\n\n一位地質學家走進一家酒吧，酒保問他：「嗨，今天要喝點什麼？」\n\n地質學家回答：「給我一杯『夸克』(Quark)！」\n\n...等等，抱歉，那是物理學家的笑話。我的版本是：\n\n為什麼地質學家從來不賭博？\n\n...因為他們太了解什麼叫做「斷層」(Fault) 了！ 😄"}
     ]

core/callbacks.py

@@ -4,8 +4,10 @@ import contextlib
 import traceback
 from datetime import datetime
 import pytz
-import difflib  # ✨ 1. 導入 difflib 函式庫
+import difflib
 
+# --- ✨ 1. 匯入所有外部服務的函式 ---
+from services import cwa_service, news_service, pws_service, usgs_service
 from core.visits import get_current_visit_count
 from core.notifications import send_line_notification_in_background
 from config.data import KNOWLEDGE_BASE
@@ -17,37 +19,28 @@ def execute_user_code(code_string, source_lab):
     status = "✅ 成功"
     error_info = ""
     fig = None
-    
     try:
         with contextlib.redirect_stdout(string_io):
             local_scope = {}
-            # Pre-import necessary libraries for the user's code
             exec("import matplotlib.pyplot as plt; import numpy as np; import cartopy.crs as ccrs; import cartopy.feature as cfeature; from matplotlib.ticker import FixedFormatter", local_scope)
             exec(code_string, local_scope)
-        
         console_output = string_io.getvalue()
         fig = local_scope.get('fig')
-        
         if fig is None:
             status = "⚠️ 警告"
             error_info = "程式碼執行完畢，但未找到 'fig' 物件。"
             return None, f"{error_info}\nPrint 輸出:\n{console_output}"
-        
         success_message = f"✅ 程式碼執行成功！\n\n--- Console Output ---\n{console_output}"
         return fig, success_message
-
     except Exception:
         status = "❌ 失敗"
         error_info = traceback.format_exc()
         final_message = f"❌ 程式碼執行失敗！\n\n--- Error Traceback ---\n{error_info}"
         return None, final_message
-        
     finally:
-        # This block will always run, regardless of success or failure
         tz = pytz.timezone('Asia/Taipei')
         current_time = datetime.now(tz).strftime('%H:%M:%S')
         visit_count = get_current_visit_count()
-        
         notification_text = (
             f"🔬 程式碼實驗室互動！\n\n"
             f"時間: {current_time}\n"
@@ -56,49 +49,47 @@ def execute_user_code(code_string, source_lab):
             f"總載入數: {visit_count}"
         )
         if status == "❌ 失敗":
-            # Add specific error type to notification for quick debugging
             error_type = error_info.strip().split('\n')[-1]
             notification_text += f"\n錯誤類型: {error_type}"
-            
         send_line_notification_in_background(notification_text)
 
+# --- ✨ 2. 建立一個「工具清單」，將關鍵字與服務函式對應 ---
+LIVE_TOOLS = [
+    {"keywords": ["新聞", "今日新聞", "news"], "function": news_service.fetch_today_news, "name": "今日新聞"},
+    {"keywords": ["cwa地震", "顯著地震", "有感地震"], "function": cwa_service.fetch_significant_earthquakes, "name": "CWA 顯著有感地震"},
+    {"keywords": ["地震預警", "cwa alarm"], "function": cwa_service.fetch_cwa_alarm_list, "name": "CWA 地震預警"},
+    {"keywords": ["全球地震", "usgs"], "function": usgs_service.fetch_global_last24h_text, "name": "全球顯著地震"},
+    {"keywords": ["pws發布", "pws info"], "function": pws_service.fetch_latest_pws_info, "name": "PWS 發布情形"},
+    {"keywords": ["pws地震", "pws alert"], "function": pws_service.fetch_cwa_pws_earthquake_info, "name": "PWS 地震警報"},
+]
 
-# --- ✨ 2. 加入新的模糊比對函式 ---
-def find_best_match(user_input, knowledge_base, threshold=0.6):
-    """
-    Finds the best matching answer from the knowledge base using fuzzy string matching.
-    """
+def find_best_match_from_kb(user_input, knowledge_base, threshold=0.6):
+    """(原有的函式) 從靜態知識庫中尋找最佳匹配的答案。"""
     best_score = 0
     best_answer = "這個問題很有趣，不過我的知識庫目前還沒有收錄相關的答案。您可以試著問我關於**課程評分、Anaconda安裝、Colab與Codespaces的差別**等問題！"
     best_match_keyword = None
-
-    # Iterate through all keywords in the knowledge base
     for category, entries in knowledge_base.items():
         for entry in entries:
             for keyword in entry['keywords']:
-                # Calculate similarity score
                 score = difflib.SequenceMatcher(None, user_input.lower(), keyword.lower()).ratio()
                 if score > best_score:
                     best_score = score
                     best_match_keyword = keyword
                     best_answer = entry['answer']
-
-    # If the best score is above the threshold, return the answer directly.
     if best_score >= threshold:
         return best_answer
-    # If the score is too low, but we found a potential match, ask for confirmation.
     elif best_match_keyword:
         return f"這個問題我不是很確定，您是指 **「{best_match_keyword}」** 嗎？\n\n我目前找到的相關資料如下：\n\n{best_answer}"
-    # If the knowledge base was empty or no match was found at all.
     else:
         return best_answer
 
-
+# --- ✨ 3. 更新主函式，優先檢查是否觸發工具，再查詢知識庫 ---
 def ai_chatbot_with_kb(message, history):
     """
-    Handles chatbot interaction by calling the fuzzy matching function and sends a notification.
+    處理聊天機器人互動的主函式。
+    優先檢查是否觸發即時工具，若無，則查詢靜態知識庫。
     """
-    # Notification logic (remains the same)
+    # (通知邏輯維持不變)
     tz = pytz.timezone('Asia/Taipei')
     current_time = datetime.now(tz).strftime('%H:%M:%S')
     visit_count = get_current_visit_count()
@@ -110,6 +101,18 @@ def ai_chatbot_with_kb(message, history):
     )
     send_line_notification_in_background(notification_text)
 
-    # --- ✨ 3. 使用新的模糊比對函式取代舊的搜尋邏輯 ---
-    # The old exact-match logic is now replaced by a single call to our new function.
-    return find_best_match(message.strip(), KNOWLEDGE_BASE)
+    user_message = message.lower().strip()
+
+    # 步驟 1: 檢查是否觸發任何即時工具
+    for tool in LIVE_TOOLS:
+        for keyword in tool["keywords"]:
+            if keyword in user_message:
+                try:
+                    # 如果觸發，執行對應的函式並回傳結果
+                    print(f"🔍 觸發即時工具：{tool['name']}")
+                    return tool["function"]()
+                except Exception as e:
+                    return f"❌ 執行「{tool['name']}」工具時發生錯誤：{e}"
+
+    # 步驟 2: 如果沒有觸發工具，則查詢靜態知識庫 (使用舊有的模糊比對)
+    return find_best_match_from_kb(user_message, KNOWLEDGE_BASE)

services/cwa_service.py

@@ -0,0 +1,149 @@
+# 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))
+
+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 ("未知", "未知")
+    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, "未知")
+
+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()
+
+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)
+        # 假設 TAIPEI_TZ 已在檔案開頭定義
+        df["Time"] = pd.to_datetime(df["Time"], errors="coerce").dt.tz_localize(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:
+    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()
+        
+        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:
+        raise e
+

services/news_service.py

@@ -0,0 +1,75 @@
+# news_service.py
+import requests
+import json
+from datetime import datetime
+
+# 從 config.py 匯入 API URL
+from config import PTS_NEWS_API
+
+def fetch_today_news() -> str:
+    """
+    從公視新聞網 API 擷取今日新聞並格式化為易於閱讀的文字訊息。
+    """
+    try:
+        # 發送 API 請求
+        response = requests.get(PTS_NEWS_API, timeout=20)
+        response.raise_for_status()  # 確保請求成功
+
+        # 解析 JSON 回應
+        data = response.json()
+        
+        source = data.get("source", "未知來源")
+        last_updated_raw = data.get("last_updated")
+
+        # 格式化資料更新時間
+        if last_updated_raw:
+            try:
+                # 處理可能包含時區資訊的 ISO 格式時間
+                last_updated_dt = datetime.fromisoformat(last_updated_raw.replace("Z", "+00:00"))
+                last_updated_str = last_updated_dt.strftime('%Y-%m-%d %H:%M')
+            except (ValueError, TypeError):
+                last_updated_str = str(last_updated_raw)
+        else:
+            last_updated_str = "未知"
+            
+        # 建立訊息標頭
+        lines = [
+            f"📰 今日新聞摘要 ({source})",
+            f"   (資料更新: {last_updated_str})",
+            "─" * 25
+        ]
+
+        articles = data.get("articles", [])
+        if not articles:
+            lines.append("\nℹ️ 目前沒有可顯示的新聞。")
+            return "\n".join(lines)
+            
+        # 逐條解析並格式化新聞
+        for i, article in enumerate(articles[:10]): # 最多顯示 10 則新聞
+            title = article.get("title", "無標題").strip()
+            link = article.get("link", "#")
+            summary = article.get("summary", "無摘要。").strip()
+
+            # 組合單條新聞訊息
+            article_lines = [
+                f"📌 {title}",
+                f"📝 摘要: {summary}",
+                f"🔗 閱讀全文: {link}"
+            ]
+            
+            lines.append("\n".join(article_lines))
+            
+            # 在新聞之間加入分隔線
+            if i < len(articles) - 1:
+                lines.append("-" * 15)
+
+        return "\n\n".join(lines)
+
+    except requests.exceptions.Timeout:
+        return "❌ 新聞查詢失敗：連線超時。"
+    except requests.exceptions.RequestException as e:
+        return f"❌ 新聞查詢失敗：網路連線錯誤。\n錯誤訊息：{e}"
+    except json.JSONDecodeError:
+        return "❌ 新聞查詢失敗：無法解析伺服器回傳的資料格式。"
+    except Exception as e:
+        return f"❌ 新聞查詢失敗：發生未預期的錯誤。\n錯誤訊息：{e}"

services/pws_service.py

@@ -0,0 +1,132 @@
+# pws_service.py (Robust Version)
+import requests
+import json
+from datetime import datetime
+from config import PWS_API_URL, CWA_PWS_EARTHQUAKE_API
+
+def fetch_latest_pws_info() -> str:
+    """
+    從 MCP PWS 伺服器擷取最新的 PWS (Public Weather Service) 發布情形。
+    此版本經過優化，能穩定處理 Gradio API 的 Server-Sent Events (SSE) 串流。
+    """
+    final_report = None
+    
+    try:
+        # 使用 stream=True 保持連線，接收伺服器持續發送的事件
+        with requests.get(PWS_API_URL, timeout=20, stream=True) as r:
+            r.raise_for_status()  # 確保 HTTP 連線成功
+
+            # 迭代處理從伺服器接收到的每一行資料
+            for line in r.iter_lines():
+                if line:
+                    decoded_line = line.decode('utf-8')
+                    
+                    # SSE 事件的資料以 "data: " 開頭
+                    if decoded_line.startswith('data:'):
+                        json_data_str = decoded_line[len('data:'):].strip()
+                        
+                        try:
+                            data = json.loads(json_data_str)
+                            # Gradio API 在完成時會發送 'process_completed' 訊息
+                            if data.get("msg") == "process_completed":
+                                # 最終結果通常在 output['data'][0] 中
+                                output_data = data.get("output", {}).get("data", [])
+                                if output_data:
+                                    final_report = output_data[0]
+                                break # 收到最終結果，跳出迴圈
+
+                        except (json.JSONDecodeError, KeyError, IndexError):
+                            # 如果中途的訊息格式不符或不含所需資料，靜默忽略並繼續等待
+                            continue
+
+        # 迴圈結束後，檢查是否成功取得最終報告
+        if final_report:
+            header = "🛰️ 最新 PWS 發布情形:"
+            separator = "-" * 20
+            return f"{header}\n{separator}\n{final_report}"
+        else:
+            return "❌ PWS 查詢失敗：已連接伺服器，但未收到有效的最終報告資料。"
+
+    except requests.exceptions.Timeout:
+        return f"❌ PWS 查詢失敗：連線超時，伺服器沒有在 20 秒內回應。"
+    except requests.exceptions.RequestException as e:
+        return f"❌ PWS 查詢失敗：網路連線錯誤。\n錯誤訊息：{e}"
+    except Exception as e:
+        return f"❌ PWS 查詢失敗：發生未預期的錯誤。\n錯誤訊息：{e}"
+
+def fetch_cwa_pws_earthquake_info() -> str:
+    """從指定的 API 端點擷取最新的 CWA PWS 地震訊息並格式化輸出。"""
+    try:
+        response = requests.get(CWA_PWS_EARTHQUAKE_API, timeout=15)
+        response.raise_for_status()
+        
+        data = response.json()
+        
+        source = data.get("source", "未知來源")
+        last_updated_raw = data.get("last_updated")
+        
+        # 格式化資料更新時間
+        if last_updated_raw:
+            try:
+                last_updated_dt = datetime.fromisoformat(last_updated_raw)
+                last_updated_str = last_updated_dt.strftime('%Y-%m-%d %H:%M')
+            except (ValueError, TypeError):
+                last_updated_str = last_updated_raw
+        else:
+            last_updated_str = "未知"
+
+        # 建立訊息標頭
+        lines = [
+            f"📢 最新地震PWS警報 ({source})",
+            f"   (資料更新時間: {last_updated_str})",
+            "─" * 25
+        ]
+
+        alerts = data.get("alerts", [])
+        if not alerts:
+            lines.append("\n✅ 目前無最新的地震PWS警報。")
+            return "\n".join(lines)
+
+        # 逐條解析並格式化警報
+        for i, alert in enumerate(alerts):
+            title = alert.get("title", "無標題")
+            publish_time_raw = alert.get("publish_time")
+            
+            # 格式化警報發布時間
+            if publish_time_raw:
+                try:
+                    publish_time_dt = datetime.fromisoformat(publish_time_raw)
+                    publish_time_str = publish_time_dt.strftime('%Y-%m-%d %H:%M')
+                except (ValueError, TypeError):
+                    publish_time_str = publish_time_raw
+            else:
+                publish_time_str = "未知"
+
+            summary = alert.get("summary", "無摘要。").strip()
+            affected_counties = alert.get("affected_counties", [])
+            affected_str = ", ".join(affected_counties) if affected_counties else "未提供"
+
+            # 組合單條警報訊息
+            alert_lines = [
+                f"🚨 {title} ({publish_time_str})",
+                f"🗺️ 影響地區: {affected_str}",
+                f"💬 內容: {summary}"
+            ]
+            
+            # 將單條警報加入主訊息列表
+            lines.append("\n".join(alert_lines))
+            
+            # 在警報之間加入分隔線
+            if i < len(alerts) - 1:
+                lines.append("-" * 15)
+
+        return "\n\n".join(lines)
+
+    except requests.exceptions.Timeout:
+        return "❌ PWS 地震訊息查詢失敗：連線超時。"
+    except requests.exceptions.RequestException as e:
+        return f"❌ PWS 地震訊息查詢失敗：網路連線錯誤。\n錯誤訊息：{e}"
+    except json.JSONDecodeError:
+        return "❌ PWS 地震訊息查詢失敗：無法解析伺服器回傳的資料格式。"
+    except Exception as e:
+        return f"❌ PWS 地震訊息查詢失敗：發生未預期的錯誤。\n錯誤訊息：{e}"

services/usgs_service.py

@@ -0,0 +1,78 @@
+# usgs_service.py
+import requests
+import pandas as pd
+from datetime import datetime, timedelta, timezone
+from config import USGS_API_BASE_URL, CURRENT_YEAR
+
+def _iso(dt: datetime) -> str:
+    """將 datetime 物件格式化為 USGS API 需要的 ISO 8601 字串。"""
+    return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
+
+def fetch_global_last24h_text(min_mag: float = 5.0, limit: int = 10) -> str:
+    """從 USGS 擷取過去 24 小時的全球顯著地震。"""
+    now_utc = datetime.now(timezone.utc)
+    since = now_utc - timedelta(hours=24)
+    params = {
+        "format": "geojson",
+        "starttime": _iso(since),
+        "endtime": _iso(now_utc),
+        "minmagnitude": float(min_mag),
+        "limit": int(limit),
+        "orderby": "time",
+    }
+    try:
+        r = requests.get(USGS_API_BASE_URL, params=params, timeout=15)
+        r.raise_for_status()
+        features = r.json().get("features", [])
+        if not features:
+            return f"✅ 過去 24 小時內，全球無規模 {min_mag} 以上的顯著地震。"
+        
+        lines = [f"🚨 近 24 小時全球顯著地震 (M≥{min_mag}):", "-" * 20]
+        for f in features:
+            p = f["properties"]
+            t_utc = datetime.fromtimestamp(p["time"] / 1000, tz=timezone.utc)
+            
+            lines.append(
+                # [修改] 將 "震級" 改為 "規模"
+                f"規模: {p['mag']:.1f} | 日期時間: {t_utc.strftime('%Y-%m-%d %H:%M')} (UTC)\n"
+                f"地點: {p.get('place', 'N/A')}\n"
+                f"報告連結: {p.get('url', '無')}"
+            )
+        return "\n\n".join(lines)
+    except Exception as e:
+        return f"❌ 查詢失敗：{e}"
+
+def fetch_taiwan_df_this_year(min_mag: float = 5.0) -> pd.DataFrame | str:
+    """從USGS擷取今年以來台灣區域的顯著地震。"""
+    now_utc = datetime.now(timezone.utc)
+    start_of_year_utc = datetime(now_utc.year, 1, 1, tzinfo=timezone.utc)
+    params = {
+        "format": "geojson", "starttime": _iso(start_of_year_utc), "endtime": _iso(now_utc),
+        "minmagnitude": float(min_mag),
+        "minlatitude": 21, "maxlatitude": 26,
+        "minlongitude": 119, "maxlongitude": 123,
+        "limit": 250,
+        "orderby": "time",
+    }
+    try:
+        r = requests.get(USGS_API_BASE_URL, params=params, timeout=20)
+        r.raise_for_status()
+        features = r.json().get("features", [])
+        if not features:
+            return f"✅ 今年 ({CURRENT_YEAR} 年) 以來，台灣區域無 M≥{min_mag:.1f} 的顯著地震。"
+        
+        rows = []
+        for f in features:
+            p = f["properties"]
+            lon, lat, *_ = f["geometry"]["coordinates"]
+            rows.append({
+                "latitude": lat, 
+                "longitude": lon, 
+                "magnitude": p["mag"],
+                "place": p.get("place", ""), 
+                "time_utc": datetime.fromtimestamp(p["time"]/1000, tz=timezone.utc),
+                "url": p.get("url", "")
+            })
+        return pd.DataFrame(rows)
+    except Exception as e:
+        return f"❌ 查詢失敗: {e}"