mice/engine/score_api_client.py

#!/usr/bin/env python3
"""
Score API Client for Mice Game
Client module to integrate with the FastAPI score server
"""

import json
from typing import Optional, List, Dict, Any
import time

# Try to import requests; if unavailable, provide a minimal urllib-based fallback.
try:
    import requests  # type: ignore
    _HAS_REQUESTS = True
except Exception:
    requests = None
    _HAS_REQUESTS = False
from typing import Optional, List, Dict, Any
import time


class ScoreAPIClient:
    """Client for communicating with the Mice Game Score API"""
    
    def __init__(self, api_base_url: str = "http://localhost:8000", timeout: int = 5):
        """
        Initialize the API client
        
        Args:
            api_base_url: Base URL of the API server
            timeout: Request timeout in seconds
        """
        self.api_base_url = api_base_url.rstrip('/')
        self.timeout = timeout
    
    def _make_request(self, method: str, endpoint: str, data: Optional[Dict] = None) -> Optional[Dict[str, Any]]:
        """
        Make HTTP request to API
        
        Args:
            method: HTTP method (GET, POST, etc.)
            endpoint: API endpoint
            data: Request data for POST requests
        
        Returns:
            Response JSON or None if error
        """
        url = f"{self.api_base_url}{endpoint}"
        
        try:
            if _HAS_REQUESTS:
                if method.upper() == "GET":
                    response = requests.get(url, timeout=self.timeout)
                elif method.upper() == "POST":
                    response = requests.post(url, json=data, timeout=self.timeout)
                else:
                    raise ValueError(f"Unsupported HTTP method: {method}")

                if response.status_code == 200:
                    return response.json()
                elif response.status_code in [400, 404, 409]:
                    return {"error": True, "status": response.status_code, "detail": response.json()}
                else:
                    return {"error": True, "status": response.status_code, "detail": "Server error"}
            else:
                # urllib fallback for environments without requests (e.g., Pyodide without wheel)
                from urllib.request import Request, urlopen
                from urllib.error import URLError, HTTPError
                import urllib.parse

                if method.upper() == 'GET':
                    req = Request(url, method='GET')
                    try:
                        with urlopen(req, timeout=self.timeout) as resp:
                            body = resp.read()
                            try:
                                return json.loads(body.decode('utf-8'))
                            except Exception:
                                return None
                    except HTTPError as he:
                        try:
                            detail = json.loads(he.read().decode('utf-8'))
                        except Exception:
                            detail = str(he)
                        return {"error": True, "status": he.code, "detail": detail}
                    except URLError:
                        return {"error": True, "detail": "Could not connect to score server"}
                elif method.upper() == 'POST':
                    data_bytes = json.dumps(data).encode('utf-8') if data is not None else None
                    req = Request(url, data=data_bytes, method='POST')
                    req.add_header('Content-Type', 'application/json')
                    try:
                        with urlopen(req, timeout=self.timeout) as resp:
                            body = resp.read()
                            try:
                                return json.loads(body.decode('utf-8'))
                            except Exception:
                                return None
                    except HTTPError as he:
                        try:
                            detail = json.loads(he.read().decode('utf-8'))
                        except Exception:
                            detail = str(he)
                        return {"error": True, "status": he.code, "detail": detail}
                    except URLError:
                        return {"error": True, "detail": "Could not connect to score server"}
                else:
                    raise ValueError(f"Unsupported HTTP method: {method}")

        except Exception as e:
            return {"error": True, "detail": str(e)}
    
    def signup_user(self, device_id: str, user_id: str) -> Dict[str, Any]:
        """
        Register a new user
        
        Args:
            device_id: Device identifier
            user_id: User identifier
        
        Returns:
            Response dictionary with success/error status
        """
        endpoint = f"/signup/{device_id}/{user_id}"
        response = self._make_request("POST", endpoint)
        
        if response is None:
            return {"success": False, "message": "Failed to connect to server"}
        
        if response.get("error"):
            return {"success": False, "message": response.get("detail", "Unknown error")}
        
        return response
    
    def submit_score(self, device_id: str, user_id: str, score: int, game_completed: bool = True) -> Dict[str, Any]:
        """
        Submit a score for a user
        
        Args:
            device_id: Device identifier
            user_id: User identifier
            score: Game score
            game_completed: Whether the game was completed
        
        Returns:
            Response dictionary with success/error status
        """
        endpoint = f"/score/{device_id}/{user_id}"
        data = {
            "user_id": user_id,
            "device_id": device_id,
            "score": score,
            "game_completed": game_completed
        }
        
        response = self._make_request("POST", endpoint, data)
        
        if response is None:
            return {"success": False, "message": "Failed to connect to server"}
        
        if response.get("error"):
            return {"success": False, "message": response.get("detail", "Unknown error")}
        
        return response
    
    def get_device_users(self, device_id: str) -> List[Dict[str, Any]]:
        """
        Get all users registered for a device
        
        Args:
            device_id: Device identifier
        
        Returns:
            List of user dictionaries
        """
        endpoint = f"/users/{device_id}"
        response = self._make_request("GET", endpoint)
        
        if response is None:
            return []
        
        # Check if it's an error response (dict with error field) or success (list)
        if isinstance(response, dict) and response.get("error"):
            return []
        
        # If it's a list (successful response), return it
        if isinstance(response, list):
            return response
        
        return []
    
    def get_user_scores(self, device_id: str, user_id: str, limit: int = 10) -> List[Dict[str, Any]]:
        """
        Get recent scores for a user
        
        Args:
            device_id: Device identifier
            user_id: User identifier
            limit: Maximum number of scores to return
        
        Returns:
            List of score dictionaries
        """
        endpoint = f"/scores/{device_id}/{user_id}?limit={limit}"
        response = self._make_request("GET", endpoint)
        
        if response is None:
            return []
        
        # Check if it's an error response (dict with error field) or success (list)
        if isinstance(response, dict) and response.get("error"):
            return []
        
        # If it's a list (successful response), return it
        if isinstance(response, list):
            return response
        
        return []
    
    def get_leaderboard(self, device_id: str, limit: int = 10) -> List[Dict[str, Any]]:
        """
        Get leaderboard for a device
        
        Args:
            device_id: Device identifier
            limit: Maximum number of entries to return
        
        Returns:
            List of leaderboard entries
        """
        endpoint = f"/leaderboard/{device_id}?limit={limit}"
        response = self._make_request("GET", endpoint)
        
        if response is None:
            return []
        
        # Check if it's an error response (dict with error field) or success (list)
        if isinstance(response, dict) and response.get("error"):
            return []
        
        # If it's a list (successful response), return it
        if isinstance(response, list):
            return response
        
        return []
    
    def get_global_leaderboard(self, limit: int = 10) -> List[Dict[str, Any]]:
        """
        Get global leaderboard across all devices
        
        Args:
            limit: Maximum number of entries to return
        
        Returns:
            List of global leaderboard entries
        """
        endpoint = f"/leaderboard/global/top?limit={limit}"
        response = self._make_request("GET", endpoint)
        
        if response is None:
            return []
        
        # Check if it's an error response (dict with error field) or success (list)
        if isinstance(response, dict) and response.get("error"):
            return []
        
        # If it's a list (successful response), return it
        if isinstance(response, list):
            return response
        
        return []
    
    def is_server_available(self) -> bool:
        """
        Check if the API server is available
        
        Returns:
            True if server is reachable, False otherwise
        """
        response = self._make_request("GET", "/")
        return response is not None and not response.get("error")
    
    def user_exists(self, device_id: str, user_id: str) -> bool:
        """
        Check if a user is registered for a device
        
        Args:
            device_id: Device identifier
            user_id: User identifier
        
        Returns:
            True if user exists, False otherwise
        """
        users = self.get_device_users(device_id)
        return any(user["user_id"] == user_id for user in users)


# Convenience functions for easy integration
def create_api_client(api_url: str = "http://localhost:8000") -> ScoreAPIClient:
    """Create and return an API client instance"""
    return ScoreAPIClient(api_url)

def test_connection(api_url: str = "http://localhost:8000") -> bool:
    """Test if the API server is available"""
    client = ScoreAPIClient(api_url)
    return client.is_server_available()


# Example usage and testing
if __name__ == "__main__":
    # Example usage
    print("Testing Score API Client...")
    
    # Create client
    client = ScoreAPIClient()
    
    # Test server connection
    if not client.is_server_available():
        print("ERROR: API server is not available. Start it with: python score_api.py")
        exit(1)
    
    print("API server is available!")
    
    # Example device and user
    device_id = "DEV-CLIENT01"
    user_id = "ClientTestUser"
    
    # Test user signup
    print(f"\nTesting user signup: {user_id}")
    result = client.signup_user(device_id, user_id)
    print(f"Signup result: {result}")
    
    # Test score submission
    print(f"\nTesting score submission...")
    result = client.submit_score(device_id, user_id, 1750, True)
    print(f"Score submission result: {result}")
    
    # Test getting users
    print(f"\nGetting users for device {device_id}:")
    users = client.get_device_users(device_id)
    for user in users:
        print(f"  User: {user['user_id']}, Best Score: {user['best_score']}")
    
    # Test getting user scores
    print(f"\nGetting scores for {user_id}:")
    scores = client.get_user_scores(device_id, user_id)
    for score in scores:
        print(f"  Score: {score['score']}, Time: {score['timestamp']}")
    
    # Test leaderboard
    print(f"\nLeaderboard for device {device_id}:")
    leaderboard = client.get_leaderboard(device_id)
    for entry in leaderboard:
        print(f"  Rank {entry['rank']}: {entry['user_id']} - {entry['best_score']} pts")
    
    print("\nClient testing completed!")