File size: 9,620 Bytes
cd9bca9 |
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 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 |
"""
LINE Bot 服務 - 整合業務查詢功能
處理來自 LINE 官方帳號的用戶訊息
"""
import logging
from typing import Dict, Any, Optional
from backend.services.business_query_service import BusinessQueryService
from backend.services.database_service import DatabaseService
logger = logging.getLogger(__name__)
class LineBotService:
"""LINE Bot 服務類"""
def __init__(self):
self.business_query_service = BusinessQueryService()
self.db_service = DatabaseService()
def handle_text_message(self, user_id: str, message_text: str, display_name: str = None) -> Dict[str, Any]:
"""
處理 LINE 用戶的文字訊息
Args:
user_id: LINE 用戶ID
message_text: 用戶發送的訊息內容
display_name: 用戶顯示名稱（可選）
Returns:
包含回應訊息和相關資料的字典
"""
try:
# 記錄用戶訊息
logger.info(f"收到來自用戶 {user_id} 的訊息: {message_text}")
# 檢查是否為特殊指令
if message_text.lower() in ['help', '幫助', '說明', '指令']:
return self._handle_help_command(user_id)
elif message_text.lower() in ['menu', '選單', '功能']:
return self._handle_menu_command(user_id)
# 確保用戶資料存在
self._ensure_user_profile(user_id, display_name)
# 處理業務查詢
query_result = self.business_query_service.process_user_query(message_text, user_id)
# 準備回應
response = {
"type": "text",
"text": query_result["response_message"],
"user_id": user_id,
"success": query_result["success"],
"intent": query_result["intent"],
"confidence": query_result["confidence"]
}
# 如果查詢成功且有資料，可以添加快速回覆按鈕
if query_result["success"] and query_result["data"]:
response["quick_reply"] = self._generate_quick_reply_buttons(query_result["intent"])
return response
except Exception as e:
logger.error(f"處理 LINE 訊息時發生錯誤: {str(e)}")
return {
"type": "text",
"text": "抱歉，系統暫時無法處理您的請求，請稍後再試。",
"user_id": user_id,
"success": False,
"error": str(e)
}
def _handle_help_command(self, user_id: str) -> Dict[str, Any]:
"""處理幫助指令"""
help_message = self.business_query_service.get_help_message()
return {
"type": "text",
"text": help_message,
"user_id": user_id,
"success": True,
"intent": "help"
}
def _handle_menu_command(self, user_id: str) -> Dict[str, Any]:
"""處理選單指令"""
menu_message = """
🏪 智能查詢系統主選單
請選擇您需要的功能：
1️⃣ 商品查詢
輸入：查詢商品 [商品名稱]
2️⃣ 庫存查詢
輸入：庫存查詢 [商品名稱]
3️⃣ 訂單查詢
輸入：我的訂單
4️⃣ 低庫存警告
輸入：低庫存商品
5️⃣ 業務統計
輸入：業務摘要
💡 您也可以直接用自然語言提問！
例如：「iPhone 還有多少庫存？」
輸入「幫助」查看詳細說明
"""
return {
"type": "text",
"text": menu_message.strip(),
"user_id": user_id,
"success": True,
"intent": "menu",
"quick_reply": {
"items": [
{"type": "action", "action": {"type": "message", "label": "商品查詢", "text": "查詢商品"}},
{"type": "action", "action": {"type": "message", "label": "庫存查詢", "text": "庫存查詢"}},
{"type": "action", "action": {"type": "message", "label": "我的訂單", "text": "我的訂單"}},
{"type": "action", "action": {"type": "message", "label": "低庫存", "text": "低庫存商品"}},
{"type": "action", "action": {"type": "message", "label": "業務統計", "text": "業務摘要"}}
]
}
}
def _ensure_user_profile(self, user_id: str, display_name: str = None):
"""確保用戶資料存在於資料庫中"""
try:
# 檢查用戶是否已存在
existing_user = self.db_service.get_user_profile(user_id)
if not existing_user:
# 建立新用戶資料
user_data = {
"user_id": user_id,
"display_name": display_name or f"用戶_{user_id[:8]}",
"created_at": "now()"
}
success = self.db_service.create_user_profile(user_data)
if success:
logger.info(f"已建立新用戶資料: {user_id}")
else:
logger.warning(f"建立用戶資料失敗: {user_id}")
except Exception as e:
logger.error(f"處理用戶資料時發生錯誤: {str(e)}")
def _generate_quick_reply_buttons(self, intent: str) -> Dict[str, Any]:
"""根據查詢意圖生成快速回覆按鈕"""
if intent == "product_search":
return {
"items": [
{"type": "action", "action": {"type": "message", "label": "查看庫存", "text": "庫存查詢"}},
{"type": "action", "action": {"type": "message", "label": "相關商品", "text": "查詢商品"}},
{"type": "action", "action": {"type": "message", "label": "主選單", "text": "選單"}}
]
}
elif intent == "inventory_check":
return {
"items": [
{"type": "action", "action": {"type": "message", "label": "低庫存警告", "text": "低庫存商品"}},
{"type": "action", "action": {"type": "message", "label": "商品詳情", "text": "查詢商品"}},
{"type": "action", "action": {"type": "message", "label": "主選單", "text": "選單"}}
]
}
elif intent == "order_search":
return {
"items": [
{"type": "action", "action": {"type": "message", "label": "業務統計", "text": "業務摘要"}},
{"type": "action", "action": {"type": "message", "label": "主選單", "text": "選單"}}
]
}
else:
return {
"items": [
{"type": "action", "action": {"type": "message", "label": "商品查詢", "text": "查詢商品"}},
{"type": "action", "action": {"type": "message", "label": "庫存查詢", "text": "庫存查詢"}},
{"type": "action", "action": {"type": "message", "label": "主選單", "text": "選單"}}
]
}
def handle_postback_action(self, user_id: str, postback_data: str) -> Dict[str, Any]:
"""處理 Postback 動作（按鈕點擊等）"""
try:
# 解析 postback 資料
if postback_data.startswith("query_"):
query_type = postback_data.replace("query_", "")
if query_type == "products":
return self.handle_text_message(user_id, "查詢商品")
elif query_type == "inventory":
return self.handle_text_message(user_id, "庫存查詢")
elif query_type == "orders":
return self.handle_text_message(user_id, "我的訂單")
elif query_type == "summary":
return self.handle_text_message(user_id, "業務摘要")
# 預設回應
return self._handle_menu_command(user_id)
except Exception as e:
logger.error(f"處理 Postback 動作時發生錯誤: {str(e)}")
return {
"type": "text",
"text": "抱歉，無法處理此操作。",
"user_id": user_id,
"success": False,
"error": str(e)
}
def get_user_activity_summary(self, user_id: str, days: int = 7) -> Dict[str, Any]:
"""取得用戶活動摘要"""
try:
# 這裡可以實作用戶活動統計
# 例如：查詢次數、常用功能等
summary = {
"user_id": user_id,
"period_days": days,
"total_queries": 0, # 從資料庫查詢
"most_used_features": [], # 最常使用的功能
"last_activity": None # 最後活動時間
}
return {
"success": True,
"data": summary
}
except Exception as e:
logger.error(f"取得用戶活動摘要時發生錯誤: {str(e)}")
return {
"success": False,
"error": str(e)
} |