forked from Hay1tsme/artemis
573 lines
20 KiB
Python
573 lines
20 KiB
Python
import logging
|
|
import json
|
|
from datetime import datetime, timedelta
|
|
from time import strftime
|
|
|
|
import pytz
|
|
from typing import Dict, Any
|
|
|
|
from core.config import CoreConfig
|
|
from titles.chuni.const import ChuniConstants
|
|
from titles.chuni.database import ChuniData
|
|
from titles.chuni.config import ChuniConfig
|
|
|
|
class ChuniBase():
|
|
def __init__(self, core_cfg: CoreConfig, game_cfg: ChuniConfig) -> None:
|
|
self.core_cfg = core_cfg
|
|
self.game_cfg = game_cfg
|
|
self.data = ChuniData(core_cfg)
|
|
self.date_time_format = "%Y-%m-%d %H:%M:%S"
|
|
self.logger = logging.getLogger("chuni")
|
|
self.game = ChuniConstants.GAME_CODE
|
|
self.version = ChuniConstants.VER_CHUNITHM
|
|
|
|
def handle_game_login_api_request(self, data: Dict) -> Dict:
|
|
#self.data.base.log_event("chuni", "login", logging.INFO, {"version": self.version, "user": data["userId"]})
|
|
return { "returnCode": 1 }
|
|
|
|
def handle_game_logout_api_request(self, data: Dict) -> Dict:
|
|
#self.data.base.log_event("chuni", "logout", logging.INFO, {"version": self.version, "user": data["userId"]})
|
|
return { "returnCode": 1 }
|
|
|
|
def handle_get_game_charge_api_request(self, data: Dict) -> Dict:
|
|
game_charge_list = self.data.static.get_enabled_charges(self.version)
|
|
|
|
charges = []
|
|
for x in range(len(game_charge_list)):
|
|
charges.append({
|
|
"orderId": x,
|
|
"chargeId": game_charge_list[x]["chargeId"],
|
|
"price": 1,
|
|
"startDate": "2017-12-05 07:00:00.0",
|
|
"endDate": "2099-12-31 00:00:00.0",
|
|
"salePrice": 1,
|
|
"saleStartDate": "2017-12-05 07:00:00.0",
|
|
"saleEndDate": "2099-12-31 00:00:00.0"
|
|
})
|
|
return {
|
|
"length": len(charges),
|
|
"gameChargeList": charges
|
|
}
|
|
|
|
def handle_get_game_event_api_request(self, data: Dict) -> Dict:
|
|
game_events = self.data.static.get_enabled_events(self.version)
|
|
|
|
event_list = []
|
|
for evt_row in game_events:
|
|
tmp = {}
|
|
tmp["id"] = evt_row["eventId"]
|
|
tmp["type"] = evt_row["type"]
|
|
tmp["startDate"] = "2017-12-05 07:00:00.0"
|
|
tmp["endDate"] = "2099-12-31 00:00:00.0"
|
|
event_list.append(tmp)
|
|
|
|
return {
|
|
"type": data["type"],
|
|
"length": len(event_list),
|
|
"gameEventList": event_list
|
|
}
|
|
|
|
def handle_get_game_idlist_api_request(self, data: Dict) -> Dict:
|
|
return { "type": data["type"], "length": 0, "gameIdlistList": [] }
|
|
|
|
def handle_get_game_message_api_request(self, data: Dict) -> Dict:
|
|
return { "type": data["type"], "length": "0", "gameMessageList": [] }
|
|
|
|
def handle_get_game_ranking_api_request(self, data: Dict) -> Dict:
|
|
return { "type": data["type"], "gameRankingList": [] }
|
|
|
|
def handle_get_game_sale_api_request(self, data: Dict) -> Dict:
|
|
return { "type": data["type"], "length": 0, "gameSaleList": [] }
|
|
|
|
def handle_get_game_setting_api_request(self, data: Dict) -> Dict:
|
|
reboot_start = datetime.strftime(datetime.now() - timedelta(hours=4), self.date_time_format)
|
|
reboot_end = datetime.strftime(datetime.now() - timedelta(hours=3), self.date_time_format)
|
|
return {
|
|
"gameSetting": {
|
|
"dataVersion": "1.00.00",
|
|
"isMaintenance": "false",
|
|
"requestInterval": 10,
|
|
"rebootStartTime": reboot_start,
|
|
"rebootEndTime": reboot_end,
|
|
"isBackgroundDistribute": "false",
|
|
"maxCountCharacter": 300,
|
|
"maxCountItem": 300,
|
|
"maxCountMusic": 300,
|
|
},
|
|
"isDumpUpload": "false",
|
|
"isAou": "false",
|
|
}
|
|
|
|
def handle_get_user_activity_api_request(self, data: Dict) -> Dict:
|
|
user_activity_list = self.data.profile.get_profile_activity(data["userId"], data["kind"])
|
|
|
|
activity_list = []
|
|
|
|
for activity in user_activity_list:
|
|
tmp = activity._asdict()
|
|
tmp.pop("user")
|
|
tmp["id"] = tmp["activityId"]
|
|
tmp.pop("activityId")
|
|
activity_list.append(tmp)
|
|
|
|
return {
|
|
"userId": data["userId"],
|
|
"length": len(activity_list),
|
|
"kind": data["kind"],
|
|
"userActivityList": activity_list
|
|
}
|
|
|
|
def handle_get_user_character_api_request(self, data: Dict) -> Dict:
|
|
characters = self.data.item.get_characters(data["userId"])
|
|
if characters is None: return {}
|
|
next_idx = -1
|
|
|
|
characterList = []
|
|
for x in range(int(data["nextIndex"]), len(characters)):
|
|
tmp = characters[x]._asdict()
|
|
tmp.pop("user")
|
|
tmp.pop("id")
|
|
characterList.append(tmp)
|
|
|
|
if len(characterList) >= int(data["maxCount"]):
|
|
break
|
|
|
|
if len(characterList) >= int(data["maxCount"]) and len(characters) > int(data["maxCount"]) + int(data["nextIndex"]):
|
|
next_idx = int(data["maxCount"]) + int(data["nextIndex"]) + 1
|
|
|
|
return {
|
|
"userId": data["userId"],
|
|
"length": len(characterList),
|
|
"nextIndex": next_idx,
|
|
"userCharacterList": characterList
|
|
}
|
|
|
|
def handle_get_user_charge_api_request(self, data: Dict) -> Dict:
|
|
user_charge_list = self.data.profile.get_profile_charge(data["userId"])
|
|
|
|
charge_list = []
|
|
for charge in user_charge_list:
|
|
tmp = charge._asdict()
|
|
tmp.pop("id")
|
|
tmp.pop("user")
|
|
charge_list.append(tmp)
|
|
|
|
return {
|
|
"userId": data["userId"],
|
|
"length": len(charge_list),
|
|
"userChargeList": charge_list
|
|
}
|
|
|
|
def handle_get_user_course_api_request(self, data: Dict) -> Dict:
|
|
user_course_list = self.data.score.get_courses(data["userId"])
|
|
if user_course_list is None:
|
|
return {
|
|
"userId": data["userId"],
|
|
"length": 0,
|
|
"nextIndex": -1,
|
|
"userCourseList": []
|
|
}
|
|
|
|
course_list = []
|
|
next_idx = int(data["nextIndex"])
|
|
max_ct = int(data["maxCount"])
|
|
|
|
for x in range(next_idx, len(user_course_list)):
|
|
tmp = user_course_list[x]._asdict()
|
|
tmp.pop("user")
|
|
tmp.pop("id")
|
|
course_list.append(tmp)
|
|
|
|
if len(user_course_list) >= max_ct:
|
|
break
|
|
|
|
if len(user_course_list) >= max_ct:
|
|
next_idx = next_idx + max_ct
|
|
else:
|
|
next_idx = -1
|
|
|
|
return {
|
|
"userId": data["userId"],
|
|
"length": len(course_list),
|
|
"nextIndex": next_idx,
|
|
"userCourseList": course_list
|
|
}
|
|
|
|
def handle_get_user_data_api_request(self, data: Dict) -> Dict:
|
|
p = self.data.profile.get_profile_data(data["userId"], self.version)
|
|
if p is None: return {}
|
|
|
|
profile = p._asdict()
|
|
profile.pop("id")
|
|
profile.pop("user")
|
|
profile.pop("version")
|
|
|
|
return {
|
|
"userId": data["userId"],
|
|
"userData": profile
|
|
}
|
|
|
|
def handle_get_user_data_ex_api_request(self, data: Dict) -> Dict:
|
|
p = self.data.profile.get_profile_data_ex(data["userId"], self.version)
|
|
if p is None: return {}
|
|
|
|
profile = p._asdict()
|
|
profile.pop("id")
|
|
profile.pop("user")
|
|
profile.pop("version")
|
|
|
|
return {
|
|
"userId": data["userId"],
|
|
"userDataEx": profile
|
|
}
|
|
|
|
def handle_get_user_duel_api_request(self, data: Dict) -> Dict:
|
|
user_duel_list = self.data.item.get_duels(data["userId"])
|
|
if user_duel_list is None: return {}
|
|
|
|
duel_list = []
|
|
for duel in user_duel_list:
|
|
tmp = duel._asdict()
|
|
tmp.pop("id")
|
|
tmp.pop("user")
|
|
duel_list.append(tmp)
|
|
|
|
return {
|
|
"userId": data["userId"],
|
|
"length": len(duel_list),
|
|
"userDuelList": duel_list
|
|
}
|
|
|
|
def handle_get_user_favorite_item_api_request(self, data: Dict) -> Dict:
|
|
return {
|
|
"userId": data["userId"],
|
|
"length": 0,
|
|
"kind": data["kind"],
|
|
"nextIndex": -1,
|
|
"userFavoriteItemList": []
|
|
}
|
|
|
|
def handle_get_user_favorite_music_api_request(self, data: Dict) -> Dict:
|
|
"""
|
|
This is handled via the webui, which we don't have right now
|
|
"""
|
|
|
|
return {
|
|
"userId": data["userId"],
|
|
"length": 0,
|
|
"userFavoriteMusicList": []
|
|
}
|
|
|
|
def handle_get_user_item_api_request(self, data: Dict) -> Dict:
|
|
kind = int(int(data["nextIndex"]) / 10000000000)
|
|
next_idx = int(int(data["nextIndex"]) % 10000000000)
|
|
user_item_list = self.data.item.get_items(data["userId"], kind)
|
|
|
|
if user_item_list is None or len(user_item_list) == 0:
|
|
return {"userId": data["userId"], "nextIndex": -1, "itemKind": kind, "userItemList": []}
|
|
|
|
items: list[Dict[str, Any]] = []
|
|
for i in range(next_idx, len(user_item_list)):
|
|
tmp = user_item_list[i]._asdict()
|
|
tmp.pop("user")
|
|
tmp.pop("id")
|
|
items.append(tmp)
|
|
if len(items) >= int(data["maxCount"]):
|
|
break
|
|
|
|
xout = kind * 10000000000 + next_idx + len(items)
|
|
|
|
if len(items) < int(data["maxCount"]): nextIndex = 0
|
|
else: nextIndex = xout
|
|
|
|
return {"userId": data["userId"], "nextIndex": nextIndex, "itemKind": kind, "length": len(items), "userItemList": items}
|
|
|
|
def handle_get_user_login_bonus_api_request(self, data: Dict) -> Dict:
|
|
"""
|
|
Unsure how to get this to trigger...
|
|
"""
|
|
return {
|
|
"userId": data["userId"],
|
|
"length": 2,
|
|
"userLoginBonusList": [
|
|
{
|
|
"presetId": '10',
|
|
"bonusCount": '0',
|
|
"lastUpdateDate": "1970-01-01 09:00:00",
|
|
"isWatched": "true"
|
|
},
|
|
{
|
|
"presetId": '20',
|
|
"bonusCount": '0',
|
|
"lastUpdateDate": "1970-01-01 09:00:00",
|
|
"isWatched": "true"
|
|
},
|
|
]
|
|
}
|
|
|
|
def handle_get_user_map_api_request(self, data: Dict) -> Dict:
|
|
user_map_list = self.data.item.get_maps(data["userId"])
|
|
if user_map_list is None: return {}
|
|
|
|
map_list = []
|
|
for map in user_map_list:
|
|
tmp = map._asdict()
|
|
tmp.pop("id")
|
|
tmp.pop("user")
|
|
map_list.append(tmp)
|
|
|
|
return {
|
|
"userId": data["userId"],
|
|
"length": len(map_list),
|
|
"userMapList": map_list
|
|
}
|
|
|
|
def handle_get_user_music_api_request(self, data: Dict) -> Dict:
|
|
music_detail = self.data.score.get_scores(data["userId"])
|
|
if music_detail is None:
|
|
return {
|
|
"userId": data["userId"],
|
|
"length": 0,
|
|
"nextIndex": -1,
|
|
"userMusicList": [] #240
|
|
}
|
|
song_list = []
|
|
next_idx = int(data["nextIndex"])
|
|
max_ct = int(data["maxCount"])
|
|
|
|
for x in range(next_idx, len(music_detail)):
|
|
found = False
|
|
tmp = music_detail[x]._asdict()
|
|
tmp.pop("user")
|
|
tmp.pop("id")
|
|
|
|
for song in song_list:
|
|
if song["userMusicDetailList"][0]["musicId"] == tmp["musicId"]:
|
|
found = True
|
|
song["userMusicDetailList"].append(tmp)
|
|
song["length"] = len(song["userMusicDetailList"])
|
|
|
|
if not found:
|
|
song_list.append({
|
|
"length": 1,
|
|
"userMusicDetailList": [tmp]
|
|
})
|
|
|
|
if len(song_list) >= max_ct:
|
|
break
|
|
|
|
if len(song_list) >= max_ct:
|
|
next_idx += max_ct
|
|
else:
|
|
next_idx = 0
|
|
|
|
return {
|
|
"userId": data["userId"],
|
|
"length": len(song_list),
|
|
"nextIndex": next_idx,
|
|
"userMusicList": song_list #240
|
|
}
|
|
|
|
def handle_get_user_option_api_request(self, data: Dict) -> Dict:
|
|
p = self.data.profile.get_profile_option(data["userId"])
|
|
|
|
option = p._asdict()
|
|
option.pop("id")
|
|
option.pop("user")
|
|
|
|
return {
|
|
"userId": data["userId"],
|
|
"userGameOption": option
|
|
}
|
|
|
|
def handle_get_user_option_ex_api_request(self, data: Dict) -> Dict:
|
|
p = self.data.profile.get_profile_option_ex(data["userId"])
|
|
|
|
option = p._asdict()
|
|
option.pop("id")
|
|
option.pop("user")
|
|
|
|
return {
|
|
"userId": data["userId"],
|
|
"userGameOptionEx": option
|
|
}
|
|
|
|
def read_wtf8(self, src):
|
|
return bytes([ord(c) for c in src]).decode("utf-8")
|
|
|
|
def handle_get_user_preview_api_request(self, data: Dict) -> Dict:
|
|
profile = self.data.profile.get_profile_preview(data["userId"], self.version)
|
|
if profile is None: return None
|
|
profile_character = self.data.item.get_character(data["userId"], profile["characterId"])
|
|
|
|
if profile_character is None:
|
|
chara = {}
|
|
else:
|
|
chara = profile_character._asdict()
|
|
chara.pop("id")
|
|
chara.pop("user")
|
|
|
|
return {
|
|
"userId": data["userId"],
|
|
# Current Login State
|
|
"isLogin": False,
|
|
"lastLoginDate": profile["lastPlayDate"],
|
|
# User Profile
|
|
"userName": profile["userName"],
|
|
"reincarnationNum": profile["reincarnationNum"],
|
|
"level": profile["level"],
|
|
"exp": profile["exp"],
|
|
"playerRating": profile["playerRating"],
|
|
"lastGameId": profile["lastGameId"],
|
|
"lastRomVersion": profile["lastRomVersion"],
|
|
"lastDataVersion": profile["lastDataVersion"],
|
|
"lastPlayDate": profile["lastPlayDate"],
|
|
"trophyId": profile["trophyId"],
|
|
"nameplateId": profile["nameplateId"],
|
|
# Current Selected Character
|
|
"userCharacter": chara,
|
|
# User Game Options
|
|
"playerLevel": profile["playerLevel"],
|
|
"rating": profile["rating"],
|
|
"headphone": profile["headphone"],
|
|
"chargeState": "1",
|
|
"userNameEx": profile["userName"],
|
|
}
|
|
|
|
def handle_get_user_recent_rating_api_request(self, data: Dict) -> Dict:
|
|
recet_rating_list = self.data.profile.get_profile_recent_rating(data["userId"])
|
|
if recet_rating_list is None:
|
|
return {
|
|
"userId": data["userId"],
|
|
"length": 0,
|
|
"userRecentRatingList": [],
|
|
}
|
|
|
|
return {
|
|
"userId": data["userId"],
|
|
"length": len(recet_rating_list["recentRating"]),
|
|
"userRecentRatingList": recet_rating_list["recentRating"],
|
|
}
|
|
|
|
def handle_get_user_region_api_request(self, data: Dict) -> Dict:
|
|
# TODO: Region
|
|
return {
|
|
"userId": data["userId"],
|
|
"length": 0,
|
|
"userRegionList": [],
|
|
}
|
|
|
|
def handle_get_user_team_api_request(self, data: Dict) -> Dict:
|
|
# TODO: Team
|
|
return {
|
|
"userId": data["userId"],
|
|
"teamId": 0
|
|
}
|
|
|
|
def handle_get_team_course_setting_api_request(self, data: Dict) -> Dict:
|
|
return {
|
|
"userId": data["userId"],
|
|
"length": 0,
|
|
"nextIndex": 0,
|
|
"teamCourseSettingList": [],
|
|
}
|
|
|
|
def handle_get_team_course_rule_api_request(self, data: Dict) -> Dict:
|
|
return {
|
|
"userId": data["userId"],
|
|
"length": 0,
|
|
"nextIndex": 0,
|
|
"teamCourseRuleList": [],
|
|
}
|
|
|
|
def handle_upsert_user_all_api_request(self, data: Dict) -> Dict:
|
|
upsert = data["upsertUserAll"]
|
|
user_id = data["userId"]
|
|
|
|
if "userData" in upsert:
|
|
try:
|
|
upsert["userData"][0]["userName"] = self.read_wtf8(upsert["userData"][0]["userName"])
|
|
except: pass
|
|
|
|
self.data.profile.put_profile_data(user_id, self.version, upsert["userData"][0])
|
|
if "userDataEx" in upsert:
|
|
self.data.profile.put_profile_data_ex(user_id, self.version, upsert["userDataEx"][0])
|
|
if "userGameOption" in upsert:
|
|
self.data.profile.put_profile_option(user_id, upsert["userGameOption"][0])
|
|
if "userGameOptionEx" in upsert:
|
|
self.data.profile.put_profile_option_ex(user_id, upsert["userGameOptionEx"][0])
|
|
if "userRecentRatingList" in upsert:
|
|
self.data.profile.put_profile_recent_rating(user_id, upsert["userRecentRatingList"])
|
|
|
|
if "userCharacterList" in upsert:
|
|
for character in upsert["userCharacterList"]:
|
|
self.data.item.put_character(user_id, character)
|
|
|
|
if "userMapList" in upsert:
|
|
for map in upsert["userMapList"]:
|
|
self.data.item.put_map(user_id, map)
|
|
|
|
if "userCourseList" in upsert:
|
|
for course in upsert["userCourseList"]:
|
|
self.data.score.put_course(user_id, course)
|
|
|
|
if "userDuelList" in upsert:
|
|
for duel in upsert["userDuelList"]:
|
|
self.data.item.put_duel(user_id, duel)
|
|
|
|
if "userItemList" in upsert:
|
|
for item in upsert["userItemList"]:
|
|
self.data.item.put_item(user_id, item)
|
|
|
|
if "userActivityList" in upsert:
|
|
for activity in upsert["userActivityList"]:
|
|
self.data.profile.put_profile_activity(user_id, activity)
|
|
|
|
if "userChargeList" in upsert:
|
|
for charge in upsert["userChargeList"]:
|
|
self.data.profile.put_profile_charge(user_id, charge)
|
|
|
|
if "userMusicDetailList" in upsert:
|
|
for song in upsert["userMusicDetailList"]:
|
|
self.data.score.put_score(user_id, song)
|
|
|
|
if "userPlaylogList" in upsert:
|
|
for playlog in upsert["userPlaylogList"]:
|
|
self.data.score.put_playlog(user_id, playlog)
|
|
|
|
if "userTeamPoint" in upsert:
|
|
# TODO: team stuff
|
|
pass
|
|
|
|
if "userMapAreaList" in upsert:
|
|
for map_area in upsert["userMapAreaList"]:
|
|
self.data.item.put_map_area(user_id, map_area)
|
|
|
|
if "userOverPowerList" in upsert:
|
|
for overpower in upsert["userOverPowerList"]:
|
|
self.data.profile.put_profile_overpower(user_id, overpower)
|
|
|
|
if "userEmoneyList" in upsert:
|
|
for emoney in upsert["userEmoneyList"]:
|
|
self.data.profile.put_profile_emoney(user_id, emoney)
|
|
|
|
return { "returnCode": "1" }
|
|
|
|
def handle_upsert_user_chargelog_api_request(self, data: Dict) -> Dict:
|
|
return { "returnCode": "1" }
|
|
|
|
def handle_upsert_client_bookkeeping_api_request(self, data: Dict) -> Dict:
|
|
return { "returnCode": "1" }
|
|
|
|
def handle_upsert_client_develop_api_request(self, data: Dict) -> Dict:
|
|
return { "returnCode": "1" }
|
|
|
|
def handle_upsert_client_error_api_request(self, data: Dict) -> Dict:
|
|
return { "returnCode": "1" }
|
|
|
|
def handle_upsert_client_setting_api_request(self, data: Dict) -> Dict:
|
|
return { "returnCode": "1" }
|
|
|
|
def handle_upsert_client_testmode_api_request(self, data: Dict) -> Dict:
|
|
return { "returnCode": "1" }
|