from typing import List, Optional, Dict, Any, Tuple, Set from bemani.common import ( APIConstants, GameConstants, VersionConstants, DBConstants, Parallel, ) from bemani.data.interfaces import APIProviderInterface from bemani.data.api.base import BaseGlobalData from bemani.data.mysql.user import UserData from bemani.data.mysql.music import MusicData from bemani.data.remoteuser import RemoteUser from bemani.data.types import UserID, Score, Song class GlobalMusicData(BaseGlobalData): def __init__( self, api: APIProviderInterface, user: UserData, music: MusicData ) -> None: super().__init__(api) self.user = user self.music = music def __get_cardids(self, userid: UserID) -> List[str]: if RemoteUser.is_remote(userid): return [RemoteUser.userid_to_card(userid)] else: return self.user.get_cards(userid) def __min(self, int1: int, int2: int) -> int: # -1 is used as a 'no value' so it should not overwrite a 0 if int1 == -1: return int2 if int2 == -1: return int1 return min(int1, int2) def __max(self, int1: int, int2: int) -> int: return max(int1, int2) def __format_ddr_score( self, version: int, songid: int, songchart: int, data: Dict[str, Any] ) -> Score: halo = { "none": DBConstants.DDR_HALO_NONE, "gfc": DBConstants.DDR_HALO_GOOD_FULL_COMBO, "fc": DBConstants.DDR_HALO_GREAT_FULL_COMBO, "pfc": DBConstants.DDR_HALO_PERFECT_FULL_COMBO, "mfc": DBConstants.DDR_HALO_MARVELOUS_FULL_COMBO, }.get(data.get("halo"), DBConstants.DDR_HALO_NONE) rank = { "AAA": DBConstants.DDR_RANK_AAA, "AA+": DBConstants.DDR_RANK_AA_PLUS, "AA": DBConstants.DDR_RANK_AA, "AA-": DBConstants.DDR_RANK_AA_MINUS, "A+": DBConstants.DDR_RANK_A_PLUS, "A": DBConstants.DDR_RANK_A, "A-": DBConstants.DDR_RANK_A_MINUS, "B+": DBConstants.DDR_RANK_B_PLUS, "B": DBConstants.DDR_RANK_B, "B-": DBConstants.DDR_RANK_B_MINUS, "C+": DBConstants.DDR_RANK_C_PLUS, "C": DBConstants.DDR_RANK_C, "C-": DBConstants.DDR_RANK_C_MINUS, "D+": DBConstants.DDR_RANK_D_PLUS, "D": DBConstants.DDR_RANK_D, "E": DBConstants.DDR_RANK_E, }.get(data.get("rank"), DBConstants.DDR_RANK_E) ghost = "" trace: List[int] = [] if version == VersionConstants.DDR_ACE: # DDR Ace is specia ghost = "".join([str(x) for x in data.get("ghost", [])]) else: trace = [int(x) for x in data.get("ghost", [])] return Score( -1, songid, songchart, int(data.get("points", 0)), int(data.get("timestamp", -1)), self.__max(int(data.get("timestamp", -1)), int(data.get("updated", -1))), -1, # No location for remote play 1, # No play info for remote play { "combo": int(data.get("combo", -1)), "rank": rank, "halo": halo, "ghost": ghost, "trace": trace, }, ) def __format_iidx_score( self, version: int, songid: int, songchart: int, data: Dict[str, Any] ) -> Score: status = { "np": DBConstants.IIDX_CLEAR_STATUS_NO_PLAY, "failed": DBConstants.IIDX_CLEAR_STATUS_FAILED, "ac": DBConstants.IIDX_CLEAR_STATUS_ASSIST_CLEAR, "ec": DBConstants.IIDX_CLEAR_STATUS_EASY_CLEAR, "nc": DBConstants.IIDX_CLEAR_STATUS_CLEAR, "hc": DBConstants.IIDX_CLEAR_STATUS_HARD_CLEAR, "exhc": DBConstants.IIDX_CLEAR_STATUS_EX_HARD_CLEAR, "fc": DBConstants.IIDX_CLEAR_STATUS_FULL_COMBO, }.get(data.get("status"), DBConstants.IIDX_CLEAR_STATUS_NO_PLAY) return Score( -1, songid, songchart, int(data.get("points", 0)), int(data.get("timestamp", -1)), self.__max(int(data.get("timestamp", -1)), int(data.get("updated", -1))), -1, # No location for remote play 1, # No play info for remote play { "clear_status": status, "ghost": bytes([int(b) for b in data.get("ghost", [])]), "miss_count": int(data.get("miss", -1)), "pgreats": int(data.get("pgreat", -1)), "greats": int(data.get("great", -1)), }, ) def __format_jubeat_score( self, version: int, songid: int, songchart: int, data: Dict[str, Any] ) -> Score: status = { "failed": DBConstants.JUBEAT_PLAY_MEDAL_FAILED, "cleared": DBConstants.JUBEAT_PLAY_MEDAL_CLEARED, "nfc": DBConstants.JUBEAT_PLAY_MEDAL_NEARLY_FULL_COMBO, "fc": DBConstants.JUBEAT_PLAY_MEDAL_FULL_COMBO, "nec": DBConstants.JUBEAT_PLAY_MEDAL_NEARLY_EXCELLENT, "exc": DBConstants.JUBEAT_PLAY_MEDAL_EXCELLENT, }.get(data.get("status"), DBConstants.JUBEAT_PLAY_MEDAL_FAILED) return Score( -1, songid, songchart, int(data.get("points", 0)), int(data.get("timestamp", -1)), self.__max(int(data.get("timestamp", -1)), int(data.get("updated", -1))), -1, # No location for remote play 1, # No play info for remote play { "medal": status, "combo": int(data.get("combo", -1)), "ghost": [int(x) for x in data.get("ghost", [])], "music_rate": int(data.get("music_rate")), }, ) def __format_museca_score( self, version: int, songid: int, songchart: int, data: Dict[str, Any] ) -> Score: rank = { "death": DBConstants.MUSECA_GRADE_DEATH, "poor": DBConstants.MUSECA_GRADE_POOR, "mediocre": DBConstants.MUSECA_GRADE_MEDIOCRE, "good": DBConstants.MUSECA_GRADE_GOOD, "great": DBConstants.MUSECA_GRADE_GREAT, "excellent": DBConstants.MUSECA_GRADE_EXCELLENT, "superb": DBConstants.MUSECA_GRADE_SUPERB, "masterpiece": DBConstants.MUSECA_GRADE_MASTERPIECE, "perfect": DBConstants.MUSECA_GRADE_PERFECT, }.get(data.get("rank"), DBConstants.MUSECA_GRADE_DEATH) status = { "failed": DBConstants.MUSECA_CLEAR_TYPE_FAILED, "cleared": DBConstants.MUSECA_CLEAR_TYPE_CLEARED, "fc": DBConstants.MUSECA_CLEAR_TYPE_FULL_COMBO, }.get(data.get("status"), DBConstants.MUSECA_CLEAR_TYPE_FAILED) return Score( -1, songid, songchart, int(data.get("points", 0)), int(data.get("timestamp", -1)), self.__max(int(data.get("timestamp", -1)), int(data.get("updated", -1))), -1, # No location for remote play 1, # No play info for remote play { "grade": rank, "clear_type": status, "combo": int(data.get("combo", -1)), "stats": { "btn_rate": int(data.get("buttonrate", -1)), "long_rate": int(data.get("longrate", -1)), "vol_rate": int(data.get("volrate", -1)), }, }, ) def __format_popn_score( self, version: int, songid: int, songchart: int, data: Dict[str, Any] ) -> Score: status = { "cf": DBConstants.POPN_MUSIC_PLAY_MEDAL_CIRCLE_FAILED, "df": DBConstants.POPN_MUSIC_PLAY_MEDAL_DIAMOND_FAILED, "sf": DBConstants.POPN_MUSIC_PLAY_MEDAL_STAR_FAILED, "ec": DBConstants.POPN_MUSIC_PLAY_MEDAL_EASY_CLEAR, "cc": DBConstants.POPN_MUSIC_PLAY_MEDAL_CIRCLE_CLEARED, "dc": DBConstants.POPN_MUSIC_PLAY_MEDAL_DIAMOND_CLEARED, "sc": DBConstants.POPN_MUSIC_PLAY_MEDAL_STAR_CLEARED, "cfc": DBConstants.POPN_MUSIC_PLAY_MEDAL_CIRCLE_FULL_COMBO, "dfc": DBConstants.POPN_MUSIC_PLAY_MEDAL_DIAMOND_FULL_COMBO, "sfc": DBConstants.POPN_MUSIC_PLAY_MEDAL_STAR_FULL_COMBO, "p": DBConstants.POPN_MUSIC_PLAY_MEDAL_PERFECT, }.get(data.get("status"), DBConstants.POPN_MUSIC_PLAY_MEDAL_CIRCLE_FAILED) return Score( -1, songid, songchart, int(data.get("points", 0)), int(data.get("timestamp", -1)), self.__max(int(data.get("timestamp", -1)), int(data.get("updated", -1))), -1, # No location for remote play 1, # No play info for remote play { "medal": status, "combo": int(data.get("combo", -1)), }, ) def __format_reflec_score( self, version: int, songid: int, songchart: int, data: Dict[str, Any] ) -> Score: status = { "np": DBConstants.REFLEC_BEAT_CLEAR_TYPE_NO_PLAY, "failed": DBConstants.REFLEC_BEAT_CLEAR_TYPE_FAILED, "cleared": DBConstants.REFLEC_BEAT_CLEAR_TYPE_CLEARED, "hc": DBConstants.REFLEC_BEAT_CLEAR_TYPE_HARD_CLEARED, "shc": DBConstants.REFLEC_BEAT_CLEAR_TYPE_S_HARD_CLEARED, }.get(data.get("status"), DBConstants.REFLEC_BEAT_CLEAR_TYPE_NO_PLAY) halo = { "none": DBConstants.REFLEC_BEAT_COMBO_TYPE_NONE, "ac": DBConstants.REFLEC_BEAT_COMBO_TYPE_ALMOST_COMBO, "fc": DBConstants.REFLEC_BEAT_COMBO_TYPE_FULL_COMBO, "fcaj": DBConstants.REFLEC_BEAT_COMBO_TYPE_FULL_COMBO_ALL_JUST, }.get(data.get("halo"), DBConstants.REFLEC_BEAT_COMBO_TYPE_NONE) return Score( -1, songid, songchart, int(data.get("points", 0)), int(data.get("timestamp", -1)), self.__max(int(data.get("timestamp", -1)), int(data.get("updated", -1))), -1, # No location for remote play 1, # No play info for remote play { "achievement_rate": int(data.get("rate", -1)), "clear_type": status, "combo_type": halo, "miss_count": int(data.get("miss", -1)), "combo": int(data.get("combo", -1)), }, ) def __format_sdvx_score( self, version: int, songid: int, songchart: int, data: Dict[str, Any] ) -> Score: status = { "np": DBConstants.SDVX_CLEAR_TYPE_NO_PLAY, "failed": DBConstants.SDVX_CLEAR_TYPE_FAILED, "cleared": DBConstants.SDVX_CLEAR_TYPE_CLEAR, "hc": DBConstants.SDVX_CLEAR_TYPE_HARD_CLEAR, "uc": DBConstants.SDVX_CLEAR_TYPE_ULTIMATE_CHAIN, "puc": DBConstants.SDVX_CLEAR_TYPE_PERFECT_ULTIMATE_CHAIN, }.get(data.get("status"), DBConstants.SDVX_CLEAR_TYPE_NO_PLAY) rank = { "E": DBConstants.SDVX_GRADE_NO_PLAY, "D": DBConstants.SDVX_GRADE_D, "C": DBConstants.SDVX_GRADE_C, "B": DBConstants.SDVX_GRADE_B, "A": DBConstants.SDVX_GRADE_A, "A+": DBConstants.SDVX_GRADE_A_PLUS, "AA": DBConstants.SDVX_GRADE_AA, "AA+": DBConstants.SDVX_GRADE_AA_PLUS, "AAA": DBConstants.SDVX_GRADE_AAA, "AAA+": DBConstants.SDVX_GRADE_AAA_PLUS, "S": DBConstants.SDVX_GRADE_S, }.get(data.get("rank"), DBConstants.SDVX_GRADE_NO_PLAY) return Score( -1, songid, songchart, int(data.get("points", 0)), int(data.get("timestamp", -1)), self.__max(int(data.get("timestamp", -1)), int(data.get("updated", -1))), -1, # No location for remote play 1, # No play info for remote play { "grade": rank, "clear_type": status, "combo": int(data.get("combo", -1)), "stats": { "btn_rate": int(data.get("buttonrate", -1)), "long_rate": int(data.get("longrate", -1)), "vol_rate": int(data.get("volrate", -1)), }, }, ) def __format_score( self, game: GameConstants, version: int, songid: int, songchart: int, data: Dict[str, Any], ) -> Optional[Score]: if game == GameConstants.DDR: return self.__format_ddr_score(version, songid, songchart, data) if game == GameConstants.IIDX: return self.__format_iidx_score(version, songid, songchart, data) if game == GameConstants.JUBEAT: return self.__format_jubeat_score(version, songid, songchart, data) if game == GameConstants.MUSECA: return self.__format_museca_score(version, songid, songchart, data) if game == GameConstants.POPN_MUSIC: return self.__format_popn_score(version, songid, songchart, data) if game == GameConstants.REFLEC_BEAT: return self.__format_reflec_score(version, songid, songchart, data) if game == GameConstants.SDVX: return self.__format_sdvx_score(version, songid, songchart, data) return None def __merge_ddr_score( self, version: int, oldscore: Score, newscore: Score ) -> Score: return Score( -1, oldscore.id, oldscore.chart, self.__max(oldscore.points, newscore.points), self.__max(oldscore.timestamp, newscore.timestamp), self.__max( self.__max(oldscore.update, newscore.update), self.__max(oldscore.timestamp, newscore.timestamp), ), oldscore.location, # Always propagate location from local setup if possible oldscore.plays + newscore.plays, { "rank": self.__max(oldscore.data["rank"], newscore.data["rank"]), "halo": self.__max(oldscore.data["halo"], newscore.data["halo"]), "ghost": oldscore.data.get("ghost") if oldscore.points > newscore.points else newscore.data.get("ghost"), "trace": oldscore.data.get("trace") if oldscore.points > newscore.points else newscore.data.get("trace"), "combo": self.__max(oldscore.data["combo"], newscore.data["combo"]), }, ) def __merge_iidx_score( self, version: int, oldscore: Score, newscore: Score ) -> Score: return Score( -1, oldscore.id, oldscore.chart, self.__max(oldscore.points, newscore.points), self.__max(oldscore.timestamp, newscore.timestamp), self.__max( self.__max(oldscore.update, newscore.update), self.__max(oldscore.timestamp, newscore.timestamp), ), oldscore.location, # Always propagate location from local setup if possible oldscore.plays + newscore.plays, { "clear_status": self.__max( oldscore.data["clear_status"], newscore.data["clear_status"] ), "ghost": oldscore.data.get("ghost") if oldscore.points > newscore.points else newscore.data.get("ghost"), "miss_count": self.__min( oldscore.data.get_int("miss_count", -1), newscore.data.get_int("miss_count", -1), ), "pgreats": oldscore.data.get_int("pgreats", -1) if oldscore.points > newscore.points else newscore.data.get_int("pgreats", -1), "greats": oldscore.data.get_int("greats", -1) if oldscore.points > newscore.points else newscore.data.get_int("greats", -1), }, ) def __merge_jubeat_score( self, version: int, oldscore: Score, newscore: Score ) -> Score: rate = self.__max( oldscore.data.get("music_rate", -1), newscore.data.get("music_rate", -1) ) return Score( -1, oldscore.id, oldscore.chart, self.__max(oldscore.points, newscore.points), self.__max(oldscore.timestamp, newscore.timestamp), self.__max( self.__max(oldscore.update, newscore.update), self.__max(oldscore.timestamp, newscore.timestamp), ), oldscore.location, # Always propagate location from local setup if possible oldscore.plays + newscore.plays, { "ghost": oldscore.data.get("ghost") if oldscore.points > newscore.points else newscore.data.get("ghost"), "combo": self.__max(oldscore.data["combo"], newscore.data["combo"]), "medal": self.__max(oldscore.data["medal"], newscore.data["medal"]), # Conditionally include this if we have any info for it. **({"music_rate": rate} if rate >= 0 else {}), }, ) def __merge_museca_score( self, version: int, oldscore: Score, newscore: Score ) -> Score: return Score( -1, oldscore.id, oldscore.chart, self.__max(oldscore.points, newscore.points), self.__max(oldscore.timestamp, newscore.timestamp), self.__max( self.__max(oldscore.update, newscore.update), self.__max(oldscore.timestamp, newscore.timestamp), ), oldscore.location, # Always propagate location from local setup if possible oldscore.plays + newscore.plays, { "grade": self.__max(oldscore.data["grade"], newscore.data["grade"]), "clear_type": self.__max( oldscore.data["clear_type"], newscore.data["clear_type"] ), "combo": self.__max(oldscore.data["combo"], newscore.data["combo"]), "stats": oldscore.data["stats"] if oldscore.points > newscore.points else newscore.data["stats"], }, ) def __merge_popn_score( self, version: int, oldscore: Score, newscore: Score ) -> Score: return Score( -1, oldscore.id, oldscore.chart, self.__max(oldscore.points, newscore.points), self.__max(oldscore.timestamp, newscore.timestamp), self.__max( self.__max(oldscore.update, newscore.update), self.__max(oldscore.timestamp, newscore.timestamp), ), oldscore.location, # Always propagate location from local setup if possible oldscore.plays + newscore.plays, { "combo": self.__max(oldscore.data["combo"], newscore.data["combo"]), "medal": self.__max(oldscore.data["medal"], newscore.data["medal"]), }, ) def __merge_reflec_score( self, version: int, oldscore: Score, newscore: Score ) -> Score: return Score( -1, oldscore.id, oldscore.chart, self.__max(oldscore.points, newscore.points), self.__max(oldscore.timestamp, newscore.timestamp), self.__max( self.__max(oldscore.update, newscore.update), self.__max(oldscore.timestamp, newscore.timestamp), ), oldscore.location, # Always propagate location from local setup if possible oldscore.plays + newscore.plays, { "clear_type": self.__max( oldscore.data["clear_type"], newscore.data["clear_type"] ), "combo_type": self.__max( oldscore.data["combo_type"], newscore.data["combo_type"] ), "miss_count": self.__min( oldscore.data.get_int("miss_count", -1), newscore.data.get_int("miss_count", -1), ), "combo": self.__max(oldscore.data["combo"], newscore.data["combo"]), "achievement_rate": self.__max( oldscore.data["achievement_rate"], newscore.data["achievement_rate"] ), }, ) def __merge_sdvx_score( self, version: int, oldscore: Score, newscore: Score ) -> Score: return Score( -1, oldscore.id, oldscore.chart, self.__max(oldscore.points, newscore.points), self.__max(oldscore.timestamp, newscore.timestamp), self.__max( self.__max(oldscore.update, newscore.update), self.__max(oldscore.timestamp, newscore.timestamp), ), oldscore.location, # Always propagate location from local setup if possible oldscore.plays + newscore.plays, { "grade": self.__max(oldscore.data["grade"], newscore.data["grade"]), "clear_type": self.__max( oldscore.data["clear_type"], newscore.data["clear_type"] ), "combo": self.__max( oldscore.data.get_int("combo", 1), newscore.data.get_int("combo", -1), ), "stats": oldscore.data["stats"] if oldscore.points > newscore.points else newscore.data["stats"], }, ) def __merge_score( self, game: GameConstants, version: int, oldscore: Score, newscore: Score ) -> Score: if oldscore.id != newscore.id or oldscore.chart != newscore.chart: raise Exception( "Logic error! Tried to merge scores from different song/charts!" ) if game == GameConstants.DDR: return self.__merge_ddr_score(version, oldscore, newscore) if game == GameConstants.IIDX: return self.__merge_iidx_score(version, oldscore, newscore) if game == GameConstants.JUBEAT: return self.__merge_jubeat_score(version, oldscore, newscore) if game == GameConstants.MUSECA: return self.__merge_museca_score(version, oldscore, newscore) if game == GameConstants.POPN_MUSIC: return self.__merge_popn_score(version, oldscore, newscore) if game == GameConstants.REFLEC_BEAT: return self.__merge_reflec_score(version, oldscore, newscore) if game == GameConstants.SDVX: return self.__merge_sdvx_score(version, oldscore, newscore) return oldscore def get_score( self, game: GameConstants, version: int, userid: UserID, songid: int, songchart: int, ) -> Optional[Score]: # Helper function so we can iterate over all servers for a single card def get_scores_for_card(cardid: str) -> List[Score]: return Parallel.flatten( Parallel.call( [client.get_records for client in self.clients], game, version, APIConstants.ID_TYPE_INSTANCE, [songid, songchart, cardid], ) ) relevant_cards = self.__get_cardids(userid) if RemoteUser.is_remote(userid): # No need to look up local score for this user scores = Parallel.flatten( Parallel.map( get_scores_for_card, relevant_cards, ) ) localscore = None else: localscore, scores = Parallel.execute( [ lambda: self.music.get_score( game, version, userid, songid, songchart ), lambda: Parallel.flatten( Parallel.map( get_scores_for_card, relevant_cards, ) ), ] ) topscore = localscore for score in scores: if int(score["song"]) != songid: continue if int(score["chart"]) != songchart: continue newscore = self.__format_score(game, version, songid, songchart, score) if topscore is None: # No merging needed topscore = newscore continue topscore = self.__merge_score(game, version, topscore, newscore) return topscore def get_scores( self, game: GameConstants, version: int, userid: UserID, since: Optional[int] = None, until: Optional[int] = None, ) -> List[Score]: relevant_cards = self.__get_cardids(userid) if RemoteUser.is_remote(userid): # No need to look up local score for this user scores = Parallel.flatten( Parallel.call( [client.get_records for client in self.clients], game, version, APIConstants.ID_TYPE_CARD, relevant_cards, since, until, ) ) localscores: List[Score] = [] else: localscores, scores = Parallel.execute( [ lambda: self.music.get_scores(game, version, userid, since, until), lambda: Parallel.flatten( Parallel.call( [client.get_records for client in self.clients], game, version, APIConstants.ID_TYPE_CARD, relevant_cards, since, until, ) ), ] ) allscores: Dict[int, Dict[int, Score]] = {} def add_score(score: Score) -> None: if score.id not in allscores: allscores[score.id] = {} allscores[score.id][score.chart] = score def get_score(songid: int, songchart: int) -> Optional[Score]: return allscores.get(songid, {}).get(songchart) # First, seed with local scores for score in localscores: add_score(score) # Second, merge in remote scorse for remotescore in scores: songid = int(remotescore["song"]) chart = int(remotescore["chart"]) newscore = self.__format_score(game, version, songid, chart, remotescore) oldscore = get_score(songid, chart) if oldscore is None: add_score(newscore) else: add_score(self.__merge_score(game, version, oldscore, newscore)) # Finally, flatten and return finalscores: List[Score] = [] for songid in allscores: for chart in allscores[songid]: finalscores.append(allscores[songid][chart]) return finalscores def __merge_global_scores( self, game: GameConstants, version: int, localcards: List[Tuple[str, UserID]], localscores: List[Tuple[UserID, Score]], remotescores: List[Dict[str, Any]], ) -> List[Tuple[UserID, Score]]: card_to_id = {cardid: userid for (cardid, userid) in localcards} allscores: Dict[UserID, Dict[int, Dict[int, Score]]] = {} def add_score(userid: UserID, score: Score) -> None: if userid not in allscores: allscores[userid] = {} if score.id not in allscores[userid]: allscores[userid][score.id] = {} allscores[userid][score.id][score.chart] = score def get_score(userid: UserID, songid: int, songchart: int) -> Optional[Score]: return allscores.get(userid, {}).get(songid, {}).get(songchart) # First, seed with local scores for userid, score in localscores: add_score(userid, score) # Second, merge in remote scorse for remotescore in remotescores: # Figure out the userid of this score cardids = sorted([card.upper() for card in remotescore.get("cards", [])]) if len(cardids) == 0: continue for cardid in cardids: if cardid in card_to_id: userid = card_to_id[cardid] break else: userid = RemoteUser.card_to_userid(cardids[0]) songid = int(remotescore["song"]) chart = int(remotescore["chart"]) newscore = self.__format_score(game, version, songid, chart, remotescore) oldscore = get_score(userid, songid, chart) if oldscore is None: add_score(userid, newscore) else: add_score(userid, self.__merge_score(game, version, oldscore, newscore)) # Finally, flatten and return finalscores: List[Tuple[UserID, Score]] = [] for userid in allscores: for songid in allscores[userid]: for chart in allscores[userid][songid]: finalscores.append((userid, allscores[userid][songid][chart])) return finalscores def get_all_scores( self, game: GameConstants, version: Optional[int] = None, userid: Optional[UserID] = None, songid: Optional[int] = None, songchart: Optional[int] = None, since: Optional[int] = None, until: Optional[int] = None, ) -> List[Tuple[UserID, Score]]: # First, pass off to local-only if this was called with parameters we don't support if version is None or userid is not None or songid is None: return self.music.get_all_scores( game, version, userid, songid, songchart, since, until ) # Now, figure out the request key based on passed in parameters if songchart is None: songkey = [songid] else: songkey = [songid, songchart] # Now, fetch all the scores remotely and locally localcards, localscores, remotescores = Parallel.execute( [ self.user.get_all_cards, lambda: self.music.get_all_scores( game, version, userid, songid, songchart, since, until ), lambda: Parallel.flatten( Parallel.call( [client.get_records for client in self.clients], game, version, APIConstants.ID_TYPE_SONG, songkey, since, until, ) ), ] ) return self.__merge_global_scores( game, version, localcards, localscores, remotescores ) def __merge_global_records( self, game: GameConstants, version: int, localcards: List[Tuple[str, UserID]], localscores: List[Tuple[UserID, Score]], remotescores: List[Dict[str, Any]], ) -> List[Tuple[UserID, Score]]: card_to_id = {cardid: userid for (cardid, userid) in localcards} allscores: Dict[int, Dict[int, Tuple[UserID, Score]]] = {} def add_score(userid: UserID, score: Score) -> None: if score.id not in allscores: allscores[score.id] = {} allscores[score.id][score.chart] = (userid, score) def get_score( songid: int, songchart: int ) -> Tuple[Optional[UserID], Optional[Score]]: return allscores.get(songid, {}).get(songchart, (None, None)) # First, seed with local records for userid, score in localscores: add_score(userid, score) # Second, merge in remote records for remotescore in remotescores: # Figure out the userid of this score cardids = sorted([card.upper() for card in remotescore.get("cards", [])]) if len(cardids) == 0: continue for cardid in cardids: if cardid in card_to_id: userid = card_to_id[cardid] break else: userid = RemoteUser.card_to_userid(cardids[0]) songid = int(remotescore["song"]) chart = int(remotescore["chart"]) newscore = self.__format_score(game, version, songid, chart, remotescore) oldid, oldscore = get_score(songid, chart) if oldscore is None: add_score(userid, newscore) else: # if IDs are the same then we should merge them if oldid == userid: add_score( userid, self.__merge_score(game, version, oldscore, newscore) ) else: # if the IDs are different we need to check which score actually belongs if newscore.points > oldscore.points: add_score(userid, newscore) # Finally, flatten and return finalscores: List[Tuple[UserID, Score]] = [] for songid in allscores: for chart in allscores[songid]: finalscores.append( (allscores[songid][chart][0], allscores[songid][chart][1]) ) return finalscores def get_all_records( self, game: GameConstants, version: Optional[int] = None, userlist: Optional[List[UserID]] = None, locationlist: Optional[List[int]] = None, ) -> List[Tuple[UserID, Score]]: # First, pass off to local-only if this was called with parameters we don't support if version is None or userlist is not None or locationlist is not None: return self.music.get_all_records(game, version, userlist, locationlist) # Now, fetch all records remotely and locally localcards, localscores, remotescores = Parallel.execute( [ self.user.get_all_cards, lambda: self.music.get_all_records( game, version, userlist, locationlist ), lambda: Parallel.flatten( Parallel.call( [client.get_records for client in self.clients], game, version, APIConstants.ID_TYPE_SERVER, [], ) ), ] ) return self.__merge_global_records( game, version, localcards, localscores, remotescores ) def get_clear_rates( self, game: GameConstants, version: int, songid: Optional[int] = None, songchart: Optional[int] = None, ) -> Dict[int, Dict[int, Dict[str, int]]]: """ Given an optional songid, or optional songid and songchart, looks up clear rates in remote servers that are connected to us. If neither id or chart is given, looks up global clear rates. If songid is given, looks up clear rates for each chart for the song. If songid and chart is given, looks up clear rates for that song/chart. Returns a dictionary keyed by songid, whos values are a dictionary keyed by chart, whos values are a dictionary containing integer counts keyed by 'plays', 'clears', and 'combos'. An example is as follows: { musicid: { chart: { plays: total plays, clears: total clears, combos: total full combos, }, }, } """ if songid is None and songchart is None: statistics = Parallel.flatten( Parallel.call( [client.get_statistics for client in self.clients], game, version, APIConstants.ID_TYPE_SERVER, [], ) ) elif songid is not None: if songchart is None: ids = [songid] else: ids = [songid, songchart] statistics = Parallel.flatten( Parallel.call( [client.get_statistics for client in self.clients], game, version, APIConstants.ID_TYPE_SONG, ids, ) ) else: statistics = [] retval: Dict[int, Dict[int, Dict[str, int]]] = {} for stat in statistics: songid = stat.get("song") songchart = stat.get("chart") if songid is None or songchart is None: continue songid = int(songid) songchart = int(songchart) if songid not in retval: retval[songid] = {} if songchart not in retval[songid]: retval[songid][songchart] = { "plays": 0, "clears": 0, "combos": 0, } def get_val(v: str) -> int: out = stat.get(v, -1) if out < 0: out = 0 return out retval[songid][songchart]["plays"] += get_val("plays") retval[songid][songchart]["clears"] += get_val("clears") retval[songid][songchart]["combos"] += get_val("combos") return retval def __format_ddr_song( self, version: int, songid: int, songchart: int, name: Optional[str], artist: Optional[str], genre: Optional[str], data: Dict[str, Any], ) -> Song: return Song( game=GameConstants.DDR, version=version, songid=songid, songchart=songchart, name=name, artist=artist, genre=genre, data={ "groove": { "air": int(data["groove"]["air"]), "chaos": int(data["groove"]["chaos"]), "freeze": int(data["groove"]["freeze"]), "stream": int(data["groove"]["stream"]), "voltage": int(data["groove"]["voltage"]), }, "bpm_min": int(data["bpm_min"]), "bpm_max": int(data["bpm_max"]), "category": int(data["category"]), "difficulty": int(data["difficulty"]), "edit_id": int(data["editid"]), }, ) def __format_iidx_song( self, version: int, songid: int, songchart: int, name: Optional[str], artist: Optional[str], genre: Optional[str], data: Dict[str, Any], ) -> Song: return Song( game=GameConstants.IIDX, version=version, songid=songid, songchart=songchart, name=name, artist=artist, genre=genre, data={ "bpm_min": int(data["bpm_min"]), "bpm_max": int(data["bpm_max"]), "notecount": int(data["notecount"]), "difficulty": int(data["difficulty"]), }, ) def __format_jubeat_song( self, version: int, songid: int, songchart: int, name: Optional[str], artist: Optional[str], genre: Optional[str], data: Dict[str, Any], ) -> Song: defaultcategory = { 1: VersionConstants.JUBEAT, 2: VersionConstants.JUBEAT_RIPPLES, 3: VersionConstants.JUBEAT_KNIT, 4: VersionConstants.JUBEAT_COPIOUS, 5: VersionConstants.JUBEAT_SAUCER, 6: VersionConstants.JUBEAT_PROP, 7: VersionConstants.JUBEAT_QUBELL, 8: VersionConstants.JUBEAT_CLAN, 9: VersionConstants.JUBEAT_FESTO, }.get(int(songid / 10000000), VersionConstants.JUBEAT) # Map the category to the version numbers defined on BEMAPI. categorymapping = { "1": VersionConstants.JUBEAT, "2": VersionConstants.JUBEAT_RIPPLES, "2a": VersionConstants.JUBEAT_RIPPLES_APPEND, "3": VersionConstants.JUBEAT_KNIT, "3a": VersionConstants.JUBEAT_KNIT_APPEND, "4": VersionConstants.JUBEAT_COPIOUS, "4a": VersionConstants.JUBEAT_COPIOUS_APPEND, "5": VersionConstants.JUBEAT_SAUCER, "5a": VersionConstants.JUBEAT_SAUCER_FULFILL, "6": VersionConstants.JUBEAT_PROP, "7": VersionConstants.JUBEAT_QUBELL, "8": VersionConstants.JUBEAT_CLAN, "9": VersionConstants.JUBEAT_FESTO, } return Song( game=GameConstants.JUBEAT, version=version, songid=songid, songchart=songchart, name=name, artist=artist, genre=genre, data={ "bpm_min": int(data["bpm_min"]), "bpm_max": int(data["bpm_max"]), "difficulty": int(data["difficulty"]), "version": categorymapping.get( data.get("category", "0"), defaultcategory ), }, ) def __format_museca_song( self, version: int, songid: int, songchart: int, name: Optional[str], artist: Optional[str], genre: Optional[str], data: Dict[str, Any], ) -> Song: return Song( game=GameConstants.MUSECA, version=version, songid=songid, songchart=songchart, name=name, artist=artist, genre=genre, data={ "bpm_min": int(data["bpm_min"]), "bpm_max": int(data["bpm_max"]), "limited": int(data["limited"]), "difficulty": int(data["difficulty"]), }, ) def __format_popn_song( self, version: int, songid: int, songchart: int, name: Optional[str], artist: Optional[str], genre: Optional[str], data: Dict[str, Any], ) -> Song: return Song( game=GameConstants.POPN_MUSIC, version=version, songid=songid, songchart=songchart, name=name, artist=artist, genre=genre, data={ "difficulty": int(data["difficulty"]), "category": str(data["category"]), }, ) def __format_reflec_song( self, version: int, songid: int, songchart: int, name: Optional[str], artist: Optional[str], genre: Optional[str], data: Dict[str, Any], ) -> Song: return Song( game=GameConstants.REFLEC_BEAT, version=version, songid=songid, songchart=songchart, name=name, artist=artist, genre=genre, data={ "difficulty": int(data["difficulty"]), "folder": int(data["category"]), "chart_id": str(data["musicid"]), }, ) def __format_sdvx_song( self, version: int, songid: int, songchart: int, name: Optional[str], artist: Optional[str], genre: Optional[str], data: Dict[str, Any], ) -> Song: return Song( game=GameConstants.SDVX, version=version, songid=songid, songchart=songchart, name=name, artist=artist, genre=genre, data={ "bpm_min": int(data["bpm_min"]), "bpm_max": int(data["bpm_max"]), "limited": int(data["limited"]), "difficulty": int(data["difficulty"]), }, ) def __format_song( self, game: GameConstants, version: int, songid: int, songchart: int, name: Optional[str], artist: Optional[str], genre: Optional[str], data: Dict[str, Any], ) -> Optional[Song]: if game == GameConstants.DDR: return self.__format_ddr_song( version, songid, songchart, name, artist, genre, data ) if game == GameConstants.IIDX: return self.__format_iidx_song( version, songid, songchart, name, artist, genre, data ) if game == GameConstants.JUBEAT: return self.__format_jubeat_song( version, songid, songchart, name, artist, genre, data ) if game == GameConstants.MUSECA: return self.__format_museca_song( version, songid, songchart, name, artist, genre, data ) if game == GameConstants.POPN_MUSIC: return self.__format_popn_song( version, songid, songchart, name, artist, genre, data ) if game == GameConstants.REFLEC_BEAT: return self.__format_reflec_song( version, songid, songchart, name, artist, genre, data ) if game == GameConstants.SDVX: return self.__format_sdvx_song( version, songid, songchart, name, artist, genre, data ) return None def get_all_songs( self, game: GameConstants, version: Optional[int] = None, ) -> List[Song]: """ Given a game and a version, look up all song/chart combos associated with that game. Parameters: game - Enum value representing a game series. version - Integer representing which version of the game. Returns: A list of Song objects detailing the song information for each song. """ if version is None: # We could do a ton of work to support this by iterating over all versions # and combining, but this isn't going to be used in that manner, so lets # skip that for now. return [] catalogs: List[Dict[str, List[Dict[str, Any]]]] = Parallel.call( [client.get_catalog for client in self.clients], game, version ) retval: List[Song] = [] seen: Set[str] = set() for catalog in catalogs: for entry in catalog.get("songs", []): song = self.__format_song( game, version, int(entry["song"]), int(entry["chart"]), str(entry["title"] if entry["title"] is not None else "") or None, str(entry["artist"] if entry["artist"] is not None else "") or None, str(entry["genre"] if entry["genre"] is not None else "") or None, entry, ) if song is None: continue key = f"{song.id}_{song.chart}" if key in seen: continue retval.append(song) seen.add(key) return retval