""" 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) }