Rename entrypoints and add logging for pagination.
This commit is contained in:
parent
4f8ec53033
commit
2d396737e4
10
README.md
10
README.md
@ -4,13 +4,13 @@
|
||||
|
||||
```sh
|
||||
pipx install pleb-saberlist
|
||||
replay_ranked_ss
|
||||
player_scores_by_stars
|
||||
leaderboard_songs_by_stars
|
||||
```
|
||||
|
||||
### Example
|
||||
|
||||
```sh
|
||||
replay_ranked_ss
|
||||
Enter the playerid (Default: 76561199407393962):
|
||||
Enter the minimum starlevel to include on the playlist (Default: 5): 6
|
||||
Enter the maximum starlevel to include on the playlist (Default: 7.0):
|
||||
@ -35,6 +35,12 @@ pip install --editable .
|
||||
|
||||
## Tips
|
||||
|
||||
Count results
|
||||
|
||||
```shell
|
||||
jq '.songs | length' < playlist.bplist
|
||||
```
|
||||
|
||||
Avoid printing covers in console.
|
||||
|
||||
```shell
|
||||
|
@ -33,6 +33,6 @@ Homepage = "https://git.satstack.dev/blee/beatsaber-playlist-tool"
|
||||
|
||||
# https://setuptools.pypa.io/en/latest/userguide/entry_point.html
|
||||
[project.scripts]
|
||||
replay_ranked_ss = "saberlist.scoresaber:replay_ranked"
|
||||
ranked_leaderboards_ss = "saberlist.scoresaber:ranked_leaderboards"
|
||||
player_scores_by_stars = "saberlist.scoresaber:replay_ranked"
|
||||
leaderboard_songs_by_stars = "saberlist.scoresaber:leaderboard_songs"
|
||||
#replay_ranked_bl = "saberlist.beatleader:replay_ranked"
|
@ -2,8 +2,16 @@ import json
|
||||
import asyncio
|
||||
from pyscoresaber import ScoreSaberAPI, ScoreSort
|
||||
|
||||
import logging
|
||||
logging.basicConfig(
|
||||
format='%(asctime)s %(levelname)s: %(message)s',
|
||||
datefmt='%Y-%m-%d %H:%M:%S',
|
||||
level=logging.DEBUG
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def filter_and_sort_scores_by_stars(scores, min_stars=0.1, max_stars=float('inf')):
|
||||
|
||||
def filter_and_sort_scores_by_stars(scores, min_stars: float, max_stars: float):
|
||||
# Exclude scores outside the specified star range
|
||||
filtered_scores = [score for score in scores if min_stars <= score.leaderboard.stars < max_stars]
|
||||
# Sort the remaining scores by stars
|
||||
@ -37,6 +45,7 @@ def leaderboards_to_playlist(leaderboards: list, playlist_title: str, playlist_a
|
||||
playlist_json = json.dumps(playlist, indent=4)
|
||||
with open(f"{playlist_title}.bplist", 'w') as file:
|
||||
file.write(playlist_json)
|
||||
logging.info(f"Playlist written to {playlist_title}.bplist")
|
||||
return playlist_json
|
||||
|
||||
def scores_to_playlist(scores, playlist_title, playlist_author = "SaberList Tool"):
|
||||
@ -64,11 +73,12 @@ def scores_to_playlist(scores, playlist_title, playlist_author = "SaberList Tool
|
||||
playlist_json = json.dumps(playlist, indent=4)
|
||||
with open(f"{playlist_title}.bplist", 'w') as file:
|
||||
file.write(playlist_json)
|
||||
logging.info(f"Playlist written to {playlist_title}.bplist")
|
||||
|
||||
return playlist_json
|
||||
|
||||
async def async_replay_ranked():
|
||||
scoresaber = ScoreSaberAPI()
|
||||
scoresaber = ExtendedScoreSaberAPI()
|
||||
try:
|
||||
await scoresaber.start() # Initialize the API client
|
||||
|
||||
@ -137,18 +147,34 @@ async def async_leaderboards(min_stars: float, max_stars: float):
|
||||
from math import ceil
|
||||
from typing import AsyncIterable, List
|
||||
|
||||
from pyscoresaber.models import LeaderboardInfo
|
||||
from pyscoresaber.models import LeaderboardInfo, PlayerScore
|
||||
class ExtendedScoreSaberAPI(ScoreSaberAPI):
|
||||
async def leaderboards_all(self, **params) -> AsyncIterable[List[LeaderboardInfo]]:
|
||||
page = 1
|
||||
max_page = -1
|
||||
while page <= max_page or max_page == -1:
|
||||
logging.info(f"Fetching page {page}/{max_page} of leaderboards")
|
||||
leaderboards = await self.leaderboards(**params, page=page)
|
||||
|
||||
if max_page == -1:
|
||||
max_page = ceil(leaderboards.metadata.total / leaderboards.metadata.items_per_page)
|
||||
logging.info(f"Got {len(leaderboards.leaderboards)} songs from leaderboard page {page}")
|
||||
|
||||
yield leaderboards.leaderboards
|
||||
page += 1
|
||||
|
||||
async def player_scores_all(self, player_id: int, score_sort: ScoreSort) -> AsyncIterable[List[PlayerScore]]:
|
||||
page = 1
|
||||
max_page = -1
|
||||
|
||||
while page < max_page or max_page == -1:
|
||||
try:
|
||||
leaderboards = await self.leaderboards(**params, page=page)
|
||||
if max_page == -1:
|
||||
max_page = ceil(leaderboards.metadata.total / leaderboards.metadata.items_per_page)
|
||||
yield leaderboards.leaderboards
|
||||
page += 1
|
||||
except Exception as e:
|
||||
print(f"An error occurred while fetching page {page}: {e}")
|
||||
break
|
||||
logging.info(f"Fetching page {page}/{max_page} of player scores")
|
||||
recent_scores = await self.player_scores(player_id, sort=score_sort, limit=100, page=page)
|
||||
|
||||
if max_page == -1:
|
||||
max_page = ceil(recent_scores.metadata.total / 100) + 1
|
||||
logging.info(f"Got {len(recent_scores.player_scores)} scores")
|
||||
|
||||
yield recent_scores.player_scores
|
||||
|
||||
page += 1
|
||||
|
Loading…
x
Reference in New Issue
Block a user