cwadayi commited on
Commit
2c25eaa
·
verified ·
1 Parent(s): 8302d50

Update command_handler.py

Browse files
Files changed (1) hide show
  1. command_handler.py +15 -70
command_handler.py CHANGED
@@ -7,103 +7,48 @@ from cwa_service import fetch_cwa_alarm_list
7
  from usgs_service import fetch_global_last24h_text, fetch_taiwan_df_this_year
8
  from plotting_service import create_and_save_map
9
  from ai_service import generate_ai_text
10
- from weather_service import fetch_weather_by_location # [新增] 匯入天氣服務
11
  from config import CURRENT_YEAR, HF_SPACE_URL
12
 
13
- def get_help_message() -> TextMessage:
14
- """回傳包含所有可用指令的說明訊息。"""
15
- text = (
16
- "📖 指令列表\n\n"
17
- "【地震資訊】\n"
18
- "• /global - 查詢全球近24小時顯著地震\n"
19
- "• /taiwan - 顯示今年台灣顯著地震列表\n"
20
- "• /map - 繪製台灣顯著地震地圖\n"
21
- "• /alert - 獲取 CWA 最新地震預警\n\n"
22
- "【AI 與工具】\n"
23
- "• /ai <問題> - 與 AI 助理對話\n"
24
- "• /weather <地點> - 查詢即時天氣\n\n"
25
- "【基本指令】\n"
26
- "• /info - 關於此機器人\n"
27
- "• /help - 顯示此說明"
28
- )
29
  return TextMessage(text=text)
30
 
31
- def get_info_message() -> TextMessage:
32
- """回傳機器人資訊。"""
33
- text = (
34
- "🤖 關於我\n\n"
35
- "我是一個多功能助理機器人,提供地震查詢、天氣資訊與 AI 對話功能。\n\n"
36
- "• 版本: 2.0\n"
37
- "• 資料來源: CWA, USGS, Hugging Face\n"
38
- "• 開發者: dayichen"
39
- )
40
  return TextMessage(text=text)
41
 
42
- def get_taiwan_earthquake_list() -> TextMessage:
43
- """回傳近期的台灣地震文字列表。"""
44
- result = fetch_taiwan_df_this_year()
45
- if isinstance(result, pd.DataFrame):
46
- count = len(result)
47
- lines = [f"🇹🇼 今年 ({CURRENT_YEAR} 年) 台灣區域顯著地震 (M≥5.0),共 {count} 筆:", "-" * 20]
48
- for _, row in result.head(15).iterrows():
49
- t = row["time_utc"].strftime("%Y-%m-%d %H:%M")
50
- lines.append(f"震級: {row['magnitude']:.1f} | 日期時間: {t} (UTC)\n地點: {row['place']}")
51
- if count > 15:
52
- lines.append(f"... (還有 {count - 15} 筆,請用 /map 查看全部)")
53
- reply_text = "\n\n".join(lines)
54
- else:
55
- reply_text = result
56
  return TextMessage(text=reply_text)
57
 
58
- def get_taiwan_earthquake_map(base_url: str) -> list:
59
- """產生並回傳台灣地震地圖的訊息。"""
60
- result = fetch_taiwan_df_this_year()
61
- if isinstance(result, pd.DataFrame):
62
- filename = create_and_save_map(result)
63
- image_url = f"{(HF_SPACE_URL or base_url)}/static/{filename}"
64
- return [
65
  TextMessage(text="🗺️ 已為您繪製今年台灣區域 M≥5.0 地震分佈圖(UTC)。"),
66
  ImageMessage(original_content_url=image_url, preview_image_url=image_url),
67
  ]
68
- else:
69
- return [TextMessage(text=result)]
70
 
71
  def process_message(user_message_raw: str, request_base_url: str) -> list:
72
  """處理使用者的文字訊息並回傳一個包含回覆訊息的列表。"""
73
  user_message = (user_message_raw or "").strip()
74
 
75
- # 將舊指令對應到新指令,提供向下相容
76
- if user_message.lower() in ["地震", "quake"]:
77
- user_message = "/global"
78
- elif user_message.lower() in ["台灣地震", "臺灣地震"]:
79
- user_message = "/taiwan"
80
- elif user_message.lower() in ["台灣地震畫圖", "臺灣地震畫圖"]:
81
- user_message = "/map"
82
- elif user_message.lower() == "地震預警":
83
- user_message = "/alert"
84
 
85
- # [重構] 改為處理以 / 開頭的指令
86
  if user_message.startswith('/'):
87
  parts = user_message.split(' ', 1)
88
  command = parts[0].lower()
89
  arg = parts[1] if len(parts) > 1 else ""
90
 
91
- if command == '/help':
92
- return [get_help_message()]
93
- elif command == '/info':
94
- return [get_info_message()]
95
- elif command == '/global':
96
- return [TextMessage(text=fetch_global_last24h_text())]
97
- elif command == '/taiwan':
98
- return [get_taiwan_earthquake_list()]
99
- elif command == '/map':
100
- return get_taiwan_earthquake_map(request_base_url)
101
  elif command == '/alert':
102
  return [TextMessage(text=fetch_cwa_alarm_list(limit=5))]
103
  elif command == '/weather':
104
  if not arg:
105
  return [TextMessage(text="請輸入地點,例如:/weather 台北")]
106
- return [TextMessage(text=fetch_weather_by_location(arg))]
 
107
  elif command == '/ai':
108
  if not arg:
109
  return [TextMessage(text="請輸入問題,例如:/ai 台灣最高的山是哪座?")]
 
7
  from usgs_service import fetch_global_last24h_text, fetch_taiwan_df_this_year
8
  from plotting_service import create_and_save_map
9
  from ai_service import generate_ai_text
10
+ from weather_service import fetch_forecast_by_location # [修改] 更新匯入的函式名稱
11
  from config import CURRENT_YEAR, HF_SPACE_URL
12
 
13
+ # ... (get_help_message, get_info_message 等函式不變) ...
14
+ def get_help_message():
15
+ # ...
 
 
 
 
 
 
 
 
 
 
 
 
 
16
  return TextMessage(text=text)
17
 
18
+ def get_info_message():
19
+ # ...
 
 
 
 
 
 
 
20
  return TextMessage(text=text)
21
 
22
+ def get_taiwan_earthquake_list():
23
+ # ...
 
 
 
 
 
 
 
 
 
 
 
 
24
  return TextMessage(text=reply_text)
25
 
26
+ def get_taiwan_earthquake_map(base_url: str):
27
+ # ...
28
+ return [
 
 
 
 
29
  TextMessage(text="🗺️ 已為您繪製今年台灣區域 M≥5.0 地震分佈圖(UTC)。"),
30
  ImageMessage(original_content_url=image_url, preview_image_url=image_url),
31
  ]
 
 
32
 
33
  def process_message(user_message_raw: str, request_base_url: str) -> list:
34
  """處理使用者的文字訊息並回傳一個包含回覆訊息的列表。"""
35
  user_message = (user_message_raw or "").strip()
36
 
37
+ # ... (舊指令相容性轉換不變) ...
 
 
 
 
 
 
 
 
38
 
 
39
  if user_message.startswith('/'):
40
  parts = user_message.split(' ', 1)
41
  command = parts[0].lower()
42
  arg = parts[1] if len(parts) > 1 else ""
43
 
44
+ # ... (其他 command 判斷不變) ...
 
 
 
 
 
 
 
 
 
45
  elif command == '/alert':
46
  return [TextMessage(text=fetch_cwa_alarm_list(limit=5))]
47
  elif command == '/weather':
48
  if not arg:
49
  return [TextMessage(text="請輸入地點,例如:/weather 台北")]
50
+ # [修改] 呼叫新的天氣預報函式
51
+ return [TextMessage(text=fetch_forecast_by_location(arg))]
52
  elif command == '/ai':
53
  if not arg:
54
  return [TextMessage(text="請輸入問題,例如:/ai 台灣最高的山是哪座?")]