core/callbacks.py

# core/callbacks.py
import io
import contextlib
import traceback
from datetime import datetime
import pytz
import difflib
import 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 顯著有感地震"},
    {"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.strip()

    # 步驟 1: 檢查是否為快捷指令
    if user_message.startswith('#'):
        # ✨ 新增 elif 區塊來處理 #7
        if user_message == '#7':
            print("💡 觸發進階查詢說明 (#7)")
            return "好的，請依照以下格式提供查詢的日期範圍與規模：\n`查詢 YYYY-MM-DD 到 YYYY-MM-DD 規模 X.X 以上地震`\n\n例如：`查詢 2024-04-01 到 2024-04-07 規模 6.0 以上地震`"
        
        try:
            tool_index = int(user_message[1:]) - 1
            if 0 <= tool_index < len(LIVE_TOOLS):
                tool = LIVE_TOOLS[tool_index]
                print(f"🚀 觸發快捷指令：{tool['name']}")
                return tool["function"]()
            else:
                return f"⚠️ 無效的快捷指令！請輸入 #1 到 #{len(LIVE_TOOLS)} 之間的數字，或 #7 查看進階查詢說明。"
        except (ValueError, IndexError):
            return "⚠️ 快捷指令格式錯誤！請輸入像 `#1` 這樣的格式。"

    # 步驟 2: 檢查是否為進階地震查詢格式
    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:
            return usgs_service.fetch_usgs_earthquakes_by_date(start_date, end_date, float(magnitude))
        except Exception as e:
            return f"❌ 執行進階查詢時發生錯誤：{e}"

    # 步驟 3: 如果都不是，則檢查是否觸發即時工具 (關鍵字比對)
    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}"

    # 步驟 4: 如果都沒有觸發，則查詢靜態知識庫
    return find_best_match_from_kb(user_message_lower, KNOWLEDGE_BASE)