File size: 6,207 Bytes
cd9bca9 0503b45 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 |
"""
業務查詢服務 - 整合 NLP 分析和資料庫查詢
用於處理 LINE 用戶的自然語言業務查詢
"""
import logging
from typing import Dict, Any, Optional, List
from backend.services.nlp_service import NLPService
from backend.services.database_service import DatabaseService
from backend.models.schemas import DatabaseResult
logger = logging.getLogger(__name__)
class BusinessQueryService:
"""業務查詢服務 - 整合自然語言處理和資料庫查詢"""
def __init__(self):
self.nlp_service = NLPService()
self.db_service = DatabaseService()
def process_user_query(self, user_message: str, user_id: str = None) -> Dict[str, Any]:
"""
處理用戶查詢的主要入口點
Args:
user_message: 用戶的自然語言查詢
user_id: 用戶ID（可選）
Returns:
包含查詢結果和格式化回應的字典
"""
try:
# 1. 使用 NLP 服務分析用戶查詢
analysis_result = self.nlp_service.analyze_business_query(user_message, user_id)
logger.info(f"NLP 分析結果 - 意圖: {analysis_result.intent}, 信心度: {analysis_result.confidence}")
# 2. 根據分析結果執行對應的資料庫查詢
db_result = self._execute_database_query(analysis_result)
# 3. 格式化回應訊息
response_message = self.nlp_service.format_response_message(db_result, analysis_result.intent)
# 4. 儲存查詢記錄（如果有用戶ID）
if user_id:
self.db_service.save_message(user_id, user_message, "query")
return {
"success": db_result.success,
"intent": analysis_result.intent,
"confidence": analysis_result.confidence,
"entities": analysis_result.entities,
"response_message": response_message,
"data": db_result.data,
"count": db_result.count,
"error": db_result.error
}
except Exception as e:
logger.error(f"處理用戶查詢時發生錯誤: {str(e)}")
return {
"success": False,
"intent": "unknown",
"confidence": 0.0,
"entities": {},
"response_message": f"抱歉，處理您的查詢時發生錯誤：{str(e)}",
"data": [],
"count": 0,
"error": str(e)
}
def _execute_database_query(self, analysis_result) -> DatabaseResult:
"""根據 NLP 分析結果執行對應的資料庫查詢"""
try:
method_name = analysis_result.parameters.get("method")
if method_name == "search_products":
return self.db_service.search_products(
query_text=analysis_result.parameters.get("query_text"),
category=analysis_result.parameters.get("category"),
limit=analysis_result.parameters.get("limit", 10)
)
elif method_name == "check_inventory":
return self.db_service.check_inventory(
product_name=analysis_result.parameters.get("product_name"),
category=analysis_result.parameters.get("category")
)
elif method_name == "search_orders":
return self.db_service.search_orders(
user_id=analysis_result.parameters.get("user_id"),
status=analysis_result.parameters.get("status"),
limit=analysis_result.parameters.get("limit", 10)
)
elif method_name == "get_low_stock_products":
return self.db_service.get_low_stock_products(
threshold=analysis_result.parameters.get("threshold", 10)
)
elif method_name == "get_business_summary":
return self.db_service.get_business_summary()
else:
# 使用通用的自然語言查詢處理
return self.db_service.process_natural_language_query(
analysis_result.parameters.get("query_text", ""),
analysis_result.parameters.get("user_id")
)
except Exception as e:
logger.error(f"執行資料庫查詢時發生錯誤: {str(e)}")
return DatabaseResult(
success=False,
error=f"資料庫查詢失敗: {str(e)}"
)
def get_query_suggestions(self, user_id: str = None) -> List[str]:
"""提供查詢建議"""
suggestions = [
"查詢商品 iPhone",
"庫存查詢 筆記型電腦",
"我的訂單",
"低庫存商品",
"業務統計摘要",
"查詢客戶資料",
"今天的銷售報表"
]
return suggestions
def get_help_message(self) -> str:
"""取得幫助訊息"""
help_text = """
🤖 智能查詢助手使用說明：
📦 商品查詢：
• "查詢商品 iPhone"
• "有什麼筆記型電腦"
• "商品價格查詢"
📊 庫存查詢：
• "庫存查詢 iPhone"
• "筆記型電腦還有多少"
• "查詢存貨狀況"
📋 訂單查詢：
• "我的訂單"
• "查詢訂單狀態"
• "訂單編號 ORD001"
⚠️ 庫存警告：
• "低庫存商品"
• "缺貨商品查詢"
📈 業務統計：
• "業務摘要"
• "統計報表"
• "總計資料"
💡 您可以用自然語言提問，系統會智能理解您的需求！
"""
return help_text.strip()
def validate_user_permissions(self, user_id: str, query_type: str) -> bool:
"""驗證用戶權限（可根據需要擴展）"""
# 這裡可以根據用戶角色和查詢類型來驗證權限
# 目前先返回 True，允許所有查詢
return True |