""" 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}") # 🔧 後門測試：直接與 DeepSeek V3 聊天 if message_text.startswith('##test##'): return self._handle_deepseek_test(user_id, message_text[8:].strip()) # 檢查是否為特殊指令 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) } def _handle_deepseek_test(self, user_id: str, test_message: str) -> Dict[str, Any]: """ 🔧 後門測試：直接與 DeepSeek V3 聊天 使用方式：在 LINE 中發送 "##test##你好" """ try: import asyncio import httpx from backend.config import settings if not test_message: return { "type": "text", "text": "🔧 DeepSeek V3 測試模式\n\n使用方式：##test##[您的訊息]\n\n範例：##test##你好，請介紹一下自己", "user_id": user_id, "success": True, "intent": "deepseek_test" } # 同步調用異步函數的方法 def sync_call_deepseek(): return asyncio.run(self._call_deepseek_api(test_message)) try: response_text = sync_call_deepseek() return { "type": "text", "text": f"🤖 DeepSeek V3 回應：\n\n{response_text}", "user_id": user_id, "success": True, "intent": "deepseek_test" } except Exception as api_error: return { "type": "text", "text": f"❌ DeepSeek V3 API 錯誤：\n{str(api_error)}\n\n請檢查：\n1. API Key 設定\n2. 網路連線\n3. OpenRouter 服務狀態", "user_id": user_id, "success": False, "intent": "deepseek_test", "error": str(api_error) } except Exception as e: logger.error(f"DeepSeek 測試錯誤: {str(e)}") return { "type": "text", "text": f"❌ 測試功能錯誤：{str(e)}", "user_id": user_id, "success": False, "intent": "deepseek_test", "error": str(e) } async def _call_deepseek_api(self, message: str) -> str: """ 直接調用 DeepSeek V3 API """ from backend.config import settings api_key = settings.OPENROUTER_API_KEY model = settings.OPENROUTER_MODEL if not api_key: return "❌ OPENROUTER_API_KEY 未設定" headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", "HTTP-Referer": "https://huggingface.co/spaces", "X-Title": "LINE Bot DeepSeek Test" } try: async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post( "https://openrouter.ai/api/v1/chat/completions", headers=headers, json={ "model": model, "messages": [ { "role": "system", "content": "你是 DeepSeek V3，一個強大的中文 AI 助手。請用繁體中文回應，並在回應開頭說明你是 DeepSeek V3。" }, { "role": "user", "content": message } ], "temperature": 0.7, "max_tokens": 500 } ) if response.status_code == 200: result = response.json() content = result["choices"][0]["message"]["content"] return content else: return f"❌ API 錯誤 {response.status_code}: {response.text}" except httpx.TimeoutException: return "❌ API 請求超時，請稍後再試" except Exception as e: return f"❌ 網路錯誤: {str(e)}"