Yarninator/bgg_api
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Created statistic cache

Browse source
This commit is contained in:
Yarne Coppens 2024-08-25 10:49:54 +02:00
parent 0b412b8951
commit 93c5e65af5
2 changed files with 157 additions and 32 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
src/main.py
View file

@ -1,4 +1,4 @@
from typing import Union
from typing import Union, Dict
from pydantic import BaseModel
from sqlmodel import Session
from threading import Thread
@ -25,6 +25,7 @@ def refresh_data():
data_connection.delete_database()
data_connection.create_db_and_tables()
with Session(data_connection.get_db_engine()) as session:
statistic_creator.clear_cache()
data_connection.get_user_collection(session)
data_connection.get_user_owned_collection(session)
data_connection.get_user_wishlist_collection(session)
@ -219,8 +220,8 @@ def get_winrate(player_name: str | None = None, session: Session = Depends(get_s
return statistic_to_return
@app.get('/statistics/winrate_over_time', response_model=statistic_classes.TimeLineStatistic)
def get_winrate_over_time(player_name: str, day_step: int = 1, session: Session=Depends(get_session)):
@app.get('/statistics/winrate_over_time', response_model=Union[statistic_classes.TimeLineStatistic, Dict[str,statistic_classes.TimeLineStatistic]])
def get_winrate_over_time(player_name: str | None = None, day_step: int = 1, session: Session=Depends(get_session)):
statistic_to_return = statistic_creator.get_winrate_over_time(session, player_name, day_step)
return statistic_to_return

182
src/modules/statistic_creator.py
View file

@ -1,14 +1,44 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Union, Dict
if TYPE_CHECKING:
from src.main import BoardgameFilterParams, PlayFilterParams
import hashlib
from src.classes import statistic_classes
from src.modules import data_connection
from datetime import date, timedelta, datetime
from sqlmodel import Session
cached_statistics = {}
def clear_cache():
global cached_statistics
cached_statistics = {}
def get_from_cache(argument_dict: dict):
#Session is random for each request
if 'session' in argument_dict:
del argument_dict['session']
md5hash = hashlib.md5(str(argument_dict).encode('utf-8')).hexdigest()
if md5hash in cached_statistics:
return md5hash, cached_statistics[md5hash]
else:
return md5hash, None
def get_total_owned_games(session: Session, filtering_query: BoardgameFilterParams = None) -> statistic_classes.NumberStatistic:
statistic_name = 'Amount of games in owned collection'
md5hash, cached_statistic = get_from_cache({**locals()})
if cached_statistic != None:
return cached_statistic
owned_collection = data_connection.get_user_owned_collection(session)
if filtering_query != None:
@ -17,16 +47,27 @@ def get_total_owned_games(session: Session, filtering_query: BoardgameFilterPara
total_owned_games = len(owned_collection)
statistic_dict = {
"name":"Amount of games in owned collection",
"name":statistic_name,
"result":total_owned_games
}
statistic_to_return = statistic_classes.NumberStatistic.model_validate(statistic_dict)
cached_statistics[md5hash] = statistic_to_return
return statistic_to_return
def get_total_owned_collection_cost(session: Session, filtering_query: BoardgameFilterParams = None) -> statistic_classes.NumberStatistic:
statistic_name = 'Total cost of the owned collection'
md5hash, cached_statistic = get_from_cache({**locals()})
if cached_statistic != None:
return cached_statistic
owned_collection = data_connection.get_user_owned_collection(session)
if filtering_query != None:
@ -36,16 +77,27 @@ def get_total_owned_collection_cost(session: Session, filtering_query: Boardgame
total_cost = sum([boardgame.owned_info.price_paid for boardgame in owned_collection])
statistic_dict = {
"name":"Total cost of the owned collection",
"name":statistic_name,
"result":total_cost
}
statistic_to_return = statistic_classes.NumberStatistic.model_validate(statistic_dict)
cached_statistics[md5hash] = statistic_to_return
return statistic_to_return
def get_amount_of_games_over_time(session: Session, filtering_query: BoardgameFilterParams = None, day_step: int = 1) -> statistic_classes.TimeLineStatistic:
statistic_name = 'Amount of games in owned collection over time'
md5hash, cached_statistic = get_from_cache({**locals()})
if cached_statistic != None:
return cached_statistic
def daterange(start_date: date, end_date: date, day_step):
days = int((end_date - start_date).days)
for n in range(0, days, day_step):
@ -65,16 +117,27 @@ def get_amount_of_games_over_time(session: Session, filtering_query: BoardgameFi
timeline_dict[current_date] = len(games_in_collection_at_date)
statistic_dict = {
"name":"Amount of games in owned collection over time",
"name":statistic_name,
"result":timeline_dict
}
statistic_to_return = statistic_classes.TimeLineStatistic.model_validate(statistic_dict)
cached_statistics[md5hash] = statistic_to_return
return statistic_to_return
def get_amount_of_games_played_per_year(session: Session, filtering_query: PlayFilterParams = None) -> statistic_classes.TimeLineStatistic:
statistic_name = 'Amount of games played per year'
md5hash, cached_statistic = get_from_cache({**locals()})
if cached_statistic != None:
return cached_statistic
all_plays = data_connection.get_plays(session)
all_plays.sort(key= lambda x: x.play_date)
@ -99,15 +162,26 @@ def get_amount_of_games_played_per_year(session: Session, filtering_query: PlayF
statistic_dict = {
"name":"Amount of games played per year",
"name":statistic_name,
"result":years_plays_dict
}
statistic_to_return = statistic_classes.TimeLineStatistic.model_validate(statistic_dict)
cached_statistics[md5hash] = statistic_to_return
return statistic_to_return
def get_most_expensive_games(session: Session, filtering_query: BoardgameFilterParams = None, top_amount: int = 10) -> statistic_classes.GamesStatistic:
statistic_name = 'Most expensive games'
md5hash, cached_statistic = get_from_cache({**locals()})
if cached_statistic != None:
return cached_statistic
most_expensive_games = data_connection.get_user_owned_collection(session)
most_expensive_games = filtering_query.do_filtering(most_expensive_games)
@ -117,16 +191,27 @@ def get_most_expensive_games(session: Session, filtering_query: BoardgameFilterP
most_expensive_games = most_expensive_games[0:top_amount]
statistic_dict = {
"name":"Most expensive games",
"name":statistic_name,
"result":most_expensive_games
}
statistic_to_return = statistic_classes.GamesStatistic.model_validate(statistic_dict)
cached_statistics[md5hash] = statistic_to_return
return statistic_to_return
def get_shelf_of_shame(session: Session, filtering_query: BoardgameFilterParams = None) -> statistic_classes.GamesStatistic:
statistic_name = "Shelf of Shame"
md5hash, cached_statistic = get_from_cache({**locals()})
if cached_statistic != None:
return cached_statistic
boardgames_in_collection = data_connection.get_user_collection(session)
owned_boardgames = data_connection.get_user_owned_collection(session)
@ -145,24 +230,33 @@ def get_shelf_of_shame(session: Session, filtering_query: BoardgameFilterParams
owned_boardgames_no_plays.sort(key=lambda x: x.name)
statistic_dict = {
"name":"Shelf of Shame",
"name":statistic_name,
"result":owned_boardgames_no_plays
}
statistic_to_return = statistic_classes.GamesStatistic.model_validate(statistic_dict)
cached_statistics[md5hash] = statistic_to_return
return statistic_to_return
def get_winrate(session: Session, player_name: str | None = None):
statistic_name = 'Player winrate'
md5hash, cached_statistic = get_from_cache({**locals()})
if cached_statistic != None:
return cached_statistic
if player_name == None:
players_to_calculate = data_connection.get_all_players(session)
else:
players_to_calculate = [data_connection.get_player(player_name.title(), session)]
statistic_dict = {
'name': 'Player winrate',
'name': statistic_name,
'result': {}
}
@ -177,43 +271,73 @@ def get_winrate(session: Session, player_name: str | None = None):
statistic_to_return = statistic_classes.PlayerStatistic.model_validate(statistic_dict)
cached_statistics[md5hash] = statistic_to_return
return statistic_to_return
def get_winrate_over_time(session: Session, player_name: str, day_step = 1):
def get_winrate_over_time(session: Session, player_name: str | None = None, day_step = 1) -> Dict[str,statistic_classes.TimeLineStatistic]:
statistic_name = 'Player winrate over time'
md5hash, cached_statistic = get_from_cache({**locals()})
if cached_statistic != None:
return cached_statistic
def daterange(start_date: date, end_date: date, day_step):
days = int((end_date - start_date).days)
for n in range(0, days, day_step):
yield start_date + timedelta(n)
if player_name == None:
wanted_players = data_connection.get_all_players(session)
else:
wanted_players = [data_connection.get_player(player_name, session)]
wanted_player = data_connection.get_player(player_name, session)
all_playplayers = [playplayer for playplayer in wanted_player.playplayers]
all_playplayers.sort(key=lambda x: x.play.play_date)
dict_to_return = {}
start_date = all_playplayers[0].play.play_date
start_date = datetime.today().date()
timeline_dict = {}
for wanted_player in wanted_players:
for current_date in daterange(start_date, date.today(), day_step):
playplayers_at_date = list(filter(lambda playplayer: playplayer.play.play_date <= current_date, all_playplayers))
total_games_played = len(playplayers_at_date)
total_games_won = 0
for playplayer in playplayers_at_date:
if playplayer.has_won:
total_games_won += 1
timeline_dict[current_date] = total_games_won / total_games_played
all_playplayers = [playplayer for playplayer in wanted_player.playplayers]
all_playplayers.sort(key=lambda x: x.play.play_date)
if start_date > all_playplayers[0].play.play_date:
start_date = all_playplayers[0].play.play_date
for wanted_player in wanted_players:
all_playplayers = [playplayer for playplayer in wanted_player.playplayers]
all_playplayers.sort(key=lambda x: x.play.play_date)
timeline_dict = {}
for current_date in daterange(start_date, date.today(), day_step):
playplayers_at_date = list(filter(lambda playplayer: playplayer.play.play_date <= current_date, all_playplayers))
total_games_played = len(playplayers_at_date)
total_games_won = 0
for playplayer in playplayers_at_date:
if playplayer.has_won:
total_games_won += 1
if total_games_played != 0:
timeline_dict[current_date] = total_games_won / total_games_played
else:
timeline_dict[current_date] = 0
statistic_dict = {
'name': 'Player winrate over time',
'result': timeline_dict
}
statistic_dict = {
'name': statistic_name,
'result': timeline_dict
}
statistic_to_return = statistic_classes.TimeLineStatistic.model_validate(statistic_dict)
statistic_to_return = statistic_classes.TimeLineStatistic.model_validate(statistic_dict)
return statistic_to_return
dict_to_return[wanted_player.name] = statistic_to_return
cached_statistics[md5hash] = dict_to_return
return dict_to_return