From efd249d808f5091adda58537850829f63790ce2a Mon Sep 17 00:00:00 2001 From: Kevin Trocolli Date: Sat, 8 Jun 2024 17:26:51 -0400 Subject: [PATCH] maimai: some housekeeping --- titles/mai2/base.py | 249 ++++++++++++++++++++++++++++++++++++ titles/mai2/buddies.py | 13 -- titles/mai2/festival.py | 14 +- titles/mai2/festivalplus.py | 19 --- titles/mai2/universe.py | 206 ----------------------------- 5 files changed, 250 insertions(+), 251 deletions(-) diff --git a/titles/mai2/base.py b/titles/mai2/base.py index 90597d1..ee46ae9 100644 --- a/titles/mai2/base.py +++ b/titles/mai2/base.py @@ -4,6 +4,7 @@ import logging from base64 import b64decode from os import path, stat, remove from PIL import ImageFile +from random import randint import pytz from core.config import CoreConfig @@ -886,3 +887,251 @@ class Mai2Base: self.logger.error(f"Failed to delete {out_name}.bin, please remove it manually") return {'returnCode': ret_code, 'apiName': 'UploadUserPhotoApi'} + + async def handle_get_user_favorite_item_api_request(self, data: Dict) -> Dict: + user_id = data.get("userId", 0) + kind = data.get("kind", 2) # 1 is fav music, 2 is rival user IDs + next_index = data.get("nextIndex", 0) + max_ct = data.get("maxCount", 100) # always 100 + is_all = data.get("isAllFavoriteItem", False) # always false + + """ + class userFavoriteItemList: + orderId: int, never checked + id: int, either song ID for kind 1, or rival user ID for kind 2 + """ + return { + "userId": user_id, + "kind": kind, + "nextIndex": 0, + "userFavoriteItemList": [], + } + + async def handle_get_user_recommend_rate_music_api_request(self, data: Dict) -> Dict: + """ + userRecommendRateMusicIdList: list[int] + """ + return {"userId": data["userId"], "userRecommendRateMusicIdList": []} + + async def handle_get_user_recommend_select_music_api_request(self, data: Dict) -> Dict: + """ + userRecommendSelectionMusicIdList: list[int] + """ + return {"userId": data["userId"], "userRecommendSelectionMusicIdList": []} + + async def handle_get_user_new_item_api_request(self, data: Dict) -> Dict: + # TODO: Added in 1.41, implement this? + user_id = data["userId"] + version = data.get("version", 1041000) + user_playlog_list = data.get("userPlaylogList", []) + + return { + "userId": user_id, + "itemKind": -1, + "itemId": -1, + } + + # CardMaker support added in Universe + async def handle_cm_get_user_preview_api_request(self, data: Dict) -> Dict: + p = await self.data.profile.get_profile_detail(data["userId"], self.version) + if p is None: + return {} + + return { + "userName": p["userName"], + "rating": p["playerRating"], + # hardcode lastDataVersion for CardMaker + "lastDataVersion": "1.20.00", # Future versiohs should replace this with the correct version + # checks if the user is still logged in + "isLogin": False, + "isExistSellingCard": True, + } + + async def handle_cm_get_user_data_api_request(self, data: Dict) -> Dict: + # user already exists, because the preview checks that already + p = await self.data.profile.get_profile_detail(data["userId"], self.version) + + cards = await self.data.card.get_user_cards(data["userId"]) + if cards is None or len(cards) == 0: + # This should never happen + self.logger.error( + f"handle_get_user_data_api_request: Internal error - No cards found for user id {data['userId']}" + ) + return {} + + # get the dict representation of the row so we can modify values + user_data = p._asdict() + + # remove the values the game doesn't want + user_data.pop("id") + user_data.pop("user") + user_data.pop("version") + + return {"userId": data["userId"], "userData": user_data} + + async def handle_cm_login_api_request(self, data: Dict) -> Dict: + return {"returnCode": 1} + + async def handle_cm_logout_api_request(self, data: Dict) -> Dict: + return {"returnCode": 1} + + async def handle_cm_get_selling_card_api_request(self, data: Dict) -> Dict: + selling_cards = await self.data.static.get_enabled_cards(self.version) + if selling_cards is None: + return {"length": 0, "sellingCardList": []} + + selling_card_list = [] + for card in selling_cards: + tmp = card._asdict() + tmp.pop("id") + tmp.pop("version") + tmp.pop("cardName") + tmp.pop("enabled") + + tmp["startDate"] = datetime.strftime( + tmp["startDate"], Mai2Constants.DATE_TIME_FORMAT + ) + tmp["endDate"] = datetime.strftime( + tmp["endDate"], Mai2Constants.DATE_TIME_FORMAT + ) + tmp["noticeStartDate"] = datetime.strftime( + tmp["noticeStartDate"], Mai2Constants.DATE_TIME_FORMAT + ) + tmp["noticeEndDate"] = datetime.strftime( + tmp["noticeEndDate"], Mai2Constants.DATE_TIME_FORMAT + ) + + selling_card_list.append(tmp) + + return {"length": len(selling_card_list), "sellingCardList": selling_card_list} + + async def handle_cm_get_user_card_api_request(self, data: Dict) -> Dict: + user_cards = await self.data.item.get_cards(data["userId"]) + if user_cards is None: + return {"returnCode": 1, "length": 0, "nextIndex": 0, "userCardList": []} + + max_ct = data["maxCount"] + next_idx = data["nextIndex"] + start_idx = next_idx + end_idx = max_ct + start_idx + + if len(user_cards[start_idx:]) > max_ct: + next_idx += max_ct + else: + next_idx = 0 + + card_list = [] + for card in user_cards: + tmp = card._asdict() + tmp.pop("id") + tmp.pop("user") + + tmp["startDate"] = datetime.strftime( + tmp["startDate"], Mai2Constants.DATE_TIME_FORMAT + ) + tmp["endDate"] = datetime.strftime( + tmp["endDate"], Mai2Constants.DATE_TIME_FORMAT + ) + card_list.append(tmp) + + return { + "returnCode": 1, + "length": len(card_list[start_idx:end_idx]), + "nextIndex": next_idx, + "userCardList": card_list[start_idx:end_idx], + } + + async def handle_cm_get_user_item_api_request(self, data: Dict) -> Dict: + await self.handle_get_user_item_api_request(data) + + async def handle_cm_get_user_character_api_request(self, data: Dict) -> Dict: + characters = await self.data.item.get_characters(data["userId"]) + + chara_list = [] + for chara in characters: + chara_list.append( + { + "characterId": chara["characterId"], + # no clue why those values are even needed + "point": 0, + "count": 0, + "level": chara["level"], + "nextAwake": 0, + "nextAwakePercent": 0, + "favorite": False, + "awakening": chara["awakening"], + "useCount": chara["useCount"], + } + ) + + return { + "returnCode": 1, + "length": len(chara_list), + "userCharacterList": chara_list, + } + + async def handle_cm_get_user_card_print_error_api_request(self, data: Dict) -> Dict: + return {"length": 0, "userPrintDetailList": []} + + async def handle_cm_upsert_user_print_api_request(self, data: Dict) -> Dict: + user_id = data["userId"] + upsert = data["userPrintDetail"] + + # set a random card serial number + serial_id = "".join([str(randint(0, 9)) for _ in range(20)]) + + # calculate start and end date of the card + start_date = datetime.utcnow() + end_date = datetime.utcnow() + timedelta(days=15) + + user_card = upsert["userCard"] + await self.data.item.put_card( + user_id, + user_card["cardId"], + user_card["cardTypeId"], + user_card["charaId"], + user_card["mapId"], + # add the correct start date and also the end date in 15 days + start_date, + end_date, + ) + + # get the profile extend to save the new bought card + extend = await self.data.profile.get_profile_extend(user_id, self.version) + if extend: + extend = extend._asdict() + # parse the selectedCardList + # 6 = Freedom Pass, 4 = Gold Pass (cardTypeId) + selected_cards: List = extend["selectedCardList"] + + # if no pass is already added, add the corresponding pass + if not user_card["cardTypeId"] in selected_cards: + selected_cards.insert(0, user_card["cardTypeId"]) + + extend["selectedCardList"] = selected_cards + await self.data.profile.put_profile_extend(user_id, self.version, extend) + + # properly format userPrintDetail for the database + upsert.pop("userCard") + upsert.pop("serialId") + upsert["printDate"] = datetime.strptime(upsert["printDate"], "%Y-%m-%d") + + await self.data.item.put_user_print_detail(user_id, serial_id, upsert) + + return { + "returnCode": 1, + "orderId": 0, + "serialId": serial_id, + "startDate": datetime.strftime(start_date, Mai2Constants.DATE_TIME_FORMAT), + "endDate": datetime.strftime(end_date, Mai2Constants.DATE_TIME_FORMAT), + } + + async def handle_cm_upsert_user_printlog_api_request(self, data: Dict) -> Dict: + return { + "returnCode": 1, + "orderId": 0, + "serialId": data["userPrintlog"]["serialId"], + } + + async def handle_cm_upsert_buy_card_api_request(self, data: Dict) -> Dict: + return {"returnCode": 1} diff --git a/titles/mai2/buddies.py b/titles/mai2/buddies.py index f04b215..38049a1 100644 --- a/titles/mai2/buddies.py +++ b/titles/mai2/buddies.py @@ -17,16 +17,3 @@ class Mai2Buddies(Mai2FestivalPlus): # hardcode lastDataVersion for CardMaker user_data["lastDataVersion"] = "1.40.00" return user_data - - async def handle_get_user_new_item_api_request(self, data: Dict) -> Dict: - # TODO: Added in 1.41, implement this? - user_id = data["userId"] - version = data.get("version", 1041000) - user_playlog_list = data.get("userPlaylogList", []) - - return { - "userId": user_id, - "itemKind": -1, - "itemId": -1, - } - diff --git a/titles/mai2/festival.py b/titles/mai2/festival.py index 94ce3ec..451f0ba 100644 --- a/titles/mai2/festival.py +++ b/titles/mai2/festival.py @@ -20,18 +20,6 @@ class Mai2Festival(Mai2UniversePlus): async def handle_user_login_api_request(self, data: Dict) -> Dict: user_login = await super().handle_user_login_api_request(data) - # useless? + # TODO: Make use of this user_login["Bearer"] = "ARTEMiSTOKEN" return user_login - - async def handle_get_user_recommend_rate_music_api_request(self, data: Dict) -> Dict: - """ - userRecommendRateMusicIdList: list[int] - """ - return {"userId": data["userId"], "userRecommendRateMusicIdList": []} - - async def handle_get_user_recommend_select_music_api_request(self, data: Dict) -> Dict: - """ - userRecommendSelectionMusicIdList: list[int] - """ - return {"userId": data["userId"], "userRecommendSelectionMusicIdList": []} diff --git a/titles/mai2/festivalplus.py b/titles/mai2/festivalplus.py index 375d546..3314e34 100644 --- a/titles/mai2/festivalplus.py +++ b/titles/mai2/festivalplus.py @@ -17,22 +17,3 @@ class Mai2FestivalPlus(Mai2Festival): # hardcode lastDataVersion for CardMaker user_data["lastDataVersion"] = "1.35.00" return user_data - - async def handle_get_user_favorite_item_api_request(self, data: Dict) -> Dict: - user_id = data.get("userId", 0) - kind = data.get("kind", 2) - next_index = data.get("nextIndex", 0) - max_ct = data.get("maxCount", 100) - is_all = data.get("isAllFavoriteItem", False) - - """ - class userFavoriteItemList: - orderId: int - id: int - """ - return { - "userId": user_id, - "kind": kind, - "nextIndex": 0, - "userFavoriteItemList": [], - } diff --git a/titles/mai2/universe.py b/titles/mai2/universe.py index 00353d3..8ed0d88 100644 --- a/titles/mai2/universe.py +++ b/titles/mai2/universe.py @@ -1,8 +1,6 @@ from typing import Any, List, Dict from random import randint from datetime import datetime, timedelta -import pytz -import json from core.config import CoreConfig from titles.mai2.splashplus import Mai2SplashPlus @@ -14,207 +12,3 @@ class Mai2Universe(Mai2SplashPlus): def __init__(self, cfg: CoreConfig, game_cfg: Mai2Config) -> None: super().__init__(cfg, game_cfg) self.version = Mai2Constants.VER_MAIMAI_DX_UNIVERSE - - async def handle_cm_get_user_preview_api_request(self, data: Dict) -> Dict: - p = await self.data.profile.get_profile_detail(data["userId"], self.version) - if p is None: - return {} - - return { - "userName": p["userName"], - "rating": p["playerRating"], - # hardcode lastDataVersion for CardMaker - "lastDataVersion": "1.20.00", - # checks if the user is still logged in - "isLogin": False, - "isExistSellingCard": True, - } - - async def handle_cm_get_user_data_api_request(self, data: Dict) -> Dict: - # user already exists, because the preview checks that already - p = await self.data.profile.get_profile_detail(data["userId"], self.version) - - cards = await self.data.card.get_user_cards(data["userId"]) - if cards is None or len(cards) == 0: - # This should never happen - self.logger.error( - f"handle_get_user_data_api_request: Internal error - No cards found for user id {data['userId']}" - ) - return {} - - # get the dict representation of the row so we can modify values - user_data = p._asdict() - - # remove the values the game doesn't want - user_data.pop("id") - user_data.pop("user") - user_data.pop("version") - - return {"userId": data["userId"], "userData": user_data} - - async def handle_cm_login_api_request(self, data: Dict) -> Dict: - return {"returnCode": 1} - - async def handle_cm_logout_api_request(self, data: Dict) -> Dict: - return {"returnCode": 1} - - async def handle_cm_get_selling_card_api_request(self, data: Dict) -> Dict: - selling_cards = await self.data.static.get_enabled_cards(self.version) - if selling_cards is None: - return {"length": 0, "sellingCardList": []} - - selling_card_list = [] - for card in selling_cards: - tmp = card._asdict() - tmp.pop("id") - tmp.pop("version") - tmp.pop("cardName") - tmp.pop("enabled") - - tmp["startDate"] = datetime.strftime( - tmp["startDate"], Mai2Constants.DATE_TIME_FORMAT - ) - tmp["endDate"] = datetime.strftime( - tmp["endDate"], Mai2Constants.DATE_TIME_FORMAT - ) - tmp["noticeStartDate"] = datetime.strftime( - tmp["noticeStartDate"], Mai2Constants.DATE_TIME_FORMAT - ) - tmp["noticeEndDate"] = datetime.strftime( - tmp["noticeEndDate"], Mai2Constants.DATE_TIME_FORMAT - ) - - selling_card_list.append(tmp) - - return {"length": len(selling_card_list), "sellingCardList": selling_card_list} - - async def handle_cm_get_user_card_api_request(self, data: Dict) -> Dict: - user_cards = await self.data.item.get_cards(data["userId"]) - if user_cards is None: - return {"returnCode": 1, "length": 0, "nextIndex": 0, "userCardList": []} - - max_ct = data["maxCount"] - next_idx = data["nextIndex"] - start_idx = next_idx - end_idx = max_ct + start_idx - - if len(user_cards[start_idx:]) > max_ct: - next_idx += max_ct - else: - next_idx = 0 - - card_list = [] - for card in user_cards: - tmp = card._asdict() - tmp.pop("id") - tmp.pop("user") - - tmp["startDate"] = datetime.strftime( - tmp["startDate"], Mai2Constants.DATE_TIME_FORMAT - ) - tmp["endDate"] = datetime.strftime( - tmp["endDate"], Mai2Constants.DATE_TIME_FORMAT - ) - card_list.append(tmp) - - return { - "returnCode": 1, - "length": len(card_list[start_idx:end_idx]), - "nextIndex": next_idx, - "userCardList": card_list[start_idx:end_idx], - } - - async def handle_cm_get_user_item_api_request(self, data: Dict) -> Dict: - await super().handle_get_user_item_api_request(data) - - async def handle_cm_get_user_character_api_request(self, data: Dict) -> Dict: - characters = await self.data.item.get_characters(data["userId"]) - - chara_list = [] - for chara in characters: - chara_list.append( - { - "characterId": chara["characterId"], - # no clue why those values are even needed - "point": 0, - "count": 0, - "level": chara["level"], - "nextAwake": 0, - "nextAwakePercent": 0, - "favorite": False, - "awakening": chara["awakening"], - "useCount": chara["useCount"], - } - ) - - return { - "returnCode": 1, - "length": len(chara_list), - "userCharacterList": chara_list, - } - - async def handle_cm_get_user_card_print_error_api_request(self, data: Dict) -> Dict: - return {"length": 0, "userPrintDetailList": []} - - async def handle_cm_upsert_user_print_api_request(self, data: Dict) -> Dict: - user_id = data["userId"] - upsert = data["userPrintDetail"] - - # set a random card serial number - serial_id = "".join([str(randint(0, 9)) for _ in range(20)]) - - # calculate start and end date of the card - start_date = datetime.utcnow() - end_date = datetime.utcnow() + timedelta(days=15) - - user_card = upsert["userCard"] - await self.data.item.put_card( - user_id, - user_card["cardId"], - user_card["cardTypeId"], - user_card["charaId"], - user_card["mapId"], - # add the correct start date and also the end date in 15 days - start_date, - end_date, - ) - - # get the profile extend to save the new bought card - extend = await self.data.profile.get_profile_extend(user_id, self.version) - if extend: - extend = extend._asdict() - # parse the selectedCardList - # 6 = Freedom Pass, 4 = Gold Pass (cardTypeId) - selected_cards: List = extend["selectedCardList"] - - # if no pass is already added, add the corresponding pass - if not user_card["cardTypeId"] in selected_cards: - selected_cards.insert(0, user_card["cardTypeId"]) - - extend["selectedCardList"] = selected_cards - await self.data.profile.put_profile_extend(user_id, self.version, extend) - - # properly format userPrintDetail for the database - upsert.pop("userCard") - upsert.pop("serialId") - upsert["printDate"] = datetime.strptime(upsert["printDate"], "%Y-%m-%d") - - await self.data.item.put_user_print_detail(user_id, serial_id, upsert) - - return { - "returnCode": 1, - "orderId": 0, - "serialId": serial_id, - "startDate": datetime.strftime(start_date, Mai2Constants.DATE_TIME_FORMAT), - "endDate": datetime.strftime(end_date, Mai2Constants.DATE_TIME_FORMAT), - } - - async def handle_cm_upsert_user_printlog_api_request(self, data: Dict) -> Dict: - return { - "returnCode": 1, - "orderId": 0, - "serialId": data["userPrintlog"]["serialId"], - } - - async def handle_cm_upsert_buy_card_api_request(self, data: Dict) -> Dict: - return {"returnCode": 1}