Add cutoff date to mapper's maps, and other changes

This commit is contained in:
Brian Lee 2025-02-23 08:55:47 -08:00
parent 362603d160
commit 0787ad6bee
7 changed files with 546 additions and 27 deletions

61
docs/BeatLeaderPlayers.md Normal file
View File

@ -0,0 +1,61 @@
# BL Players
```python
from helpers.SimpleBeatLeaderAPI import SimpleBeatLeaderAPI
beatleader_api = SimpleBeatLeaderAPI()
player_data = beatleader_api.get_players()
# Filter players with high accuracy and rank better than 1000
high_accuracy_players = [
player for player in player_data
if player['scoreStats']['averageRankedAccuracy'] > 0.96 and
player['rank'] > 1000 and
player['scoreStats']['totalPlayCount'] > 1000
]
# Sort the filtered list by rank in descending order
high_accuracy_players_sorted = sorted(high_accuracy_players, key=lambda player: player['rank'])
for player in high_accuracy_players_sorted:
rank = player['rank']
name = player['name']
player_id = player['id']
accuracy = player['scoreStats']['averageRankedAccuracy'] * 100
total_plays = player['scoreStats']['totalPlayCount']
print(f" {rank}: {player_id} {accuracy:.2f}%, {total_plays} plays: {name}")
```
Results:
1002: 76561198108621262 96.84%, 2799 plays: Nano
1039: 76561198067254301 97.40%, 2083 plays: Muz
1106: 76561198020334769 96.70%, 1272 plays: Rexxz
1140: 76561198103016268 97.17%, 2635 plays: ACC | Wookie
1144: 76561198074878770 96.91%, 1007 plays: Jormungandr
1158: 76561198154593513 97.05%, 1283 plays: ZLQ
1182: 76561198398878358 96.51%, 1717 plays: Rimu
1221: 76561198141764746 97.71%, 1688 plays: ACC | Jani
1303: 76561198890357715 97.91%, 1824 plays: mojo
1333: 76561197977159578 96.35%, 1656 plays: Ando
1344: 76561198092977917 97.05%, 2906 plays: bluewingoo
1356: 76561199100866994 96.30%, 1800 plays: Dollface
1411: 76561198040296837 96.59%, 2043 plays: BSFR | Vred
1460: 76561198162613446 96.54%, 1581 plays: RockaX
1599: 76561198212831695 96.20%, 1274 plays: Xmpo
1798: 76561198139760882 96.13%, 1084 plays: W33talik
1907: 76561198084876475 96.51%, 1090 plays: NeroMJ
1932: 76561198163148965 96.37%, 1345 plays: Hygoto
2048: 76561198358170446 98.24%, 2130 plays: 19:47
2088: 76561198163772169 97.82%, 1239 plays: Kunya
2227: 76561198200272334 96.06%, 1442 plays: Arflic
2231: 76561197993806676 96.40%, 2296 plays: Taoh Rihze
2323: 76561199017645460 96.57%, 2033 plays: GF | Soogy
2353: 76561197999550197 96.42%, 1272 plays: ACC | Ixsen
2551: 3767819413276402 96.22%, 1192 plays: rileywip
2563: 76561198052914112 96.98%, 1365 plays: BlueFlame_MK
2607: 76561198023638450 96.23%, 22687 plays: Craedien
2763: 76561198017723775 96.63%, 1131 plays: ika
3272: 76561199157033165 96.09%, 1936 plays: Lumberjack462
3309: 76561199407393962 96.27%, 1777 plays: pleb

View File

@ -55,9 +55,9 @@ map_data = beatsaver_api.get_maps(year=2024, month=9)
```python
from helpers.SimpleBeatLeaderAPI import SimpleBeatLeaderAPI
player_id = "76561199407393962"
beatleader_api = SimpleBeatLeaderAPI()
player_id = "76561199407393962"
scores_data = beatleader_api.get_player_scores(player_id).get('data', [])
acc_graph = beatleader_api.get_player_accgraph(player_id)
@ -70,10 +70,13 @@ filtered_data[0]
```python
from helpers.SimpleBeatSaverAPI import SimpleBeatSaverAPI
beat_saver_api = SimpleBeatSaverAPI()
from saberlist.playlist_strategies.beatsaver import *
beatsaver_api = SimpleBeatSaverAPI()
environment_maps = beatsaver_api.get_maps_by_environment()
curated_songs = beatsaver_api.get_curated_songs(use_cache=False)
mapper_maps = beatsaver_api.get_mapper_maps(mapper_id=4285738, use_cache=False)
curated_songs = beat_saver_api.get_curated_songs(use_cache=False)
mapper_maps = beat_saver_api.get_mapper_maps(mapper_id=4285738, use_cache=False)
```
## ScoreSaberAPI

View File

@ -51,7 +51,10 @@ class SimpleBeatLeaderAPI:
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, use_cache=True, page_size=100, max_pages=None):
def get_player_scores(self, player_id,
use_cache=True,
page_size=100,
max_pages=None):
cache_file = self._get_cache_filename(player_id)
if use_cache and self._is_cache_valid(cache_file):
@ -134,7 +137,12 @@ class SimpleBeatLeaderAPI:
logging.error(f"Error fetching player info for ID {player_id}: {e}")
return None
def get_leaderboard(self, hash, diff="ExpertPlus", mode="Standard", use_cache=True, page=1, count=10) -> list[dict]:
def get_leaderboard(self, hash,
diff="ExpertPlus",
mode="Standard",
use_cache=True,
page=1,
count=10) -> list[dict]:
"""
Retrieve leaderboard for a specific map, with caching.
@ -178,7 +186,11 @@ class SimpleBeatLeaderAPI:
logging.error(f"Error fetching leaderboard for hash {hash}, diff {diff}, mode {mode}: {e}")
return None
def get_player_accgraph(self, player_id, use_cache=True, context="general", include_unranked=False, type="acc"):
def get_player_accgraph(self, player_id,
use_cache=True,
context="general",
include_unranked=False,
type="acc"):
"""
Retrieve graph data for a specific player.
@ -220,7 +232,9 @@ class SimpleBeatLeaderAPI:
logging.error(f"Error fetching acc graph for player {player_id}: {e}")
return None
def get_ranked_maps(self, stars_from=5, stars_to=10, use_cache=True):
def get_ranked_maps(self, stars_from=5,
stars_to=10,
use_cache=True):
"""
Retrieve all ranked maps within the specified star range, handling pagination and caching.
@ -292,4 +306,67 @@ class SimpleBeatLeaderAPI:
json.dump(all_maps, f, indent=4)
logging.debug(f"Cached {len(all_maps)} ranked maps to {cache_file}")
return all_maps
return all_maps
def get_players(self,
max_pages=80,
count=50,
use_cache=True):
"""
Retrieve a list of players ordered by play rank.
This method fetches players from the API using sortBy=0 (ordered by play rank) and paginates through
the results up to a maximum of 'max_pages' pages. The 'count' parameter defines the number of players
per page (default: 50). Results are cached on disk in the configured cache directory.
:param max_pages: Maximum number of pages to retrieve (default: 80)
:param count: Number of players per page (default: 50)
:param use_cache: Whether to use cached data if available (default: True)
:return: List of player data.
"""
cache_file = os.path.join(self.CACHE_DIR, f"players_{max_pages}.json")
if use_cache and self._is_cache_valid(cache_file):
logging.debug("Using cached player data.")
with open(cache_file, 'r') as f:
return json.load(f)
logging.debug(f"Fetching fresh player data ordered by play rank (max_pages: {max_pages}, count: {count}).")
url = f"{self.BASE_URL}/players"
all_players = []
page = 1
total_items = None
while page <= max_pages:
params = {
"sortBy": 0,
"page": page,
"count": count
}
try:
response = self.session.get(url, params=params)
response.raise_for_status()
data = response.json()
# Set total_items from the metadata returned in the first request.
if total_items is None:
total_items = data.get("metadata", {}).get("total", 0)
players_page = data.get("data", [])
all_players.extend(players_page)
# Stop if we've fetched all available players or reached max_pages.
if len(all_players) >= total_items or not players_page:
break
page += 1
sleep(1) # Respect API rate limits
except requests.exceptions.RequestException as e:
logging.error(f"Error fetching players on page {page}: {e}")
break
with open(cache_file, 'w') as f:
json.dump(all_players, f)
logging.debug(f"Cached player data to {cache_file}")
return all_players

View File

@ -66,7 +66,10 @@ class SimpleBeatSaverAPI:
return self.CACHE_DIR
def get_curated_songs(self, use_cache=True) -> list[dict]:
def get_curated_songs(
self,
use_cache=True
) -> list[dict]:
"""
Retrieve curated songs from BeatSaver.
@ -127,7 +130,11 @@ class SimpleBeatSaverAPI:
return processed_songs
def get_followed_mappers(self, user_id: int = 243016, use_cache=True) -> list[dict]:
def get_followed_mappers(
self,
user_id: int = 243016,
use_cache=True
) -> list[dict]:
"""
Retrieve list of mappers followed by a specific user.
@ -159,7 +166,29 @@ class SimpleBeatSaverAPI:
logging.error(f"Error fetching followed mappers: {e}")
return []
def get_mapper_maps(self, mapper_id: int, use_cache=True) -> list[dict]:
"""Paste this into ipython to get sample data:
from helpers.SimpleBeatSaverAPI import SimpleBeatSaverAPI
from saberlist.playlist_strategies.beatsaver import *
BASE_URL = "https://api.beatsaver.com"
import requests
mapper_id = 29945
page = 0
session = requests.Session()
url = f"{BASE_URL}/search/text/{page}"
params = {
'collaborator': str(mapper_id),
'automapper': 'true',
'sortOrder': 'Latest'
}
response = session.get(url, params=params)
data = response.json()
data['docs'][0]
"""
def get_mapper_maps(
self,
mapper_id: int,
use_cache=True
) -> list[dict]:
"""
Retrieve all maps created by a specific mapper.
@ -197,6 +226,7 @@ class SimpleBeatSaverAPI:
'hash': version['hash'],
'key': song['id'],
'songName': song['metadata']['songName'],
'date': song['lastPublishedAt'] # e.g. 2024-10-20T22:49:05.842454Z
}
processed_songs.append(song_info)
@ -216,3 +246,118 @@ class SimpleBeatSaverAPI:
json.dump(processed_songs, f)
return processed_songs
def get_maps_by_environment(
self,
environment_name: str = None,
max_pages: int = 10,
use_cache: bool = True
) -> list[dict]:
"""
Retrieve all maps with a given environment name from BeatSaver.
:param environment_name: The name of the environment to filter maps by. If None, prompts user to select.
:param max_pages: Maximum number of pages to fetch. If None, fetch all available pages.
:param use_cache: Whether to use cached data if available (default: True)
:return: List of dictionaries containing map information
"""
if not environment_name:
environment_name = self.prompt_for_environment()
cache_file = os.path.join(self.CACHE_DIR, f"environment_{environment_name}_maps.json")
if use_cache and self._is_cache_valid(cache_file):
logging.debug(f"Using cached data for environment '{environment_name}'")
with open(cache_file, 'r') as f:
return json.load(f)
processed_maps = []
page = 0
while True:
url = f"{self.BASE_URL}/search/text/{page}"
params = {
"environments": f"{environment_name}Environment"
}
try:
response = self.session.get(url, params=params)
response.raise_for_status()
data = response.json()
# Process the response to extract relevant information
for song in data.get('docs', []):
for version in song.get('versions', []):
map_info = {
'hash': version['hash'],
'key': song['id'],
'songName': song['metadata']['songName'],
'environment': environment_name,
}
processed_maps.append(map_info)
# Check if we've reached the last page
if page >= data['info']['pages'] - 1:
break
page += 1
if max_pages is not None and page >= max_pages:
break
sleep(1) # Respectful delay between requests
except requests.exceptions.RequestException as e:
logging.error(f"Error fetching maps for environment '{environment_name}': {e}")
return []
# Cache the results
with open(cache_file, 'w') as f:
json.dump(processed_maps, f)
logging.debug(f"Cached {len(processed_maps)} maps for environment '{environment_name}'")
return processed_maps
def prompt_for_environment(self) -> str:
"""
Prompt the user to select a map environment from a list of known environments.
:return: Selected environment name as a string
"""
# We're only interested in V3 environments and beyond
known_environments = [
'Weave',
'Pyro',
'EDM',
'TheSecond',
'Lizzo',
'TheWeeknd',
'RockMixtape',
'Dragons2',
'Panic2',
'Queen',
'LinkinPark2',
'TheRollingStones',
'Lattice',
'DaftPunk',
'HipHop',
'Collider',
'Britney',
'Monstercat2',
'Metallica',
]
print("Please select a map environment from the list below:")
for idx, env in enumerate(known_environments, start=1):
print(f"{idx}. {env}")
while True:
try:
choice = int(input("Enter the number corresponding to your choice: "))
if 1 <= choice <= len(known_environments):
selected_environment = known_environments[choice - 1]
logging.debug(f"Selected environment: {selected_environment}")
return selected_environment
else:
print(f"Please enter a number between 1 and {len(known_environments)}.")
except ValueError:
print("Invalid input. Please enter a valid number.")

View File

@ -39,6 +39,8 @@ from saberlist.playlist_strategies.accuracy import (
from saberlist.playlist_strategies.beatsaver import (
playlist_strategy_beatsaver_curated,
playlist_strategy_beatsaver_mappers,
playlist_strategy_beatsaver_environment,
playlist_strategy_beatsaver_all_environments,
)
def saberlist() -> None:
@ -115,6 +117,30 @@ def saberlist() -> None:
playlist_data, playlist_title = playlist_strategy_beatsaver_mappers(SimpleBeatSaverAPI())
playlist_builder = PlaylistBuilder(covers_dir='./covers/pajamas')
elif strategy == 'beatsaver_environment':
environment_name = args.environment_name
playlist_data, playlist_title = playlist_strategy_beatsaver_environment(
SimpleBeatSaverAPI(cache_expiry_days=CACHE_EXPIRY_DAYS),
environment_name=environment_name
)
playlist_builder = PlaylistBuilder(covers_dir='./covers/beatsaver')
elif strategy == 'beatsaver_all_environments':
playlists = playlist_strategy_beatsaver_all_environments(
SimpleBeatSaverAPI(cache_expiry_days=CACHE_EXPIRY_DAYS)
)
builder = PlaylistBuilder(covers_dir='./covers/beatsaver')
for env, (data, title) in playlists.items():
if data:
builder.create_playlist(
data,
playlist_title=f"{title}",
playlist_author="SaberList Tool"
)
else:
logging.info(f"No new songs to add for environment '{env}'.")
return
elif strategy == 'blank_playlist':
playlist_data = []
playlist_title = input("Enter playlist title: ")
@ -228,6 +254,23 @@ def parse_args_subcommands():
action="store_true",
help="Reset the history for blank_playlist (usually unnecessary)")
# -------- beatsaver_environment --------
parser_bs_env = subparsers.add_parser("beatsaver_environment",
help="Generate a playlist for a specific BeatSaver environment")
parser_bs_env.add_argument("-r", "--reset",
action="store_true",
help="Reset the history for beatsaver_environment")
parser_bs_env.add_argument("environment_name",
type=str,
help="Name of the BeatSaver environment (e.g., 'Weave', 'Pyro')")
# -------- beatsaver_all_environments --------
parser_bs_all_env = subparsers.add_parser("beatsaver_all_environments",
help="Generate playlists for all known BeatSaver environments")
parser_bs_all_env.add_argument("-r", "--reset",
action="store_true",
help="Reset the history for all beatsaver environments")
# If no arguments passed, print help
if len(sys.argv) == 1:
parser.print_help(sys.stderr)

View File

@ -1,8 +1,8 @@
from statistics import mean
from typing import Dict, Any, List
from typing import Dict, Any, List, Tuple, Optional
import logging
import os
from datetime import datetime, timedelta
from datetime import datetime, timedelta, timezone
import calendar
from dotenv import load_dotenv
@ -235,16 +235,25 @@ def map_leaders_by_month(month: int = 9, year: int = 2024, game_modes: List[str]
def playlist_strategy_beatsaver_mappers(
api: SimpleBeatSaverAPI,
song_count: int = 50
song_count: int = 50,
date_threshold: Optional[datetime] = None
) -> List[Dict[str, Any]]:
"""
Build a playlist from maps created by specified mappers on BeatSaver,
interleaving maps from different mappers for variety.
Maps with a publication date older than the date_threshold are skipped.
:param api: SimpleBeatSaverAPI instance for making API calls
:param song_count: Number of songs to include in the playlist
:param date_threshold: datetime threshold; maps published before this date are skipped.
Defaults to January 1, 2024, in UTC.
:return: Tuple of (playlist data, playlist title)
"""
from datetime import timezone
if date_threshold is None:
date_threshold = datetime(2024, 1, 1, tzinfo=timezone.utc)
history = load_history()
history.setdefault('beatsaver_mappers', {})
history.setdefault('playlist_counts', {})
@ -256,11 +265,12 @@ def playlist_strategy_beatsaver_mappers(
new_count = current_count + 1
history['playlist_counts'][count_key] = new_count
# Collect maps by mapper
# Collect maps by mapper with date filtering
maps_by_mapper = {}
for mapper_id in mapper_ids:
logging.info(f"Fetching maps for mapper ID: {mapper_id}")
mapper_maps = api.get_mapper_maps(mapper_id=mapper_id, use_cache=False)
# mapper_maps = api.get_mapper_maps(mapper_id=mapper_id, use_cache=False)
mapper_maps = api.get_mapper_maps(mapper_id=mapper_id)
if not mapper_maps:
logging.warning(f"No maps found for mapper ID: {mapper_id}")
@ -268,12 +278,31 @@ def playlist_strategy_beatsaver_mappers(
logging.info(f"Found {len(mapper_maps)} maps from mapper ID: {mapper_id}")
# Filter out maps that are in history
eligible_maps = [
song for song in mapper_maps
if song.get('hash') not in history['beatsaver_mappers']
]
eligible_maps = []
for song in mapper_maps:
# Skip if the song already exists in history
if song.get('hash') in history['beatsaver_mappers']:
continue
# Filter out maps with a date older than the provided threshold.
date_str = song.get('date')
if not date_str:
logging.warning(f"Skipping song {song.get('songName')} as it has no date.")
continue
try:
# Convert the ISO date string (which ends with 'Z') into a timezone-aware datetime
song_date = datetime.fromisoformat(date_str.replace("Z", "+00:00"))
except Exception as e:
logging.warning(f"Could not parse date '{date_str}' for song {song.get('songName')}: {e}")
continue
if song_date < date_threshold:
logging.debug(f"Skipping song {song.get('songName')} published back in {song_date.strftime('%b %Y')}")
continue
eligible_maps.append(song)
if eligible_maps:
maps_by_mapper[mapper_id] = eligible_maps
@ -306,3 +335,154 @@ def playlist_strategy_beatsaver_mappers(
playlist_title = f"mappers-{new_count:02d}"
return playlist_data, playlist_title
def playlist_strategy_beatsaver_environment(
api: SimpleBeatSaverAPI,
environment_name: str,
song_count: int = 50,
max_pages: int = 10
) -> Tuple[List[Dict[str, Any]], str]:
"""
Build a playlist from maps that match the specified environment.
Args:
api (SimpleBeatSaverAPI): Instance for BeatSaver API calls.
environment_name (str): The map environment to filter by.
song_count (int): Max number of songs to include (default: 50).
max_pages (int): Maximum number of pages to fetch from the API (default: 10).
Returns:
A tuple containing:
- playlist_data: A list of dictionaries with song details.
- playlist_title: A unique playlist title in the format "<environment>-XX".
"""
history = load_history()
history.setdefault('beatsaver_environment', {})
history.setdefault('playlist_counts', {})
if environment_name not in history['beatsaver_environment']:
history['beatsaver_environment'][environment_name] = {}
count_key = f"beatsaver_environment-{environment_name}"
current_count = history['playlist_counts'].get(count_key, 0)
new_count = current_count + 1
history['playlist_counts'][count_key] = new_count
maps = api.get_maps_by_environment(
environment_name=environment_name,
max_pages=max_pages,
use_cache=True
)
if not maps:
logging.error(f"No maps found for environment: {environment_name}.")
return [], f"{environment_name.lower()}-{new_count:02d}"
playlist_data = []
for song in maps:
song_hash = song.get('hash')
if song_hash in history['beatsaver_environment'][environment_name]:
logging.debug(f"Skipping song {song_hash} for environment {environment_name} as it is in history.")
continue
playlist_data.append(song)
logging.info(f"Song added: {song.get('songName', 'Unknown')} from environment {environment_name}")
history['beatsaver_environment'][environment_name][song_hash] = True
if len(playlist_data) >= song_count:
break
if not playlist_data:
logging.info(f"No new songs found for environment {environment_name} based on history.")
else:
logging.info(f"Total songs added for environment {environment_name}: {len(playlist_data)}")
save_history(history)
playlist_title = f"{environment_name.lower()}-{new_count:02d}"
return playlist_data, playlist_title
def playlist_strategy_beatsaver_all_environments(
api: SimpleBeatSaverAPI,
song_count: int = 50,
max_pages: int = 10
) -> Dict[str, Tuple[List[Dict[str, Any]], str]]:
"""
Build playlists for every known environment.
Args:
api (SimpleBeatSaverAPI): Instance for BeatSaver API calls.
song_count (int): Max number of songs per environment playlist (default: 50).
max_pages (int): Maximum pages to fetch from the API per environment (default: 10).
Returns:
A dictionary mapping each environment (str) to a tuple:
(playlist_data, playlist_title)
"""
known_environments = [
'Weave',
'Pyro',
'EDM',
'TheSecond',
'Lizzo',
'TheWeeknd',
'RockMixtape',
'Dragons2',
'Panic2',
'Queen',
'LinkinPark2',
'TheRollingStones',
'Lattice',
'DaftPunk',
'HipHop',
'Collider',
'Britney',
'Monstercat2',
'Metallica',
]
playlists: Dict[str, Tuple[List[Dict[str, Any]], str]] = {}
history = load_history()
history.setdefault('beatsaver_environment', {})
history.setdefault('playlist_counts', {})
for environment in known_environments:
if environment not in history['beatsaver_environment']:
history['beatsaver_environment'][environment] = {}
count_key = f"beatsaver_environment-{environment}"
current_count = history['playlist_counts'].get(count_key, 0)
new_count = current_count + 1
history['playlist_counts'][count_key] = new_count
maps = api.get_maps_by_environment(
environment_name=environment,
max_pages=max_pages,
use_cache=True
)
if not maps:
logging.error(f"No maps found for environment: {environment}.")
playlists[environment] = ([], f"{environment.lower()}-{new_count:02d}")
continue
playlist_data: List[Dict[str, Any]] = []
for song in maps:
song_hash = song.get('hash')
if song_hash in history['beatsaver_environment'][environment]:
logging.debug(f"Skipping song {song_hash} for environment {environment} as it is in history.")
continue
playlist_data.append(song)
logging.info(f"Song added: {song.get('songName', 'Unknown')} from environment {environment}")
history['beatsaver_environment'][environment][song_hash] = True
if len(playlist_data) >= song_count:
break
if not playlist_data:
logging.info(f"No new songs found for environment {environment} based on history.")
else:
logging.info(f"Total songs added for environment {environment}: {len(playlist_data)}")
playlist_title = f"{environment.lower()}-{new_count:02d}"
playlists[environment] = (playlist_data, playlist_title)
save_history(history)
return playlists

View File

@ -53,12 +53,22 @@ def prompt_for_player_id(default_id: str = '76561199407393962') -> str:
def prompt_for_mapper_ids() -> List[int]:
default_mapper_ids = [
4285547, # Avexus
4330286, # VoltageO
4294118, # Spinvvy
29945, # Uncouth
104443, # Rxerti
113133, # Cush
120215, # Jonas_0_0
145971, # Bellus
202784, # Najoko
4284542, # PogU (ForsenCDPogU)
4284220, # Z-ANESaber
4284904, # xScaramouche
4285547, # Avexus
4285738, # Lekrkoekj
113133 # Cush
4287112, # Jabob
4293090, # Kassi
4294118, # Spinvvy
4326041, # minsiii
4330286, # VoltageO
]
prompt = f"Enter mapper IDs (Default: {default_mapper_ids}): "
mapper_ids = input(prompt).strip() or ",".join(map(str, default_mapper_ids))