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.
This commit is contained in:
Yûki VACHOT
parent
6548f11884
commit
4f74343efc
3 changed files with 450 additions and 422 deletions
|
|
@ -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",
|
||||
]
|
||||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue