修正查詢

backend/services/enhanced_product_service.py

@@ -227,6 +227,14 @@ class EnhancedProductService:
 
             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}")
@@ -246,11 +254,20 @@ class EnhancedProductService:
                     ])
 
                 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"✅ 搜尋條件已應用")
+                    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:
@@ -259,8 +276,19 @@ class EnhancedProductService:
             # 優先顯示有庫存的商品
             query = query.order_by(Product.stock.desc())
 
-            products = query.options(joinedload(Product.category)).limit(limit).all()
-            logger.info(f"📦 查詢到 {len(products)} 個商品")
+            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:
@@ -303,7 +331,10 @@ class EnhancedProductService:
             )
         finally:
             if db:
+                logger.info(f"🔒 關閉資料庫連接")
                 close_database_session(db)
+            else:
+                logger.warning(f"⚠️ 沒有資料庫連接需要關閉")
     
     def _get_stock_status(self, stock: int) -> str:
         """獲取庫存狀態"""

debug_sql_query.py

@@ -0,0 +1,195 @@
+"""
+調試 SQL 查詢 - 檢查實際執行的 SQL
+"""
+
+def debug_sql_query():
+    """調試 SQL 查詢邏輯"""
+    
+    print("🔍 調試 SQL 查詢邏輯")
+    print("=" * 50)
+    
+    # 模擬關鍵字
+    keywords = ['貓砂', '礦砂', '豆腐砂', '水晶砂', '木屑砂', 'litter', '貓']
+    
+    print(f"關鍵字: {keywords}")
+    
+    # 生成 SQL 查詢條件
+    search_conditions = []
+    
+    for keyword in keywords:
+        search_conditions.extend([
+            f"productName ILIKE '%{keyword}%'",
+            f"productCode ILIKE '%{keyword}%'",
+            f"barcode ILIKE '%{keyword}%'"
+        ])
+    
+    print(f"\n生成的搜尋條件數量: {len(search_conditions)}")
+    print("前10個條件:")
+    for i, condition in enumerate(search_conditions[:10], 1):
+        print(f"  {i}. {condition}")
+    
+    # 組合完整的 SQL
+    where_clause = " OR ".join(search_conditions)
+    full_sql = f"""
+SELECT id, productCode, productName, stock, warehouse, category_id
+FROM products 
+WHERE is_deleted = false 
+AND ({where_clause})
+ORDER BY stock DESC
+LIMIT 5;
+"""
+    
+    print(f"\n完整 SQL 查詢:")
+    print(full_sql)
+    
+    # 測試與實際商品的匹配
+    print(f"\n🛍️ 測試與實際商品的匹配")
+    print("=" * 50)
+    
+    # 實際商品資料
+    products = [
+        {
+            "id": 1,
+            "productCode": "OL1100-1",
+            "productName": "毆力天然犬種300g 室內成犬無榖小顆粒",
+            "stock": 100
+        },
+        {
+            "id": 2,
+            "productCode": "SW-06-01",
+            "productName": "Shovel well豪好鏟 破碎型礦砂",
+            "stock": 50
+        },
+        {
+            "id": 3,
+            "productCode": "TL-03",
+            "productName": "美國極冠貓砂 薰衣草12kg",
+            "stock": 48
+        },
+        {
+            "id": 4,
+            "productCode": "SL11002",
+            "productName": "首領汪 膠原鴨舌 5入彭湃包",
+            "stock": 100
+        }
+    ]
+    
+    matched_products = []
+    
+    for product in products:
+        product_name = product["productName"].lower()
+        product_code = product["productCode"].lower()
+        
+        # 檢查是否匹配任一關鍵字
+        for keyword in keywords:
+            keyword_lower = keyword.lower()
+            
+            if (keyword_lower in product_name or 
+                keyword_lower in product_code):
+                matched_products.append({
+                    "product": product,
+                    "matched_keyword": keyword
+                })
+                print(f"✅ 匹配: {product['productName']} (關鍵字: '{keyword}')")
+                break
+    
+    print(f"\n總共匹配 {len(matched_products)} 個商品")
+    
+    if not matched_products:
+        print("❌ 沒有匹配的商品")
+        
+        # 逐個測試關鍵字
+        print("\n🔍 逐個測試關鍵字:")
+        for keyword in keywords:
+            print(f"\n關鍵字: '{keyword}'")
+            found = False
+            for product in products:
+                if keyword.lower() in product["productName"].lower():
+                    print(f"  ✅ 匹配商品: {product['productName']}")
+                    found = True
+            if not found:
+                print(f"  ❌ 無匹配商品")
+
+def debug_ilike_logic():
+    """調試 ILIKE 邏輯"""
+    
+    print(f"\n🔍 調試 ILIKE 邏輯")
+    print("=" * 50)
+    
+    # 測試字串
+    test_strings = [
+        "美國極冠貓砂 薰衣草12kg",
+        "Shovel well豪好鏟 破碎型礦砂",
+        "毆力天然犬種300g 室內成犬無榖小顆粒",
+        "首領汪 膠原鴨舌 5入彭湃包"
+    ]
+    
+    # 測試關鍵字
+    test_keywords = ["貓砂", "礦砂", "貓", "狗", "犬"]
+    
+    print("ILIKE 匹配測試:")
+    for keyword in test_keywords:
+        print(f"\n關鍵字: '{keyword}'")
+        for string in test_strings:
+            # 模擬 ILIKE '%keyword%' 邏輯
+            if keyword.lower() in string.lower():
+                print(f"  ✅ '{string}' 包含 '{keyword}'")
+            else:
+                print(f"  ❌ '{string}' 不包含 '{keyword}'")
+
+def debug_or_logic():
+    """調試 OR 邏輯"""
+    
+    print(f"\n🔍 調試 OR 邏輯")
+    print("=" * 50)
+    
+    # 模擬 SQLAlchemy OR 查詢
+    products = [
+        {"productName": "美國極冠貓砂 薰衣草12kg", "productCode": "TL-03"},
+        {"productName": "Shovel well豪好鏟 破碎型礦砂", "productCode": "SW-06-01"},
+        {"productName": "毆力天然犬種300g 室內成犬無榖小顆粒", "productCode": "OL1100-1"},
+        {"productName": "首領汪 膠原鴨舌 5入彭湃包", "productCode": "SL11002"}
+    ]
+    
+    keywords = ['貓砂', '礦砂']
+    
+    print(f"使用關鍵字: {keywords}")
+    print("OR 邏輯測試:")
+    
+    for product in products:
+        conditions_met = []
+        
+        for keyword in keywords:
+            # 檢查 productName
+            if keyword.lower() in product["productName"].lower():
+                conditions_met.append(f"productName contains '{keyword}'")
+            
+            # 檢查 productCode  
+            if keyword.lower() in product["productCode"].lower():
+                conditions_met.append(f"productCode contains '{keyword}'")
+        
+        if conditions_met:
+            print(f"  ✅ {product['productName']}")
+            print(f"     滿足條件: {', '.join(conditions_met)}")
+        else:
+            print(f"  ❌ {product['productName']} (無匹配)")
+
+def main():
+    """主函數"""
+    print("🚀 開始 SQL 查詢調試")
+    print("=" * 60)
+    
+    debug_sql_query()
+    debug_ilike_logic()
+    debug_or_logic()
+    
+    print("\n" + "=" * 60)
+    print("✅ SQL 調試完成！")
+    print("\n💡 如果邏輯測試都正確，問題可能在於:")
+    print("1. 資料庫連接問題")
+    print("2. SQLAlchemy 查詢執行問題") 
+    print("3. 事務或會話問題")
+    print("4. 權限問題")
+
+if __name__ == "__main__":
+    main()

test_database_connection.py

@@ -0,0 +1,211 @@
+"""
+測試資料庫連接和商品查詢
+"""
+
+import sys
+import os
+
+# 添加專案根目錄到 Python 路徑
+sys.path.append(os.path.dirname(os.path.abspath(__file__)))
+
+def test_database_connection():
+    """測試資料庫連接"""
+    try:
+        from backend.database.connection import get_database_session, close_database_session
+        from backend.database.models import Product, Category
+        from sqlalchemy.orm import joinedload
+        from sqlalchemy import text
+        
+        print("🔗 測試資料庫連接...")
+        db = get_database_session()
+        
+        if not db:
+            print("❌ 無法獲取資料庫連接")
+            return False
+        
+        print("✅ 資料庫連接成功")
+        
+        # 1. 測試原始 SQL 查詢
+        print("\n1. 測試原始 SQL 查詢...")
+        try:
+            result = db.execute(text("SELECT COUNT(*) as count FROM products WHERE is_deleted = false"))
+            count = result.fetchone()[0]
+            print(f"   商品總數 (SQL): {count}")
+        except Exception as e:
+            print(f"   ❌ SQL 查詢失敗: {str(e)}")
+            return False
+        
+        # 2. 測試 ORM 查詢
+        print("\n2. 測試 ORM 查詢...")
+        try:
+            products = db.query(Product).filter(Product.is_deleted == False).all()
+            print(f"   商品總數 (ORM): {len(products)}")
+            
+            if products:
+                print("   前3個商品:")
+                for i, product in enumerate(products[:3], 1):
+                    print(f"     {i}. {product.productName} (編號: {product.productCode}, 庫存: {product.stock})")
+            
+        except Exception as e:
+            print(f"   ❌ ORM 查詢失敗: {str(e)}")
+            import traceback
+            traceback.print_exc()
+            return False
+        
+        # 3. 測試關聯查詢
+        print("\n3. 測試關聯查詢...")
+        try:
+            products_with_category = db.query(Product).options(
+                joinedload(Product.category)
+            ).filter(Product.is_deleted == False).limit(3).all()
+            
+            print(f"   關聯查詢成功，取得 {len(products_with_category)} 個商品")
+            for product in products_with_category:
+                category_name = product.category.name if product.category else "無分類"
+                print(f"     - {product.productName} (分類: {category_name})")
+                
+        except Exception as e:
+            print(f"   ❌ 關聯查詢失敗: {str(e)}")
+            import traceback
+            traceback.print_exc()
+            return False
+        
+        # 4. 測試貓砂相關商品查詢
+        print("\n4. 測試貓砂相關商品查詢...")
+        try:
+            from sqlalchemy import or_
+            
+            keywords = ['貓砂', '礦砂', '貓']
+            search_filters = []
+            
+            for keyword in keywords:
+                search_filters.extend([
+                    Product.productName.ilike(f"%{keyword}%"),
+                    Product.productCode.ilike(f"%{keyword}%")
+                ])
+            
+            cat_products = db.query(Product).filter(
+                Product.is_deleted == False
+            ).filter(
+                or_(*search_filters)
+            ).all()
+            
+            print(f"   找到 {len(cat_products)} 個貓砂相關商品:")
+            for product in cat_products:
+                print(f"     ✅ {product.productName} (庫存: {product.stock})")
+                
+        except Exception as e:
+            print(f"   ❌ 貓砂查詢失敗: {str(e)}")
+            import traceback
+            traceback.print_exc()
+            return False
+        
+        close_database_session(db)
+        return True
+        
+    except Exception as e:
+        print(f"❌ 資料庫測試失敗: {str(e)}")
+        import traceback
+        traceback.print_exc()
+        return False
+
+def test_enhanced_product_service():
+    """測試 Enhanced Product Service"""
+    try:
+        from backend.services.enhanced_product_service import EnhancedProductService
+        
+        print("\n🛍️ 測試 Enhanced Product Service...")
+        
+        service = EnhancedProductService()
+        
+        # 測試推薦查詢
+        print("\n測試推薦查詢: '貓砂還有嗎?'")
+        result = service.get_product_recommendations(
+            query_text="貓砂還有嗎?",
+            limit=5
+        )
+        
+        print(f"查詢結果:")
+        print(f"  成功: {result.success}")
+        print(f"  數量: {result.count}")
+        print(f"  錯誤: {result.error}")
+        
+        if result.success and result.data:
+            print(f"  找到的商品:")
+            for item in result.data:
+                print(f"    - {item.get('product_name')} (庫存: {item.get('current_stock')})")
+        
+        return result.success
+        
+    except Exception as e:
+        print(f"��� Enhanced Product Service 測試失敗: {str(e)}")
+        import traceback
+        traceback.print_exc()
+        return False
+
+def test_pydantic_ai_service():
+    """測試 Pydantic AI Service"""
+    try:
+        from backend.services.pydantic_ai_service import ProductQueryService
+        
+        print("\n🤖 測試 Pydantic AI Service...")
+        
+        service = ProductQueryService()
+        
+        if not service.is_available():
+            print("⚠️ Pydantic AI 服務不可用 (可能缺少 GROQ_API_KEY)")
+            return False
+        
+        # 測試同步查詢
+        print("\n測試同步查詢: '貓砂還有嗎?'")
+        result = service.process_product_query_sync(
+            user_message="貓砂還有嗎?",
+            user_id="test_user"
+        )
+        
+        print(f"查詢結果:")
+        print(f"  成功: {result.get('success')}")
+        print(f"  模式: {result.get('mode')}")
+        print(f"  錯誤: {result.get('error')}")
+        print(f"  回應: {result.get('text', '')[:100]}...")
+        
+        return result.get('success', False)
+        
+    except Exception as e:
+        print(f"❌ Pydantic AI Service 測試失敗: {str(e)}")
+        import traceback
+        traceback.print_exc()
+        return False
+
+def main():
+    """主函數"""
+    print("🚀 開始資料庫連接和服務測試")
+    print("=" * 60)
+    
+    # 1. 測試資料庫連接
+    db_success = test_database_connection()
+    
+    if not db_success:
+        print("\n❌ 資料庫連接測試失敗，停止後續測試")
+        return
+    
+    # 2. 測試 Enhanced Product Service
+    service_success = test_enhanced_product_service()
+    
+    # 3. 測試 Pydantic AI Service
+    ai_success = test_pydantic_ai_service()
+    
+    print("\n" + "=" * 60)
+    print("📊 測試結果總結:")
+    print(f"  資料庫連接: {'✅ 成功' if db_success else '❌ 失敗'}")
+    print(f"  Enhanced Product Service: {'✅ 成功' if service_success else '❌ 失敗'}")
+    print(f"  Pydantic AI Service: {'✅ 成功' if ai_success else '❌ 失敗'}")
+    
+    if db_success and service_success:
+        print("\n💡 如果服務測試成功但 LineBot 仍然沒有回應，問題可能在於:")
+        print("  1. LineBot 路由邏輯")
+        print("  2. 回應格式化")
+        print("  3. LINE API 回應")
+
+if __name__ == "__main__":
+    main()