Upload backend/scripts/test_api_endpoint.py with huggingface_hub

backend/scripts/test_api_endpoint.py

@@ -0,0 +1,147 @@
+#!/usr/bin/env python
+"""
+Script để test chatbot API endpoint.
+"""
+import os
+import requests
+import json
+import time
+from typing import Dict, Any
+
+
+API_BASE_URL = os.environ.get("API_BASE_URL", "http://localhost:8000").rstrip("/")
+
+
+def test_health_endpoint():
+    """Test health endpoint."""
+    print("="*60)
+    print("Test Health Endpoint")
+    print("="*60)
+    
+    try:
+        response = requests.get(f"{API_BASE_URL}/api/chatbot/health/", timeout=5)
+        print(f"Status Code: {response.status_code}")
+        if response.status_code == 200:
+            data = response.json()
+            print(f"Status: {data.get('status', 'N/A')}")
+            print(f"Service: {data.get('service', 'N/A')}")
+            print(f"Classifier Loaded: {data.get('classifier_loaded', False)}")
+            return True
+        else:
+            print(f"Error: {response.text}")
+            return False
+    except requests.exceptions.ConnectionError:
+        print("❌ Cannot connect to server. Is Django server running?")
+        print("   Start server with: cd backend/hue_portal && POSTGRES_HOST=localhost POSTGRES_PORT=5433 python manage.py runserver")
+        return False
+    except Exception as e:
+        print(f"❌ Error: {e}")
+        return False
+
+
+def test_chatbot_api(query: str, expected_intent: str = None) -> Dict[str, Any]:
+    """Test chatbot API with a query."""
+    print(f"\n📝 Query: {query}")
+    
+    start_time = time.time()
+    try:
+        response = requests.post(
+            f"{API_BASE_URL}/api/chat/",
+            json={"message": query},
+            headers={"Content-Type": "application/json"},
+            timeout=30
+        )
+        latency_ms = (time.time() - start_time) * 1000
+        
+        print(f"   Status Code: {response.status_code}")
+        
+        if response.status_code == 200:
+            data = response.json()
+            intent = data.get('intent', 'N/A')
+            confidence = data.get('confidence', 0)
+            count = data.get('count', 0)
+            message_preview = data.get('message', '')[:100]
+            
+            print(f"   ✅ Intent: {intent}")
+            print(f"   ✅ Confidence: {confidence:.4f}")
+            print(f"   ✅ Results: {count}")
+            print(f"   ✅ Latency: {latency_ms:.2f}ms")
+            print(f"   ✅ Message preview: {message_preview}...")
+            
+            if expected_intent and intent != expected_intent:
+                print(f"   ⚠️ Expected intent: {expected_intent}, got: {intent}")
+            
+            return {
+                "success": True,
+                "intent": intent,
+                "confidence": confidence,
+                "count": count,
+                "latency_ms": latency_ms
+            }
+        else:
+            print(f"   ❌ Error: {response.text}")
+            return {"success": False, "error": response.text}
+            
+    except requests.exceptions.ConnectionError:
+        print("   ❌ Cannot connect to server")
+        return {"success": False, "error": "Connection error"}
+    except Exception as e:
+        print(f"   ❌ Error: {e}")
+        return {"success": False, "error": str(e)}
+
+
+def main():
+    print("="*60)
+    print("Chatbot API Endpoint Test")
+    print("="*60)
+    
+    # Test health endpoint first
+    if not test_health_endpoint():
+        print("\n⚠️ Health check failed. Please start Django server first.")
+        return
+    
+    # Test chatbot API with various queries
+    print("\n" + "="*60)
+    print("Test Chatbot API Endpoint")
+    print("="*60)
+    
+    test_cases = [
+        ("Làm thủ tục cư trú cần gì?", "search_procedure"),
+        ("Cảnh báo lừa đảo giả danh công an", "search_advisory"),
+        ("Thủ tục PCCC như thế nào?", "search_procedure"),
+        ("Mức phạt vượt đèn đỏ", "search_fine"),
+        ("Địa chỉ công an tỉnh", "search_office"),
+        ("Lừa đảo mạo danh cán bộ", "search_advisory"),
+    ]
+    
+    results = []
+    for query, expected_intent in test_cases:
+        result = test_chatbot_api(query, expected_intent)
+        results.append(result)
+        time.sleep(0.5)  # Small delay between requests
+    
+    # Summary
+    print("\n" + "="*60)
+    print("Test Summary")
+    print("="*60)
+    
+    successful = sum(1 for r in results if r.get("success", False))
+    total = len(results)
+    avg_latency = sum(r.get("latency_ms", 0) for r in results if r.get("success", False)) / successful if successful > 0 else 0
+    
+    print(f"Successful: {successful}/{total}")
+    print(f"Average Latency: {avg_latency:.2f}ms")
+    
+    # Intent accuracy
+    correct_intents = sum(1 for i, (_, expected) in enumerate(test_cases) 
+                         if results[i].get("intent") == expected)
+    print(f"Intent Accuracy: {correct_intents}/{total} ({correct_intents/total*100:.1f}%)")
+    
+    print("\n" + "="*60)
+    print("Test Complete")
+    print("="*60)
+
+
+if __name__ == "__main__":
+    main()
+