File size: 7,354 Bytes
85ef28f |
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 |
"""
測試資料庫連接和商品查詢
"""
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()
|