Update core/callbacks.py

core/callbacks.py

@@ -1,116 +1,132 @@
-# core/callbacks.py
-import io
-import contextlib
-import traceback
-from datetime import datetime
-import pytz
-import difflib
-
-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
-
-# (execute_user_code 函式維持不變，此處省略以節省篇幅)
-def execute_user_code(code_string, source_lab):
-    """Executes user-provided code in a restricted environment and sends a notification."""
-    string_io = io.StringIO()
-    status = "✅ 成功"
-    error_info = ""
-    fig = None
-    try:
-        with contextlib.redirect_stdout(string_io):
-            local_scope = {}
-            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:
-        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"
-            f"實驗室: {source_lab}\n"
-            f"執行狀態: {status}\n"
-            f"總載入數: {visit_count}"
-        )
-        if status == "❌ 失敗":
-            error_type = error_info.strip().split('\n')[-1]
-            notification_text += f"\n錯誤類型: {error_type}"
-        send_line_notification_in_background(notification_text)
-
-LIVE_TOOLS = [
-    {"keywords": ["新聞", "今日新聞", "news"], "function": news_service.fetch_today_news, "name": "今日新聞"},
-    {"keywords": ["cwa地震", "顯著地震", "有感地震"], "function": cwa_service.fetch_significant_earthquakes, "name": "CWA 顯著有感地震"},
-    # --- ✨ 在這裡新增了 "eew" 和 "現在有地震預警嗎？" ---
-    {"keywords": ["地震預警", "cwa alarm", "eew", "現在有地震預警嗎？"], "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", "pws", "最新的 pws 地震警報"], "function": pws_service.fetch_cwa_pws_earthquake_info, "name": "PWS 地震警報"},
-]
-
-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
-    for category, entries in knowledge_base.items():
-        for entry in entries:
-            for keyword in entry['keywords']:
-                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 best_score >= threshold:
-        return best_answer
-    elif best_match_keyword:
-        return f"這個問題我不是很確定，您是指 **「{best_match_keyword}」** 嗎？\n\n我目前找到的相關資料如下：\n\n{best_answer}"
-    else:
-        return best_answer
-
-def ai_chatbot_with_kb(message, history):
-    """
-    處理聊天機器人互動的主函式。
-    優先檢查是否觸發即時工具，若無，則查詢靜態知識庫。
-    """
-    # (通知邏輯維持不變)
-    tz = pytz.timezone('Asia/Taipei')
-    current_time = datetime.now(tz).strftime('%H:%M:%S')
-    visit_count = get_current_visit_count()
-    notification_text = (
-        f"🤖 AI 助教被提問！\n\n"
-        f"時間: {current_time}\n"
-        f"使用者問題:\n「{message}」\n\n"
-        f"總載入數: {visit_count}"
-    )
-    send_line_notification_in_background(notification_text)
-
-    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)
+# core/callbacks.py
+import io
+import contextlib
+import traceback
+from datetime import datetime
+import pytz
+import difflib
+import re # ✨ 1. 匯入 re 模組
+
+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
+
+# (execute_user_code 函式維持不變，此處省略以節省篇幅)
+def execute_user_code(code_string, source_lab):
+    """Executes user-provided code in a restricted environment and sends a notification."""
+    string_io = io.StringIO()
+    status = "✅ 成功"
+    error_info = ""
+    fig = None
+    try:
+        with contextlib.redirect_stdout(string_io):
+            local_scope = {}
+            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:
+        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"
+            f"實驗室: {source_lab}\n"
+            f"執行狀態: {status}\n"
+            f"總載入數: {visit_count}"
+        )
+        if status == "❌ 失敗":
+            error_type = error_info.strip().split('\n')[-1]
+            notification_text += f"\n錯誤類型: {error_type}"
+        send_line_notification_in_background(notification_text)
+
+LIVE_TOOLS = [
+    {"keywords": ["新聞", "今日新聞", "news"], "function": news_service.fetch_today_news, "name": "今日新聞"},
+    {"keywords": ["cwa地震", "顯著地震", "有感地震"], "function": cwa_service.fetch_significant_earthquakes, "name": "CWA 顯著有感地震"},
+    # --- ✨ 在這裡新增了 "eew" 和 "現在有地震預警嗎？" ---
+    {"keywords": ["地震預警", "cwa alarm", "eew", "現在有地震預警嗎？"], "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", "pws", "最新的 pws 地震警報"], "function": pws_service.fetch_cwa_pws_earthquake_info, "name": "PWS 地震警報"},
+]
+
+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
+    for category, entries in knowledge_base.items():
+        for entry in entries:
+            for keyword in entry['keywords']:
+                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 best_score >= threshold:
+        return best_answer
+    elif best_match_keyword:
+        return f"這個問題我不是很確定，您是指 **「{best_match_keyword}」** 嗎？\n\n我目前找到的相關資料如下：\n\n{best_answer}"
+    else:
+        return best_answer
+
+# --- ✨ 2. 這是修改後的核心函式 ---
+def ai_chatbot_with_kb(message, history):
+    """
+    處理聊天機器人互動的主函式。
+    優先檢查進階查詢格式，其次檢查即時工具，最後查詢靜態知識庫。
+    """
+    # (通知邏輯維持不變)
+    tz = pytz.timezone('Asia/Taipei')
+    current_time = datetime.now(tz).strftime('%H:%M:%S')
+    visit_count = get_current_visit_count()
+    notification_text = (
+        f"🤖 AI 助教被提問！\n\n"
+        f"時間: {current_time}\n"
+        f"使用者問題:\n「{message}」\n\n"
+        f"總載入數: {visit_count}"
+    )
+    send_line_notification_in_background(notification_text)
+
+    user_message = message.strip() # 保留大小寫以符合可能的未來需求，但比對時再轉小寫
+
+    # 步驟 1: 檢查是否為進階地震查詢格式
+    # 使用正規表示式來匹配 "查詢 YYYY-MM-DD 到 YYYY-MM-DD 規模 X.X 以上地震"
+    pattern = r"查詢\s*(\d{4}-\d{2}-\d{2})\s*到\s*(\d{4}-\d{2}-\d{2})\s*規模\s*(\d+(?:\.\d+)?)\s*以上地震"
+    match = re.search(pattern, user_message)
+    
+    if match:
+        start_date, end_date, magnitude = match.groups()
+        print(f"🔍 觸發進階地震查詢：{start_date} 到 {end_date}, M≥{magnitude}")
+        try:
+            # 呼叫 usgs_service 中的新函式
+            return usgs_service.fetch_usgs_earthquakes_by_date(start_date, end_date, float(magnitude))
+        except Exception as e:
+            return f"❌ 執行進階查詢時發生錯誤：{e}"
+
+    # 步驟 2: 如果不是進階查詢，則檢查是否觸發任何即時工具 (關鍵字比對)
+    user_message_lower = user_message.lower()
+    for tool in LIVE_TOOLS:
+        for keyword in tool["keywords"]:
+            if keyword in user_message_lower:
+                try:
+                    print(f"🔍 觸發即時工具：{tool['name']}")
+                    return tool["function"]()
+                except Exception as e:
+                    return f"❌ 執行「{tool['name']}」工具時發生錯誤：{e}"
+
+    # 步驟 3: 如果都沒有觸發，則查詢靜態知識庫
+    return find_best_match_from_kb(user_message_lower, KNOWLEDGE_BASE)