Spaces:
Sleeping
Sleeping
File size: 4,870 Bytes
071beff 240ea75 2edbd99 071beff 417ae3a 5a97175 417ae3a 2edbd99 417ae3a 8ac5ed8 417ae3a 071beff 240ea75 2edbd99 240ea75 071beff 417ae3a 240ea75 417ae3a 240ea75 417ae3a 2edbd99 417ae3a 071beff |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 |
# command_handler.py
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"
"• /global - 查詢全球近24小時顯著地震\n"
"• /taiwan - 顯示今年台灣顯著地震列表\n"
"• /map - 取得台灣顯著地震地圖連結\n"
"• /alert - 獲取 CWA 最新地震預警\n"
"• /significant - CWA 最新顯著有感地震\n\n" # [移除] /local 指令說明
"【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.4\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)]
def process_message(user_message_raw: str, request_base_url: str) -> list:
"""處理使用者的文字訊息並回傳一個包含回覆訊息的列表。"""
user_message = (user_message_raw or "").strip()
# 將舊指令對應到新指令，提供向下相容
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 == '/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 == '/local': ...
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))]
|