2506 lines
91 KiB
Python
2506 lines
91 KiB
Python
from datetime import datetime, timedelta
|
||
import os
|
||
from random import choice
|
||
from typing import Any, Dict, List
|
||
import json
|
||
import logging
|
||
|
||
from core.config import CoreConfig
|
||
from titles.idac.const import IDACConstants
|
||
from titles.idac.config import IDACConfig
|
||
from titles.idac.base import IDACBase
|
||
|
||
|
||
class IDACSeason2(IDACBase):
|
||
def __init__(self, core_cfg: CoreConfig, game_cfg: IDACConfig) -> None:
|
||
super().__init__(core_cfg, game_cfg)
|
||
self.version = IDACConstants.VER_IDAC_SEASON_2
|
||
|
||
# load the play stamps and timetrial events into memory
|
||
self.stamp_info = []
|
||
if self.game_config.stamp.enable:
|
||
for stamp in self.game_config.stamp.enabled_stamps:
|
||
if not os.path.exists(f"./titles/idac/data/stamps/{stamp}.json"):
|
||
self.logger.warning(f"Stamp {stamp} is enabled but does not exist!")
|
||
continue
|
||
|
||
with open(
|
||
f"./titles/idac/data/stamps/{stamp}.json", encoding="UTF-8"
|
||
) as f:
|
||
self.logger.debug(f"Loading stamp {stamp}")
|
||
self.stamp_info.append(self._fix_dates(json.load(f)))
|
||
|
||
self.timetrial_event = {}
|
||
self.timetrial_event_id = None
|
||
if self.game_config.timetrial.enable:
|
||
timetrial = self.game_config.timetrial.enabled_timetrial
|
||
if timetrial is not None:
|
||
if not os.path.exists(f"./titles/idac/data/timetrial/{timetrial}.json"):
|
||
self.logger.warning(
|
||
f"Timetrial {timetrial} is enabled but does not exist!"
|
||
)
|
||
else:
|
||
self.logger.debug(f"Loading timetrial {timetrial}")
|
||
with open(
|
||
f"./titles/idac/data/timetrial/{timetrial}.json",
|
||
encoding="UTF-8",
|
||
) as f:
|
||
self.timetrial_event = self._fix_dates(json.load(f))
|
||
|
||
# required for saving
|
||
self.timetrial_event_id = self.timetrial_event.get(
|
||
"timetrial_event_id"
|
||
)
|
||
|
||
def handle_alive_get_request(self, data: Dict, headers: Dict):
|
||
return {
|
||
"status_code": "0",
|
||
# 1 = success, 0 = failed
|
||
"server_status": 1,
|
||
"force_reboot_time": int(datetime.now().timestamp()) - 86400,
|
||
}
|
||
|
||
def _fix_dates(self, input: dict):
|
||
"""
|
||
Fix "start_dt" and "end_dt" dates in a JSON file.
|
||
"""
|
||
output = {}
|
||
|
||
self.logger.debug(f"Fixing dates in {type(input)}")
|
||
for key, value in input.items():
|
||
if key in {"start_dt", "end_dt"}:
|
||
if isinstance(value, str):
|
||
value = int(datetime.strptime(value, "%Y-%m-%d").timestamp())
|
||
|
||
output[key] = value
|
||
return output
|
||
|
||
def handle_boot_getconfigdata_request(self, data: Dict, headers: Dict):
|
||
"""
|
||
category:
|
||
1 = D Coin
|
||
3 = Car Dressup Token
|
||
5 = Avatar Dressup Token
|
||
6 = Tachometer
|
||
7 = Aura
|
||
8 = Aura Color
|
||
9 = Avatar Face
|
||
10 = Avatar Eye
|
||
11 = Avatar Mouth
|
||
12 = Avatar Hair
|
||
13 = Avatar Glasses
|
||
14 = Avatar Face accessories
|
||
15 = Avatar Body
|
||
18 = Avatar Background
|
||
21 = Chat Stamp
|
||
22 = Keychain
|
||
24 = Title
|
||
25 = FullTune Ticket
|
||
26 = Paper Cup
|
||
27 = BGM
|
||
28 = Drifting Text
|
||
31 = Start Menu BG
|
||
32 = Car Color/Paint
|
||
33 = Aura Level
|
||
34 = FullTune Ticket Fragment
|
||
35 = Underneon Lights
|
||
"""
|
||
version = headers["device_version"]
|
||
ver_str = version.replace(".", "")[:3]
|
||
|
||
if self.core_cfg.server.is_develop:
|
||
domain_api_game = f"http://{self.core_cfg.title.hostname}:{self.core_cfg.title.port}/SDGT/{ver_str}/"
|
||
else:
|
||
domain_api_game = f"http://{self.core_cfg.title.hostname}/SDGT/{ver_str}/"
|
||
|
||
return {
|
||
"status_code": "0",
|
||
"free_continue_enable": 1,
|
||
"free_continue_new": 1,
|
||
"free_continue_play": 1,
|
||
"difference_time_to_jp": 0,
|
||
# has to match the game asset version to show theory of street
|
||
"asset_version": "1",
|
||
# option version? MV01?
|
||
"optional_version": "1",
|
||
"disconnect_offset": 0,
|
||
"boost_balance_version": "0",
|
||
"time_release_number": "0",
|
||
"play_stamp_enable": 1,
|
||
"play_stamp_bonus_coin": 1,
|
||
"gacha_chara_needs": 1,
|
||
"both_win_system_control": 1,
|
||
"subcard_system_congrol": 1,
|
||
"server_maintenance_start_hour": 0,
|
||
"server_maintenance_start_minutes": 0,
|
||
"server_maintenance_end_hour": 0,
|
||
"server_maintenance_end_minutes": 0,
|
||
"domain_api_game": domain_api_game,
|
||
"domain_matching": f"http://{self.core_cfg.title.hostname}:{self.game_config.server.matching}",
|
||
"domain_echo1": f"{self.core_cfg.title.hostname}:{self.game_config.server.echo1}",
|
||
"domain_echo2": f"{self.core_cfg.title.hostname}:{self.game_config.server.echo2}",
|
||
"domain_ping": f"{self.core_cfg.title.hostname}",
|
||
"battle_gift_event_master": [],
|
||
"round_event": [
|
||
{
|
||
"round_event_id": 30,
|
||
"round_event_nm": f"{self.core_cfg.server.name} Event",
|
||
"start_dt": int(
|
||
datetime.strptime("2023-01-01", "%Y-%m-%d").timestamp()
|
||
),
|
||
"end_dt": int(
|
||
datetime.strptime("2029-01-01", "%Y-%m-%d").timestamp()
|
||
),
|
||
"round_start_rank": 0,
|
||
"save_filename": "0",
|
||
# https://info-initialdac.sega.jp/1898/
|
||
"vscount": [
|
||
{
|
||
"reward_upper_limit": 10,
|
||
"reward_lower_limit": 10,
|
||
"reward": [{"reward_category": 21, "reward_type": 483}],
|
||
},
|
||
{
|
||
"reward_upper_limit": 40,
|
||
"reward_lower_limit": 40,
|
||
"reward": [{"reward_category": 21, "reward_type": 484}],
|
||
},
|
||
{
|
||
"reward_upper_limit": 80,
|
||
"reward_lower_limit": 80,
|
||
"reward": [{"reward_category": 22, "reward_type": 516}],
|
||
},
|
||
{
|
||
"reward_upper_limit": 120,
|
||
"reward_lower_limit": 120,
|
||
"reward": [{"reward_category": 21, "reward_type": 461}],
|
||
},
|
||
{
|
||
"reward_upper_limit": 180,
|
||
"reward_lower_limit": 180,
|
||
"reward": [{"reward_category": 21, "reward_type": 462}],
|
||
}
|
||
],
|
||
"rank": [],
|
||
"point": [],
|
||
"playable_course_list": [
|
||
{"course_id": 4, "course_day": 0},
|
||
{"course_id": 4, "course_day": 1},
|
||
{"course_id": 6, "course_day": 0},
|
||
{"course_id": 6, "course_day": 1},
|
||
{"course_id": 8, "course_day": 0},
|
||
{"course_id": 8, "course_day": 1},
|
||
{"course_id": 10, "course_day": 0},
|
||
{"course_id": 10, "course_day": 1},
|
||
{"course_id": 12, "course_day": 0},
|
||
{"course_id": 12, "course_day": 1},
|
||
{"course_id": 14, "course_day": 0},
|
||
{"course_id": 14, "course_day": 1},
|
||
{"course_id": 16, "course_day": 0},
|
||
{"course_id": 16, "course_day": 1},
|
||
{"course_id": 18, "course_day": 0},
|
||
{"course_id": 18, "course_day": 1},
|
||
{"course_id": 20, "course_day": 0},
|
||
{"course_id": 20, "course_day": 1},
|
||
{"course_id": 22, "course_day": 0},
|
||
{"course_id": 22, "course_day": 1},
|
||
{"course_id": 24, "course_day": 0},
|
||
{"course_id": 24, "course_day": 1},
|
||
{"course_id": 26, "course_day": 0},
|
||
{"course_id": 26, "course_day": 1},
|
||
{"course_id": 36, "course_day": 0},
|
||
{"course_id": 36, "course_day": 1},
|
||
{"course_id": 38, "course_day": 0},
|
||
{"course_id": 38, "course_day": 1},
|
||
{"course_id": 40, "course_day": 0},
|
||
{"course_id": 40, "course_day": 1},
|
||
{"course_id": 42, "course_day": 0},
|
||
{"course_id": 42, "course_day": 1},
|
||
{"course_id": 44, "course_day": 0},
|
||
{"course_id": 44, "course_day": 1},
|
||
{"course_id": 46, "course_day": 0},
|
||
{"course_id": 46, "course_day": 1},
|
||
{"course_id": 48, "course_day": 0},
|
||
{"course_id": 48, "course_day": 1},
|
||
{"course_id": 50, "course_day": 0},
|
||
{"course_id": 50, "course_day": 1},
|
||
{"course_id": 52, "course_day": 0},
|
||
{"course_id": 52, "course_day": 1},
|
||
{"course_id": 54, "course_day": 0},
|
||
{"course_id": 54, "course_day": 1},
|
||
{"course_id": 56, "course_day": 0},
|
||
{"course_id": 56, "course_day": 1},
|
||
{"course_id": 58, "course_day": 0},
|
||
{"course_id": 58, "course_day": 1},
|
||
{"course_id": 68, "course_day": 0},
|
||
{"course_id": 68, "course_day": 1},
|
||
{"course_id": 70, "course_day": 0},
|
||
{"course_id": 70, "course_day": 1},
|
||
],
|
||
}
|
||
],
|
||
"last_round_event": [],
|
||
"last_round_event_ranking": [],
|
||
"round_event_exp": [],
|
||
"stamp_info": self.stamp_info,
|
||
# 0 = use default data, 1+ = server version of timereleasedata response
|
||
"timerelease_no": 3,
|
||
# 0 = use default data, 1+ = server version of gachadata response
|
||
"timerelease_avatar_gacha_no": 3,
|
||
"takeover_reward": [],
|
||
"subcard_judge": [
|
||
{
|
||
"condition_id": 1,
|
||
"lower_rank": 0,
|
||
"higher_rank": 10,
|
||
"condition_start": 2,
|
||
"condition_end": 3,
|
||
}
|
||
],
|
||
"special_promote": [{"counter": 1, "online_rank_id": 1}],
|
||
"matching_id": 1,
|
||
"matching_group": [
|
||
{
|
||
"group_id": 1,
|
||
"group_percent": 1,
|
||
}
|
||
],
|
||
"timetrial_disp_date": int(
|
||
datetime.strptime("2023-10-01", "%Y-%m-%d").timestamp()
|
||
),
|
||
# price for every car
|
||
"buy_car_need_cash": 5000,
|
||
# number of buyable shop/customization time limits
|
||
"time_extension_limit": 1,
|
||
"collabo_id": 0,
|
||
"driver_debut_end_date": int(
|
||
datetime.strptime("2029-01-01", "%Y-%m-%d").timestamp()
|
||
),
|
||
"online_battle_param1": 1,
|
||
"online_battle_param2": 1,
|
||
"online_battle_param3": 1,
|
||
"online_battle_param4": 1,
|
||
"online_battle_param5": 1,
|
||
"online_battle_param6": 1,
|
||
"online_battle_param7": 1,
|
||
"online_battle_param8": 1,
|
||
"theory_open_version": "1.30",
|
||
"theory_close_version": "1.50",
|
||
"special_mode_data": {
|
||
"start_dt": int(
|
||
datetime.strptime("2023-01-01", "%Y-%m-%d").timestamp()
|
||
),
|
||
"end_dt": int(datetime.strptime("2029-01-01", "%Y-%m-%d").timestamp()),
|
||
"story_type": 4, # touhou special event
|
||
},
|
||
"timetrial_event_data": self.timetrial_event,
|
||
}
|
||
|
||
def handle_boot_bookkeep_request(self, data: Dict, headers: Dict):
|
||
pass
|
||
|
||
def handle_boot_getgachadata_request(self, data: Dict, headers: Dict):
|
||
"""
|
||
Reward category types:
|
||
9: Face
|
||
10: Eye
|
||
11: Mouth
|
||
12: Hair
|
||
13: Glasses
|
||
14: Face accessories
|
||
15: Body
|
||
18: Background
|
||
"""
|
||
|
||
with open("./titles/idac/data/avatarGacha.json", encoding="UTF-8") as f:
|
||
avatar_gacha_data = json.load(f)
|
||
|
||
# avatar_gacha_data = {
|
||
# "status_code": "0",
|
||
# "avatar_gacha_data": [
|
||
# {
|
||
# "avatar_gacha_id": 0,
|
||
# "avatar_gacha_nm": "Standard",
|
||
# "gacha_type": 0,
|
||
# "save_filename": "0",
|
||
# "use_ticket_cnt": 1,
|
||
# "start_dt": int(
|
||
# datetime.strptime("2019-01-01", "%Y-%m-%d").timestamp()
|
||
# ),
|
||
# "end_dt": int(
|
||
# datetime.strptime("2029-01-01", "%Y-%m-%d").timestamp()
|
||
# ),
|
||
# "gacha_reward": [
|
||
# {
|
||
# "reward_id": 117,
|
||
# "reward_type": 118,
|
||
# "reward_category": 18,
|
||
# "rate": 1000,
|
||
# "pickup_flag": 0,
|
||
# },
|
||
# ],
|
||
# }
|
||
# ],
|
||
# }
|
||
|
||
self.logger.debug(
|
||
f'Available avatar gacha items: {len(avatar_gacha_data["avatar_gacha_data"][0]["gacha_reward"])}'
|
||
)
|
||
|
||
return avatar_gacha_data
|
||
|
||
def handle_boot_gettimereleasedata_request(self, data: Dict, headers: Dict):
|
||
"""
|
||
timerelease chapter:
|
||
1 = Story: 1, 2, 3, 4, 5, 6, 7, 8, 9, 19 (Chapter 10), (29 Chapter 11 lol?)
|
||
2 = MF Ghost: 10, 11, 12, 13, 14, 15
|
||
3 = Bunta: 15, 16, 17, 18, 19, 20, (21, 21, 22?)
|
||
4 = Special Event: 23, 24, 25, 26, 27, 28 (Touhou Project)
|
||
"""
|
||
path = "./titles/idac/data/"
|
||
|
||
# 1.00.00 is default
|
||
device_version_data = headers.get("device_version", "1.00.00")
|
||
device_version = int(device_version_data.replace(".", "")[:-2])
|
||
|
||
timerelease_filename = f"timeRelease_v{device_version:04d}"
|
||
timerelease_path = f"{path}{timerelease_filename}.json"
|
||
|
||
# if the file doesn't exist, try to find the next lowest version
|
||
if not os.path.exists(timerelease_path):
|
||
while device_version > 100:
|
||
device_version -= 1
|
||
timerelease_filename = f"timeRelease_v{device_version:04d}"
|
||
timerelease_path = f"{path}{timerelease_filename}.json"
|
||
|
||
# if the file exists, break out of the loop
|
||
if os.path.exists(timerelease_path):
|
||
break
|
||
|
||
self.logger.debug(f"Using time release file: {timerelease_filename}")
|
||
# load the time release data
|
||
with open(f"{path}{timerelease_filename}.json") as f:
|
||
time_release_data = json.load(f)
|
||
|
||
return time_release_data
|
||
|
||
def handle_advertise_getrankingdata_request(self, data: Dict, headers: Dict):
|
||
best_data = []
|
||
for last_update in data.get("last_update_date"):
|
||
course_id = last_update.get("course_id")
|
||
|
||
ranking = self.data.item.get_time_trial_ranking_by_course(
|
||
self.version, course_id
|
||
)
|
||
ranking_data = []
|
||
for i, rank in enumerate(ranking):
|
||
user_id = rank["user"]
|
||
|
||
# get the username, country and store from the profile
|
||
profile = self.data.profile.get_profile(user_id, self.version)
|
||
arcade = self.data.arcade.get_arcade(profile["store"])
|
||
|
||
if arcade is None:
|
||
arcade = {}
|
||
arcade["name"] = self.core_cfg.server.name
|
||
|
||
# should never happen
|
||
if profile is None:
|
||
continue
|
||
|
||
ranking_data.append(
|
||
{
|
||
"course_id": course_id,
|
||
"rank": i + 1,
|
||
"username": profile["username"],
|
||
"value": rank["goal_time"],
|
||
# gat the store name from the profile
|
||
"store": arcade["name"],
|
||
# get the country id from the profile, 9 is JPN
|
||
"country": profile["country"],
|
||
"style_car_id": rank["style_car_id"],
|
||
# convert the datetime to a timestamp
|
||
"play_dt": int(rank["play_dt"].timestamp()),
|
||
"section_time_1": rank["section_time_1"],
|
||
"section_time_2": rank["section_time_2"],
|
||
"section_time_3": rank["section_time_3"],
|
||
"section_time_4": rank["section_time_4"],
|
||
"mission": rank["mission"],
|
||
}
|
||
)
|
||
|
||
best_data.append(
|
||
{
|
||
"course_id": course_id,
|
||
"ranking_data": ranking_data,
|
||
}
|
||
)
|
||
|
||
return {
|
||
"status_code": "0",
|
||
"national_best_data": best_data,
|
||
"shop_best_data": best_data,
|
||
"rank_management_flag": 0,
|
||
}
|
||
|
||
def handle_login_checklock_request(self, data: Dict, headers: Dict):
|
||
user_id = data["id"]
|
||
access_code = data["accesscode"]
|
||
is_new_player = 0
|
||
|
||
# check that the user_id from access_code matches the user_id
|
||
if user_id == self.data.card.get_user_id_from_card(access_code):
|
||
lock_result = 1
|
||
|
||
# check if an IDAC profile already exists
|
||
p = self.data.profile.get_profile(user_id, self.version)
|
||
is_new_player = 1 if p is None else 0
|
||
else:
|
||
lock_result = 0
|
||
user_id = ""
|
||
|
||
# other: in use
|
||
return {
|
||
"status_code": "0",
|
||
# 0 = already in use, 1 = good, 2 = too new
|
||
"lock_result": lock_result,
|
||
"lock_date": int(datetime.now().timestamp()),
|
||
"daily_play": 1,
|
||
"session": f"{user_id}",
|
||
"shared_security_key": "a",
|
||
"session_procseq": "a",
|
||
"new_player": is_new_player,
|
||
"server_status": 1,
|
||
}
|
||
|
||
def handle_login_unlock_request(self, data: Dict, headers: Dict):
|
||
return {
|
||
"status_code": "0",
|
||
"lock_result": 1,
|
||
}
|
||
|
||
def handle_login_relock_request(self, data: Dict, headers: Dict):
|
||
return {
|
||
"status_code": "0",
|
||
"lock_result": 1,
|
||
"lock_date": int(datetime.now().timestamp()),
|
||
}
|
||
|
||
def handle_login_guestplay_request(self, data: Dict, headers: Dict):
|
||
# TODO
|
||
pass
|
||
|
||
def _generate_story_data(self, user_id: int) -> Dict:
|
||
stories = self.data.item.get_stories(user_id)
|
||
|
||
story_data = []
|
||
for s in stories:
|
||
chapter_id = s["chapter"]
|
||
episodes = self.data.item.get_story_episodes(user_id, chapter_id)
|
||
|
||
episode_data = []
|
||
for e in episodes:
|
||
episode_id = e["episode"]
|
||
difficulties = self.data.item.get_story_episode_difficulties(
|
||
user_id, episode_id
|
||
)
|
||
|
||
difficulty_data = []
|
||
for d in difficulties:
|
||
difficulty_data.append(
|
||
{
|
||
"difficulty": d["difficulty"],
|
||
"play_count": d["play_count"],
|
||
"clear_count": d["clear_count"],
|
||
"play_status": d["play_status"],
|
||
"play_score": d["play_score"],
|
||
}
|
||
)
|
||
|
||
episode_data.append(
|
||
{
|
||
"episode": e["episode"],
|
||
"play_status": e["play_status"],
|
||
"difficulty_data": difficulty_data,
|
||
}
|
||
)
|
||
|
||
story_data.append(
|
||
{
|
||
"story_type": s["story_type"],
|
||
"chapter": s["chapter"],
|
||
"loop_count": s["loop_count"],
|
||
"episode_data": episode_data,
|
||
}
|
||
)
|
||
|
||
return story_data
|
||
|
||
def _generate_special_data(self, user_id: int) -> Dict:
|
||
# 4 = special mode
|
||
specials = self.data.item.get_best_challenges_by_vs_type(user_id, story_type=4)
|
||
|
||
special_data = []
|
||
for s in specials:
|
||
special_data.append(
|
||
{
|
||
"story_type": s["story_type"],
|
||
"vs_type": s["vs_type"],
|
||
"max_clear_lv": s["max_clear_lv"],
|
||
"last_play_lv": s["last_play_lv"],
|
||
# change to last_play_course_id?
|
||
"last_play_course_id": s["course_id"],
|
||
}
|
||
)
|
||
|
||
return special_data
|
||
|
||
def _generate_challenge_data(self, user_id: int) -> Dict:
|
||
# challenge mode (Bunta challenge only right now)
|
||
challenges = self.data.item.get_best_challenges_by_vs_type(
|
||
user_id, story_type=3
|
||
)
|
||
|
||
challenge_data = []
|
||
for c in challenges:
|
||
challenge_data.append(
|
||
{
|
||
"story_type": c["story_type"],
|
||
"vs_type": c["vs_type"],
|
||
"max_clear_lv": c["max_clear_lv"],
|
||
"last_play_lv": c["last_play_lv"],
|
||
# change to last_play_course_id?
|
||
"last_play_course_id": c["course_id"],
|
||
"play_count": c["play_count"],
|
||
}
|
||
)
|
||
|
||
return challenge_data
|
||
|
||
def _save_stock_data(self, user_id: int, stock_data: Dict):
|
||
updated_stock_data = {}
|
||
for k, v in stock_data.items():
|
||
if v != "":
|
||
updated_stock_data[k] = v
|
||
|
||
if updated_stock_data:
|
||
self.data.profile.put_profile_stock(
|
||
user_id, self.version, updated_stock_data
|
||
)
|
||
|
||
def handle_user_getdata_request(self, data: Dict, headers: Dict):
|
||
user_id = int(headers["session"])
|
||
|
||
# get the user's profile, can never be None
|
||
p = self.data.profile.get_profile(user_id, self.version)
|
||
user_data = p._asdict()
|
||
arcade = self.data.arcade.get_arcade(user_data["store"])
|
||
|
||
del user_data["id"]
|
||
del user_data["user"]
|
||
del user_data["version"]
|
||
user_data["id"] = user_id
|
||
user_data["store_name"] = (
|
||
self.core_cfg.server.name if arcade is None else arcade["name"]
|
||
)
|
||
user_data["last_play_date"] = int(user_data["last_play_date"].timestamp())
|
||
user_data["create_date"] = int(user_data["create_date"].timestamp())
|
||
|
||
# get the user's rank
|
||
r = self.data.profile.get_profile_rank(user_id, self.version)
|
||
rank_data = r._asdict()
|
||
del rank_data["id"]
|
||
del rank_data["user"]
|
||
del rank_data["version"]
|
||
|
||
# add the mode_rank_data to the user_data
|
||
user_data["mode_rank_data"] = rank_data
|
||
|
||
# get the user's avatar
|
||
a = self.data.profile.get_profile_avatar(user_id)
|
||
avatar_data = a._asdict()
|
||
del avatar_data["id"]
|
||
del avatar_data["user"]
|
||
|
||
# get the user's stock
|
||
s = self.data.profile.get_profile_stock(user_id, self.version)
|
||
stock_data = s._asdict()
|
||
del stock_data["id"]
|
||
del stock_data["user"]
|
||
del stock_data["version"]
|
||
|
||
# get the user's config
|
||
c = self.data.profile.get_profile_config(user_id)
|
||
config_data = c._asdict()
|
||
del config_data["id"]
|
||
del config_data["user"]
|
||
config_data["id"] = config_data.pop("config_id")
|
||
|
||
# get the user's ticket
|
||
tickets: list = self.data.item.get_tickets(user_id)
|
||
|
||
"""
|
||
ticket_id:
|
||
3 = Car Dressup Points
|
||
5 = Avatar Dressup Points
|
||
25 = Full Tune Tickets
|
||
34 = Full Tune Fragments
|
||
"""
|
||
|
||
ticket_data = []
|
||
for ticket in tickets:
|
||
ticket_data.append(
|
||
{
|
||
"ticket_id": ticket["ticket_id"],
|
||
"ticket_cnt": ticket["ticket_cnt"],
|
||
}
|
||
)
|
||
|
||
# get the user's course, required for the "course proeficiency"
|
||
courses = self.data.item.get_courses(user_id)
|
||
course_data = []
|
||
for course in courses:
|
||
course_data.append(
|
||
{
|
||
"id": 0, # no clue, always 0?
|
||
"course_id": course["course_id"],
|
||
"run_counts": course["run_counts"],
|
||
# "course proeficiency" in exp points
|
||
"skill_level_exp": course["skill_level_exp"],
|
||
}
|
||
)
|
||
|
||
# get the profile theory data
|
||
theory_data = {}
|
||
theory = self.data.profile.get_profile_theory(user_id, self.version)
|
||
if theory is not None:
|
||
theory_data = theory._asdict()
|
||
del theory_data["id"]
|
||
del theory_data["user"]
|
||
del theory_data["version"]
|
||
|
||
# get the users theory course data
|
||
theory_course_data = []
|
||
theory_courses = self.data.item.get_theory_courses(user_id)
|
||
for course in theory_courses:
|
||
tmp = course._asdict()
|
||
del tmp["id"]
|
||
del tmp["user"]
|
||
tmp["update_dt"] = int(tmp["update_dt"].timestamp())
|
||
|
||
theory_course_data.append(tmp)
|
||
|
||
# get the users theory partner data
|
||
theory_partner_data = []
|
||
theory_partners = self.data.item.get_theory_partners(user_id)
|
||
for partner in theory_partners:
|
||
tmp = partner._asdict()
|
||
del tmp["id"]
|
||
del tmp["user"]
|
||
|
||
theory_partner_data.append(tmp)
|
||
|
||
# get the users theory running pram data
|
||
theory_running_pram_data = []
|
||
theory_running = self.data.item.get_theory_running(user_id)
|
||
for running in theory_running:
|
||
tmp = running._asdict()
|
||
del tmp["id"]
|
||
del tmp["user"]
|
||
|
||
theory_running_pram_data.append(tmp)
|
||
|
||
# get the users vs info data
|
||
vs_info_data = []
|
||
vs_info = self.data.item.get_vs_infos(user_id)
|
||
for vs in vs_info:
|
||
vs_info_data.append(
|
||
{
|
||
"battle_mode": 1,
|
||
"vs_cnt": 1,
|
||
"vs_win": vs["win_flg"],
|
||
"invalid": 0,
|
||
"str": 0,
|
||
"str_now": 0,
|
||
"lose_now": 0,
|
||
"vs_history": vs["vs_history"],
|
||
"course_select_priority": 0,
|
||
"vsinfo_course_data": [
|
||
{
|
||
"course_id": vs["course_id"],
|
||
"vs_cnt": 1,
|
||
"vs_win": vs["win_flg"],
|
||
}
|
||
],
|
||
}
|
||
)
|
||
|
||
# get the user's car
|
||
cars = self.data.item.get_cars(self.version, user_id, only_pickup=True)
|
||
fulltune_count = 0
|
||
total_car_parts_count = 0
|
||
car_data = []
|
||
for car in cars:
|
||
tmp = car._asdict()
|
||
del tmp["id"]
|
||
del tmp["user"]
|
||
del tmp["version"]
|
||
|
||
car_data.append(tmp)
|
||
# tune_level of 16 means fully tuned, so add 1 to fulltune_count
|
||
if car["tune_level"] >= 16:
|
||
fulltune_count += 1
|
||
|
||
# add the number of car parts to total_car_parts_count?
|
||
# total_car_parts_count += tmp["total_car_parts_count"]
|
||
|
||
car_data.append(tmp)
|
||
|
||
# update user profile car count
|
||
user_data["have_car_cnt"] = len(car_data)
|
||
|
||
# get the user's play stamps
|
||
stamps = self.data.item.get_stamps(user_id)
|
||
stamp_event_data = []
|
||
for stamp in stamps:
|
||
tmp = stamp._asdict()
|
||
del tmp["id"]
|
||
del tmp["user"]
|
||
|
||
now = datetime.now()
|
||
|
||
# create timestamp for today at 1am
|
||
this_day = now.replace(hour=1, minute=0, second=0, microsecond=0)
|
||
|
||
# check if this_day is greater than or equal to create_date_daily
|
||
if this_day >= tmp["create_date_daily"]:
|
||
# reset the daily stamp
|
||
tmp["create_date_daily"] = now
|
||
tmp["daily_bonus"] = 0
|
||
|
||
# create a timestamp for this monday at 1am
|
||
this_monday = now - timedelta(days=now.weekday())
|
||
this_monday = this_monday.replace(hour=1, minute=0, second=0, microsecond=0)
|
||
|
||
# check if this_monday is greater than or equal to create_date_weekly
|
||
if this_monday >= tmp["create_date_weekly"]:
|
||
# reset the weekly stamp
|
||
tmp["create_date_weekly"] = now
|
||
tmp["weekly_bonus"] = 0
|
||
|
||
# update the play stamp in the database
|
||
self.data.item.put_stamp(user_id, tmp)
|
||
|
||
del tmp["create_date_daily"]
|
||
del tmp["create_date_weekly"]
|
||
stamp_event_data.append(tmp)
|
||
|
||
# get the user's timetrial event data
|
||
timetrial_event_data = {}
|
||
timetrial = self.data.item.get_timetrial_event(user_id, self.timetrial_event_id)
|
||
if timetrial is not None:
|
||
timetrial_event_data = {
|
||
"timetrial_event_id": timetrial["timetrial_event_id"],
|
||
"point": timetrial["point"],
|
||
}
|
||
|
||
return {
|
||
"status_code": "0",
|
||
"user_base_data": user_data,
|
||
"avatar_data": avatar_data,
|
||
"pick_up_car_data": car_data,
|
||
"story_data": self._generate_story_data(user_id),
|
||
"vsinfo_data": vs_info_data,
|
||
"stock_data": stock_data,
|
||
"mission_data": {
|
||
"id": 0,
|
||
"achieve_flag": 0,
|
||
"received_flag": 0,
|
||
"update_dt": int(datetime.now().timestamp() - 86400),
|
||
},
|
||
"weekly_mission_data": [],
|
||
"course_data": course_data,
|
||
"toppatu_event_data": {
|
||
"id": 0,
|
||
"event_id": 0,
|
||
"count1": 0,
|
||
"count2": 0,
|
||
"count3": 0,
|
||
"accept_flag": 0,
|
||
},
|
||
"event_data": {
|
||
"id": 0,
|
||
"active_event_id": 0,
|
||
"dialog_show_date": int(datetime.now().timestamp() - 86400),
|
||
"show_start_dialog_flag": 1,
|
||
"show_progress_dialog_flag": 1,
|
||
"show_end_dialog_flag": 1,
|
||
"end_event_id": 0,
|
||
},
|
||
"rewards_data": {},
|
||
"login_bonus_data": {
|
||
"gacha_id": 0,
|
||
"gacha_item_id": 0,
|
||
"category": 0,
|
||
"type": 0,
|
||
},
|
||
"frozen_data": {"frozen_status": 2},
|
||
"penalty_data": {"penalty_flag": 0, "penalty_2_level": 0},
|
||
"config_data": config_data,
|
||
"battle_gift_data": [],
|
||
"ticket_data": ticket_data,
|
||
"round_event": [],
|
||
"last_round_event": [],
|
||
"past_round_event": [],
|
||
"total_round_point": 0,
|
||
"stamp_event_data": stamp_event_data,
|
||
"avatar_gacha_lottery_data": {"avatar_gacha_id": 0},
|
||
"fulltune_count": fulltune_count,
|
||
"total_car_parts_count": total_car_parts_count,
|
||
"car_layout_count": [],
|
||
"car_style_count": [],
|
||
"car_use_count": [],
|
||
"maker_use_count": [],
|
||
"story_course": [{"course_id": 0, "count": 1}],
|
||
# TODO!
|
||
# "driver_debut": {
|
||
# "play_count": 137,
|
||
# "daily_play": 5,
|
||
# "last_play_dt": 0,
|
||
# "use_start_date": 0,
|
||
# "use_end_date": 0,
|
||
# "use_dt": 0,
|
||
# "ticket_cnt": 0,
|
||
# "ticket_get_bit": 0,
|
||
# },
|
||
"theory_data": theory_data,
|
||
"theory_course_data": theory_course_data,
|
||
"theory_partner_data": theory_partner_data,
|
||
"theory_running_pram_data": theory_running_pram_data,
|
||
"special_mode_data": self._generate_special_data(user_id),
|
||
"challenge_mode_data": self._generate_challenge_data(user_id),
|
||
"season_rewards_data": [],
|
||
"timetrial_event_data": timetrial_event_data,
|
||
"special_mode_hint_data": {"story_type": 0, "hint_display_flag": 0},
|
||
}
|
||
|
||
def handle_timetrial_getbestrecordpreta_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
for car_id in data["car_ids"]:
|
||
pass
|
||
|
||
course_mybest_data = []
|
||
courses = self.data.item.get_time_trial_user_best_courses(self.version, user_id)
|
||
for course in courses:
|
||
course_mybest_data.append(
|
||
{
|
||
"course_id": course["course_id"],
|
||
# local rank, store rank, worldwide rank?
|
||
"rank": 1,
|
||
# no clue
|
||
"member": 10000,
|
||
# goal_time in ms
|
||
"value": course["goal_time"],
|
||
# total number of entries per course?
|
||
"total": 10,
|
||
"store": self.core_cfg.server.name,
|
||
# use car_id from request?
|
||
"car_id": 0,
|
||
"style_car_id": course["style_car_id"],
|
||
"play_dt": course["play_dt"].timestamp(),
|
||
"section_time_1": course["section_time_1"],
|
||
"section_time_2": course["section_time_2"],
|
||
"section_time_3": course["section_time_3"],
|
||
"section_time_4": course["section_time_4"],
|
||
# no clue
|
||
"mission": course["mission"],
|
||
}
|
||
)
|
||
|
||
course_pickup_car_best_data = []
|
||
courses = self.data.item.get_time_trial_courses(self.version)
|
||
for course in courses:
|
||
car_list = []
|
||
best_cars = self.data.item.get_time_trial_best_cars_by_course(
|
||
self.version, course["course_id"], user_id
|
||
)
|
||
|
||
for i, car in enumerate(best_cars):
|
||
car_list.append(
|
||
{
|
||
"rank": i + 1,
|
||
# no clue
|
||
"member": user_id,
|
||
"value": car["goal_time"],
|
||
"store": self.core_cfg.server.name,
|
||
# use car_id from request?
|
||
"car_id": 0,
|
||
"style_car_id": car["style_car_id"],
|
||
"play_dt": car["play_dt"].timestamp(),
|
||
"section_time_1": car["section_time_1"],
|
||
"section_time_2": car["section_time_2"],
|
||
"section_time_3": car["section_time_3"],
|
||
"section_time_4": car["section_time_4"],
|
||
"mission": car["mission"],
|
||
}
|
||
)
|
||
|
||
course_pickup_car_best_data.append(
|
||
{
|
||
"course_id": course["course_id"],
|
||
"car_list": car_list,
|
||
}
|
||
)
|
||
|
||
return {
|
||
"status_code": "0",
|
||
"course_mybest_data": course_mybest_data,
|
||
"course_pickup_car_best_data": course_pickup_car_best_data,
|
||
}
|
||
|
||
def handle_timetrial_getbestrecordprerace_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
course_id = data["course_id"]
|
||
for car in data["car_ids"]:
|
||
# TODO: get the best record for this car
|
||
style_car_id = car["style_car_id"]
|
||
|
||
# Not sure if this is actually correct
|
||
ranking = self.data.item.get_time_trial_ranking_by_course(
|
||
self.version, course_id
|
||
)
|
||
course_best_data = []
|
||
for i, rank in enumerate(ranking):
|
||
car_user_id = rank["user"]
|
||
|
||
# get the username, country and store from the profile
|
||
profile = self.data.profile.get_profile(car_user_id, self.version)
|
||
arcade = self.data.arcade.get_arcade(profile["store"])
|
||
|
||
if arcade is None:
|
||
arcade = {}
|
||
arcade["name"] = self.core_cfg.server.name
|
||
|
||
# should never happen
|
||
if profile is None:
|
||
continue
|
||
|
||
course_best_data.append(
|
||
{
|
||
"course_id": course_id,
|
||
"rank": i + 1,
|
||
"member": car_user_id,
|
||
"value": rank["goal_time"],
|
||
"store": arcade["name"],
|
||
# use car_id from request?
|
||
"car_id": 0,
|
||
"style_car_id": rank["style_car_id"],
|
||
"play_dt": rank["play_dt"].timestamp(),
|
||
"section_time_1": rank["section_time_1"],
|
||
"section_time_2": rank["section_time_2"],
|
||
"section_time_3": rank["section_time_3"],
|
||
"section_time_4": rank["section_time_4"],
|
||
"mission": rank["mission"],
|
||
}
|
||
)
|
||
|
||
best_cars = self.data.item.get_time_trial_best_cars_by_course(
|
||
self.version, course_id
|
||
)
|
||
|
||
car_list = []
|
||
for i, rank in enumerate(best_cars):
|
||
car_user_id = rank["user"]
|
||
# get the username, country and store from the profile
|
||
profile = self.data.profile.get_profile(car_user_id, self.version)
|
||
arcade = self.data.arcade.get_arcade(profile["store"])
|
||
|
||
if arcade is None:
|
||
arcade = {}
|
||
arcade["name"] = self.core_cfg.server.name
|
||
|
||
# should never happen
|
||
if profile is None:
|
||
continue
|
||
|
||
car_list.append(
|
||
{
|
||
"rank": i + 1,
|
||
# no clue
|
||
"member": car_user_id,
|
||
"value": rank["goal_time"],
|
||
"store": arcade["name"],
|
||
# use car_id from request?
|
||
"car_id": 0,
|
||
"style_car_id": rank["style_car_id"],
|
||
"play_dt": rank["play_dt"].timestamp(),
|
||
"section_time_1": rank["section_time_1"],
|
||
"section_time_2": rank["section_time_2"],
|
||
"section_time_3": rank["section_time_3"],
|
||
"section_time_4": rank["section_time_4"],
|
||
"mission": rank["mission"],
|
||
}
|
||
)
|
||
|
||
return {
|
||
"status_code": "0",
|
||
"course_car_best_data": [{"course_id": course_id, "car_list": car_list}],
|
||
"course_best_data": course_best_data,
|
||
}
|
||
|
||
def handle_user_createaccount_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
car_data: Dict = data.pop("car_obj")
|
||
parts_data: List = car_data.pop("parts_list")
|
||
avatar_data: Dict = data.pop("avatar_obj")
|
||
config_data: Dict = data.pop("config_obj")
|
||
|
||
rank_data: Dict = data.pop("mode_rank_data")
|
||
stock_data: Dict = data.pop("takeover_stock_obj")
|
||
takeover_ticket_list: List = data.pop("takeover_ticket")
|
||
|
||
# not required?
|
||
use_ticket = data.pop("use_ticket")
|
||
|
||
# save profile in database
|
||
data["store"] = headers.get("a_store", 0)
|
||
data["country"] = headers.get("a_country", 0)
|
||
data["asset_version"] = headers.get("asset_version", 1)
|
||
self.data.profile.put_profile(user_id, self.version, data)
|
||
|
||
# save rank data in database
|
||
self.data.profile.put_profile_rank(user_id, self.version, rank_data)
|
||
|
||
# save stock data in database
|
||
self._save_stock_data(user_id, stock_data)
|
||
|
||
# save tickets in database
|
||
for ticket in takeover_ticket_list:
|
||
self.data.item.put_ticket(user_id, ticket)
|
||
|
||
config_data["config_id"] = config_data.pop("id")
|
||
self.data.profile.put_profile_config(user_id, config_data)
|
||
self.data.profile.put_profile_avatar(user_id, avatar_data)
|
||
|
||
# save car data and car parts in database
|
||
car_data["parts_list"] = parts_data
|
||
self.data.item.put_car(user_id, self.version, car_data)
|
||
|
||
return {"status_code": "0"}
|
||
|
||
def handle_user_updatelogin_request(self, data: Dict, headers: Dict):
|
||
pass
|
||
|
||
def handle_timetrial_getcarbest_request(self, data: Dict, headers: Dict):
|
||
pass
|
||
|
||
def handle_factory_avatargacharesult_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
stock_data: Dict = data.pop("stock_obj")
|
||
use_ticket_cnt = data["use_ticket_cnt"]
|
||
|
||
# save stock data in database
|
||
self._save_stock_data(user_id, stock_data)
|
||
|
||
# get the user's ticket
|
||
tickets: list = self.data.item.get_tickets(user_id)
|
||
ticket_list = []
|
||
for ticket in tickets:
|
||
# avatar tickets
|
||
if ticket["ticket_id"] == 5:
|
||
ticket_data = {
|
||
"ticket_id": ticket["ticket_id"],
|
||
"ticket_cnt": ticket["ticket_cnt"] - use_ticket_cnt,
|
||
}
|
||
|
||
# update the ticket in the database
|
||
self.data.item.put_ticket(user_id, ticket_data)
|
||
ticket_list.append(ticket_data)
|
||
|
||
continue
|
||
|
||
ticket_list.append(
|
||
{
|
||
"ticket_id": ticket["ticket_id"],
|
||
"ticket_cnt": ticket["ticket_cnt"],
|
||
}
|
||
)
|
||
|
||
return {"status_code": "0", "ticket_data": ticket_list}
|
||
|
||
def handle_factory_savefavoritecar_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
# save favorite cars in database
|
||
for car in data["pickup_on_car_ids"]:
|
||
self.data.item.put_car(user_id, self.version, car)
|
||
|
||
for car in data["pickup_off_car_ids"]:
|
||
self.data.item.put_car(
|
||
user_id,
|
||
self.version,
|
||
{"style_car_id": car["style_car_id"], "pickup_seq": 0},
|
||
)
|
||
|
||
return {"status_code": "0"}
|
||
|
||
def handle_factory_updatemultiplecustomizeresult_request(
|
||
self, data: Dict, headers: Dict
|
||
):
|
||
user_id = headers["session"]
|
||
|
||
car_list = data.pop("car_list")
|
||
ticket_data: List = data.pop("ticket_data")
|
||
|
||
# unused
|
||
total_car_parts_count = data.pop("total_car_parts_count")
|
||
|
||
# save tickets in database
|
||
for ticket in ticket_data:
|
||
self.data.item.put_ticket(user_id, ticket)
|
||
|
||
for car in car_list:
|
||
# save car data and car parts in database
|
||
self.data.item.put_car(user_id, self.version, car)
|
||
|
||
return {"status_code": "0"}
|
||
|
||
def handle_factory_updatecustomizeresult_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
parts_data: List = data.pop("parts_list")
|
||
ticket_data: List = data.pop("ticket_data")
|
||
|
||
# save tickets in database
|
||
for ticket in ticket_data:
|
||
self.data.item.put_ticket(user_id, ticket)
|
||
|
||
# save car data in database
|
||
data["parts_list"] = parts_data
|
||
self.data.item.put_car(user_id, self.version, data)
|
||
|
||
return {"status_code": "0"}
|
||
|
||
def handle_factory_getcardata_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
cars = self.data.item.get_cars(self.version, user_id)
|
||
car_data = []
|
||
for car in cars:
|
||
tmp = car._asdict()
|
||
del tmp["id"]
|
||
del tmp["user"]
|
||
del tmp["version"]
|
||
|
||
car_data.append(tmp)
|
||
|
||
return {
|
||
"status_code": "0",
|
||
"car_data": car_data,
|
||
}
|
||
|
||
def handle_factory_renamebefore_request(self, data: Dict, headers: Dict):
|
||
pass
|
||
|
||
def handle_factory_buycarresult_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
parts_data: List = data.pop("parts_list")
|
||
pickup_on_list: List = data.pop("pickup_on_car_ids")
|
||
pickup_off_list: List = data.pop("pickup_off_car_ids")
|
||
|
||
style_car_id = data.get("style_car_id")
|
||
|
||
# get the pickup_seq for the new car
|
||
pickup_seq = 0
|
||
# save favorite cars in database
|
||
for car in pickup_on_list:
|
||
# if the new car is a favorite get the new pickup_seqn for later
|
||
if car["style_car_id"] == style_car_id:
|
||
pickup_seq = car["pickup_seq"]
|
||
else:
|
||
self.data.item.put_car(user_id, self.version, car)
|
||
|
||
data["pickup_seq"] = pickup_seq
|
||
|
||
cash = data.pop("cash")
|
||
total_cash = data.pop("total_cash")
|
||
|
||
# save the new cash in database
|
||
self.data.profile.put_profile(
|
||
user_id, self.version, {"total_cash": total_cash, "cash": cash}
|
||
)
|
||
|
||
# full tune ticket
|
||
use_ticket = data.pop("use_ticket")
|
||
if use_ticket:
|
||
# get the user's tickets, full tune ticket id is 25
|
||
ticket = self.data.item.get_ticket(user_id, ticket_id=25)
|
||
|
||
# update the ticket in the database
|
||
self.data.item.put_ticket(
|
||
user_id,
|
||
{
|
||
"ticket_id": ticket["ticket_id"],
|
||
"ticket_cnt": ticket["ticket_cnt"] - 1,
|
||
},
|
||
)
|
||
|
||
# also set the tune_level to 16 (fully tuned)
|
||
data["tune_level"] = 16
|
||
|
||
# save car data and car parts in database
|
||
data["parts_list"] = parts_data
|
||
self.data.item.put_car(user_id, self.version, data)
|
||
|
||
for car in pickup_off_list:
|
||
self.data.item.put_car(
|
||
user_id,
|
||
self.version,
|
||
{"style_car_id": car["style_car_id"], "pickup_seq": 0},
|
||
)
|
||
|
||
# get the user's car
|
||
cars = self.data.item.get_cars(self.version, user_id)
|
||
fulltune_count = 0
|
||
total_car_parts_count = 0
|
||
for car in cars:
|
||
# tune_level of 16 means fully tuned, so add 1 to fulltune_count
|
||
if car["tune_level"] >= 16:
|
||
fulltune_count += 1
|
||
|
||
# add the number of car parts to total_car_parts_count
|
||
# total_car_parts_count += car["total_car_parts_count"]
|
||
|
||
# get the user's ticket
|
||
tickets = self.data.item.get_tickets(user_id)
|
||
ticket_data = []
|
||
for ticket in tickets:
|
||
ticket_data.append(
|
||
{
|
||
"ticket_id": ticket["ticket_id"],
|
||
"ticket_cnt": ticket["ticket_cnt"],
|
||
}
|
||
)
|
||
|
||
return {
|
||
"status_code": "0",
|
||
"ticket_data": ticket_data,
|
||
"fulltune_count": fulltune_count,
|
||
"total_car_parts_count": total_car_parts_count,
|
||
"car_layout_count": [],
|
||
"car_style_count": [],
|
||
}
|
||
|
||
def handle_factory_renameresult_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
new_username = data.get("username")
|
||
|
||
# save new username in database
|
||
if new_username:
|
||
self.data.profile.put_profile(user_id, self.version, data)
|
||
|
||
return {"status_code": "0"}
|
||
|
||
def handle_factory_updatecustomizeavatar_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
avatar_data: Dict = data.pop("avatar_obj")
|
||
stock_data: Dict = data.pop("stock_obj")
|
||
|
||
# update the stock data in database
|
||
self._save_stock_data(user_id, stock_data)
|
||
|
||
# save avatar data and avatar parts in database
|
||
self.data.profile.put_profile_avatar(user_id, avatar_data)
|
||
|
||
return {"status_code": "0"}
|
||
|
||
def handle_factory_updatecustomizeuser_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
stock_data: Dict = data.pop("stock_obj")
|
||
|
||
# update the stock data in database
|
||
self._save_stock_data(user_id, stock_data)
|
||
|
||
# update profile data and config in database
|
||
self.data.profile.put_profile(user_id, self.version, data)
|
||
|
||
return {"status_code": "0"}
|
||
|
||
def handle_user_updatestampinfo_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
stamp_event_data = data.pop("stamp_event_data")
|
||
for stamp in stamp_event_data:
|
||
self.data.item.put_stamp(user_id, stamp)
|
||
|
||
return {"status_code": "0"}
|
||
|
||
def handle_user_updatetimetrialresult_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
stock_data: Dict = data.pop("stock_obj")
|
||
ticket_data: List = data.pop("ticket_data")
|
||
reward_dist_data: Dict = data.pop("reward_dist_obj")
|
||
driver_debut_data = data.pop("driver_debut_obj")
|
||
rank_data: Dict = data.pop("mode_rank_obj")
|
||
|
||
# time trial event points
|
||
event_point = data.pop("event_point")
|
||
|
||
# save stock data in database
|
||
self._save_stock_data(user_id, stock_data)
|
||
|
||
# save tickets in database
|
||
for ticket in ticket_data:
|
||
self.data.item.put_ticket(user_id, ticket)
|
||
|
||
# save mode rank data in database
|
||
rank_data.update(reward_dist_data)
|
||
self.data.profile.put_profile_rank(user_id, self.version, rank_data)
|
||
|
||
# get the profile data, update total_play and daily_play, and save it
|
||
profile = self.data.profile.get_profile(user_id, self.version)
|
||
total_play = profile["total_play"] + 1
|
||
|
||
# update profile
|
||
self.data.profile.put_profile(
|
||
user_id,
|
||
self.version,
|
||
{
|
||
"total_play": total_play,
|
||
"last_play_date": datetime.now(),
|
||
"aura_id": data.pop("aura_id"),
|
||
"aura_color_id": data.pop("aura_color_id"),
|
||
"aura_line_id": data.pop("aura_line_id"),
|
||
"cash": data.pop("cash"),
|
||
"total_cash": data.pop("total_cash"),
|
||
"dressup_point": data.pop("dressup_point"),
|
||
"avatar_point": data.pop("avatar_point"),
|
||
"mileage": data.pop("mileage"),
|
||
},
|
||
)
|
||
|
||
# get the use_count and story_use_count of the used car
|
||
style_car_id = data.get("style_car_id")
|
||
car_mileage = data.pop("car_mileage")
|
||
used_car = self.data.item.get_car(user_id, self.version, style_car_id)._asdict()
|
||
|
||
# increase the use_count and story_use_count of the used car
|
||
used_car["use_count"] += 1
|
||
used_car["timetrial_use_count"] += 1
|
||
used_car["car_mileage"] = car_mileage
|
||
|
||
# save the used car in database
|
||
self.data.item.put_car(user_id, self.version, used_car)
|
||
|
||
# skill_level_exp is the "course proeficiency" and is saved
|
||
# in the course table
|
||
course_id = data.get("course_id")
|
||
run_counts = 1
|
||
skill_level_exp = data.pop("skill_level_exp")
|
||
|
||
# get the course data
|
||
course = self.data.item.get_course(user_id, course_id)
|
||
if course:
|
||
# update run_counts
|
||
run_counts = course["run_counts"] + 1
|
||
|
||
self.data.item.put_course(
|
||
user_id,
|
||
{
|
||
"course_id": course_id,
|
||
"run_counts": run_counts,
|
||
"skill_level_exp": skill_level_exp,
|
||
},
|
||
)
|
||
|
||
goal_time = data.get("goal_time")
|
||
# grab the ranking data and count the numbers of rows with a faster time
|
||
# than the current goal_time
|
||
course_rank = self.data.item.get_time_trial_ranking_by_course(
|
||
self.version, course_id, limit=None
|
||
)
|
||
course_rank = len([r for r in course_rank if r["goal_time"] < goal_time]) + 1
|
||
|
||
car_course_rank = self.data.item.get_time_trial_ranking_by_course(
|
||
self.version, course_id, style_car_id, limit=None
|
||
)
|
||
car_course_rank = (
|
||
len([r for r in car_course_rank if r["goal_time"] < goal_time]) + 1
|
||
)
|
||
|
||
# only update the time if its better than the best time and also not 0
|
||
if data.get("goal_time") > 0:
|
||
# get the current best goal time
|
||
best_time_trial = (
|
||
self.data.item.get_time_trial_user_best_time_by_course_car(
|
||
self.version, user_id, course_id, style_car_id
|
||
)
|
||
)
|
||
|
||
if (
|
||
best_time_trial is None
|
||
or data.get("goal_time") < best_time_trial["goal_time"]
|
||
):
|
||
# now finally save the time trial with updated timestamp
|
||
data["play_dt"] = datetime.now()
|
||
self.data.item.put_time_trial(self.version, user_id, data)
|
||
|
||
# update the timetrial event points
|
||
self.data.item.put_timetrial_event(
|
||
user_id, self.timetrial_event_id, event_point
|
||
)
|
||
|
||
return {
|
||
"status_code": "0",
|
||
"course_rank": course_rank,
|
||
"course_car_rank": car_course_rank,
|
||
"location_course_store_rank": course_rank,
|
||
"car_use_count": [],
|
||
"maker_use_count": [],
|
||
"timetrial_event_data": {
|
||
"timetrial_event_id": self.timetrial_event_id,
|
||
"point": event_point,
|
||
},
|
||
}
|
||
|
||
def handle_user_updatestoryresult_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
stock_data: Dict = data.pop("stock_obj")
|
||
ticket_data: List = data.pop("ticket_data")
|
||
reward_dist_data: Dict = data.pop("reward_dist_obj")
|
||
driver_debut_data = data.pop("driver_debut_obj")
|
||
rank_data: Dict = data.pop("mode_rank_obj")
|
||
# stamp_event_data = data.pop("stamp_event_data")
|
||
|
||
# save stock data in database
|
||
self._save_stock_data(user_id, stock_data)
|
||
|
||
# save tickets in database
|
||
for ticket in ticket_data:
|
||
self.data.item.put_ticket(user_id, ticket)
|
||
|
||
# save mode rank data in database
|
||
rank_data.update(reward_dist_data)
|
||
self.data.profile.put_profile_rank(user_id, self.version, rank_data)
|
||
|
||
# save the current story progress in database
|
||
max_loop = data.get("chapter_loop_max")
|
||
chapter_id = data.get("chapter")
|
||
|
||
episode_id = data.get("episode")
|
||
difficulty = data.get("difficulty")
|
||
play_status = data.get("play_status")
|
||
|
||
# get the current loop from the database
|
||
story_data = self.data.item.get_story(user_id, chapter_id)
|
||
# 1 = active, 2+ = cleared?
|
||
loop_count = 1
|
||
if story_data:
|
||
loop_count = story_data["loop_count"]
|
||
|
||
# if the played difficulty is smaller than loop_count you cannot clear
|
||
# (play_status = 2) the episode otherwise the following difficulties
|
||
# won't earn any EXP?
|
||
if difficulty < loop_count:
|
||
play_status = 1
|
||
|
||
# if the episode has already been cleared, set the play_status to 2
|
||
# so it won't be set to unplayed (play_status = 1)
|
||
episode_data = self.data.item.get_story_episode(user_id, episode_id)
|
||
if episode_data:
|
||
if play_status < episode_data["play_status"]:
|
||
play_status = 2
|
||
|
||
# save the current episode progress in database
|
||
self.data.item.put_story_episode(
|
||
user_id,
|
||
chapter_id,
|
||
{
|
||
"episode": episode_id,
|
||
"play_status": play_status,
|
||
},
|
||
)
|
||
|
||
if loop_count < max_loop and data.get("chapter_clear") == 1:
|
||
# increase the loop count
|
||
loop_count += 1
|
||
|
||
# for the current chapter set all episode play_status back to 1
|
||
self.data.item.put_story_episode_play_status(user_id, chapter_id, 1)
|
||
|
||
self.data.item.put_story(
|
||
user_id,
|
||
{
|
||
"story_type": data.get("story_type"),
|
||
"chapter": chapter_id,
|
||
"loop_count": loop_count,
|
||
},
|
||
)
|
||
|
||
# save the current episode difficulty progress in database
|
||
self.data.item.put_story_episode_difficulty(
|
||
user_id,
|
||
episode_id,
|
||
{
|
||
"difficulty": difficulty,
|
||
"play_count": 1, # no idea where this comes from
|
||
"clear_count": 1, # no idea where this comes from
|
||
"play_status": data.get("play_status"),
|
||
"play_score": data.get("play_score"),
|
||
},
|
||
)
|
||
|
||
# get the use_count and story_use_count of the used car
|
||
style_car_id = data.get("style_car_id")
|
||
car_mileage = data.get("car_mileage")
|
||
used_car = self.data.item.get_car(user_id, self.version, style_car_id)._asdict()
|
||
|
||
# increase the use_count and story_use_count of the used car
|
||
used_car["use_count"] += 1
|
||
used_car["story_use_count"] += 1
|
||
used_car["car_mileage"] = car_mileage
|
||
|
||
# save the used car in database
|
||
self.data.item.put_car(user_id, self.version, used_car)
|
||
|
||
# get the profile data, update total_play and daily_play, and save it
|
||
profile = self.data.profile.get_profile(user_id, self.version)
|
||
total_play = profile["total_play"] + 1
|
||
|
||
# save user profile in database
|
||
self.data.profile.put_profile(
|
||
user_id,
|
||
self.version,
|
||
{
|
||
"total_play": total_play,
|
||
"last_play_date": datetime.now(),
|
||
"mileage": data.pop("mileage"),
|
||
"cash": data.pop("cash"),
|
||
"total_cash": data.pop("total_cash"),
|
||
"dressup_point": data.pop("dressup_point"),
|
||
"avatar_point": data.pop("avatar_point"),
|
||
"aura_id": data.pop("aura_id"),
|
||
"aura_color_id": data.pop("aura_color_id"),
|
||
"aura_line_id": data.pop("aura_line_id"),
|
||
},
|
||
)
|
||
|
||
return {
|
||
"status_code": "0",
|
||
"story_data": self._generate_story_data(user_id),
|
||
"car_use_count": [],
|
||
"maker_use_count": [],
|
||
}
|
||
|
||
def handle_user_updatespecialmoderesult_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
stock_data: Dict = data.pop("stock_obj")
|
||
ticket_data: List = data.pop("ticket_data")
|
||
reward_dist_data: Dict = data.pop("reward_dist_obj")
|
||
driver_debut_data = data.pop("driver_debut_obj")
|
||
rank_data: Dict = data.pop("mode_rank_obj")
|
||
|
||
# unused
|
||
hint_display_flag: int = data.pop("hint_display_flag")
|
||
|
||
# get the vs use count from database and update it
|
||
style_car_id = data.pop("style_car_id")
|
||
car_data = self.data.item.get_car(user_id, self.version, style_car_id)
|
||
story_use_count = car_data["story_use_count"] + 1
|
||
|
||
# save car data in database
|
||
self.data.item.put_car(
|
||
user_id,
|
||
self.version,
|
||
{
|
||
"style_car_id": style_car_id,
|
||
"car_mileage": data.pop("car_mileage"),
|
||
"story_use_count": story_use_count,
|
||
},
|
||
)
|
||
|
||
# get the profile data, update total_play and daily_play, and save it
|
||
profile = self.data.profile.get_profile(user_id, self.version)
|
||
total_play = profile["total_play"] + 1
|
||
|
||
# save user profile in database
|
||
self.data.profile.put_profile(
|
||
user_id,
|
||
self.version,
|
||
{
|
||
"total_play": total_play,
|
||
"last_play_date": datetime.now(),
|
||
"mileage": data.pop("mileage"),
|
||
"cash": data.pop("cash"),
|
||
"total_cash": data.pop("total_cash"),
|
||
"dressup_point": data.pop("dressup_point"),
|
||
"avatar_point": data.pop("avatar_point"),
|
||
"aura_id": data.pop("aura_id"),
|
||
"aura_color_id": data.pop("aura_color_id"),
|
||
"aura_line_id": data.pop("aura_line_id"),
|
||
},
|
||
)
|
||
|
||
# save stock data in database
|
||
self._save_stock_data(user_id, stock_data)
|
||
|
||
# save ticket data in database
|
||
for ticket in ticket_data:
|
||
self.data.item.put_ticket(user_id, ticket)
|
||
|
||
# save mode_rank and reward_dist data in database
|
||
rank_data.update(reward_dist_data)
|
||
self.data.profile.put_profile_rank(user_id, self.version, rank_data)
|
||
|
||
# finally save the special mode with story_type=4 in database
|
||
self.data.item.put_challenge(user_id, data)
|
||
|
||
return {
|
||
"status_code": "0",
|
||
"special_mode_data": self._generate_special_data(user_id),
|
||
"car_use_count": [],
|
||
"maker_use_count": [],
|
||
}
|
||
|
||
def handle_user_updatechallengemoderesult_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
stock_data: Dict = data.pop("stock_obj")
|
||
ticket_data: List = data.pop("ticket_data")
|
||
reward_dist_data: Dict = data.pop("reward_dist_obj")
|
||
driver_debut_data = data.pop("driver_debut_obj")
|
||
rank_data: Dict = data.pop("mode_rank_obj")
|
||
|
||
# get the vs use count from database and update it
|
||
style_car_id = data.get("style_car_id")
|
||
car_data = self.data.item.get_car(user_id, self.version, style_car_id)
|
||
story_use_count = car_data["story_use_count"] + 1
|
||
|
||
# save car data in database
|
||
self.data.item.put_car(
|
||
user_id,
|
||
self.version,
|
||
{
|
||
"style_car_id": style_car_id,
|
||
"car_mileage": data.pop("car_mileage"),
|
||
"story_use_count": story_use_count,
|
||
},
|
||
)
|
||
|
||
# get the profile data, update total_play and daily_play, and save it
|
||
profile = self.data.profile.get_profile(user_id, self.version)
|
||
total_play = profile["total_play"] + 1
|
||
|
||
# save user profile in database
|
||
self.data.profile.put_profile(
|
||
user_id,
|
||
self.version,
|
||
{
|
||
"total_play": total_play,
|
||
"last_play_date": datetime.now(),
|
||
"mileage": data.pop("mileage"),
|
||
"cash": data.pop("cash"),
|
||
"total_cash": data.pop("total_cash"),
|
||
"dressup_point": data.pop("dressup_point"),
|
||
"avatar_point": data.pop("avatar_point"),
|
||
"aura_id": data.pop("aura_id"),
|
||
"aura_color_id": data.pop("aura_color_id"),
|
||
"aura_line_id": data.pop("aura_line_id"),
|
||
},
|
||
)
|
||
|
||
# save stock data in database
|
||
self._save_stock_data(user_id, stock_data)
|
||
|
||
# save ticket data in database
|
||
for ticket in ticket_data:
|
||
self.data.item.put_ticket(user_id, ticket)
|
||
|
||
# save mode_rank and reward_dist data in database
|
||
rank_data.update(reward_dist_data)
|
||
self.data.profile.put_profile_rank(user_id, self.version, rank_data)
|
||
|
||
# get the challenge mode data from database
|
||
challenge_data = self.data.item.get_challenge(
|
||
user_id, data.get("vs_type"), data.get("play_difficulty")
|
||
)
|
||
|
||
if challenge_data:
|
||
# update play count
|
||
play_count = challenge_data["play_count"] + 1
|
||
data["play_count"] = play_count
|
||
|
||
# finally save the challenge mode with story_type=3 in database
|
||
self.data.item.put_challenge(user_id, data)
|
||
|
||
return {
|
||
"status_code": "0",
|
||
"challenge_mode_data": self._generate_challenge_data(user_id),
|
||
"car_use_count": [],
|
||
"maker_use_count": [],
|
||
}
|
||
|
||
def _generate_time_trial_data(self, season_id: int, user_id: int) -> List[Dict]:
|
||
# get the season time trial data from database
|
||
timetrial_data = []
|
||
|
||
courses = self.data.item.get_courses(user_id)
|
||
if courses is None or len(courses) == 0:
|
||
return {"status_code": "0", "timetrial_data": timetrial_data}
|
||
|
||
for course in courses:
|
||
# grab the course id and course proeficiency
|
||
course_id = course["course_id"]
|
||
skill_level_exp = course["skill_level_exp"]
|
||
|
||
# get the best time for the current course for the current user
|
||
best_trial = self.data.item.get_time_trial_best_ranking_by_course(
|
||
season_id, user_id, course_id
|
||
)
|
||
if not best_trial:
|
||
continue
|
||
|
||
goal_time = best_trial["goal_time"]
|
||
# get the rank for the current course
|
||
course_rank = self.data.item.get_time_trial_ranking_by_course(
|
||
season_id, course_id, limit=None
|
||
)
|
||
course_rank = (
|
||
len([r for r in course_rank if r["goal_time"] < goal_time]) + 1
|
||
)
|
||
|
||
timetrial_data.append(
|
||
{
|
||
"style_car_id": best_trial["style_car_id"],
|
||
"course_id": course_id,
|
||
"skill_level_exp": skill_level_exp,
|
||
"goal_time": goal_time,
|
||
"rank": course_rank,
|
||
"rank_dt": int(best_trial["play_dt"].timestamp()),
|
||
}
|
||
)
|
||
|
||
return timetrial_data
|
||
|
||
def handle_user_getpastseasontadata_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
season_id = data.get("season_id")
|
||
|
||
# so to get the season 1 data just subtract 1 from the season id
|
||
past_timetrial_data = self._generate_time_trial_data(season_id - 1, user_id)
|
||
|
||
# TODO: get the current season timetrial data somehow, because after requesting
|
||
# GetPastSeasonTAData the game will NOT request GetTAData?!
|
||
return {
|
||
"status_code": "0",
|
||
"season_id": season_id,
|
||
"past_season_timetrial_data": past_timetrial_data,
|
||
}
|
||
|
||
def handle_user_gettadata_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
timetrial_data = self._generate_time_trial_data(self.version, user_id)
|
||
|
||
# TODO: get the past season timetrial data somehow, because after requesting
|
||
# GetTAData the game will NOT request GetPastSeasonTAData?!
|
||
return {
|
||
"status_code": "0",
|
||
"timetrial_data": timetrial_data,
|
||
# "past_season_timetrial_data": timetrial_data,
|
||
}
|
||
|
||
def handle_user_updatecartune_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
# full tune ticket
|
||
use_ticket = data.pop("use_ticket")
|
||
if use_ticket:
|
||
# get the user's tickets, full tune ticket id is 25
|
||
ticket = self.data.item.get_ticket(user_id, ticket_id=25)
|
||
|
||
# update the ticket in the database
|
||
self.data.item.put_ticket(
|
||
user_id,
|
||
{
|
||
"ticket_id": ticket["ticket_id"],
|
||
"ticket_cnt": ticket["ticket_cnt"] - 1,
|
||
},
|
||
)
|
||
|
||
# also set the tune_level to 16 (fully tuned)
|
||
data["tune_level"] = 16
|
||
|
||
self.data.item.put_car(user_id, self.version, data)
|
||
|
||
return {
|
||
"status_code": "0",
|
||
"story_data": self._generate_story_data(user_id),
|
||
"car_use_count": [],
|
||
"maker_use_count": [],
|
||
}
|
||
|
||
def handle_log_saveplaylog_request(self, data: Dict, headers: Dict):
|
||
pass
|
||
|
||
def handle_log_saveendlog_request(self, data: Dict, headers: Dict):
|
||
pass
|
||
|
||
def handle_user_updatemoderesult_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
config_data: Dict = data.pop("config_obj")
|
||
stock_data: Dict = data.pop("stock_obj")
|
||
ticket_data: List = data.pop("ticket_data")
|
||
reward_dist_data: Dict = data.pop("reward_dist_obj")
|
||
|
||
# not required?
|
||
mode_id = data.pop("mode_id")
|
||
standby_play_flag = data.pop("standby_play_flag")
|
||
tips_list = data.pop("tips_list")
|
||
|
||
# save stock data in database
|
||
self._save_stock_data(user_id, stock_data)
|
||
|
||
# save tickets in database
|
||
for ticket in ticket_data:
|
||
self.data.item.put_ticket(user_id, ticket)
|
||
|
||
# save rank dist data in database
|
||
self.data.profile.put_profile_rank(user_id, self.version, reward_dist_data)
|
||
|
||
# update profile data and config in database
|
||
self.data.profile.put_profile(user_id, self.version, data)
|
||
config_data["config_id"] = config_data.pop("id")
|
||
self.data.profile.put_profile_config(user_id, config_data)
|
||
|
||
return {"status_code": "0", "server_status": 1}
|
||
|
||
def _generate_theory_rival_data(
|
||
self, user_list: list, course_id: int, req_user_id: int
|
||
) -> list:
|
||
rival_data = []
|
||
for user_id in user_list:
|
||
# if not enough players are available just use the data from the req_user
|
||
if user_id == -1:
|
||
profile = self.data.profile.get_profile(req_user_id, self.version)
|
||
profile = profile._asdict()
|
||
# set the name to CPU
|
||
profile["username"] = f"ＣＰＵ"
|
||
# also reset stamps to default
|
||
profile["country"] = 9
|
||
profile["store"] = 0
|
||
profile["stamp_key_assign_0"] = 0
|
||
profile["stamp_key_assign_1"] = 1
|
||
profile["stamp_key_assign_2"] = 2
|
||
profile["stamp_key_assign_3"] = 3
|
||
profile["mytitle_id"] = 0
|
||
else:
|
||
profile = self.data.profile.get_profile(user_id, self.version)
|
||
|
||
rank = self.data.profile.get_profile_rank(profile["user"], self.version)
|
||
|
||
avatars = [
|
||
{
|
||
"sex": 0,
|
||
"face": 1,
|
||
"eye": 1,
|
||
"mouth": 1,
|
||
"hair": 1,
|
||
"glasses": 0,
|
||
"face_accessory": 0,
|
||
"body": 1,
|
||
"body_accessory": 0,
|
||
"behind": 0,
|
||
"bg": 1,
|
||
"effect": 0,
|
||
"special": 0,
|
||
},
|
||
{
|
||
"sex": 0,
|
||
"face": 1,
|
||
"eye": 1,
|
||
"mouth": 1,
|
||
"hair": 19,
|
||
"glasses": 0,
|
||
"face_accessory": 0,
|
||
"body": 2,
|
||
"body_accessory": 0,
|
||
"behind": 0,
|
||
"bg": 1,
|
||
"effect": 0,
|
||
"special": 0,
|
||
},
|
||
{
|
||
"sex": 1,
|
||
"face": 91,
|
||
"eye": 265,
|
||
"mouth": 13,
|
||
"hair": 369,
|
||
"glasses": 0,
|
||
"face_accessory": 0,
|
||
"body": 113,
|
||
"body_accessory": 0,
|
||
"behind": 0,
|
||
"bg": 1,
|
||
"effect": 0,
|
||
"special": 0,
|
||
},
|
||
{
|
||
"sex": 1,
|
||
"face": 91,
|
||
"eye": 265,
|
||
"mouth": 13,
|
||
"hair": 387,
|
||
"glasses": 0,
|
||
"face_accessory": 0,
|
||
"body": 114,
|
||
"body_accessory": 0,
|
||
"behind": 0,
|
||
"bg": 1,
|
||
"effect": 0,
|
||
"special": 0,
|
||
},
|
||
]
|
||
|
||
if user_id == -1:
|
||
# get a random avatar from the list and some random car from all users
|
||
avatar = choice(avatars)
|
||
car = self.data.item.get_random_car(self.version)
|
||
else:
|
||
avatar = self.data.profile.get_profile_avatar(profile["user"])
|
||
car = self.data.item.get_random_user_car(profile["user"], self.version)
|
||
|
||
parts_list = []
|
||
for part in car["parts_list"]:
|
||
parts_list.append(part["parts"])
|
||
|
||
course = self.data.item.get_theory_course(profile["user"], course_id)
|
||
powerhose_lv = 0
|
||
if course:
|
||
powerhose_lv = course["powerhouse_lv"]
|
||
|
||
theory_running = self.data.item.get_theory_running_by_course(
|
||
profile["user"], course_id
|
||
)
|
||
|
||
# normally it's 127 after the first play so we set it to 128
|
||
attack = 128
|
||
defense = 128
|
||
safety = 128
|
||
runaway = 128
|
||
trick_flag = 0
|
||
if theory_running and user_id != -1:
|
||
attack = theory_running["attack"]
|
||
defense = theory_running["defense"]
|
||
safety = theory_running["safety"]
|
||
runaway = theory_running["runaway"]
|
||
trick_flag = theory_running["trick_flag"]
|
||
|
||
# get the time trial ranking medal
|
||
eval_id = 0
|
||
time_trial = self.data.item.get_time_trial_best_ranking_by_course(
|
||
self.version, profile["user"], course_id
|
||
)
|
||
if time_trial:
|
||
eval_id = time_trial["eval_id"]
|
||
|
||
arcade = self.data.arcade.get_arcade(profile["store"])
|
||
if arcade is None:
|
||
arcade = {}
|
||
arcade["name"] = self.core_cfg.server.name
|
||
|
||
rival_data.append(
|
||
{
|
||
"id": profile["user"],
|
||
"name": profile["username"],
|
||
"grade": rank["grade"],
|
||
# only needed for power match
|
||
"powerhouseLv": powerhose_lv,
|
||
"mytitleId": profile["mytitle_id"],
|
||
"country": profile["country"],
|
||
"auraId": profile["aura_id"],
|
||
"auraColor": profile["aura_color_id"],
|
||
"auraLine": profile["aura_line_id"],
|
||
# not sure?
|
||
"roundRanking": 0,
|
||
"storeName": arcade["name"],
|
||
"sex": avatar["sex"],
|
||
"face": avatar["face"],
|
||
"eye": avatar["eye"],
|
||
"mouth": avatar["mouth"],
|
||
"hair": avatar["hair"],
|
||
"glasses": avatar["glasses"],
|
||
"faceAccessory": avatar["face_accessory"],
|
||
"body": avatar["body"],
|
||
"bodyAccessory": avatar["body_accessory"],
|
||
"behind": avatar["behind"],
|
||
"bg": avatar["bg"],
|
||
"effect": avatar["effect"],
|
||
"special": avatar["special"],
|
||
"styleCarId": car["style_car_id"],
|
||
"color": car["color"],
|
||
"bureau": car["bureau"],
|
||
"kana": car["kana"],
|
||
"sNo": car["s_no"],
|
||
"lNo": car["l_no"],
|
||
"tuneLv": car["tune_level"],
|
||
"carFlag": car["car_flag"],
|
||
"tunePoint": car["tune_point"],
|
||
"infinityTune": car["infinity_tune"],
|
||
"tuneParts": car["tune_parts"],
|
||
"partsList": parts_list,
|
||
"partsCount": car["equip_parts_count"],
|
||
"stamp0": profile["stamp_key_assign_0"],
|
||
"stamp1": profile["stamp_key_assign_1"],
|
||
"stamp2": profile["stamp_key_assign_2"],
|
||
"stamp3": profile["stamp_key_assign_3"],
|
||
"attack": attack,
|
||
"defense": defense,
|
||
"safety": safety,
|
||
"runaway": runaway,
|
||
"trickFlg": trick_flag,
|
||
# time trial ranking medal
|
||
"taEval": eval_id,
|
||
}
|
||
)
|
||
|
||
return rival_data
|
||
|
||
def handle_theory_matching_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
course_id = data.pop("course_id")
|
||
# no idea why thats needed?
|
||
grade = data.pop("grade")
|
||
|
||
# number of auto_matches and power_matches, official values are:
|
||
count_auto_match = 9
|
||
count_power_match = 3
|
||
|
||
# required for the power_match list?
|
||
powerhose_lv = data.pop("powerhouse_lv")
|
||
|
||
# get random profiles for auto match
|
||
profiles = self.data.profile.get_different_random_profiles(
|
||
user_id, self.version, count=count_auto_match
|
||
)
|
||
|
||
user_list = [profile["user"] for profile in profiles]
|
||
# if user_list is not count_auto_match long, fill it up with -1
|
||
while len(user_list) < count_auto_match:
|
||
user_list.append(-1)
|
||
|
||
auto_match = self._generate_theory_rival_data(user_list, course_id, user_id)
|
||
|
||
# get profiles with the same powerhouse_lv for power match
|
||
theory_courses = self.data.item.get_theory_course_by_powerhouse_lv(
|
||
user_id, course_id, powerhose_lv, count=count_power_match
|
||
)
|
||
user_list = [course["user"] for course in theory_courses]
|
||
|
||
# if user_list is not count_power_match long, fill it up with -1
|
||
while len(user_list) < count_power_match:
|
||
user_list.append(-1)
|
||
|
||
power_match = self._generate_theory_rival_data(user_list, course_id, user_id)
|
||
|
||
return {
|
||
"status_code": "0",
|
||
"server_status": 1,
|
||
"rival_data": {
|
||
"auto_match": auto_match,
|
||
"power_match": power_match,
|
||
},
|
||
}
|
||
|
||
def handle_user_updatetheoryresult_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
stock_data: Dict = data.pop("stock_obj")
|
||
ticket_data: List = data.pop("ticket_data")
|
||
reward_dist_data: Dict = data.pop("reward_dist_obj")
|
||
rank_data: Dict = data.pop("mode_rank_obj")
|
||
driver_debut_data: Dict = data.pop("driver_debut_obj")
|
||
|
||
# save stock data in database
|
||
self._save_stock_data(user_id, stock_data)
|
||
|
||
# save tickets in database
|
||
for ticket in ticket_data:
|
||
self.data.item.put_ticket(user_id, ticket)
|
||
|
||
# save rank dist data in database
|
||
rank_data.update(reward_dist_data)
|
||
self.data.profile.put_profile_rank(user_id, self.version, rank_data)
|
||
|
||
# save the profile theory data in database
|
||
play_count = 1
|
||
play_count_multi = 1
|
||
win_count = 0
|
||
win_count_multi = 0
|
||
|
||
theory_data = self.data.profile.get_profile_theory(user_id, self.version)
|
||
if theory_data:
|
||
play_count = theory_data["play_count"] + 1
|
||
play_count_multi = theory_data["play_count_multi"] + 1
|
||
win_count = theory_data["win_count"]
|
||
win_count_multi = theory_data["win_count_multi"]
|
||
|
||
# check all advantages and see if one of them is larger than 0
|
||
# if so, we won
|
||
if (
|
||
data.get("advantage_1") > 0
|
||
or data.get("advantage_2") > 0
|
||
or data.get("advantage_3") > 0
|
||
or data.get("advantage_4") > 0
|
||
):
|
||
win_count += 1
|
||
win_count_multi += 1
|
||
|
||
self.data.profile.put_profile_theory(
|
||
user_id,
|
||
self.version,
|
||
{
|
||
"play_count": play_count,
|
||
"play_count_multi": play_count_multi,
|
||
"partner_id": data.get("partner_id"),
|
||
"partner_progress": data.get("partner_progress"),
|
||
"partner_progress_score": data.get("partner_progress_score"),
|
||
"practice_start_rank": data.get("practice_start_rank"),
|
||
"general_flag": data.get("general_flag"),
|
||
"vs_history": data.get("vs_history"),
|
||
# no idea?
|
||
"vs_history_multi": data.get("vs_history"),
|
||
"win_count": win_count,
|
||
"win_count_multi": win_count_multi,
|
||
},
|
||
)
|
||
|
||
# save theory course in database
|
||
self.data.item.put_theory_course(
|
||
user_id,
|
||
{
|
||
"course_id": data.get("course_id"),
|
||
"max_victory_grade": data.get("max_victory_grade"),
|
||
# always add 1?
|
||
"run_count": 1,
|
||
"powerhouse_lv": data.get("powerhouse_lv"),
|
||
"powerhouse_exp": data.get("powerhouse_exp"),
|
||
# not sure if the played_powerhouse_lv is the same as powerhouse_lv
|
||
"played_powerhouse_lv": data.get("powerhouse_lv"),
|
||
},
|
||
)
|
||
|
||
# save the theory partner in database
|
||
self.data.item.put_theory_partner(
|
||
user_id,
|
||
{
|
||
"partner_id": data.get("partner_id"),
|
||
"fellowship_lv": data.get("fellowship_lv"),
|
||
"fellowship_exp": data.get("fellowship_exp"),
|
||
},
|
||
)
|
||
|
||
# save the theory running in database?
|
||
self.data.item.put_theory_running(
|
||
user_id,
|
||
{
|
||
"course_id": data.get("course_id"),
|
||
"attack": data.get("attack"),
|
||
"defense": data.get("defense"),
|
||
"safety": data.get("safety"),
|
||
"runaway": data.get("runaway"),
|
||
"trick_flag": data.get("trick_flag"),
|
||
},
|
||
)
|
||
|
||
# get the use_count and theory_use_count of the used car
|
||
style_car_id = data.get("style_car_id")
|
||
car_mileage = data.get("car_mileage")
|
||
used_car = self.data.item.get_car(user_id, self.version, style_car_id)._asdict()
|
||
|
||
# increase the use_count and theory_use_count of the used car
|
||
used_car["use_count"] += 1
|
||
used_car["theory_use_count"] += 1
|
||
used_car["car_mileage"] = car_mileage
|
||
|
||
# save the used car in database
|
||
self.data.item.put_car(user_id, self.version, used_car)
|
||
|
||
# get the profile data, update total_play and daily_play, and save it
|
||
profile = self.data.profile.get_profile(user_id, self.version)
|
||
total_play = profile["total_play"] + 1
|
||
|
||
# save the profile in database
|
||
self.data.profile.put_profile(
|
||
user_id,
|
||
self.version,
|
||
{
|
||
"total_play": total_play,
|
||
"last_play_date": datetime.now(),
|
||
"mileage": data.get("mileage"),
|
||
"aura_id": data.get("aura_id"),
|
||
"aura_color_id": data.get("aura_color_id"),
|
||
"aura_line_id": data.get("aura_line_id"),
|
||
"cash": data.get("cash"),
|
||
"total_cash": data.get("total_cash"),
|
||
"dressup_point": data.get("dressup_point"),
|
||
"avatar_point": data.get("avatar_point"),
|
||
},
|
||
)
|
||
|
||
return {
|
||
"status_code": "0",
|
||
"played_powerhouse_lv": data.get("powerhouse_lv"),
|
||
"car_use_count": [],
|
||
"maker_use_count": [],
|
||
"play_count": play_count,
|
||
"play_count_multi": play_count_multi,
|
||
"win_count": win_count,
|
||
"win_count_multi": win_count_multi,
|
||
}
|
||
|
||
def handle_timetrial_getbestrecordprebattle_request(
|
||
self, data: Dict, headers: Dict
|
||
):
|
||
user_id = headers["session"]
|
||
|
||
course_pickup_car_best_data = []
|
||
courses = self.data.item.get_time_trial_courses(self.version)
|
||
for course in courses:
|
||
car_list = []
|
||
best_cars = self.data.item.get_time_trial_best_cars_by_course(
|
||
self.version, course["course_id"], user_id
|
||
)
|
||
|
||
for i, car in enumerate(best_cars):
|
||
car_list.append(
|
||
{
|
||
"rank": i + 1,
|
||
# no clue
|
||
"member": user_id,
|
||
"value": car["goal_time"],
|
||
"store": self.core_cfg.server.name,
|
||
# use car_id from request?
|
||
"car_id": 0,
|
||
"style_car_id": car["style_car_id"],
|
||
"play_dt": car["play_dt"].timestamp(),
|
||
"section_time_1": car["section_time_1"],
|
||
"section_time_2": car["section_time_2"],
|
||
"section_time_3": car["section_time_3"],
|
||
"section_time_4": car["section_time_4"],
|
||
"mission": car["mission"],
|
||
}
|
||
)
|
||
|
||
course_pickup_car_best_data.append(
|
||
{
|
||
"course_id": course["course_id"],
|
||
"car_list": car_list,
|
||
}
|
||
)
|
||
|
||
return {
|
||
"status_code": "0",
|
||
"course_pickup_car_best_data": course_pickup_car_best_data,
|
||
}
|
||
|
||
def handle_user_updateonlinebattle_request(self, data: Dict, headers: Dict):
|
||
return {
|
||
"status_code": "0",
|
||
"bothwin_penalty": 1,
|
||
}
|
||
|
||
def handle_user_updateonlinebattleresult_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
stock_data: Dict = data.pop("stock_obj")
|
||
# save stock data in database
|
||
self._save_stock_data(user_id, stock_data)
|
||
|
||
ticket_data: List = data.pop("ticket_data")
|
||
for ticket in ticket_data:
|
||
self.data.item.put_ticket(user_id, ticket)
|
||
|
||
reward_dist_data: Dict = data.pop("reward_dist_obj")
|
||
rank_data: Dict = data.pop("mode_rank_obj")
|
||
|
||
# save rank dist data in database
|
||
rank_data.update(reward_dist_data)
|
||
self.data.profile.put_profile_rank(user_id, self.version, rank_data)
|
||
|
||
driver_debut_data = data.pop("driver_debut_obj")
|
||
|
||
# get the use_count and net_vs_use_count of the used car
|
||
style_car_id = data.get("style_car_id")
|
||
car_mileage = data.pop("car_mileage")
|
||
used_car = self.data.item.get_car(user_id, self.version, style_car_id)._asdict()
|
||
|
||
# increase the use_count and net_vs_use_count of the used car
|
||
used_car["use_count"] += 1
|
||
used_car["net_vs_use_count"] += 1
|
||
used_car["car_mileage"] = car_mileage
|
||
|
||
# save the used car in database
|
||
self.data.item.put_car(user_id, self.version, used_car)
|
||
|
||
# get the profile data, update total_play and daily_play, and save it
|
||
profile = self.data.profile.get_profile(user_id, self.version)
|
||
total_play = profile["total_play"] + 1
|
||
|
||
# save the profile in database
|
||
self.data.profile.put_profile(
|
||
user_id,
|
||
self.version,
|
||
{
|
||
"total_play": total_play,
|
||
"last_play_date": datetime.now(),
|
||
"mileage": data.pop("mileage"),
|
||
"aura_id": data.pop("aura_id"),
|
||
"aura_color_id": data.pop("aura_color_id"),
|
||
"aura_line_id": data.pop("aura_line_id"),
|
||
"cash": data.pop("cash"),
|
||
"total_cash": data.pop("total_cash"),
|
||
"dressup_point": data.pop("dressup_point"),
|
||
"avatar_point": data.pop("avatar_point"),
|
||
},
|
||
)
|
||
|
||
self.data.item.put_vs_info(user_id, data)
|
||
|
||
vs_info = {
|
||
"battle_mode": 0,
|
||
"vs_cnt": 1,
|
||
"vs_win": data.get("win_flg"),
|
||
"invalid": 0,
|
||
"str": 0,
|
||
"str_now": 0,
|
||
"lose_now": 0,
|
||
"vs_history": data.get("vs_history"),
|
||
"course_select_priority": data.get("course_select_priority"),
|
||
"vsinfo_course_data": [
|
||
{
|
||
"course_id": data.get("course_id"),
|
||
"vs_cnt": 1,
|
||
"vs_win": data.get("win_flg"),
|
||
}
|
||
],
|
||
}
|
||
|
||
return {
|
||
"status_code": "0",
|
||
"vsinfo_data": vs_info,
|
||
"round_event": [
|
||
{
|
||
"count": 1,
|
||
"win": 1,
|
||
"rank": 1,
|
||
"point": 1,
|
||
"total_round_point": 1,
|
||
}
|
||
],
|
||
"car_use_count": [],
|
||
"maker_use_count": [],
|
||
}
|
||
|
||
def handle_user_updatestorebattleresult_request(self, data: Dict, headers: Dict):
|
||
user_id = headers["session"]
|
||
|
||
stock_data: Dict = data.pop("stock_obj")
|
||
ticket_data: List = data.pop("ticket_data")
|
||
reward_dist_data: Dict = data.pop("reward_dist_obj")
|
||
rank_data: Dict = data.pop("mode_rank_obj")
|
||
driver_debut_data: Dict = data.pop("driver_debut_obj")
|
||
|
||
# no idea?
|
||
result = data.pop("result")
|
||
battle_gift_event_id = data.pop("battle_gift_event_id")
|
||
gift_id = data.pop("gift_id")
|
||
|
||
# save stock data in database
|
||
self._save_stock_data(user_id, stock_data)
|
||
|
||
# save tickets in database
|
||
for ticket in ticket_data:
|
||
self.data.item.put_ticket(user_id, ticket)
|
||
|
||
# save rank dist data in database
|
||
rank_data.update(reward_dist_data)
|
||
self.data.profile.put_profile_rank(user_id, self.version, rank_data)
|
||
|
||
# get the use_count and net_vs_use_count of the used car
|
||
style_car_id = data.get("style_car_id")
|
||
car_mileage = data.pop("car_mileage")
|
||
used_car = self.data.item.get_car(user_id, self.version, style_car_id)._asdict()
|
||
|
||
# increase the use_count and net_vs_use_count of the used car
|
||
used_car["use_count"] += 1
|
||
used_car["vs_use_count"] += 1
|
||
used_car["car_mileage"] = car_mileage
|
||
|
||
# save the used car in database
|
||
self.data.item.put_car(user_id, self.version, used_car)
|
||
|
||
# get the profile data, update total_play and daily_play, and save it
|
||
profile = self.data.profile.get_profile(user_id, self.version)
|
||
total_play = profile["total_play"] + 1
|
||
|
||
# save the profile in database
|
||
self.data.profile.put_profile(
|
||
user_id,
|
||
self.version,
|
||
{
|
||
"total_play": total_play,
|
||
"last_play_date": datetime.now(),
|
||
"mileage": data.pop("mileage"),
|
||
"aura_id": data.pop("aura_id"),
|
||
"aura_color_id": data.pop("aura_color_id"),
|
||
"aura_line_id": data.pop("aura_line_id"),
|
||
"cash": data.pop("cash"),
|
||
"total_cash": data.pop("total_cash"),
|
||
"dressup_point": data.pop("dressup_point"),
|
||
"avatar_point": data.pop("avatar_point"),
|
||
},
|
||
)
|
||
|
||
# save vs_info in database
|
||
self.data.item.put_vs_info(user_id, data)
|
||
|
||
vs_info = {
|
||
"battle_mode": 0,
|
||
"vs_cnt": 1,
|
||
"vs_win": data.get("win_flg"),
|
||
"invalid": 0,
|
||
"str": 0,
|
||
"str_now": 0,
|
||
"lose_now": 0,
|
||
"vs_history": data.get("vs_history"),
|
||
"course_select_priority": 0,
|
||
"vsinfo_course_data": [
|
||
{
|
||
"course_id": data.get("course_id"),
|
||
"vs_cnt": 1,
|
||
"vs_win": data.get("win_flg"),
|
||
}
|
||
],
|
||
}
|
||
|
||
return {
|
||
"status_code": "0",
|
||
"vsinfo_data": vs_info,
|
||
"car_use_count": [],
|
||
"maker_use_count": [],
|
||
}
|