326 lines
13 KiB
Python
326 lines
13 KiB
Python
from typing import Any, Dict, List, Set, Tuple
|
|
|
|
from bemani.api.exceptions import APIException
|
|
from bemani.api.objects.base import BaseObject
|
|
from bemani.common import GameConstants, VersionConstants, APIConstants, DBConstants
|
|
from bemani.data import Score, UserID
|
|
|
|
|
|
class RecordsObject(BaseObject):
|
|
def __format_ddr_record(self, record: Score) -> Dict[str, Any]:
|
|
halo = {
|
|
DBConstants.DDR_HALO_NONE: "none",
|
|
DBConstants.DDR_HALO_GOOD_FULL_COMBO: "gfc",
|
|
DBConstants.DDR_HALO_GREAT_FULL_COMBO: "fc",
|
|
DBConstants.DDR_HALO_PERFECT_FULL_COMBO: "pfc",
|
|
DBConstants.DDR_HALO_MARVELOUS_FULL_COMBO: "mfc",
|
|
}.get(record.data.get_int("halo"), "none")
|
|
rank = {
|
|
DBConstants.DDR_RANK_AAA: "AAA",
|
|
DBConstants.DDR_RANK_AA_PLUS: "AA+",
|
|
DBConstants.DDR_RANK_AA: "AA",
|
|
DBConstants.DDR_RANK_AA_MINUS: "AA-",
|
|
DBConstants.DDR_RANK_A_PLUS: "A+",
|
|
DBConstants.DDR_RANK_A: "A",
|
|
DBConstants.DDR_RANK_A_MINUS: "A-",
|
|
DBConstants.DDR_RANK_B_PLUS: "B+",
|
|
DBConstants.DDR_RANK_B: "B",
|
|
DBConstants.DDR_RANK_B_MINUS: "B-",
|
|
DBConstants.DDR_RANK_C_PLUS: "C+",
|
|
DBConstants.DDR_RANK_C: "C",
|
|
DBConstants.DDR_RANK_C_MINUS: "C-",
|
|
DBConstants.DDR_RANK_D_PLUS: "D+",
|
|
DBConstants.DDR_RANK_D: "D",
|
|
DBConstants.DDR_RANK_E: "E",
|
|
}.get(record.data.get_int("rank"), "E")
|
|
|
|
if self.version == VersionConstants.DDR_ACE:
|
|
# DDR Ace is specia
|
|
ghost = [int(x) for x in record.data.get_str("ghost")]
|
|
else:
|
|
if "trace" not in record.data:
|
|
ghost = []
|
|
else:
|
|
ghost = record.data.get_int_array("trace", len(record.data["trace"]))
|
|
|
|
return {
|
|
"rank": rank,
|
|
"halo": halo,
|
|
"combo": record.data.get_int("combo"),
|
|
"ghost": ghost,
|
|
}
|
|
|
|
def __format_iidx_record(self, record: Score) -> Dict[str, Any]:
|
|
status = {
|
|
DBConstants.IIDX_CLEAR_STATUS_NO_PLAY: "np",
|
|
DBConstants.IIDX_CLEAR_STATUS_FAILED: "failed",
|
|
DBConstants.IIDX_CLEAR_STATUS_ASSIST_CLEAR: "ac",
|
|
DBConstants.IIDX_CLEAR_STATUS_EASY_CLEAR: "ec",
|
|
DBConstants.IIDX_CLEAR_STATUS_CLEAR: "nc",
|
|
DBConstants.IIDX_CLEAR_STATUS_HARD_CLEAR: "hc",
|
|
DBConstants.IIDX_CLEAR_STATUS_EX_HARD_CLEAR: "exhc",
|
|
DBConstants.IIDX_CLEAR_STATUS_FULL_COMBO: "fc",
|
|
}.get(record.data.get_int("clear_status"), "np")
|
|
|
|
return {
|
|
"status": status,
|
|
"miss": record.data.get_int("miss_count", -1),
|
|
"ghost": [b for b in record.data.get_bytes("ghost")],
|
|
"pgreat": record.data.get_int("pgreats", -1),
|
|
"great": record.data.get_int("greats", -1),
|
|
}
|
|
|
|
def __format_jubeat_record(self, record: Score) -> Dict[str, Any]:
|
|
status = {
|
|
DBConstants.JUBEAT_PLAY_MEDAL_FAILED: "failed",
|
|
DBConstants.JUBEAT_PLAY_MEDAL_CLEARED: "cleared",
|
|
DBConstants.JUBEAT_PLAY_MEDAL_NEARLY_FULL_COMBO: "nfc",
|
|
DBConstants.JUBEAT_PLAY_MEDAL_FULL_COMBO: "fc",
|
|
DBConstants.JUBEAT_PLAY_MEDAL_NEARLY_EXCELLENT: "nec",
|
|
DBConstants.JUBEAT_PLAY_MEDAL_EXCELLENT: "exc",
|
|
}.get(record.data.get_int("medal"), "failed")
|
|
if "ghost" not in record.data:
|
|
ghost: List[int] = []
|
|
else:
|
|
ghost = record.data.get_int_array("ghost", len(record.data["ghost"]))
|
|
|
|
return {
|
|
"status": status,
|
|
"combo": record.data.get_int("combo", -1),
|
|
"ghost": ghost,
|
|
"music_rate": record.data.get_int("music_rate"),
|
|
}
|
|
|
|
def __format_museca_record(self, record: Score) -> Dict[str, Any]:
|
|
rank = {
|
|
DBConstants.MUSECA_GRADE_DEATH: "death",
|
|
DBConstants.MUSECA_GRADE_POOR: "poor",
|
|
DBConstants.MUSECA_GRADE_MEDIOCRE: "mediocre",
|
|
DBConstants.MUSECA_GRADE_GOOD: "good",
|
|
DBConstants.MUSECA_GRADE_GREAT: "great",
|
|
DBConstants.MUSECA_GRADE_EXCELLENT: "excellent",
|
|
DBConstants.MUSECA_GRADE_SUPERB: "superb",
|
|
DBConstants.MUSECA_GRADE_MASTERPIECE: "masterpiece",
|
|
DBConstants.MUSECA_GRADE_PERFECT: "perfect",
|
|
}.get(record.data.get_int("grade"), "death")
|
|
status = {
|
|
DBConstants.MUSECA_CLEAR_TYPE_FAILED: "failed",
|
|
DBConstants.MUSECA_CLEAR_TYPE_CLEARED: "cleared",
|
|
DBConstants.MUSECA_CLEAR_TYPE_FULL_COMBO: "fc",
|
|
}.get(record.data.get_int("clear_type"), "failed")
|
|
|
|
return {
|
|
"rank": rank,
|
|
"status": status,
|
|
"combo": record.data.get_int("combo", -1),
|
|
"buttonrate": record.data.get_dict("stats").get_int("btn_rate"),
|
|
"longrate": record.data.get_dict("stats").get_int("long_rate"),
|
|
"volrate": record.data.get_dict("stats").get_int("vol_rate"),
|
|
}
|
|
|
|
def __format_popn_record(self, record: Score) -> Dict[str, Any]:
|
|
status = {
|
|
DBConstants.POPN_MUSIC_PLAY_MEDAL_CIRCLE_FAILED: "cf",
|
|
DBConstants.POPN_MUSIC_PLAY_MEDAL_DIAMOND_FAILED: "df",
|
|
DBConstants.POPN_MUSIC_PLAY_MEDAL_STAR_FAILED: "sf",
|
|
DBConstants.POPN_MUSIC_PLAY_MEDAL_EASY_CLEAR: "ec",
|
|
DBConstants.POPN_MUSIC_PLAY_MEDAL_CIRCLE_CLEARED: "cc",
|
|
DBConstants.POPN_MUSIC_PLAY_MEDAL_DIAMOND_CLEARED: "dc",
|
|
DBConstants.POPN_MUSIC_PLAY_MEDAL_STAR_CLEARED: "sc",
|
|
DBConstants.POPN_MUSIC_PLAY_MEDAL_CIRCLE_FULL_COMBO: "cfc",
|
|
DBConstants.POPN_MUSIC_PLAY_MEDAL_DIAMOND_FULL_COMBO: "dfc",
|
|
DBConstants.POPN_MUSIC_PLAY_MEDAL_STAR_FULL_COMBO: "sfc",
|
|
DBConstants.POPN_MUSIC_PLAY_MEDAL_PERFECT: "p",
|
|
}.get(record.data.get_int("medal"), "cf")
|
|
|
|
return {
|
|
"status": status,
|
|
"combo": record.data.get_int("combo", -1),
|
|
}
|
|
|
|
def __format_reflec_record(self, record: Score) -> Dict[str, Any]:
|
|
status = {
|
|
DBConstants.REFLEC_BEAT_CLEAR_TYPE_NO_PLAY: "np",
|
|
DBConstants.REFLEC_BEAT_CLEAR_TYPE_FAILED: "failed",
|
|
DBConstants.REFLEC_BEAT_CLEAR_TYPE_CLEARED: "cleared",
|
|
DBConstants.REFLEC_BEAT_CLEAR_TYPE_HARD_CLEARED: "hc",
|
|
DBConstants.REFLEC_BEAT_CLEAR_TYPE_S_HARD_CLEARED: "shc",
|
|
}.get(record.data.get_int("clear_type"), "np")
|
|
halo = {
|
|
DBConstants.REFLEC_BEAT_COMBO_TYPE_NONE: "none",
|
|
DBConstants.REFLEC_BEAT_COMBO_TYPE_ALMOST_COMBO: "ac",
|
|
DBConstants.REFLEC_BEAT_COMBO_TYPE_FULL_COMBO: "fc",
|
|
DBConstants.REFLEC_BEAT_COMBO_TYPE_FULL_COMBO_ALL_JUST: "fcaj",
|
|
}.get(record.data.get_int("combo_type"), "none")
|
|
|
|
return {
|
|
"rate": record.data.get_int("achievement_rate"),
|
|
"status": status,
|
|
"halo": halo,
|
|
"combo": record.data.get_int("combo", -1),
|
|
"miss": record.data.get_int("miss_count", -1),
|
|
}
|
|
|
|
def __format_sdvx_record(self, record: Score) -> Dict[str, Any]:
|
|
status = {
|
|
DBConstants.SDVX_CLEAR_TYPE_NO_PLAY: "np",
|
|
DBConstants.SDVX_CLEAR_TYPE_FAILED: "failed",
|
|
DBConstants.SDVX_CLEAR_TYPE_CLEAR: "cleared",
|
|
DBConstants.SDVX_CLEAR_TYPE_HARD_CLEAR: "hc",
|
|
DBConstants.SDVX_CLEAR_TYPE_ULTIMATE_CHAIN: "uc",
|
|
DBConstants.SDVX_CLEAR_TYPE_PERFECT_ULTIMATE_CHAIN: "puc",
|
|
}.get(record.data.get_int("clear_type"), "np")
|
|
rank = {
|
|
DBConstants.SDVX_GRADE_NO_PLAY: "E",
|
|
DBConstants.SDVX_GRADE_D: "D",
|
|
DBConstants.SDVX_GRADE_C: "C",
|
|
DBConstants.SDVX_GRADE_B: "B",
|
|
DBConstants.SDVX_GRADE_A: "A",
|
|
DBConstants.SDVX_GRADE_A_PLUS: "A+",
|
|
DBConstants.SDVX_GRADE_AA: "AA",
|
|
DBConstants.SDVX_GRADE_AA_PLUS: "AA+",
|
|
DBConstants.SDVX_GRADE_AAA: "AAA",
|
|
DBConstants.SDVX_GRADE_AAA_PLUS: "AAA+",
|
|
DBConstants.SDVX_GRADE_S: "S",
|
|
}.get(record.data.get_int("grade"), "E")
|
|
|
|
return {
|
|
"status": status,
|
|
"rank": rank,
|
|
"combo": record.data.get_int("combo", -1),
|
|
"buttonrate": record.data.get_dict("stats").get_int("btn_rate"),
|
|
"longrate": record.data.get_dict("stats").get_int("long_rate"),
|
|
"volrate": record.data.get_dict("stats").get_int("vol_rate"),
|
|
}
|
|
|
|
def __format_record(self, cardids: List[str], record: Score) -> Dict[str, Any]:
|
|
base = {
|
|
"cards": cardids,
|
|
"song": str(record.id),
|
|
"chart": str(record.chart),
|
|
"points": record.points,
|
|
"timestamp": record.timestamp,
|
|
"updated": record.update,
|
|
}
|
|
|
|
if self.game == GameConstants.DDR:
|
|
base.update(self.__format_ddr_record(record))
|
|
if self.game == GameConstants.IIDX:
|
|
base.update(self.__format_iidx_record(record))
|
|
if self.game == GameConstants.JUBEAT:
|
|
base.update(self.__format_jubeat_record(record))
|
|
if self.game == GameConstants.MUSECA:
|
|
base.update(self.__format_museca_record(record))
|
|
if self.game == GameConstants.POPN_MUSIC:
|
|
base.update(self.__format_popn_record(record))
|
|
if self.game == GameConstants.REFLEC_BEAT:
|
|
base.update(self.__format_reflec_record(record))
|
|
if self.game == GameConstants.SDVX:
|
|
base.update(self.__format_sdvx_record(record))
|
|
|
|
return base
|
|
|
|
@property
|
|
def music_version(self) -> int:
|
|
if self.game in {GameConstants.IIDX, GameConstants.MUSECA}:
|
|
if self.omnimix:
|
|
return self.version + DBConstants.OMNIMIX_VERSION_BUMP
|
|
else:
|
|
return self.version
|
|
else:
|
|
return self.version
|
|
|
|
def fetch_v1(
|
|
self, idtype: APIConstants, ids: List[str], params: Dict[str, Any]
|
|
) -> List[Dict[str, Any]]:
|
|
since = params.get("since")
|
|
until = params.get("until")
|
|
|
|
# Fetch the scores
|
|
records: List[Tuple[UserID, Score]] = []
|
|
if idtype == APIConstants.ID_TYPE_SERVER:
|
|
# Because of the way this query works, we can't apply since/until to it directly.
|
|
# If we did, it would miss higher scores earned before since or after until, and
|
|
# incorrectly report records.
|
|
records.extend(
|
|
self.data.local.music.get_all_records(self.game, self.music_version)
|
|
)
|
|
elif idtype == APIConstants.ID_TYPE_SONG:
|
|
if len(ids) == 1:
|
|
songid = int(ids[0])
|
|
chart = None
|
|
else:
|
|
songid = int(ids[0])
|
|
chart = int(ids[1])
|
|
records.extend(
|
|
self.data.local.music.get_all_scores(
|
|
self.game,
|
|
self.music_version,
|
|
songid=songid,
|
|
songchart=chart,
|
|
since=since,
|
|
until=until,
|
|
)
|
|
)
|
|
elif idtype == APIConstants.ID_TYPE_INSTANCE:
|
|
songid = int(ids[0])
|
|
chart = int(ids[1])
|
|
cardid = ids[2]
|
|
userid = self.data.local.user.from_cardid(cardid)
|
|
if userid is not None:
|
|
score = self.data.local.music.get_score(
|
|
self.game, self.music_version, userid, songid, chart
|
|
)
|
|
if score is not None:
|
|
records.append((userid, score))
|
|
elif idtype == APIConstants.ID_TYPE_CARD:
|
|
users: Set[UserID] = set()
|
|
for cardid in ids:
|
|
userid = self.data.local.user.from_cardid(cardid)
|
|
if userid is not None:
|
|
# Don't duplicate loads for users with multiple card IDs if multiples
|
|
# of those IDs are requested.
|
|
if userid in users:
|
|
continue
|
|
users.add(userid)
|
|
|
|
records.extend(
|
|
[
|
|
(userid, score)
|
|
for score in self.data.local.music.get_scores(
|
|
self.game,
|
|
self.music_version,
|
|
userid,
|
|
since=since,
|
|
until=until,
|
|
)
|
|
]
|
|
)
|
|
else:
|
|
raise APIException("Invalid ID type!")
|
|
|
|
# Now, fetch the users, and filter out scores belonging to orphaned users
|
|
id_to_cards: Dict[UserID, List[str]] = {}
|
|
retval: List[Dict[str, Any]] = []
|
|
for (userid, record) in records:
|
|
# Postfilter for queries that can't filter. This will save on data transferred.
|
|
if since is not None:
|
|
if record.update < since:
|
|
continue
|
|
if until is not None:
|
|
if record.update >= until:
|
|
continue
|
|
|
|
if userid not in id_to_cards:
|
|
cards = self.data.local.user.get_cards(userid)
|
|
if len(cards) == 0:
|
|
# Can't add this user, skip the score
|
|
continue
|
|
|
|
id_to_cards[userid] = cards
|
|
|
|
# Format the score and add it
|
|
retval.append(self.__format_record(id_to_cards[userid], record))
|
|
|
|
return retval
|