fixed some module import issue

backend/services/business_query_service.py

@@ -4,7 +4,7 @@
 """
 
 import logging
-from typing import Dict, Any, Optional
+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

tmp_rovodev_test_groq_integration.py

@@ -1,340 +0,0 @@
-#!/usr/bin/env python3
-"""
-Groq 整合測試腳本
-測試新的路由系統和 Groq 服務
-"""
-
-import requests
-import json
-import os
-from typing import Dict, Any
-
-class GroqIntegrationTester:
-    def __init__(self, base_url: str = "http://localhost:7860"):
-        self.base_url = base_url
-        self.test_results = []
-    
-    def test_health_check(self):
-        """測試健康檢查"""
-        print("🔍 測試健康檢查...")
-        try:
-            response = requests.get(f"{self.base_url}/health")
-            success = response.status_code == 200
-            result = {
-                "test": "health_check",
-                "success": success,
-                "status_code": response.status_code,
-                "response": response.json() if success else response.text
-            }
-            self.test_results.append(result)
-            print(f"✅ 健康檢查: {response.status_code}")
-            return success
-        except Exception as e:
-            print(f"❌ 健康檢查失敗: {e}")
-            self.test_results.append({
-                "test": "health_check",
-                "success": False,
-                "error": str(e)
-            })
-            return False
-    
-    def test_chat_api(self):
-        """測試聊天 API"""
-        print("\n💬 測試聊天 API...")
-        test_messages = [
-            "你好！",
-            "今天天氣如何？",
-            "請介紹一下你自己",
-            "推薦一些好用的功能"
-        ]
-        
-        for message in test_messages:
-            try:
-                response = requests.post(
-                    f"{self.base_url}/chat",
-                    json={"message": message, "user_id": "test_user"},
-                    headers={"Content-Type": "application/json"}
-                )
-                
-                success = response.status_code == 200
-                result = {
-                    "test": "chat_api",
-                    "message": message,
-                    "success": success,
-                    "status_code": response.status_code,
-                    "response": response.json() if success else response.text
-                }
-                self.test_results.append(result)
-                
-                if success:
-                    data = response.json()
-                    print(f"✅ 聊天: {message[:20]}... -> {data.get('text', '')[:50]}...")
-                else:
-                    print(f"❌ 聊天失敗: {message[:20]}... -> {response.status_code}")
-                    
-            except Exception as e:
-                print(f"❌ 聊天 API 錯誤: {e}")
-                self.test_results.append({
-                    "test": "chat_api",
-                    "message": message,
-                    "success": False,
-                    "error": str(e)
-                })
-    
-    def test_search_api(self):
-        """測試搜尋 API"""
-        print("\n🔍 測試搜尋 API...")
-        test_queries = [
-            "iPhone 15 Pro",
-            "價格 1000-5000",
-            "庫存查詢",
-            "我的訂單",
-            "低庫存商品"
-        ]
-        
-        for query in test_queries:
-            try:
-                response = requests.post(
-                    f"{self.base_url}/search",
-                    json={"message": query, "user_id": "test_user"},
-                    headers={"Content-Type": "application/json"}
-                )
-                
-                success = response.status_code == 200
-                result = {
-                    "test": "search_api",
-                    "query": query,
-                    "success": success,
-                    "status_code": response.status_code,
-                    "response": response.json() if success else response.text
-                }
-                self.test_results.append(result)
-                
-                if success:
-                    data = response.json()
-                    print(f"✅ 搜尋: {query} -> {data.get('text', '')[:50]}...")
-                else:
-                    print(f"❌ 搜尋失敗: {query} -> {response.status_code}")
-                    
-            except Exception as e:
-                print(f"❌ 搜尋 API 錯誤: {e}")
-                self.test_results.append({
-                    "test": "search_api",
-                    "query": query,
-                    "success": False,
-                    "error": str(e)
-                })
-    
-    def test_smart_routing(self):
-        """測試智能路由"""
-        print("\n🧠 測試智能路由...")
-        test_messages = [
-            # 應該路由到聊天
-            "你好，今天天氣如何？",
-            "請介紹一下你的功能",
-            
-            # 應該路由到搜尋
-            "iPhone 還有庫存嗎？",
-            "查詢價格在 2000 到 5000 的商品",
-            "我想看看訂單狀態",
-            
-            # 前綴指令
-            "/chat 你好！",
-            "/search iPhone 15",
-            "/help"
-        ]
-        
-        for message in test_messages:
-            try:
-                response = requests.post(
-                    f"{self.base_url}/route",
-                    json={"message": message, "user_id": "test_user"},
-                    headers={"Content-Type": "application/json"}
-                )
-                
-                success = response.status_code == 200
-                result = {
-                    "test": "smart_routing",
-                    "message": message,
-                    "success": success,
-                    "status_code": response.status_code,
-                    "response": response.json() if success else response.text
-                }
-                self.test_results.append(result)
-                
-                if success:
-                    data = response.json()
-                    mode = data.get('mode', 'unknown')
-                    print(f"✅ 路由: {message[:30]}... -> {mode}")
-                else:
-                    print(f"❌ 路由失敗: {message[:30]}... -> {response.status_code}")
-                    
-            except Exception as e:
-                print(f"❌ 智能路由錯誤: {e}")
-                self.test_results.append({
-                    "test": "smart_routing",
-                    "message": message,
-                    "success": False,
-                    "error": str(e)
-                })
-    
-    def test_stats_api(self):
-        """測試統計 API"""
-        print("\n📊 測試統計 API...")
-        try:
-            response = requests.get(f"{self.base_url}/stats")
-            success = response.status_code == 200
-            result = {
-                "test": "stats_api",
-                "success": success,
-                "status_code": response.status_code,
-                "response": response.json() if success else response.text
-            }
-            self.test_results.append(result)
-            
-            if success:
-                data = response.json()
-                stats = data.get('data', {}).get('stats', {})
-                total = data.get('data', {}).get('total', 0)
-                print(f"✅ 統計: 總計 {total} 次路由")
-                for mode, count in stats.items():
-                    print(f"   {mode}: {count}")
-            else:
-                print(f"❌ 統計失敗: {response.status_code}")
-                
-        except Exception as e:
-            print(f"❌ 統計 API 錯誤: {e}")
-            self.test_results.append({
-                "test": "stats_api",
-                "success": False,
-                "error": str(e)
-            })
-    
-    def test_line_webhook_simulation(self):
-        """模擬 LINE Webhook"""
-        print("\n📱 測試 LINE Webhook 模擬...")
-        test_messages = [
-            "/chat 你好！",
-            "/search iPhone",
-            "iPhone 還有庫存嗎？",
-            "幫助"
-        ]
-        
-        for message in test_messages:
-            webhook_data = {
-                "events": [
-                    {
-                        "type": "message",
-                        "message": {
-                            "type": "text",
-                            "text": message
-                        },
-                        "source": {
-                            "type": "user",
-                            "userId": "test_user_webhook"
-                        },
-                        "replyToken": "test_reply_token"
-                    }
-                ]
-            }
-            
-            try:
-                response = requests.post(
-                    f"{self.base_url}/webhook",
-                    json=webhook_data,
-                    headers={"Content-Type": "application/json"}
-                )
-                
-                success = response.status_code == 200
-                result = {
-                    "test": "line_webhook",
-                    "message": message,
-                    "success": success,
-                    "status_code": response.status_code
-                }
-                self.test_results.append(result)
-                
-                print(f"{'✅' if success else '❌'} Webhook: {message} -> {response.status_code}")
-                
-            except Exception as e:
-                print(f"❌ Webhook 錯誤: {e}")
-                self.test_results.append({
-                    "test": "line_webhook",
-                    "message": message,
-                    "success": False,
-                    "error": str(e)
-                })
-    
-    def generate_report(self):
-        """生成測試報告"""
-        print("\n" + "="*50)
-        print("📋 測試報告")
-        print("="*50)
-        
-        total_tests = len(self.test_results)
-        successful_tests = sum(1 for result in self.test_results if result.get('success', False))
-        
-        print(f"總測試數: {total_tests}")
-        print(f"���功: {successful_tests}")
-        print(f"失敗: {total_tests - successful_tests}")
-        print(f"成功率: {successful_tests/total_tests*100:.1f}%")
-        
-        # 按測試類型分組
-        test_types = {}
-        for result in self.test_results:
-            test_type = result['test']
-            if test_type not in test_types:
-                test_types[test_type] = {'total': 0, 'success': 0}
-            test_types[test_type]['total'] += 1
-            if result.get('success', False):
-                test_types[test_type]['success'] += 1
-        
-        print("\n📊 各功能測試結果:")
-        for test_type, stats in test_types.items():
-            success_rate = stats['success'] / stats['total'] * 100
-            print(f"  {test_type}: {stats['success']}/{stats['total']} ({success_rate:.1f}%)")
-        
-        # 顯示失敗的測試
-        failed_tests = [result for result in self.test_results if not result.get('success', False)]
-        if failed_tests:
-            print(f"\n❌ 失敗的測試 ({len(failed_tests)}):")
-            for result in failed_tests:
-                print(f"  - {result['test']}: {result.get('error', 'Unknown error')}")
-        
-        return {
-            'total': total_tests,
-            'successful': successful_tests,
-            'failed': total_tests - successful_tests,
-            'success_rate': successful_tests/total_tests*100,
-            'details': test_types,
-            'failed_tests': failed_tests
-        }
-
-def run_all_tests():
-    """執行所有測試"""
-    print("🚀 開始 Groq 整合測試")
-    print("="*50)
-    
-    tester = GroqIntegrationTester()
-    
-    # 執行所有測試
-    tester.test_health_check()
-    tester.test_chat_api()
-    tester.test_search_api()
-    tester.test_smart_routing()
-    tester.test_stats_api()
-    tester.test_line_webhook_simulation()
-    
-    # 生成報告
-    report = tester.generate_report()
-    
-    return report
-
-if __name__ == "__main__":
-    report = run_all_tests()
-    
-    # 保存詳細結果到文件
-    with open("tmp_rovodev_test_results.json", "w", encoding="utf-8") as f:
-        json.dump(report, f, ensure_ascii=False, indent=2)
-    
-    print(f"\n📄 詳細測試結果已保存到: tmp_rovodev_test_results.json")