File size: 14,436 Bytes
cd9bca9 81a7334 cd9bca9 81a7334 |
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 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 |
"""
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)}" |