""" 增強的商品查詢服務 - 移植自 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: db = get_database_session() # 建立基本查詢，預載入分類資訊 query = db.query(Product).options( joinedload(Product.category) ).filter(Product.is_deleted == False) # 關鍵字搜尋 - 支援商品名稱、編號、條碼 if query_text: search_terms = query_text.strip().split() for term in search_terms: search_filter = or_( Product.productName.ilike(f"%{term}%"), Product.productCode.ilike(f"%{term}%"), Product.barcode.ilike(f"%{term}%") ) query = query.filter(search_filter) # 分類篩選 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: db = get_database_session() # 分析查詢關鍵字 keywords = self._extract_keywords(query_text) # 建立推薦查詢 query = db.query(Product).filter(Product.is_deleted == False) # 多關鍵字匹配 - 使用 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}%") ]) # 使用 OR 連接所有搜尋條件 if search_filters: query = query.filter(or_(*search_filters)) # 優先顯示有庫存的商品 query = query.order_by(Product.stock.desc()) products = query.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, "recommendation_reason": f"符合關鍵字: {', '.join(keywords)}", "availability": "有庫存" if product.stock > 0 else "缺貨" }) 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_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 = ['推薦', '有沒有', '是否有', '請問', '想要', '需要', '找', '查詢', '搜尋'] # 分割並清理關鍵字 words = query_text.replace('?', '').replace('？', '').split() keywords = [word for word in words if word not in stop_words and len(word) > 1] # 擴展相關關鍵字 expanded_keywords = [] for keyword in 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']) return expanded_keywords if expanded_keywords else [query_text.strip()]