""" 增強的商品查詢服務 - 移植自 inventory_proj_routers 提供更精確的商品搜尋和庫存查詢功能 """ import logging from typing import List, Optional, Dict, Any from sqlalchemy.orm import Session, joinedload from sqlalchemy import or_, and_, func from backend.database.connection import get_database_session, close_database_session from backend.database.models import Product, Category, PurchaseOrder, SalesOrder from backend.models.schemas import DatabaseResult logger = logging.getLogger(__name__) class EnhancedProductService: """增強的商品查詢服務""" def __init__(self): pass def search_products_advanced( self, query_text: str = None, category_name: str = None, warehouse: str = None, include_stock_info: bool = True, min_stock: int = None, max_stock: int = None, limit: int = 20 ) -> DatabaseResult: """ 進階商品搜尋功能 Args: query_text: 搜尋關鍵字（商品名稱、編號、條碼） category_name: 分類名稱 warehouse: 倉庫名稱 include_stock_info: 是否包含庫存資訊 min_stock: 最小庫存量 max_stock: 最大庫存量 limit: 查詢限制數量 """ db = None try: logger.info(f"🔍 開始進階商品搜尋: '{query_text}'") db = get_database_session() # 建立基本查詢，預載入分類資訊 query = db.query(Product).options( joinedload(Product.category) ).filter(Product.is_deleted == False) # 關鍵字搜尋 - 使用智能關鍵字提取 if query_text: logger.info(f"🔑 使用關鍵字搜尋: '{query_text}'") # 使用智能關鍵字提取方法 search_terms = self._extract_keywords(query_text) logger.info(f"📝 智能提取的關鍵字: {search_terms}") if search_terms: # 建立搜尋條件 - 使用 OR 邏輯 search_filters = [] for term in search_terms: search_filters.extend([ Product.productName.ilike(f"%{term}%"), Product.productCode.ilike(f"%{term}%"), Product.barcode.ilike(f"%{term}%") ]) if search_filters: query = query.filter(or_(*search_filters)) logger.info(f"✅ 應用了 {len(search_filters)} 個搜尋條件") else: logger.warning(f"⚠️ 沒有提取到有效關鍵字") # 分類篩選 if category_name: query = query.join(Category).filter( Category.name.ilike(f"%{category_name}%") ) # 倉庫篩選 if warehouse: query = query.filter(Product.warehouse.ilike(f"%{warehouse}%")) # 庫存範圍篩選 if min_stock is not None: query = query.filter(Product.stock >= min_stock) if max_stock is not None: query = query.filter(Product.stock <= max_stock) # 執行查詢 products = query.limit(limit).all() # 轉換為字典格式 data = [] for product in products: product_data = { "id": product.id, "product_code": product.productCode, "product_name": product.productName, "unit": product.unit, "warehouse": product.warehouse, "unit_weight": product.unitWeight, "barcode": product.barcode, "category_id": product.category_id, "category_name": product.category.name if product.category else None, "created_at": product.createdAt.isoformat() if product.createdAt else None, "updated_at": product.updatedAt.isoformat() if product.updatedAt else None } # 包含庫存資訊 if include_stock_info: product_data.update({ "current_stock": product.stock, "stock_status": self._get_stock_status(product.stock), "is_low_stock": product.stock <= 10 }) data.append(product_data) return DatabaseResult( success=True, data=data, count=len(data) ) except Exception as e: logger.error(f"進階商品搜尋錯誤: {str(e)}") return DatabaseResult( success=False, error=f"商品搜尋失敗: {str(e)}" ) finally: if db: close_database_session(db) def get_products_by_category(self, category_name: str, limit: int = 20) -> DatabaseResult: """根據分類獲取商品""" db = None try: db = get_database_session() products = db.query(Product).join(Category).filter( and_( Category.name.ilike(f"%{category_name}%"), Product.is_deleted == False ) ).options(joinedload(Product.category)).limit(limit).all() data = [] for product in products: data.append({ "id": product.id, "product_code": product.productCode, "product_name": product.productName, "current_stock": product.stock, "unit": product.unit, "category_name": product.category.name if product.category else None, "warehouse": product.warehouse }) return DatabaseResult( success=True, data=data, count=len(data) ) except Exception as e: logger.error(f"分類商品查詢錯誤: {str(e)}") return DatabaseResult( success=False, error=f"分類商品查詢失敗: {str(e)}" ) finally: if db: close_database_session(db) def get_low_stock_products(self, threshold: int = 10) -> DatabaseResult: """獲取低庫存商品""" db = None try: db = get_database_session() products = db.query(Product).filter( and_( Product.stock <= threshold, Product.is_deleted == False ) ).options(joinedload(Product.category)).all() data = [] for product in products: data.append({ "id": product.id, "product_code": product.productCode, "product_name": product.productName, "current_stock": product.stock, "threshold": threshold, "unit": product.unit, "category_name": product.category.name if product.category else None, "warehouse": product.warehouse, "urgency_level": self._get_urgency_level(product.stock) }) return DatabaseResult( success=True, data=data, count=len(data) ) except Exception as e: logger.error(f"低庫存查詢錯誤: {str(e)}") return DatabaseResult( success=False, error=f"低庫存查詢失敗: {str(e)}" ) finally: if db: close_database_session(db) def get_product_recommendations(self, query_text: str, limit: int = 5) -> DatabaseResult: """ 商品推薦功能 - 基於關鍵字的智能推薦 特別針對像 "推薦貓砂" 這樣的查詢 """ db = None try: logger.info(f"🛍️ 開始商品推薦查詢: '{query_text}'") db = get_database_session() if not db: logger.error(f"❌ 無法獲取資料庫連接") return DatabaseResult(success=False, error="資料庫連接失敗") logger.info(f"✅ 資料庫連接成功") # 測試基本查詢 try: total_products = db.query(Product).filter(Product.is_deleted == False).count() logger.info(f"📊 資料庫中共有 {total_products} 個有效商品") except Exception as e: logger.error(f"❌ 基本查詢失敗: {str(e)}") return DatabaseResult(success=False, error=f"基本查詢失敗: {str(e)}") # 分析查詢關鍵字 keywords = self._extract_keywords(query_text) logger.info(f"🔑 推薦查詢關鍵字: {keywords}") # 建立推薦查詢 query = db.query(Product).filter(Product.is_deleted == False) logger.info(f"📊 基礎查詢建立完成") # 多關鍵字匹配 - 使用 OR 邏輯，任一關鍵字匹配即可 if keywords: search_filters = [] for keyword in keywords: search_filters.extend([ Product.productName.ilike(f"%{keyword}%"), Product.productCode.ilike(f"%{keyword}%"), Product.barcode.ilike(f"%{keyword}%") ]) logger.info(f"🔍 建立了 {len(search_filters)} 個搜尋條件") logger.info(f"🔑 前5個搜尋條件: {[f'productName ILIKE %{kw}%' for kw in keywords[:5]]}") # 使用 OR 連接所有搜尋條件 if search_filters: query = query.filter(or_(*search_filters)) logger.info(f"✅ 搜尋條件已應用到查詢") # 測試查詢執行 try: test_count = query.count() logger.info(f"📊 符合條件的商品數量: {test_count}") except Exception as e: logger.error(f"❌ 查詢計數失敗: {str(e)}") return DatabaseResult(success=False, error=f"查詢執行失敗: {str(e)}") else: logger.warning(f"⚠️ 沒有搜尋條件") else: logger.warning(f"⚠️ 沒有關鍵字，將返回所有商品") # 優先顯示有庫存的商品 query = query.order_by(Product.stock.desc()) try: products = query.options(joinedload(Product.category)).limit(limit).all() logger.info(f"📦 成功查詢到 {len(products)} 個商品") # 記錄每個商品的詳細資訊 for product in products: logger.info(f"📦 商品詳情: {product.productName} (編號: {product.productCode}, 庫存: {product.stock})") except Exception as e: logger.error(f"❌ 商品查詢執行失敗: {str(e)}") import traceback logger.error(f"📋 查詢錯誤詳情: {traceback.format_exc()}") return DatabaseResult(success=False, error=f"商品查詢執行失敗: {str(e)}") data = [] for product in products: # 檢查哪個關鍵字匹配了這個商品 matched_keywords = [] for keyword in keywords: if (keyword.lower() in product.productName.lower() or keyword.lower() in product.productCode.lower()): matched_keywords.append(keyword) logger.info(f"📦 商品: {product.productName} - 匹配關鍵字: {matched_keywords}") data.append({ "id": product.id, "product_code": product.productCode, "product_name": product.productName, "current_stock": product.stock, "unit": product.unit, "category_name": product.category.name if product.category else None, "warehouse": product.warehouse, "recommendation_reason": f"符合關鍵字: {', '.join(matched_keywords[:3]) if matched_keywords else '一般推薦'}", "availability": "有庫存" if product.stock > 0 else "缺貨" }) logger.info(f"✅ 推薦查詢完成，返回 {len(data)} 個商品") return DatabaseResult( success=True, data=data, count=len(data) ) except Exception as e: logger.error(f"❌ 商品推薦錯誤: {str(e)}") import traceback logger.error(f"📋 錯誤詳情: {traceback.format_exc()}") return DatabaseResult( success=False, error=f"商品推薦失敗: {str(e)}" ) finally: if db: logger.info(f"🔒 關閉資料庫連接") close_database_session(db) else: logger.warning(f"⚠️ 沒有資料庫連接需要關閉") def _get_stock_status(self, stock: int) -> str: """獲取庫存狀態""" if stock <= 0: return "缺貨" elif stock <= 5: return "庫存極低" elif stock <= 10: return "庫存偏低" elif stock <= 50: return "庫存正常" else: return "庫存充足" def _get_urgency_level(self, stock: int) -> str: """獲取緊急程度""" if stock <= 0: return "緊急" elif stock <= 3: return "高" elif stock <= 10: return "中" else: return "低" def _extract_keywords(self, query_text: str) -> List[str]: """從查詢文字中提取關鍵字，並擴展相關詞彙""" # 移除常見的查詢詞彙 stop_words = ['推薦', '有沒有', '是否有', '請問', '想要', '需要', '找', '查詢', '搜尋', '還有嗎', '還有', '嗎', '可以'] # 清理查詢文字 cleaned_text = query_text.replace('?', '').replace('？', '').strip() # 先嘗試提取核心商品詞彙 core_product_words = ['貓砂', '狗糧', '寵物', '商品', '產品', '貓', '狗', '犬', '礦砂'] extracted_core_words = [] for core_word in core_product_words: if core_word in cleaned_text: extracted_core_words.append(core_word) # 分割並清理關鍵字 words = cleaned_text.split() keywords = [] for word in words: if word not in stop_words and len(word) > 1: keywords.append(word) # 合併核心詞彙和分割的關鍵字 all_keywords = list(set(extracted_core_words + keywords)) # 如果沒有有效關鍵字，使用清理後的文字 if not all_keywords: all_keywords = [cleaned_text] # 擴展相關關鍵字 expanded_keywords = [] for keyword in all_keywords: expanded_keywords.append(keyword) # 貓砂相關擴展 if '貓砂' in keyword or '貓' in keyword: expanded_keywords.extend(['礦砂', '豆腐砂', '水晶砂', '木屑砂', 'litter', '貓砂']) # 狗糧相關擴展 if '狗糧' in keyword or '狗' in keyword: expanded_keywords.extend(['犬糧', '犬種', '狗食', 'dog']) # 寵物相關擴展 if '寵物' in keyword: expanded_keywords.extend(['貓', '狗', '犬', 'pet', 'cat']) # 商品相關擴展 if '商品' in keyword or '產品' in keyword: expanded_keywords.extend(['貓砂', '狗糧', '寵物', '食品', '用品']) # 去除重複並返回 unique_keywords = list(set(expanded_keywords)) # 記錄關鍵字提取結果用於調試 logger.info(f"關鍵字提取: '{query_text}' → 核心詞: {extracted_core_words} → 最終: {unique_keywords}") return unique_keywords