From 4f74343efc20a84231c729cd146d280dedcb5483 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Y=C3=BBki=20VACHOT?= Date: Sat, 29 Nov 2025 00:03:52 +0100 Subject: [PATCH] Refactor API module: introduced `DynamicResponse` for schema-aware responses, centralized endpoint definitions in `EndpointInfo` with metadata, and enhanced `GeoGuessrClient` with dynamic discovery and response handling. --- src/geoguessr_mcp/api/__init__.py | 12 + src/geoguessr_mcp/api/client.py | 398 +++++++++++++++---------- src/geoguessr_mcp/api/endpoints.py | 462 ++++++++++++----------------- 3 files changed, 450 insertions(+), 422 deletions(-) diff --git a/src/geoguessr_mcp/api/__init__.py b/src/geoguessr_mcp/api/__init__.py index e69de29..2b835db 100644 --- a/src/geoguessr_mcp/api/__init__.py +++ b/src/geoguessr_mcp/api/__init__.py @@ -0,0 +1,12 @@ +"""API client module for GeoGuessr communication.""" + +from .client import GeoGuessrClient, DynamicResponse +from .endpoints import Endpoints, EndpointInfo, EndpointBuilder + +__all__ = [ + "GeoGuessrClient", + "DynamicResponse", + "Endpoints", + "EndpointInfo", + "EndpointBuilder", +] \ No newline at end of file diff --git a/src/geoguessr_mcp/api/client.py b/src/geoguessr_mcp/api/client.py index f9617ce..4e38c41 100644 --- a/src/geoguessr_mcp/api/client.py +++ b/src/geoguessr_mcp/api/client.py @@ -1,201 +1,285 @@ """ -HTTP client for Geoguessr API communication. +Dynamic HTTP client for GeoGuessr API communication. + +This client automatically handles authentication, endpoint routing, +and integrates with the schema registry for dynamic response handling. """ +import logging +from typing import Any, Optional + import httpx -import logging -from typing import Optional -from ..auth.session import SessionManager -from .endpoints import EndpointBuilder, get_endpoint_info +from ..auth.session import SessionManager from ..config import settings +from ..monitoring.schema_manager import schema_registry +from .endpoints import EndpointInfo logger = logging.getLogger(__name__) -class GeoguessrClient: +class DynamicResponse: """ - Wrapper for Geoguessr API HTTP communication. + Wrapper for API responses with dynamic schema information. - This client automatically handles: - - Authentication via session manager - - Endpoint routing (main API vs. game server) - - Error handling and retries - - Logging and debugging + This class provides methods to access response data with awareness + of the current schema, making it easier for the LLM to understand + and process the data. + """ + + def __init__( + self, + data: Any, + endpoint: str, + status_code: int, + response_time_ms: float, + ): + self.data = data + self.endpoint = endpoint + self.status_code = status_code + self.response_time_ms = response_time_ms + self._schema = schema_registry.get_schema(endpoint) + + @property + def is_success(self) -> bool: + """Check if the request was successful.""" + return 200 <= self.status_code < 300 + + @property + def schema_description(self) -> str: + """Get a human-readable description of the response schema.""" + return schema_registry.generate_dynamic_description(self.endpoint) + + @property + def available_fields(self) -> list[str]: + """Get list of available fields in this response.""" + if self._schema: + return list(self._schema.fields.keys()) + if isinstance(self.data, dict): + return list(self.data.keys()) + return [] + + def get_field(self, field_name: str, default: Any = None) -> Any: + """ + Safely get a field from the response data. + + Supports nested field access using dot notation (e.g., "user.profile.name") + """ + if not isinstance(self.data, dict): + return default + + parts = field_name.split(".") + current = self.data + + for part in parts: + if isinstance(current, dict) and part in current: + current = current[part] + else: + return default + + return current + + def to_dict(self) -> dict: + """Convert response to a dictionary with metadata.""" + return { + "success": self.is_success, + "status_code": self.status_code, + "endpoint": self.endpoint, + "response_time_ms": round(self.response_time_ms, 2), + "data": self.data, + "available_fields": self.available_fields, + } + + def summarize(self, max_depth: int = 2) -> dict: + """ + Create a summarized view of the response for LLM context. + + This reduces token usage while providing essential information. + """ + def summarize_value(value: Any, depth: int) -> Any: + if depth <= 0: + if isinstance(value, (dict, list)): + return f"<{type(value).__name__} with {len(value)} items>" + return value + + if isinstance(value, dict): + return { + k: summarize_value(v, depth - 1) + for k, v in list(value.items())[:10] + } + if isinstance(value, list): + if len(value) == 0: + return [] + return [ + summarize_value(value[0], depth - 1), + f"... and {len(value) - 1} more items" if len(value) > 1 else None, + ] + if isinstance(value, str) and len(value) > 100: + return value[:100] + "..." + return value + + return { + "endpoint": self.endpoint, + "status": "success" if self.is_success else "error", + "field_count": len(self.available_fields), + "data_summary": summarize_value(self.data, max_depth), + } + + +class GeoGuessrClient: + """ + Dynamic HTTP client for GeoGuessr API. + + Features: + - Automatic authentication handling + - Dynamic response schema tracking + - Retry logic with exponential backoff + - Integrated monitoring and logging """ def __init__( self, session_manager: SessionManager, - base_url: str = settings.GEOGUESSR_API_URL, - game_server_url: str = settings.GAME_SERVER_URL, - timeout: float = 30.0 + timeout: float = settings.REQUEST_TIMEOUT, ): - """ - Initialize the Geoguessr API client. - - Args: - session_manager: Session manager for authentication - base_url: Base URL for Geoguessr API - game_server_url: URL for game server API - timeout: Request timeout in seconds - """ self.session_manager = session_manager - self.base_url = base_url - self.game_server_url = game_server_url self.timeout = timeout - async def get_authenticated_client( + async def _get_authenticated_client( self, - session_token: Optional[str] = None + session_token: Optional[str] = None, ) -> httpx.AsyncClient: - """ - Get an authenticated async HTTP client. - - Args: - session_token: Optional session token for authentication - - Returns: - Authenticated httpx.AsyncClient - - Raises: - ValueError: If no valid session is available - """ + """Get an authenticated HTTP client.""" session = await self.session_manager.get_session(session_token) if not session: raise ValueError( - "No valid session available. Please:\n" - "1. Use login() to authenticate, or\n" - "2. Set GEOGUESSR_NCFA_COOKIE environment variable" + "No valid session available. Please login first or set GEOGUESSR_NCFA_COOKIE." ) client = httpx.AsyncClient(timeout=self.timeout) - client.cookies.set( - "_ncfa", - session.ncfa_cookie, - domain="www.geoguessr.com" - ) + client.cookies.set("_ncfa", session.ncfa_cookie, domain="www.geoguessr.com") return client - def _get_base_url(self, endpoint: str, use_game_server: Optional[bool] = None) -> str: - """ - Determine the correct base URL for an endpoint. - - Args: - endpoint: API endpoint - use_game_server: Explicitly set game server usage, or auto-detect - - Returns: - Appropriate base URL - """ - if use_game_server is None: - # Auto-detect based on endpoint - use_game_server = EndpointBuilder.is_game_server_endpoint(endpoint) - - return self.game_server_url if use_game_server else self.base_url - - async def get( - self, - endpoint: str, - session_token: Optional[str] = None, - use_game_server: Optional[bool] = None, - params: Optional[dict] = None, - **kwargs - ) -> httpx.Response: - """ - Make a GET request to the Geoguessr API. - - Args: - endpoint: API endpoint (e.g., "/v3/profiles") - session_token: Optional session token - use_game_server: Whether to use game server URL (auto-detected if None) - params: Query parameters - **kwargs: Additional arguments to pass to httpx.get - - Returns: - httpx.Response - - Raises: - httpx.HTTPError: On HTTP errors - """ - base = self._get_base_url(endpoint, use_game_server) - url = f"{base}{endpoint}" - - # Get endpoint metadata for logging - metadata = get_endpoint_info(endpoint) - logger.debug( - f"GET {url} - {metadata.get('description', 'Unknown endpoint')}" + @staticmethod + def _get_base_url(endpoint: EndpointInfo) -> str: + """Get the appropriate base URL for an endpoint.""" + return ( + settings.GAME_SERVER_URL + if endpoint.use_game_server + else settings.GEOGUESSR_API_URL ) - async with await self.get_authenticated_client(session_token) as client: - response = await client.get(url, params=params, **kwargs) - response.raise_for_status() - logger.debug(f"GET {url} - Success ({response.status_code})") - return response - - async def post( - self, - endpoint: str, - session_token: Optional[str] = None, - use_game_server: Optional[bool] = None, - json_data: Optional[dict] = None, - **kwargs - ) -> httpx.Response: - """ - Make a POST request to the Geoguessr API. - - Args: - endpoint: API endpoint - session_token: Optional session token - use_game_server: Whether to use game server URL (auto-detected if None) - json_data: JSON data to send - **kwargs: Additional arguments to pass to httpx.post - - Returns: - httpx.Response - - Raises: - httpx.HTTPError: On HTTP errors - """ - base = self._get_base_url(endpoint, use_game_server) - url = f"{base}{endpoint}" - - metadata = get_endpoint_info(endpoint) - logger.debug( - f"POST {url} - {metadata.get('description', 'Unknown endpoint')}" - ) - - async with await self.get_authenticated_client(session_token) as client: - response = await client.post(url, json=json_data, **kwargs) - response.raise_for_status() - logger.debug(f"POST {url} - Success ({response.status_code})") - return response - async def request( self, - method: str, - endpoint: str, + endpoint: EndpointInfo, session_token: Optional[str] = None, - use_game_server: Optional[bool] = None, - **kwargs - ) -> httpx.Response: + params: Optional[dict] = None, + json_data: Optional[dict] = None, + **kwargs, + ) -> DynamicResponse: """ - Make a generic HTTP request. + Make a request to the GeoGuessr API. Args: - method: HTTP method (GET, POST, etc.) - endpoint: API endpoint + endpoint: Endpoint info object session_token: Optional session token - use_game_server: Whether to use game server URL - **kwargs: Additional arguments to pass to httpx + params: Query parameters + json_data: JSON body for POST requests + **kwargs: Additional arguments for httpx Returns: - httpx.Response + DynamicResponse with data and schema info """ - base = self._get_base_url(endpoint, use_game_server) - url = f"{base}{endpoint}" + url = f"{self._get_base_url(endpoint)}{endpoint.path}" - async with await self.get_authenticated_client(session_token) as client: - response = await client.request(method, url, **kwargs) - response.raise_for_status() - return response + # Build params from endpoint builder if available + if endpoint.params_builder and not params: + params = endpoint.params_builder() + + logger.debug(f"{endpoint.method} {url}") + + import time + start_time = time.time() + + async with await self._get_authenticated_client(session_token) as client: + try: + if endpoint.method == "GET": + response = await client.get(url, params=params, **kwargs) + elif endpoint.method == "POST": + response = await client.post(url, json=json_data, params=params, **kwargs) + else: + response = await client.request( + endpoint.method, url, json=json_data, params=params, **kwargs + ) + + response_time = (time.time() - start_time) * 1000 + + if response.status_code == 200: + try: + data = response.json() + # Update schema registry + schema_registry.update_schema( + endpoint.path, data, response.status_code, endpoint.method + ) + except Exception: + data = response.text + else: + data = {"error": response.text, "status_code": response.status_code} + schema_registry.mark_unavailable( + endpoint.path, f"HTTP {response.status_code}", response.status_code + ) + + return DynamicResponse( + data=data, + endpoint=endpoint.path, + status_code=response.status_code, + response_time_ms=response_time, + ) + + except httpx.TimeoutException: + schema_registry.mark_unavailable(endpoint.path, "Request timeout") + raise + except Exception as e: + schema_registry.mark_unavailable(endpoint.path, str(e)) + raise + + async def get( + self, + endpoint: EndpointInfo, + session_token: Optional[str] = None, + params: Optional[dict] = None, + **kwargs, + ) -> DynamicResponse: + """Make a GET request.""" + return await self.request(endpoint, session_token, params=params, **kwargs) + + async def post( + self, + endpoint: EndpointInfo, + session_token: Optional[str] = None, + json_data: Optional[dict] = None, + **kwargs, + ) -> DynamicResponse: + """Make a POST request.""" + return await self.request(endpoint, session_token, json_data=json_data, **kwargs) + + async def get_raw( + self, + path: str, + session_token: Optional[str] = None, + use_game_server: bool = False, + params: Optional[dict] = None, + ) -> DynamicResponse: + """ + Make a raw GET request to any path. + + Useful for discovering new endpoints or accessing endpoints + not yet defined in the registry. + """ + endpoint = EndpointInfo( + path=path, + method="GET", + use_game_server=use_game_server, + description=f"Raw request to {path}", + ) + return await self.get(endpoint, session_token, params) diff --git a/src/geoguessr_mcp/api/endpoints.py b/src/geoguessr_mcp/api/endpoints.py index 777abb4..21fc747 100644 --- a/src/geoguessr_mcp/api/endpoints.py +++ b/src/geoguessr_mcp/api/endpoints.py @@ -1,368 +1,300 @@ """ -Geoguessr API Endpoints -Centralized endpoint definitions extracted from the Geoguessr API. +GeoGuessr API Endpoints Registry. +Centralized endpoint definitions with metadata for dynamic discovery and routing. """ + +from dataclasses import dataclass +from typing import Callable, Optional + from ..config import settings +@dataclass +class EndpointInfo: + """Metadata about an API endpoint.""" + path: str + method: str = "GET" + description: str = "" + auth_required: bool = True + use_game_server: bool = False + params_builder: Optional[Callable[..., dict]] = None + + class Endpoints: """ - Centralized endpoint registry for Geoguessr API. + Centralized endpoint registry for GeoGuessr API. - Usage: - url = Endpoints.PROFILES.GET_PROFILE - full_url = f"{GEOGUESSR_BASE_URL}{url}" + All endpoints are defined here with their metadata, making it easy to + maintain and extend the API coverage. """ - # ============================================================================ - # AUTHENTICATION ENDPOINTS - # ============================================================================ class AUTH: """Authentication endpoints.""" - SIGNIN = "/v3/accounts/signin" # POST + SIGNIN = EndpointInfo( + path="/v3/accounts/signin", + method="POST", + description="Sign in with email and password", + auth_required=False, + ) - # ============================================================================ - # PROFILE ENDPOINTS - # ============================================================================ class PROFILES: """User profile and stats endpoints.""" - GET_PROFILE = "/v3/profiles" # GET - Get current user profile - GET_STATS = "/v3/profiles/stats" # GET - Get user statistics - GET_EXTENDED_STATS = "/v4/stats/me" # GET - Get extended statistics - GET_ACHIEVEMENTS = "/v3/profiles/achievements" # GET - Get user achievements - GET_USER_MAPS = "/v3/profiles/maps" # GET - Get user's custom maps + GET_PROFILE = EndpointInfo( + path="/v3/profiles", + description="Get current user profile", + ) + GET_STATS = EndpointInfo( + path="/v3/profiles/stats", + description="Get user statistics", + ) + GET_EXTENDED_STATS = EndpointInfo( + path="/v4/stats/me", + description="Get extended statistics", + ) + GET_ACHIEVEMENTS = EndpointInfo( + path="/v3/profiles/achievements", + description="Get user achievements", + ) + GET_USER_MAPS = EndpointInfo( + path="/v3/profiles/maps", + description="Get user's custom maps", + ) @staticmethod - def get_public_profile(user_id: str) -> str: + def get_public_profile(user_id: str) -> EndpointInfo: """Get public profile by user ID.""" - return f"/v3/profiles/{user_id}" + return EndpointInfo( + path=f"/v3/profiles/{user_id}", + description=f"Get public profile for user {user_id}", + ) @staticmethod - def get_user_activities(user_id: str) -> str: + def get_user_activities(user_id: str) -> EndpointInfo: """Get user activities/feed.""" - return f"/v3/users/{user_id}/activities" + return EndpointInfo( + path=f"/v3/users/{user_id}/activities", + description=f"Get activities for user {user_id}", + ) - # ============================================================================ - # GAME ENDPOINTS - # ============================================================================ class GAMES: """Game-related endpoints.""" - GET_UNFINISHED_GAMES = "/v3/social/events/unfinishedgames" # GET + GET_UNFINISHED_GAMES = EndpointInfo( + path="/v3/social/events/unfinishedgames", + description="Get unfinished games", + ) @staticmethod - def get_game_details(game_token: str) -> str: + def get_game_details(game_token: str) -> EndpointInfo: """Get details for a specific game.""" - return f"/v3/games/{game_token}" + return EndpointInfo( + path=f"/v3/games/{game_token}", + description=f"Get game details for {game_token}", + ) @staticmethod - def get_streak_game(game_token: str) -> str: + def get_streak_game(game_token: str) -> EndpointInfo: """Get streak game details.""" - return f"/v3/games/streak/{game_token}" + return EndpointInfo( + path=f"/v3/games/streak/{game_token}", + description=f"Get streak game {game_token}", + ) - # ============================================================================ - # GAME SERVER ENDPOINTS (Different base URL) - # ============================================================================ class GAME_SERVER: - """Game server endpoints (use GAME_SERVER_URL as base).""" - GET_TOURNAMENTS = "/tournaments" # GET + """Game server endpoints (different base URL).""" + GET_TOURNAMENTS = EndpointInfo( + path="/tournaments", + use_game_server=True, + description="Get tournament information", + ) @staticmethod - def get_battle_royale(game_id: str) -> str: + def get_battle_royale(game_id: str) -> EndpointInfo: """Get battle royale game.""" - return f"/battle-royale/{game_id}" + return EndpointInfo( + path=f"/battle-royale/{game_id}", + use_game_server=True, + description=f"Get battle royale game {game_id}", + ) @staticmethod - def get_duel(duel_id: str) -> str: + def get_duel(duel_id: str) -> EndpointInfo: """Get duel details.""" - return f"/duels/{duel_id}" + return EndpointInfo( + path=f"/duels/{duel_id}", + use_game_server=True, + description=f"Get duel {duel_id}", + ) @staticmethod - def get_lobby(game_id: str) -> str: + def get_lobby(game_id: str) -> EndpointInfo: """Get lobby information.""" - return f"/lobby/{game_id}" + return EndpointInfo( + path=f"/lobby/{game_id}", + use_game_server=True, + description=f"Get lobby {game_id}", + ) - # ============================================================================ - # COMPETITIVE/SEASONS ENDPOINTS - # ============================================================================ class COMPETITIVE: """Competitive and season-related endpoints.""" - GET_ACTIVE_SEASON_STATS = "/v4/seasons/active/stats" # GET + GET_ACTIVE_SEASON_STATS = EndpointInfo( + path="/v4/seasons/active/stats", + description="Get active season statistics", + ) @staticmethod - def get_season_game(game_mode: str) -> str: + def get_season_game(game_mode: str) -> EndpointInfo: """Get season game for specific mode.""" - return f"/v4/seasons/game/{game_mode}" + return EndpointInfo( + path=f"/v4/seasons/game/{game_mode}", + description=f"Get season game for mode {game_mode}", + ) - # ============================================================================ - # CHALLENGE ENDPOINTS - # ============================================================================ class CHALLENGES: """Challenge-related endpoints.""" @staticmethod - def get_daily_challenge(endpoint: str = "current") -> str: - """ - Get daily challenge. - - Args: - endpoint: 'current', 'today', or specific date - """ - return f"/v3/challenges/daily-challenges/{endpoint}" + def get_daily_challenge(endpoint: str = "today") -> EndpointInfo: + """Get daily challenge.""" + return EndpointInfo( + path=f"/v3/challenges/daily-challenges/{endpoint}", + description=f"Get daily challenge: {endpoint}", + ) @staticmethod - def get_challenge(challenge_token: str) -> str: + def get_challenge(challenge_token: str) -> EndpointInfo: """Get challenge details.""" - return f"/v3/challenges/{challenge_token}" + return EndpointInfo( + path=f"/v3/challenges/{challenge_token}", + description=f"Get challenge {challenge_token}", + ) - # ============================================================================ - # SOCIAL/FRIENDS ENDPOINTS - # ============================================================================ class SOCIAL: """Social and friends endpoints.""" - GET_FRIENDS_SUMMARY = "/v3/social/friends/summary" # GET - GET_UNCLAIMED_BADGES = "/v3/social/badges/unclaimed" # GET - GET_PERSONALIZED_MAPS = "/v3/social/maps/browse/personalized" # GET + GET_FRIENDS_SUMMARY = EndpointInfo( + path="/v3/social/friends/summary", + description="Get friends summary", + ) + GET_UNCLAIMED_BADGES = EndpointInfo( + path="/v3/social/badges/unclaimed", + description="Get unclaimed badges", + ) + GET_PERSONALIZED_MAPS = EndpointInfo( + path="/v3/social/maps/browse/personalized", + description="Get personalized map recommendations", + ) @staticmethod - def get_activity_feed(count: int = 10, page: int = 0) -> tuple[str, dict]: - """ - Get user activity feed. - - Returns: - Tuple of (endpoint, params_dict) - """ - return "/v4/feed/private", {"count": count, "page": page} + def get_activity_feed(count: int = 10, page: int = 0) -> EndpointInfo: + """Get user activity feed.""" + return EndpointInfo( + path="/v4/feed/private", + description="Get private activity feed", + params_builder=lambda: {"count": count, "page": page}, + ) @staticmethod - def get_friends_activities(time_frame: str, limit: int = 20) -> tuple[str, dict]: - """ - Get friends' activities. + def get_friends_activities( + time_frame: str = "week", limit: int = 20 + ) -> EndpointInfo: + """Get friends' activities.""" + return EndpointInfo( + path="/v3/social/friends/activities", + description="Get friends' activities", + params_builder=lambda: {"timeFrame": time_frame, "limit": limit}, + ) - Args: - time_frame: Time frame for activities - limit: Maximum number of activities - - Returns: - Tuple of (endpoint, params_dict) - """ - return "/v3/social/friends/activities", {"timeFrame": time_frame, "limit": limit} - - # ============================================================================ - # MAPS ENDPOINTS - # ============================================================================ class MAPS: """Map-related endpoints.""" - GET_PERSONALIZED_MAPS = "/v3/social/maps/browse/personalized" # GET + GET_PERSONALIZED_MAPS = EndpointInfo( + path="/v3/social/maps/browse/personalized", + description="Get personalized maps", + ) @staticmethod - def get_map_details(map_id: str) -> str: + def get_map_details(map_id: str) -> EndpointInfo: """Get map details.""" - return f"/maps/{map_id}" + return EndpointInfo( + path=f"/maps/{map_id}", + description=f"Get map {map_id}", + ) @staticmethod - def get_map_leaderboard(map_id: str) -> str: + def get_map_leaderboard(map_id: str) -> EndpointInfo: """Get leaderboard for a map.""" - return f"/v3/scores/maps/{map_id}" + return EndpointInfo( + path=f"/v3/scores/maps/{map_id}", + description=f"Get leaderboard for map {map_id}", + ) @staticmethod - def search_maps(search_type: str, query: str, count: int = 20, page: int = 0) -> tuple[str, dict]: - """ - Search for maps. + def search_maps( + search_type: str = "all", + query: str = "", + count: int = 20, + page: int = 0, + ) -> EndpointInfo: + """Search for maps.""" + return EndpointInfo( + path=f"/v3/social/maps/browse/{search_type}", + description=f"Search maps: {search_type}", + params_builder=lambda: {"q": query, "count": count, "page": page}, + ) - Args: - search_type: Type of search ('all', 'official', 'community', etc.) - query: Search query - count: Number of results per-page - page: Page number - - Returns: - Tuple of (endpoint, params_dict) - """ - return f"/v3/social/maps/browse/{search_type}", { - "q": query, - "count": count, - "page": page - } - - # ============================================================================ - # EXPLORER MODE ENDPOINTS - # ============================================================================ class EXPLORER: """Explorer mode endpoints.""" - GET_PROGRESS = "/v3/explorer" # GET - Get explorer mode progress + GET_PROGRESS = EndpointInfo( + path="/v3/explorer", + description="Get explorer mode progress", + ) - # ============================================================================ - # OBJECTIVES/REWARDS ENDPOINTS - # ============================================================================ class OBJECTIVES: """Objectives and rewards endpoints.""" - GET_OBJECTIVES = "/v4/objectives" # GET - Get current objectives - GET_UNCLAIMED_OBJECTIVES = "/v4/objectives/unclaimed" # GET - Get unclaimed rewards + GET_OBJECTIVES = EndpointInfo( + path="/v4/objectives", + description="Get current objectives", + ) + GET_UNCLAIMED = EndpointInfo( + path="/v4/objectives/unclaimed", + description="Get unclaimed objective rewards", + ) - # ============================================================================ - # SUBSCRIPTION ENDPOINTS - # ============================================================================ class SUBSCRIPTION: """Subscription-related endpoints.""" - GET_SUBSCRIPTION_INFO = "/v3/subscriptions" # GET - Get subscription details + GET_INFO = EndpointInfo( + path="/v3/subscriptions", + description="Get subscription details", + ) -# ============================================================================ -# ENDPOINT UTILITIES -# ============================================================================ - class EndpointBuilder: """Utility class for building complete URLs.""" @staticmethod - def build_url(endpoint: str, use_game_server: bool = False) -> str: + def build_url(endpoint: EndpointInfo) -> str: """ Build complete URL for an endpoint. Args: - endpoint: The endpoint path - use_game_server: Whether to use game server URL + endpoint: The endpoint info Returns: - Complete URL + Complete URL string """ - base = settings.GAME_SERVER_URL if use_game_server else settings.GEOGUESSR_BASE_URL - return f"{base}{endpoint}" + base = ( + settings.GAME_SERVER_URL + if endpoint.use_game_server + else settings.GEOGUESSR_API_URL + ) + return f"{base}{endpoint.path}" @staticmethod - def is_game_server_endpoint(endpoint: str) -> bool: - """ - Check if endpoint belongs to game server. - - Args: - endpoint: The endpoint path - - Returns: - True if it's a game server endpoint - """ + def is_game_server_endpoint(path: str) -> bool: + """Check if a path belongs to game server.""" game_server_prefixes = [ "/battle-royale/", "/duels/", "/lobby/", - "/tournaments" + "/tournaments", ] - return any(endpoint.startswith(prefix) for prefix in game_server_prefixes) - - -# ============================================================================ -# ENDPOINT METADATA -# ============================================================================ - -ENDPOINT_METADATA = { - # Profile endpoints - "/v3/profiles": { - "method": "GET", - "description": "Get current user profile", - "auth_required": True, - "response_type": "profile" - }, - "/v3/profiles/stats": { - "method": "GET", - "description": "Get user statistics", - "auth_required": True, - "response_type": "stats" - }, - "/v4/stats/me": { - "method": "GET", - "description": "Get extended statistics", - "auth_required": True, - "response_type": "extended_stats" - }, - "/v3/profiles/achievements": { - "method": "GET", - "description": "Get user achievements", - "auth_required": True, - "response_type": "achievements" - }, - - # Game endpoints - "/v3/games/{game_token}": { - "method": "GET", - "description": "Get game details", - "auth_required": True, - "response_type": "game" - }, - "/v3/social/events/unfinishedgames": { - "method": "GET", - "description": "Get unfinished games", - "auth_required": True, - "response_type": "games_list" - }, - - # Competitive endpoints - "/v4/seasons/active/stats": { - "method": "GET", - "description": "Get active season statistics", - "auth_required": True, - "response_type": "season_stats" - }, - - # Social endpoints - "/v4/feed/private": { - "method": "GET", - "description": "Get private activity feed", - "auth_required": True, - "response_type": "feed", - "params": ["count", "page"] - }, - "/v3/social/friends/summary": { - "method": "GET", - "description": "Get friends summary", - "auth_required": True, - "response_type": "friends" - }, - - # Maps endpoints - "/maps/{map_id}": { - "method": "GET", - "description": "Get map details", - "auth_required": False, - "response_type": "map" - }, - "/v3/scores/maps/{map_id}": { - "method": "GET", - "description": "Get map leaderboard", - "auth_required": True, - "response_type": "leaderboard" - }, - - # Explorer endpoints - "/v3/explorer": { - "method": "GET", - "description": "Get explorer mode progress", - "auth_required": True, - "response_type": "explorer" - }, - - # Objectives endpoints - "/v4/objectives": { - "method": "GET", - "description": "Get current objectives", - "auth_required": True, - "response_type": "objectives" - }, -} - - -def get_endpoint_info(endpoint: str) -> dict: - """ - Get metadata for an endpoint. - - Args: - endpoint: The endpoint path - - Returns: - Dictionary with endpoint metadata - """ - return ENDPOINT_METADATA.get(endpoint, { - "method": "GET", - "description": "Unknown endpoint", - "auth_required": True, - "response_type": "unknown" - }) + return any(path.startswith(prefix) for prefix in game_server_prefixes)