20 KiB
20 KiB
New Python Class
We are working on this new Python class:
import json
import os
import logging
from datetime import datetime, timedelta
from typing import Optional, Dict, Any
from clients.beatleader import client as beatleader_client
from clients.beatleader.api.player_scores import player_scores_get_compact_scores
from clients.beatleader.models.compact_score_response_response_with_metadata import CompactScoreResponseResponseWithMetadata
logging.basicConfig(
format='%(asctime)s %(levelname)s: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
level=logging.DEBUG
)
class BeatLeaderAPI:
BASE_URL = "https://api.beatleader.xyz"
def __init__(self, cache_expiry_days: int = 1, cache_dir: Optional[str] = None):
self.client = beatleader_client.Client(base_url=self.BASE_URL)
self.cache_expiry_days = cache_expiry_days
self.CACHE_DIR = cache_dir or self._determine_cache_dir()
if not os.path.exists(self.CACHE_DIR):
os.makedirs(self.CACHE_DIR)
logging.info(f"Created cache directory: {self.CACHE_DIR}")
def _determine_cache_dir(self) -> str:
home_cache = os.path.expanduser("~/.cache")
beatleader_cache = os.path.join(home_cache, "beatleader")
if os.path.exists(home_cache):
if not os.path.exists(beatleader_cache):
try:
os.makedirs(beatleader_cache)
logging.info(f"Created cache directory: {beatleader_cache}")
except OSError as e:
logging.warning(f"Failed to create {beatleader_cache}: {e}")
return os.path.join(os.getcwd(), ".cache")
return beatleader_cache
else:
logging.info("~/.cache doesn't exist, using local .cache directory")
return os.path.join(os.getcwd(), ".cache")
def _get_cache_filename(self, player_id: str) -> str:
return os.path.join(self.CACHE_DIR, f"player_{player_id}_scores.json")
def _is_cache_valid(self, cache_file: str) -> bool:
if not os.path.exists(cache_file):
return False
file_modified_time = datetime.fromtimestamp(os.path.getmtime(cache_file))
return datetime.now() - file_modified_time < timedelta(days=self.cache_expiry_days)
def get_player_scores(
self,
player_id: str,
use_cache: bool = True,
count: int = 100,
sort: str = "recent",
max_pages: Optional[int] = None
) -> Dict[str, Any]:
"""
Fetches all player scores for a given player ID, handling pagination and caching.
:param player_id: The ScoreSaber player ID.
:param use_cache: Whether to use cached data if available.
:param limit: Number of scores per page.
:param sort: Sorting criteria.
:param max_pages: Maximum number of pages to fetch. Fetch all if None.
:return: A dictionary containing metadata and a list of player scores.
"""
cache_file = self._get_cache_filename(player_id)
if use_cache and self._is_cache_valid(cache_file):
logging.debug(f"Using cached data for player {player_id}")
with open(cache_file, 'r') as f:
return json.load(f)
logging.debug(f"Fetching fresh data for player {player_id}")
all_scores = []
page = 1
total_items = None
while max_pages is None or page <= max_pages:
try:
response: CompactScoreResponseResponseWithMetadata = player_scores_get_compact_scores.sync(
client=self.client,
id=player_id,
page=page,
count=count,
sort=sort
)
except Exception as e:
logging.error(f"Error fetching page {page} for player {player_id}: {e}")
return {"metadata": {}, "playerScores": []}
all_scores.extend(response.data)
if total_items is None:
total_items = response.metadata.total
logging.debug(f"Total scores to fetch: {total_items}")
logging.debug(f"Fetched page {page}: {len(response.data)} scores")
if len(all_scores) >= total_items:
break
page += 1
result = {
'metadata': {
'itemsPerPage': response.metadata.items_per_page,
'page': response.metadata.page,
'total': response.metadata.total
},
'playerScores': all_scores
}
with open(cache_file, 'w') as f:
json.dump(result, f, default=str) # default=str to handle datetime serialization
logging.info(f"Cached scores for player {player_id} at {cache_file}")
return result
Here is src/clients/beatleader/api/player_scores/player_scores_get_compact_scores.py:
from http import HTTPStatus
from typing import Any, Dict, Optional, Union, cast
import httpx
from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.compact_score_response_response_with_metadata import CompactScoreResponseResponseWithMetadata
from ...models.difficulty_status import DifficultyStatus
from ...models.leaderboard_contexts import LeaderboardContexts
from ...models.order import Order
from ...models.requirements import Requirements
from ...models.score_filter_status import ScoreFilterStatus
from ...models.scores_sort_by import ScoresSortBy
from ...types import UNSET, Response, Unset
def _get_kwargs(
id: str,
*,
sort_by: Union[Unset, ScoresSortBy] = UNSET,
order: Union[Unset, Order] = UNSET,
page: Union[Unset, int] = 1,
count: Union[Unset, int] = 8,
search: Union[Unset, str] = UNSET,
diff: Union[Unset, str] = UNSET,
mode: Union[Unset, str] = UNSET,
requirements: Union[Unset, Requirements] = UNSET,
score_status: Union[Unset, ScoreFilterStatus] = UNSET,
leaderboard_context: Union[Unset, LeaderboardContexts] = UNSET,
type: Union[Unset, DifficultyStatus] = UNSET,
modifiers: Union[Unset, str] = UNSET,
stars_from: Union[Unset, float] = UNSET,
stars_to: Union[Unset, float] = UNSET,
time_from: Union[Unset, int] = UNSET,
time_to: Union[Unset, int] = UNSET,
event_id: Union[Unset, int] = UNSET,
) -> Dict[str, Any]:
params: Dict[str, Any] = {}
json_sort_by: Union[Unset, str] = UNSET
if not isinstance(sort_by, Unset):
json_sort_by = sort_by.value
params["sortBy"] = json_sort_by
json_order: Union[Unset, str] = UNSET
if not isinstance(order, Unset):
json_order = order.value
params["order"] = json_order
params["page"] = page
params["count"] = count
params["search"] = search
params["diff"] = diff
params["mode"] = mode
json_requirements: Union[Unset, str] = UNSET
if not isinstance(requirements, Unset):
json_requirements = requirements.value
params["requirements"] = json_requirements
json_score_status: Union[Unset, str] = UNSET
if not isinstance(score_status, Unset):
json_score_status = score_status.value
params["scoreStatus"] = json_score_status
json_leaderboard_context: Union[Unset, str] = UNSET
if not isinstance(leaderboard_context, Unset):
json_leaderboard_context = leaderboard_context.value
params["leaderboardContext"] = json_leaderboard_context
json_type: Union[Unset, str] = UNSET
if not isinstance(type, Unset):
json_type = type.value
params["type"] = json_type
params["modifiers"] = modifiers
params["stars_from"] = stars_from
params["stars_to"] = stars_to
params["time_from"] = time_from
params["time_to"] = time_to
params["eventId"] = event_id
params = {k: v for k, v in params.items() if v is not UNSET and v is not None}
_kwargs: Dict[str, Any] = {
"method": "get",
"url": f"/player/{id}/scores/compact",
"params": params,
}
return _kwargs
def _parse_response(
*, client: Union[AuthenticatedClient, Client], response: httpx.Response
) -> Optional[Union[Any, CompactScoreResponseResponseWithMetadata]]:
if response.status_code == HTTPStatus.OK:
response_200 = CompactScoreResponseResponseWithMetadata.from_dict(response.json())
return response_200
if response.status_code == HTTPStatus.BAD_REQUEST:
response_400 = cast(Any, None)
return response_400
if response.status_code == HTTPStatus.NOT_FOUND:
response_404 = cast(Any, None)
return response_404
if client.raise_on_unexpected_status:
raise errors.UnexpectedStatus(response.status_code, response.content)
else:
return None
def _build_response(
*, client: Union[AuthenticatedClient, Client], response: httpx.Response
) -> Response[Union[Any, CompactScoreResponseResponseWithMetadata]]:
return Response(
status_code=HTTPStatus(response.status_code),
content=response.content,
headers=response.headers,
parsed=_parse_response(client=client, response=response),
)
def sync_detailed(
id: str,
*,
client: Union[AuthenticatedClient, Client],
sort_by: Union[Unset, ScoresSortBy] = UNSET,
order: Union[Unset, Order] = UNSET,
page: Union[Unset, int] = 1,
count: Union[Unset, int] = 8,
search: Union[Unset, str] = UNSET,
diff: Union[Unset, str] = UNSET,
mode: Union[Unset, str] = UNSET,
requirements: Union[Unset, Requirements] = UNSET,
score_status: Union[Unset, ScoreFilterStatus] = UNSET,
leaderboard_context: Union[Unset, LeaderboardContexts] = UNSET,
type: Union[Unset, DifficultyStatus] = UNSET,
modifiers: Union[Unset, str] = UNSET,
stars_from: Union[Unset, float] = UNSET,
stars_to: Union[Unset, float] = UNSET,
time_from: Union[Unset, int] = UNSET,
time_to: Union[Unset, int] = UNSET,
event_id: Union[Unset, int] = UNSET,
) -> Response[Union[Any, CompactScoreResponseResponseWithMetadata]]:
"""Retrieve player's scores in a compact form
Fetches a paginated list of scores for a specified player ID. Returns less info to save bandwith or
processing time
Args:
id (str):
sort_by (Union[Unset, ScoresSortBy]):
order (Union[Unset, Order]): Represents the order in which values will be sorted.
page (Union[Unset, int]): Default: 1.
count (Union[Unset, int]): Default: 8.
search (Union[Unset, str]):
diff (Union[Unset, str]):
mode (Union[Unset, str]):
requirements (Union[Unset, Requirements]):
score_status (Union[Unset, ScoreFilterStatus]):
leaderboard_context (Union[Unset, LeaderboardContexts]):
type (Union[Unset, DifficultyStatus]): Represents the difficulty status of a map.
modifiers (Union[Unset, str]):
stars_from (Union[Unset, float]):
stars_to (Union[Unset, float]):
time_from (Union[Unset, int]):
time_to (Union[Unset, int]):
event_id (Union[Unset, int]):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Response[Union[Any, CompactScoreResponseResponseWithMetadata]]
"""
kwargs = _get_kwargs(
id=id,
sort_by=sort_by,
order=order,
page=page,
count=count,
search=search,
diff=diff,
mode=mode,
requirements=requirements,
score_status=score_status,
leaderboard_context=leaderboard_context,
type=type,
modifiers=modifiers,
stars_from=stars_from,
stars_to=stars_to,
time_from=time_from,
time_to=time_to,
event_id=event_id,
)
response = client.get_httpx_client().request(
**kwargs,
)
return _build_response(client=client, response=response)
def sync(
id: str,
*,
client: Union[AuthenticatedClient, Client],
sort_by: Union[Unset, ScoresSortBy] = UNSET,
order: Union[Unset, Order] = UNSET,
page: Union[Unset, int] = 1,
count: Union[Unset, int] = 8,
search: Union[Unset, str] = UNSET,
diff: Union[Unset, str] = UNSET,
mode: Union[Unset, str] = UNSET,
requirements: Union[Unset, Requirements] = UNSET,
score_status: Union[Unset, ScoreFilterStatus] = UNSET,
leaderboard_context: Union[Unset, LeaderboardContexts] = UNSET,
type: Union[Unset, DifficultyStatus] = UNSET,
modifiers: Union[Unset, str] = UNSET,
stars_from: Union[Unset, float] = UNSET,
stars_to: Union[Unset, float] = UNSET,
time_from: Union[Unset, int] = UNSET,
time_to: Union[Unset, int] = UNSET,
event_id: Union[Unset, int] = UNSET,
) -> Optional[Union[Any, CompactScoreResponseResponseWithMetadata]]:
"""Retrieve player's scores in a compact form
Fetches a paginated list of scores for a specified player ID. Returns less info to save bandwith or
processing time
Args:
id (str):
sort_by (Union[Unset, ScoresSortBy]):
order (Union[Unset, Order]): Represents the order in which values will be sorted.
page (Union[Unset, int]): Default: 1.
count (Union[Unset, int]): Default: 8.
search (Union[Unset, str]):
diff (Union[Unset, str]):
mode (Union[Unset, str]):
requirements (Union[Unset, Requirements]):
score_status (Union[Unset, ScoreFilterStatus]):
leaderboard_context (Union[Unset, LeaderboardContexts]):
type (Union[Unset, DifficultyStatus]): Represents the difficulty status of a map.
modifiers (Union[Unset, str]):
stars_from (Union[Unset, float]):
stars_to (Union[Unset, float]):
time_from (Union[Unset, int]):
time_to (Union[Unset, int]):
event_id (Union[Unset, int]):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Union[Any, CompactScoreResponseResponseWithMetadata]
"""
return sync_detailed(
id=id,
client=client,
sort_by=sort_by,
order=order,
page=page,
count=count,
search=search,
diff=diff,
mode=mode,
requirements=requirements,
score_status=score_status,
leaderboard_context=leaderboard_context,
type=type,
modifiers=modifiers,
stars_from=stars_from,
stars_to=stars_to,
time_from=time_from,
time_to=time_to,
event_id=event_id,
).parsed
async def asyncio_detailed(
id: str,
*,
client: Union[AuthenticatedClient, Client],
sort_by: Union[Unset, ScoresSortBy] = UNSET,
order: Union[Unset, Order] = UNSET,
page: Union[Unset, int] = 1,
count: Union[Unset, int] = 8,
search: Union[Unset, str] = UNSET,
diff: Union[Unset, str] = UNSET,
mode: Union[Unset, str] = UNSET,
requirements: Union[Unset, Requirements] = UNSET,
score_status: Union[Unset, ScoreFilterStatus] = UNSET,
leaderboard_context: Union[Unset, LeaderboardContexts] = UNSET,
type: Union[Unset, DifficultyStatus] = UNSET,
modifiers: Union[Unset, str] = UNSET,
stars_from: Union[Unset, float] = UNSET,
stars_to: Union[Unset, float] = UNSET,
time_from: Union[Unset, int] = UNSET,
time_to: Union[Unset, int] = UNSET,
event_id: Union[Unset, int] = UNSET,
) -> Response[Union[Any, CompactScoreResponseResponseWithMetadata]]:
"""Retrieve player's scores in a compact form
Fetches a paginated list of scores for a specified player ID. Returns less info to save bandwith or
processing time
Args:
id (str):
sort_by (Union[Unset, ScoresSortBy]):
order (Union[Unset, Order]): Represents the order in which values will be sorted.
page (Union[Unset, int]): Default: 1.
count (Union[Unset, int]): Default: 8.
search (Union[Unset, str]):
diff (Union[Unset, str]):
mode (Union[Unset, str]):
requirements (Union[Unset, Requirements]):
score_status (Union[Unset, ScoreFilterStatus]):
leaderboard_context (Union[Unset, LeaderboardContexts]):
type (Union[Unset, DifficultyStatus]): Represents the difficulty status of a map.
modifiers (Union[Unset, str]):
stars_from (Union[Unset, float]):
stars_to (Union[Unset, float]):
time_from (Union[Unset, int]):
time_to (Union[Unset, int]):
event_id (Union[Unset, int]):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Response[Union[Any, CompactScoreResponseResponseWithMetadata]]
"""
kwargs = _get_kwargs(
id=id,
sort_by=sort_by,
order=order,
page=page,
count=count,
search=search,
diff=diff,
mode=mode,
requirements=requirements,
score_status=score_status,
leaderboard_context=leaderboard_context,
type=type,
modifiers=modifiers,
stars_from=stars_from,
stars_to=stars_to,
time_from=time_from,
time_to=time_to,
event_id=event_id,
)
response = await client.get_async_httpx_client().request(**kwargs)
return _build_response(client=client, response=response)
async def asyncio(
id: str,
*,
client: Union[AuthenticatedClient, Client],
sort_by: Union[Unset, ScoresSortBy] = UNSET,
order: Union[Unset, Order] = UNSET,
page: Union[Unset, int] = 1,
count: Union[Unset, int] = 8,
search: Union[Unset, str] = UNSET,
diff: Union[Unset, str] = UNSET,
mode: Union[Unset, str] = UNSET,
requirements: Union[Unset, Requirements] = UNSET,
score_status: Union[Unset, ScoreFilterStatus] = UNSET,
leaderboard_context: Union[Unset, LeaderboardContexts] = UNSET,
type: Union[Unset, DifficultyStatus] = UNSET,
modifiers: Union[Unset, str] = UNSET,
stars_from: Union[Unset, float] = UNSET,
stars_to: Union[Unset, float] = UNSET,
time_from: Union[Unset, int] = UNSET,
time_to: Union[Unset, int] = UNSET,
event_id: Union[Unset, int] = UNSET,
) -> Optional[Union[Any, CompactScoreResponseResponseWithMetadata]]:
"""Retrieve player's scores in a compact form
Fetches a paginated list of scores for a specified player ID. Returns less info to save bandwith or
processing time
Args:
id (str):
sort_by (Union[Unset, ScoresSortBy]):
order (Union[Unset, Order]): Represents the order in which values will be sorted.
page (Union[Unset, int]): Default: 1.
count (Union[Unset, int]): Default: 8.
search (Union[Unset, str]):
diff (Union[Unset, str]):
mode (Union[Unset, str]):
requirements (Union[Unset, Requirements]):
score_status (Union[Unset, ScoreFilterStatus]):
leaderboard_context (Union[Unset, LeaderboardContexts]):
type (Union[Unset, DifficultyStatus]): Represents the difficulty status of a map.
modifiers (Union[Unset, str]):
stars_from (Union[Unset, float]):
stars_to (Union[Unset, float]):
time_from (Union[Unset, int]):
time_to (Union[Unset, int]):
event_id (Union[Unset, int]):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Union[Any, CompactScoreResponseResponseWithMetadata]
"""
return (
await asyncio_detailed(
id=id,
client=client,
sort_by=sort_by,
order=order,
page=page,
count=count,
search=search,
diff=diff,
mode=mode,
requirements=requirements,
score_status=score_status,
leaderboard_context=leaderboard_context,
type=type,
modifiers=modifiers,
stars_from=stars_from,
stars_to=stars_to,
time_from=time_from,
time_to=time_to,
event_id=event_id,
)
).parsed
Please review get_player_scores(), we wonder if the sort option is done correctly.