hallway/artemis
1
0
Fork 0
forked from Hay1tsme/artemis
Code Pull Requests Activity

Move wacca from megaime develop branch, should at least partially fix #3 #4 and #5

Browse Source
This commit is contained in:
Hay1tsme
2023-02-22 22:22:03 -05:00
parent 026fcc5182
commit c3aac4c38e
13 changed files with 309 additions and 341 deletions
Download Patch File Download Diff File Expand all files Collapse all files

297
titles/wacca/base.py
View File

@ -77,9 +77,13 @@ class WaccaBase():
self.logger.info(f"{req.chipId} -> {housing_id}")
resp = HousingGetResponse(housing_id)
return resp.make()
def handle_advertise_GetRanking_request(self, data: Dict) -> Dict:
req = AdvertiseGetRankingRequest(data)
return AdvertiseGetRankingResponse().make()
def handle_housing_start_request(self, data: Dict) -> Dict:
req = HousingStartRequest(data)
req = HousingStartRequestV1(data)
resp = HousingStartResponseV1(
1,
@ -103,7 +107,69 @@ class WaccaBase():
self.logger.info(f"Log out user {req.userId} from {req.chipId}")
return BaseResponse().make()
def handle_user_status_login_request(self, data: Dict) -> List[Any]:
def handle_user_status_get_request(self, data: Dict)-> Dict:
req = UserStatusGetRequest(data)
resp = UserStatusGetV1Response()
ver_split = req.appVersion.split(".")
profile = self.data.profile.get_profile(aime_id=req.aimeId)
if profile is None:
self.logger.info(f"No user exists for aime id {req.aimeId}")
resp.profileStatus = ProfileStatus.ProfileRegister
return resp.make()
self.logger.info(f"User preview for {req.aimeId} from {req.chipId}")
if profile["last_game_ver"] is None:
profile_ver_split = ver_split
resp.lastGameVersion = req.appVersion
else:
profile_ver_split = profile["last_game_ver"].split(".")
resp.lastGameVersion = profile["last_game_ver"]
resp.userStatus.userId = profile["id"]
resp.userStatus.username = profile["username"]
resp.userStatus.xp = profile["xp"]
resp.userStatus.danLevel = profile["dan_level"]
resp.userStatus.danType = profile["dan_type"]
resp.userStatus.wp = profile["wp"]
resp.userStatus.useCount = profile["login_count"]
set_title_id = self.data.profile.get_options(WaccaConstants.OPTIONS["set_title_id"], profile["user"])
if set_title_id is None:
set_title_id = self.OPTIONS_DEFAULTS["set_title_id"]
resp.setTitleId = set_title_id
set_icon_id = self.data.profile.get_options(WaccaConstants.OPTIONS["set_title_id"], profile["user"])
if set_icon_id is None:
set_icon_id = self.OPTIONS_DEFAULTS["set_icon_id"]
resp.setIconId = set_icon_id
if int(ver_split[0]) > int(profile_ver_split[0]):
resp.versionStatus = PlayVersionStatus.VersionUpgrade
elif int(ver_split[0]) < int(profile_ver_split[0]):
resp.versionStatus = PlayVersionStatus.VersionTooNew
else:
if int(ver_split[1]) > int(profile_ver_split[1]):
resp.versionStatus = PlayVersionStatus.VersionUpgrade
elif int(ver_split[1]) < int(profile_ver_split[1]):
resp.versionStatus = PlayVersionStatus.VersionTooNew
else:
if int(ver_split[2]) > int(profile_ver_split[2]):
resp.versionStatus = PlayVersionStatus.VersionUpgrade
elif int(ver_split[2]) < int(profile_ver_split[2]):
resp.versionStatus = PlayVersionStatus.VersionTooNew
return resp.make()
def handle_user_status_login_request(self, data: Dict)-> Dict:
req = UserStatusLoginRequest(data)
resp = UserStatusLoginResponseV1()
is_new_day = False
@ -139,120 +205,8 @@ class WaccaBase():
resp.firstLoginDaily = int(is_new_day)
return resp.make()
def handle_user_status_get_request(self, data: Dict) -> List[Any]:
req = UserStatusGetRequest(data)
resp = UserStatusGetV1Response()
ver_split = req.appVersion.split(".")
profile = self.data.profile.get_profile(aime_id=req.aimeId)
if profile is None:
self.logger.info(f"No user exists for aime id {req.aimeId}")
return resp.make()
self.logger.info(f"User preview for {req.aimeId} from {req.chipId}")
if profile["last_game_ver"] is None:
profile_ver_split = ver_split
resp.lastGameVersion = req.appVersion
else:
profile_ver_split = profile["last_game_ver"].split(".")
resp.lastGameVersion = profile["last_game_ver"]
resp.userStatus.userId = profile["id"]
resp.userStatus.username = profile["username"]
resp.userStatus.xp = profile["xp"]
resp.userStatus.danLevel = profile["dan_level"]
resp.userStatus.danType = profile["dan_type"]
resp.userStatus.wp = profile["wp"]
resp.userStatus.useCount = profile["login_count"]
resp.userStatus.loginDays = profile["login_count_days"]
resp.userStatus.loginConsecutiveDays = profile["login_count_days_consec"]
set_title_id = self.data.profile.get_options(WaccaConstants.OPTIONS["set_title_id"], profile["user"])
if set_title_id is None:
set_title_id = self.OPTIONS_DEFAULTS["set_title_id"]
resp.setTitleId = set_title_id
set_icon_id = self.data.profile.get_options(WaccaConstants.OPTIONS["set_title_id"], profile["user"])
if set_icon_id is None:
set_icon_id = self.OPTIONS_DEFAULTS["set_icon_id"]
resp.setIconId = set_icon_id
if profile["last_login_date"].timestamp() < int((datetime.now().replace(hour=0,minute=0,second=0,microsecond=0) - timedelta(days=1)).timestamp()):
resp.userStatus.loginConsecutiveDays = 0
if int(ver_split[0]) > int(profile_ver_split[0]):
resp.versionStatus = PlayVersionStatus.VersionUpgrade
elif int(ver_split[0]) < int(profile_ver_split[0]):
resp.versionStatus = PlayVersionStatus.VersionTooNew
else:
if int(ver_split[1]) > int(profile_ver_split[1]):
resp.versionStatus = PlayVersionStatus.VersionUpgrade
elif int(ver_split[1]) < int(profile_ver_split[1]):
resp.versionStatus = PlayVersionStatus.VersionTooNew
else:
if int(ver_split[2]) > int(profile_ver_split[2]):
resp.versionStatus = PlayVersionStatus.VersionUpgrade
elif int(ver_split[2]) < int(profile_ver_split[2]):
resp.versionStatus = PlayVersionStatus.VersionTooNew
if profile["always_vip"]:
resp.userStatus.vipExpireTime = int((datetime.now() + timedelta(days=30)).timestamp())
elif profile["vip_expire_time"] is not None:
resp.userStatus.vipExpireTime = int(profile["vip_expire_time"].timestamp())
return resp.make()
def handle_user_status_login_request(self, data: Dict) -> List[Any]:
req = UserStatusLoginRequest(data)
resp = UserStatusLoginResponseV2()
is_new_day = False
is_consec_day = False
is_consec_day = True
if req.userId == 0:
self.logger.info(f"Guest login on {req.chipId}")
resp.lastLoginDate = 0
else:
profile = self.data.profile.get_profile(req.userId)
if profile is None:
self.logger.warn(f"Unknown user id {req.userId} attempted login from {req.chipId}")
return resp.make()
self.logger.info(f"User {req.userId} login on {req.chipId}")
last_login_time = int(profile["last_login_date"].timestamp())
resp.lastLoginDate = last_login_time
# If somebodies login timestamp < midnight of current day, then they are logging in for the first time today
if last_login_time < int(datetime.now().replace(hour=0,minute=0,second=0,microsecond=0).timestamp()):
is_new_day = True
is_consec_day = True
# If somebodies login timestamp > midnight of current day + 1 day, then they broke their daily login streak
elif last_login_time > int((datetime.now().replace(hour=0,minute=0,second=0,microsecond=0) + timedelta(days=1)).timestamp()):
is_consec_day = False
# else, they are simply logging in again on the same day, and we don't need to do anything for that
self.data.profile.session_login(req.userId, is_new_day, is_consec_day)
resp.vipInfo.pageYear = datetime.now().year
resp.vipInfo.pageMonth = datetime.now().month
resp.vipInfo.pageDay = datetime.now().day
resp.vipInfo.numItem = 1
resp.firstLoginDaily = int(is_new_day)
return resp.make()
def handle_user_status_create_request(self, data: Dict) -> List[Any]:
def handle_user_status_create_request(self, data: Dict)-> Dict:
req = UserStatusCreateRequest(data)
profileId = self.data.profile.create_profile(req.aimeId, req.username, self.version)
@ -284,7 +238,7 @@ class WaccaBase():
return UserStatusCreateResponseV2(profileId, req.username).make()
def handle_user_status_getDetail_request(self, data: Dict) -> List[Any]:
def handle_user_status_getDetail_request(self, data: Dict)-> Dict:
req = UserStatusGetDetailRequest(data)
resp = UserStatusGetDetailResponseV1()
@ -301,16 +255,7 @@ class WaccaBase():
profile_song_unlocks = self.data.item.get_song_unlocks(user_id)
profile_options = self.data.profile.get_options(user_id)
profile_trophies = self.data.item.get_trophies(user_id)
profile_tickets = self.data.item.get_tickets(user_id)
if profile["vip_expire_time"] is None:
resp.userStatus.vipExpireTime = 0
else:
resp.userStatus.vipExpireTime = int(profile["vip_expire_time"].timestamp())
if profile["always_vip"] or self.game_config.mods.always_vip:
resp.userStatus.vipExpireTime = int((self.srvtime + timedelta(days=31)).timestamp())
profile_tickets = self.data.item.get_tickets(user_id)
resp.songUpdateTime = int(profile["last_login_date"].timestamp())
resp.songPlayStatus = [profile["last_song_id"], 1]
@ -322,8 +267,6 @@ class WaccaBase():
resp.userStatus.danType = profile["dan_type"]
resp.userStatus.wp = profile["wp"]
resp.userStatus.useCount = profile["login_count"]
resp.userStatus.loginDays = profile["login_count_days"]
resp.userStatus.loginConsecutiveDays = profile["login_count_days_consec"]
if self.game_config.mods.infinite_wp:
resp.userStatus.wp = 999999
@ -346,13 +289,9 @@ class WaccaBase():
for unlock in profile_song_unlocks:
for x in range(1, unlock["highest_difficulty"] + 1):
resp.userItems.songUnlocks.append(SongUnlock(unlock["song_id"], x, 0, int(unlock["acquire_date"].timestamp())))
if x > 2:
resp.scores.append(BestScoreDetailV1(unlock["song_id"], x))
empty_scores = len(resp.scores)
for song in profile_scores:
resp.seasonInfo.cumulativeScore += song["score"]
empty_score_idx = resp.find_score_idx(song["song_id"], song["chart_id"], 0, empty_scores)
clear_cts = SongDetailClearCounts(
song["play_ct"],
@ -368,24 +307,16 @@ class WaccaBase():
song["grade_master_ct"]
)
if empty_score_idx is not None:
resp.scores[empty_score_idx].clearCounts = clear_cts
resp.scores[empty_score_idx].clearCountsSeason = clear_cts
resp.scores[empty_score_idx].gradeCounts = grade_cts
resp.scores[empty_score_idx].score = song["score"]
resp.scores[empty_score_idx].bestCombo = song["best_combo"]
resp.scores[empty_score_idx].lowestMissCtMaybe = song["lowest_miss_ct"]
resp.scores[empty_score_idx].rating = song["rating"]
else:
deets = BestScoreDetailV1(song["song_id"], song["chart_id"])
deets.clearCounts = clear_cts
deets.clearCountsSeason = clear_cts
deets.gradeCounts = grade_cts
deets.score = song["score"]
deets.bestCombo = song["best_combo"]
deets.lowestMissCtMaybe = song["lowest_miss_ct"]
deets.rating = song["rating"]
deets = BestScoreDetailV1(song["song_id"], song["chart_id"])
deets.clearCounts = clear_cts
deets.clearCountsSeason = clear_cts
deets.gradeCounts = grade_cts
deets.score = song["score"]
deets.bestCombo = song["best_combo"]
deets.lowestMissCtMaybe = song["lowest_miss_ct"]
deets.rating = song["rating"]
resp.scores.append(deets)
for trophy in profile_trophies:
resp.userItems.trophies.append(TrophyItem(trophy["trophy_id"], trophy["season"], trophy["progress"], trophy["badge_type"]))
@ -434,7 +365,7 @@ class WaccaBase():
return resp.make()
def handle_user_trial_get_request(self, data: Dict) -> List[Any]:
def handle_user_trial_get_request(self, data: Dict)-> Dict:
req = UserTrialGetRequest(data)
resp = UserTrialGetResponse()
@ -444,10 +375,6 @@ class WaccaBase():
return resp.make()
self.logger.info(f"Get trial info for user {req.profileId}")
for d in self.allowed_stages:
if d[1] > 0 and d[1] < 10:
resp.stageList.append(StageInfo(d[0], d[1]))
stages = self.data.score.get_stageup(user_id, self.version)
if stages is None:
@ -474,7 +401,7 @@ class WaccaBase():
return resp.make()
def handle_user_trial_update_request(self, data: Dict) -> List[Any]:
def handle_user_trial_update_request(self, data: Dict)-> Dict:
req = UserTrialUpdateRequest(data)
total_score = 0
@ -496,8 +423,8 @@ class WaccaBase():
# We only care about total score for best of, even if one score happens to be lower (I think)
if total_score > (old_stage["song1_score"] + old_stage["song2_score"] + old_stage["song3_score"]):
best_score1 = req.songScores[0]
best_score2 = req.songScores[2]
best_score3 = req.songScores[3]
best_score2 = req.songScores[1]
best_score3 = req.songScores[2]
else:
best_score1 = old_stage["song1_score"]
best_score2 = old_stage["song2_score"]
@ -528,9 +455,9 @@ class WaccaBase():
self.data.item.put_item(user_id, WaccaConstants.ITEM_TYPES["icon"], current_icon)
self.data.item.put_item(user_id, WaccaConstants.ITEM_TYPES["navigator"], current_nav)
self.data.profile.update_profile_playtype(req.profileId, 4, data["appVersion"][:7])
return BaseResponse.make()
return BaseResponse().make()
def handle_user_sugoroku_update_request(self, data: Dict) -> List[Any]:
def handle_user_sugoroku_update_request(self, data: Dict)-> Dict:
ver_split = data["appVersion"].split(".")
resp = BaseResponse()
@ -552,10 +479,10 @@ class WaccaBase():
self.data.profile.update_gate(user_id, req.gateId, req.page, req.progress, req.loops, mission_flg, req.totalPts)
return resp.make()
def handle_user_info_getMyroom_request(self, data: Dict) -> List[Any]:
return UserInfogetMyroomResponse().make()
def handle_user_info_getMyroom_request(self, data: Dict)-> Dict:
return UserInfogetMyroomResponseV1().make()
def handle_user_music_unlock_request(self, data: Dict) -> List[Any]:
def handle_user_music_unlock_request(self, data: Dict)-> Dict:
req = UserMusicUnlockRequest(data)
profile = self.data.profile.get_profile(req.profileId)
@ -605,29 +532,35 @@ class WaccaBase():
return UserMusicUnlockResponse(current_wp, new_tickets).make()
def handle_user_info_getRanking_request(self, data: Dict) -> List[Any]:
def handle_user_info_getRanking_request(self, data: Dict)-> Dict:
# total score, high score by song, cumulative socre, stage up score, other score, WP ranking
# This likely requies calculating standings at regular intervals and caching the results
return UserInfogetRankingResponse().make()
def handle_user_music_update_request(self, data: Dict) -> List[Any]:
req = UserMusicUpdateRequest(data)
ver_split = req.appVersion.split(".")
def handle_user_music_update_request(self, data: Dict)-> Dict:
ver_split = data["appVersion"].split(".")
if int(ver_split[0]) >= 3:
resp = UserMusicUpdateResponseV3()
req = UserMusicUpdateRequestV2(data)
elif int(ver_split[0]) >= 2:
resp = UserMusicUpdateResponseV2()
req = UserMusicUpdateRequestV2(data)
else:
resp = UserMusicUpdateResponseV1()
req = UserMusicUpdateRequestV1(data)
resp.songDetail.songId = req.songDetail.songId
resp.songDetail.difficulty = req.songDetail.difficulty
if req.profileId == 0:
self.logger.info(f"Guest score for song {req.songDetail.songId} difficulty {req.songDetail.difficulty}")
return resp.make()
profile = self.data.profile.get_profile(req.profileId)
if profile is None:
self.logger.warn(f"handle_user_music_update_request: No profile for game_id {req.profileId}")
return BaseResponse().make()
return resp.make()
user_id = profile["user"]
self.util_put_items(req.profileId, user_id, req.itemsObtained)
@ -715,18 +648,18 @@ class WaccaBase():
return resp.make()
#TODO: Coop and vs data
def handle_user_music_updateCoop_request(self, data: Dict) -> List[Any]:
def handle_user_music_updateCoop_request(self, data: Dict)-> Dict:
coop_info = data["params"][4]
return self.handle_user_music_update_request(data)
def handle_user_music_updateVersus_request(self, data: Dict) -> List[Any]:
def handle_user_music_updateVersus_request(self, data: Dict)-> Dict:
vs_info = data["params"][4]
return self.handle_user_music_update_request(data)
def handle_user_music_updateTrial_request(self, data: Dict) -> List[Any]:
def handle_user_music_updateTrial_request(self, data: Dict)-> Dict:
return self.handle_user_music_update_request(data)
def handle_user_mission_update_request(self, data: Dict) -> List[Any]:
def handle_user_mission_update_request(self, data: Dict)-> Dict:
req = UserMissionUpdateRequest(data)
page_status = req.params[1][1]
@ -742,7 +675,7 @@ class WaccaBase():
return BaseResponse().make()
def handle_user_goods_purchase_request(self, data: Dict) -> List[Any]:
def handle_user_goods_purchase_request(self, data: Dict)-> Dict:
req = UserGoodsPurchaseRequest(data)
resp = UserGoodsPurchaseResponse()
@ -775,13 +708,13 @@ class WaccaBase():
return resp.make()
def handle_competition_status_login_request(self, data: Dict) -> List[Any]:
def handle_competition_status_login_request(self, data: Dict)-> Dict:
return BaseResponse().make()
def handle_competition_status_update_request(self, data: Dict) -> List[Any]:
def handle_competition_status_update_request(self, data: Dict)-> Dict:
return BaseResponse().make()
def handle_user_rating_update_request(self, data: Dict) -> List[Any]:
def handle_user_rating_update_request(self, data: Dict)-> Dict:
req = UserRatingUpdateRequest(data)
user_id = self.data.profile.profile_to_aime_user(req.profileId)
@ -797,8 +730,8 @@ class WaccaBase():
return BaseResponse().make()
def handle_user_status_update_request(self, data: Dict) -> List[Any]:
req = UserStatusUpdateRequestV2(data)
def handle_user_status_update_request(self, data: Dict)-> Dict:
req = UserStatusUpdateRequestV1(data)
user_id = self.data.profile.profile_to_aime_user(req.profileId)
if user_id is None:
@ -807,8 +740,6 @@ class WaccaBase():
self.util_put_items(req.profileId, user_id, req.itemsRecieved)
self.data.profile.update_profile_playtype(req.profileId, req.playType.value, data["appVersion"][:7])
self.data.profile.update_profile_lastplayed(req.profileId, req.lastSongInfo.lastSongId, req.lastSongInfo.lastSongDiff,
req.lastSongInfo.lastFolderOrd, req.lastSongInfo.lastFolderId, req.lastSongInfo.lastSongOrd)
current_icon = self.data.profile.get_options(user_id, WaccaConstants.OPTIONS["set_icon_id"])
current_nav = self.data.profile.get_options(user_id, WaccaConstants.OPTIONS["set_nav_id"])
@ -826,7 +757,7 @@ class WaccaBase():
self.data.item.put_item(user_id, WaccaConstants.ITEM_TYPES["navigator"], current_nav)
return BaseResponse().make()
def handle_user_info_update_request(self, data: Dict) -> List[Any]:
def handle_user_info_update_request(self, data: Dict)-> Dict:
req = UserInfoUpdateRequest(data)
user_id = self.data.profile.profile_to_aime_user(req.profileId)
@ -845,7 +776,7 @@ class WaccaBase():
return BaseResponse().make()
def handle_user_vip_get_request(self, data: Dict) -> List[Any]:
def handle_user_vip_get_request(self, data: Dict)-> Dict:
req = UserVipGetRequest(data)
resp = UserVipGetResponse()
@ -868,7 +799,7 @@ class WaccaBase():
return resp.make()
def handle_user_vip_start_request(self, data: Dict) -> List[Any]:
def handle_user_vip_start_request(self, data: Dict)-> Dict:
req = UserVipStartRequest(data)
profile = self.data.profile.get_profile(req.profileId)