pgw00g/artemis
1
0
Fork 0
forked from Dniel97/artemis
Code Pull Requests Activity

update Event Ranking, Tech Challenge, and Music Ranking

Browse Source
This commit is contained in:
phantomlan 2023-11-07 04:38:04 +01:00 committed by phantomlan
parent 1897e8002d
commit d2e2c14074
3 changed files with 56 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
titles/ongeki/base.py
View File

@ -157,7 +157,7 @@ class OngekiBase:
return {"type": data["type"], "length": 0, "gameIdlistList": []}
def handle_get_game_ranking_api_request(self, data: Dict) -> Dict:
game_ranking_list = self.data.static.get_ranking_list()
game_ranking_list = self.data.static.get_ranking_list(self.version)
ranking_list = []
for music in game_ranking_list:
@ -168,7 +168,6 @@ class OngekiBase:
return {"length": 0, "gameRankingList": []}
return {
"type": data["type"],
#"length": len(ranking_list),
"gameRankingList": ranking_list,
}
@ -270,12 +269,13 @@ class OngekiBase:
return {"length": 0, "gameSaleList": []}
def handle_get_game_tech_music_api_request(self, data: Dict) -> Dict:
music_list = self.data.item.get_tech_music()
music_list = self.data.static.get_tech_music(self.version)
prep_music_list = []
for music in music_list:
tmp = music._asdict()
tmp.pop("id")
tmp.pop("version")
prep_music_list.append(tmp)
if prep_music_list is None:
@ -441,7 +441,7 @@ class OngekiBase:
}
def handle_get_user_tech_event_api_request(self, data: Dict) -> Dict:
user_tech_event_list = self.data.item.get_tech_event(data["userId"])
user_tech_event_list = self.data.item.get_tech_event(self.version, data["userId"])
if user_tech_event_list is None:
return {}
@ -450,6 +450,7 @@ class OngekiBase:
tmp = evt._asdict()
tmp.pop("id")
tmp.pop("user")
tmp.pop("version")
tech_evt.append(tmp)
return {
@ -459,7 +460,7 @@ class OngekiBase:
}
def handle_get_user_tech_event_ranking_api_request(self, data: Dict) -> Dict:
user_tech_event_ranks = self.data.item.get_tech_event_ranking(data["userId"])
user_tech_event_ranks = self.data.item.get_tech_event_ranking(self.version, data["userId"])
if user_tech_event_ranks is None:
return {
"userId": data["userId"],
@ -598,7 +599,7 @@ class OngekiBase:
return {"userId": data["userId"], "userData": user_data}
def handle_get_user_event_ranking_api_request(self, data: Dict) -> Dict:
user_event_ranking_list = self.data.item.get_ranking_event_ranks(data["userId"])
user_event_ranking_list = self.data.item.get_ranking_event_ranks(self.version, data["userId"])
if user_event_ranking_list is None:
return {}
@ -1027,7 +1028,7 @@ class OngekiBase:
if "userEventPointList" in upsert:
for x in upsert["userEventPointList"]:
self.data.item.put_event_point(user_id, x)
self.data.item.put_event_point(user_id, self.version, x)
if "userMissionPointList" in upsert:
for x in upsert["userMissionPointList"]:
@ -1061,10 +1062,10 @@ class OngekiBase:
if "userTechEventList" in upsert:
for x in upsert["userTechEventList"]:
self.data.item.put_tech_event(user_id, x)
self.data.item.put_tech_event(user_id, self.version, x)
# This should be updated once a day in maintenance window, but for time being we will push the update on each upsert
self.data.item.put_tech_event_ranking(user_id, x)
self.data.item.put_tech_event_ranking(user_id, self.version, x)
if "userKopList" in upsert:
for x in upsert["userKopList"]:

58
titles/ongeki/schema/item.py
View File

@ -171,10 +171,11 @@ event_point = Table(
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade")),
Column("eventId", Integer),
Column("point", Integer),
Column("version", Integer, nullable=False),
Column("eventId", Integer, nullable=False),
Column("point", Integer, nullable=False),
Column("rank", Integer),
Column("type", Integer),
Column("type", Integer, nullable=False),
Column("date", String(25)),
Column("isRankingRewarded", Boolean),
UniqueConstraint("user", "eventId", name="ongeki_user_event_point_uk"),
@ -241,9 +242,10 @@ tech_event = Table(
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade")),
Column("eventId", Integer),
Column("totalTechScore", Integer),
Column("totalPlatinumScore", Integer),
Column("version", Integer, nullable=False),
Column("eventId", Integer, nullable=False),
Column("totalTechScore", Integer, nullable=False),
Column("totalPlatinumScore", Integer, nullable=False),
Column("techRecordDate", String(25)),
Column("isRankingRewarded", Boolean),
Column("isTotalTechNewRecord", Boolean),
@ -251,27 +253,17 @@ tech_event = Table(
mysql_charset="utf8mb4",
)
tech_music = Table(
"ongeki_tech_music_list",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column("eventId", Integer),
Column("musicId", Integer),
Column("level", Integer),
UniqueConstraint("musicId", name="ongeki_tech_music_list_uk"),
mysql_charset="utf8mb4",
)
tech_ranking = Table(
"ongeki_tech_event_ranking",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column("user", ForeignKey("aime_user.id", ondelete="cascade", onupdate="cascade"), nullable=False),
Column("version", Integer, nullable=False),
Column("date", String(25)),
Column("eventId", Integer),
Column("eventId", Integer, nullable=False),
Column("rank", Integer),
Column("totalPlatinumScore", Integer),
Column("totalTechScore", Integer),
Column("totalPlatinumScore", Integer, nullable=False),
Column("totalTechScore", Integer, nullable=False),
UniqueConstraint("user", "eventId", name="ongeki_tech_event_ranking_uk"),
mysql_charset="utf8mb4",
)
@ -562,10 +554,11 @@ class OngekiItemData(BaseData):
return None
return result.fetchall()
def put_event_point(self, aime_id: int, event_point_data: Dict) -> Optional[int]:
def put_event_point(self, aime_id: int, version: int, event_point_data: Dict) -> Optional[int]:
# We update only the newest (type: 1) entry, in official spec game watches for both latest(type:1) and previous (type:2) entries to give an additional info how many ranks has player moved up or down
# This fully featured is on TODO list, at the moment we just update the tables as data comes and give out rank as request comes
event_point_data["user"] = aime_id
event_point_data["version"] = version
event_point_data["type"] = 1
event_point_time = datetime.now()
event_point_data["date"] = datetime.strftime(event_point_time, "%Y-%m-%d %H:%M")
@ -647,14 +640,6 @@ class OngekiItemData(BaseData):
return None
return result.fetchall()
def get_tech_music(self) -> Optional[List[Dict]]:
sql = select(tech_music)
result = self.execute(sql)
if result is None:
return None
return result.fetchall()
def put_tech_event(self, aime_id: int, tech_event_data: Dict) -> Optional[int]:
tech_event_data["user"] = aime_id
@ -667,8 +652,9 @@ class OngekiItemData(BaseData):
return None
return result.lastrowid
def put_tech_event_ranking(self, aime_id: int, tech_event_data: Dict) -> Optional[int]:
def put_tech_event_ranking(self, version: int, aime_id: int, tech_event_data: Dict) -> Optional[int]:
tech_event_data["user"] = aime_id
tech_event_data["version"] = version
tech_event_data.pop("isRankingRewarded")
tech_event_data.pop("isTotalTechNewRecord")
tech_event_data["date"] = tech_event_data.pop("techRecordDate")
@ -683,8 +669,8 @@ class OngekiItemData(BaseData):
return None
return result.lastrowid
def get_tech_event(self, aime_id: int) -> Optional[List[Dict]]:
sql = select(tech_event).where(tech_event.c.user == aime_id)
def get_tech_event(self, version: int, aime_id: int) -> Optional[List[Dict]]:
sql = select(tech_event).where(and_(tech_event.c.user == aime_id, tech_event.c.version == version))
result = self.execute(sql)
if result is None:
@ -774,17 +760,17 @@ class OngekiItemData(BaseData):
return result.lastrowid
def get_ranking_event_ranks(self, aime_id: int) -> Optional[List[Dict]]:
def get_ranking_event_ranks(self, version: int, aime_id: int) -> Optional[List[Dict]]:
# Calculates player rank on GameRequest from server, and sends it back, official spec would rank players in maintenance period, on TODO list
sql = select(event_point.c.id, event_point.c.user, event_point.c.eventId, event_point.c.type, func.row_number().over(partition_by=event_point.c.eventId, order_by=event_point.c.point.desc()).label('rank'), event_point.c.date, event_point.c.point)
sql = select(event_point.c.id, event_point.c.user, event_point.c.eventId, event_point.c.type, func.row_number().over(partition_by=event_point.c.eventId, order_by=event_point.c.point.desc()).label('rank'), event_point.c.date, event_point.c.point).where(event_point.c.version == version)
result = self.execute(sql)
if result is None:
self.logger.error(f"failed to rank aime_id: {aime_id} ranking event positions")
return None
return result.fetchall()
def get_tech_event_ranking(self, aime_id: int) -> Optional[List[Dict]]:
sql = select(tech_ranking.c.id, tech_ranking.c.user, tech_ranking.c.date, tech_ranking.c.eventId, func.row_number().over(partition_by=tech_ranking.c.eventId, order_by=[tech_ranking.c.totalTechScore.desc(),tech_ranking.c.totalPlatinumScore.desc()]).label('rank'), tech_ranking.c.totalTechScore, tech_ranking.c.totalPlatinumScore)
def get_tech_event_ranking(self, version: int, aime_id: int) -> Optional[List[Dict]]:
sql = select(tech_ranking.c.id, tech_ranking.c.user, tech_ranking.c.date, tech_ranking.c.eventId, func.row_number().over(partition_by=tech_ranking.c.eventId, order_by=[tech_ranking.c.totalTechScore.desc(),tech_ranking.c.totalPlatinumScore.desc()]).label('rank'), tech_ranking.c.totalTechScore, tech_ranking.c.totalPlatinumScore).where(tech_ranking.c.version == version)
result = self.execute(sql)
if result is None:
self.logger.warning(f"aime_id: {aime_id} has no tech ranking ranks")

26
titles/ongeki/schema/static.py
View File

@ -100,6 +100,7 @@ music_ranking = Table(
"ongeki_static_music_ranking_list",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column("version", Integer, nullable=False),
Column("musicId", Integer, nullable=False),
Column("point", Integer, nullable=False),
Column("userName", String(255)),
@ -136,6 +137,18 @@ present = Table(
mysql_charset="utf8mb4",
)
tech_music = Table(
"ongeki_static_tech_music",
metadata,
Column("id", Integer, primary_key=True, nullable=False),
Column("version", Integer, nullable=False),
Column("eventId", Integer, nullable=False),
Column("musicId", Integer, nullable=False),
Column("level", Integer, nullable=False),
UniqueConstraint("version", "musicId", name="ongeki_static_tech_music_uk"),
mysql_charset="utf8mb4",
)
class OngekiStaticData(BaseData):
def put_card(self, version: int, card_id: int, **card_data) -> Optional[int]:
sql = insert(cards).values(version=version, cardId=card_id, **card_data)
@ -373,8 +386,8 @@ class OngekiStaticData(BaseData):
return None
return result.fetchone()
def get_ranking_list(self) -> Optional[List[Dict]]:
sql = select(music_ranking.c.musicId.label('id'), music_ranking.c.point, music_ranking.c.userName)
def get_ranking_list(self, version: int) -> Optional[List[Dict]]:
sql = select(music_ranking.c.musicId.label('id'), music_ranking.c.point, music_ranking.c.userName).where(music_ranking.c.version == version)
result = self.execute(sql)
if result is None:
return None
@ -414,3 +427,12 @@ class OngekiStaticData(BaseData):
self.logger.warning(f"Failed to load present list")
return None
return result.fetchall()
def get_tech_music(self, version: int) -> Optional[List[Dict]]:
sql = select(tech_music).where(tech_music.c.version == version)
result = self.execute(sql)
if result is None:
return None
return result.fetchall()