import datetime from typing import Any, List, Dict import logging import json import urllib from core.config import CoreConfig from titles.diva.config import DivaConfig from titles.diva.const import DivaConstants from titles.diva.database import DivaData class DivaBase(): def __init__(self, cfg: CoreConfig, game_cfg: DivaConfig) -> None: self.core_cfg = cfg # Config file self.game_config = game_cfg self.data = DivaData(cfg) # Database self.date_time_format = "%Y-%m-%d %H:%M:%S" self.logger = logging.getLogger("diva") self.game = DivaConstants.GAME_CODE self.version = DivaConstants.VER_PROJECT_DIVA_ARCADE_FUTURE_TONE dt = datetime.datetime.now() self.time_lut=urllib.parse.quote(dt.strftime("%Y-%m-%d %H:%M:%S:16.0")) def handle_test_request(self, data: Dict) -> Dict: return "" def handle_game_init_request(self, data: Dict) -> Dict: return ( f'' ) def handle_attend_request(self, data: Dict) -> Dict: encoded = "&" params = { 'atnd_prm1': '0,1,1,0,0,0,1,0,100,0,0,0,0,0,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1', 'atnd_prm2': '30,10,100,4,1,50,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1', 'atnd_prm3': '100,0,1,1,1,1,1,1,1,1,2,3,4,1,1,1,3,4,5,1,1,1,4,5,6,1,1,1,5,6,7,4,4,4,9,10,14,5,10,10,25,20,50,30,90,5,10,10,25,20,50,30,90,5,10,10,25,20,50,30,90,5,10,10,25,20,50,30,90,5,10,10,25,20,50,30,90,10,30,1,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0', 'atnd_lut': f'{self.time_lut}', } encoded += urllib.parse.urlencode(params) encoded = encoded.replace("%2C", ",") return encoded def handle_ping_request(self, data: Dict) -> Dict: encoded = "&" params = { 'ping_b_msg': f'Welcome to {self.core_cfg.server.name} network!', 'ping_m_msg': 'xxx', 'atnd_lut': f'{self.time_lut}', 'fi_lut': f'{self.time_lut}', 'ci_lut': f'{self.time_lut}', 'qi_lut': f'{self.time_lut}', 'pvl_lut': '2021-05-22 12:08:16.0', 'shp_ctlg_lut': '2020-06-10 19:44:16.0', 'cstmz_itm_ctlg_lut': '2019-10-08 20:23:12.0', 'ngwl_lut': '2019-10-08 20:23:12.0', 'rnk_nv_lut': '2020-06-10 19:51:30.0', 'rnk_ps_lut': f'{self.time_lut}', 'bi_lut': '2020-09-18 10:00:00.0', 'cpi_lut': '2020-10-25 09:25:10.0', 'bdlol_lut': '2020-09-18 10:00:00.0', 'p_std_hc_lut': '2019-08-01 04:00:36.0', 'p_std_i_n_lut': '2019-08-01 04:00:36.0', 'pdcl_lut': '2019-08-01 04:00:36.0', 'pnml_lut': '2019-08-01 04:00:36.0', 'cinml_lut': '2019-08-01 04:00:36.0', 'rwl_lut': '2019-08-01 04:00:36.0', 'req_inv_cmd_num': '-1,-1,-1,-1,-1,-1,-1,-1,-1,-1', 'req_inv_cmd_prm1': '-1,-1,-1,-1,-1,-1,-1,-1,-1,-1', 'req_inv_cmd_prm2': '-1,-1,-1,-1,-1,-1,-1,-1,-1,-1', 'req_inv_cmd_prm3': '-1,-1,-1,-1,-1,-1,-1,-1,-1,-1', 'req_inv_cmd_prm4': '-1,-1,-1,-1,-1,-1,-1,-1,-1,-1', 'pow_save_flg': 0, 'nblss_dnt_p': 100, 'nblss_ltt_rl_vp': 1500, 'nblss_ex_ltt_flg': 1, 'nblss_dnt_st_tm': "2019-07-15 12:00:00.0", 'nblss_dnt_ed_tm': "2019-09-17 12:00:00.0", 'nblss_ltt_st_tm': "2019-09-18 12:00:00.0", 'nblss_ltt_ed_tm': "2019-09-22 12:00:00.0", } encoded += urllib.parse.urlencode(params) encoded = encoded.replace("+", "%20") encoded = encoded.replace("%2C", ",") return encoded def handle_pv_list_request(self, data: Dict) -> Dict: pvlist = "" with open(r"titles/diva/data/PvList0.dat", encoding="utf-8") as shop: lines = shop.readlines() for line in lines: pvlist += f"{line}" pvlist += "," with open(r"titles/diva/data/PvList1.dat", encoding="utf-8") as shop: lines = shop.readlines() for line in lines: pvlist += f"{line}" pvlist += "," with open(r"titles/diva/data/PvList2.dat", encoding="utf-8") as shop: lines = shop.readlines() for line in lines: pvlist += f"{line}" pvlist += "," with open(r"titles/diva/data/PvList3.dat", encoding="utf-8") as shop: lines = shop.readlines() for line in lines: pvlist += f"{line}" pvlist += "," with open(r"titles/diva/data/PvList4.dat", encoding="utf-8") as shop: lines = shop.readlines() for line in lines: pvlist += f"{line}" response = "" response += f"&pvl_lut={self.time_lut}" response += f"&pv_lst={pvlist}" return ( response ) def handle_shop_catalog_request(self, data: Dict) -> Dict: catalog = "" shopList = self.data.static.get_enabled_shops(self.version) if not shopList: with open(r"titles/diva/data/ShopCatalog.dat", encoding="utf-8") as shop: lines = shop.readlines() for line in lines: line = urllib.parse.quote(line) + "," catalog += f"{urllib.parse.quote(line)}" else: for shop in shopList: line = str(shop["shopId"]) + "," + str(shop['unknown_0']) + "," + shop['name'] + "," + str(shop['points']) + "," + shop['start_date'] + "," + shop['end_date'] + "," + str(shop["type"]) line = urllib.parse.quote(line) + "," catalog += f"{urllib.parse.quote(line)}" catalog = catalog.replace("+", "%20") response = f"&shp_ctlg_lut={self.time_lut}" response += f"&shp_ctlg={catalog[:-3]}" return ( response ) def handle_buy_module_request(self, data: Dict) -> Dict: profile = self.data.profile.get_profile(data["pd_id"], self.version) module = self.data.static.get_enabled_shop(self.version, int(data["mdl_id"])) # make sure module is available to purchase if not module: return f"&shp_rslt=0&vcld_pts={profile['vcld_pts']}" # make sure player has enough vocaloid points to buy module if profile["vcld_pts"] < int(data["mdl_price"]): return f"&shp_rslt=0&vcld_pts={profile['vcld_pts']}" new_vcld_pts = profile["vcld_pts"] - int(data["mdl_price"]) self.data.profile.update_profile( profile["user"], vcld_pts=new_vcld_pts ) self.data.module.put_module(data["pd_id"], self.version, data["mdl_id"]) # generate the mdl_have string mdl_have = self.data.module.get_modules_have_string(data["pd_id"], self.version) response = "&shp_rslt=1" response += f"&mdl_id={data['mdl_id']}" response += f"&mdl_have={mdl_have}" response += f"&vcld_pts={new_vcld_pts}" return response def handle_cstmz_itm_ctlg_request(self, data: Dict) -> Dict: catalog = "" itemList = self.data.static.get_enabled_items(self.version) if not itemList: with open(r"titles/diva/data/ItemCatalog.dat", encoding="utf-8") as item: lines = item.readlines() for line in lines: line = urllib.parse.quote(line) + "," catalog += f"{urllib.parse.quote(line)}" else: for item in itemList: line = str(item["itemId"]) + "," + str(item['unknown_0']) + "," + item['name'] + "," + str(item['points']) + "," + item['start_date'] + "," + item['end_date'] + "," + str(item["type"]) line = urllib.parse.quote(line) + "," catalog += f"{urllib.parse.quote(line)}" catalog = catalog.replace("+", "%20") response = f"&cstmz_itm_ctlg_lut={self.time_lut}" response += f"&cstmz_itm_ctlg={catalog[:-3]}" return ( response ) def handle_buy_cstmz_itm_request(self, data: Dict) -> Dict: profile = self.data.profile.get_profile(data["pd_id"], self.version) item = self.data.static.get_enabled_item(self.version, int(data["cstmz_itm_id"])) # make sure module is available to purchase if not item: return f"&shp_rslt=0&vcld_pts={profile['vcld_pts']}" # make sure player has enough vocaloid points to buy the customize item if profile["vcld_pts"] < int(data["cstmz_itm_price"]): return f"&shp_rslt=0&vcld_pts={profile['vcld_pts']}" new_vcld_pts = profile["vcld_pts"] - int(data["cstmz_itm_price"]) # save new Vocaloid Points balance self.data.profile.update_profile( profile["user"], vcld_pts=new_vcld_pts ) self.data.customize.put_customize_item(data["pd_id"], self.version, data["cstmz_itm_id"]) # generate the cstmz_itm_have string cstmz_itm_have = self.data.customize.get_customize_items_have_string(data["pd_id"], self.version) response = "&shp_rslt=1" response += f"&cstmz_itm_id={data['cstmz_itm_id']}" response += f"&cstmz_itm_have={cstmz_itm_have}" response += f"&vcld_pts={new_vcld_pts}" return response def handle_festa_info_request(self, data: Dict) -> Dict: encoded = "&" params = { 'fi_id': '1,-1', 'fi_name': f'{self.core_cfg.server.name} Opening,xxx', 'fi_kind': '0,0', 'fi_difficulty': '-1,-1', 'fi_pv_id_lst': 'ALL,ALL', 'fi_attr': '7FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF,7FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF', 'fi_add_vp': '20,0', 'fi_mul_vp': '1,1', 'fi_st': '2022-06-17 17:00:00.0,2014-07-08 18:10:11.0', 'fi_et': '2029-01-01 10:00:00.0,2014-07-08 18:10:11.0', 'fi_lut': '{self.time_lut}', } encoded += urllib.parse.urlencode(params) encoded = encoded.replace("+", "%20") encoded = encoded.replace("%2C", ",") return encoded def handle_contest_info_request(self, data: Dict) -> Dict: response = "" response += f"&ci_lut={self.time_lut}" response += "&ci_str=%2A%2A%2A,%2A%2A%2A,%2A%2A%2A,%2A%2A%2A,%2A%2A%2A,%2A%2A%2A,%2A%2A%2A,%2A%2A%2A" return ( response ) def handle_qst_inf_request(self, data: Dict) -> Dict: quest = "" questList = self.data.static.get_enabled_quests(self.version) if not questList: with open(r"titles/diva/data/QuestInfo.dat", encoding="utf-8") as shop: lines = shop.readlines() for line in lines: quest += f"{urllib.parse.quote(line)}," response = "" response += f"&qi_lut={self.time_lut}" response += f"&qhi_str={quest[:-1]}" else: for quests in questList: line = str(quests["questId"]) + "," + str(quests['quest_order']) + "," + str(quests['kind']) + "," + str(quests['unknown_0']) + "," + quests['start_datetime'] + "," + quests['end_datetime'] + "," + quests["name"] + "," + str(quests["unknown_1"]) + "," + str(quests["unknown_2"]) + "," + str(quests["quest_enable"]) quest += f"{urllib.parse.quote(line)}%0A," responseline = f"{quest[:-1]}," for i in range(len(questList),59): responseline += "%2A%2A%2A%0A," response = "" response += f"&qi_lut={self.time_lut}" response += f"&qhi_str={responseline}%2A%2A%2A" response += "&qrai_str=%2D1%2C%2D1%2C%2D1%2C%2D1%2C%2D1%2C%2D1%2C%2D1%2C%2D1%2C%2D1%2C%2D1%2C%2D1,%2D1%2C%2D1%2C%2D1%2C%2D1%2C%2D1%2C%2D1%2C%2D1%2C%2D1%2C%2D1%2C%2D1%2C%2D1,%2D1%2C%2D1%2C%2D1%2C%2D1%2C%2D1%2C%2D1%2C%2D1%2C%2D1%2C%2D1%2C%2D1%2C%2D1" return ( response ) def handle_nv_ranking_request(self, data: Dict) -> Dict: return ( f'' ) def handle_ps_ranking_request(self, data: Dict) -> Dict: return ( f'' ) def handle_ng_word_request(self, data: Dict) -> Dict: return ( f'' ) def handle_rmt_wp_list_request(self, data: Dict) -> Dict: return ( f'' ) def handle_pv_def_chr_list_request(self, data: Dict) -> Dict: return ( f'' ) def handle_pv_ng_mdl_list_request(self, data: Dict) -> Dict: return ( f'' ) def handle_cstmz_itm_ng_mdl_lst_request(self, data: Dict) -> Dict: return ( f'' ) def handle_banner_info_request(self, data: Dict) -> Dict: return ( f'' ) def handle_banner_data_request(self, data: Dict) -> Dict: return ( f'' ) def handle_cm_ply_info_request(self, data: Dict) -> Dict: return ( f'' ) def handle_pstd_h_ctrl_request(self, data: Dict) -> Dict: return ( f'' ) def handle_pstd_item_ng_lst_request(self, data: Dict) -> Dict: return ( f'' ) def handle_pre_start_request(self, data: Dict) -> str: profile = self.data.profile.get_profile(data["aime_id"], self.version) profile_shop = self.data.item.get_shop(data["aime_id"], self.version) if profile is None: return f"&ps_result=-3" else: response = "&ps_result=1" response += "&accept_idx=100" response += "&nblss_ltt_stts=-1" response += "&nblss_ltt_tckt=-1" response += "&nblss_ltt_is_opn=-1" response += f"&pd_id={data['aime_id']}" response += f"&player_name={profile['player_name']}" response += f"&sort_kind={profile['player_name']}" response += f"&lv_efct_id={profile['lv_efct_id']}" response += f"&lv_plt_id={profile['lv_plt_id']}" response += f"&lv_str={profile['lv_str']}" response += f"&lv_num={profile['lv_num']}" response += f"&lv_pnt={profile['lv_pnt']}" response += f"&vcld_pts={profile['vcld_pts']}" response += f"&skn_eqp={profile['use_pv_skn_eqp']}" response += f"&btn_se_eqp={profile['use_pv_btn_se_eqp']}" response += f"&sld_se_eqp={profile['use_pv_sld_se_eqp']}" response += f"&chn_sld_se_eqp={profile['use_pv_chn_sld_se_eqp']}" response += f"&sldr_tch_se_eqp={profile['use_pv_sldr_tch_se_eqp']}" response += f"&passwd_stat={profile['passwd_stat']}" # Store stuff to add to rework response += f"&mdl_eqp_tm={self.time_lut}" mdl_eqp_ary = "-999,-999,-999" # get the common_modules from the profile shop if profile_shop: mdl_eqp_ary = profile_shop["mdl_eqp_ary"] response += f"&mdl_eqp_ary={mdl_eqp_ary}" return response def handle_registration_request(self, data: Dict) -> Dict: self.data.profile.create_profile(self.version, data["aime_id"], data["player_name"]) return (f"&cd_adm_result=1&pd_id={data['aime_id']}") def handle_start_request(self, data: Dict) -> Dict: profile = self.data.profile.get_profile(data["pd_id"], self.version) profile_shop = self.data.item.get_shop(data["pd_id"], self.version) if profile is None: return mdl_have = "F" * 250 # generate the mdl_have string if "unlock_all_modules" is disabled if not self.game_config.mods.unlock_all_modules: mdl_have = self.data.module.get_modules_have_string(data["pd_id"], self.version) cstmz_itm_have = "F" * 250 # generate the cstmz_itm_have string if "unlock_all_items" is disabled if not self.game_config.mods.unlock_all_items: cstmz_itm_have = self.data.customize.get_customize_items_have_string(data["pd_id"], self.version) response = f"&pd_id={data['pd_id']}" response += "&start_result=1" response += "&accept_idx=100" response += f"&hp_vol={profile['hp_vol']}" response += f"&btn_se_vol={profile['btn_se_vol']}" response += f"&btn_se_vol2={profile['btn_se_vol2']}" response += f"&sldr_se_vol2={profile['sldr_se_vol2']}" response += f"&sort_kind={profile['sort_kind']}" response += f"&player_name={profile['player_name']}" response += f"&lv_num={profile['lv_num']}" response += f"&lv_pnt={profile['lv_pnt']}" response += f"&lv_efct_id={profile['lv_efct_id']}" response += f"&lv_plt_id={profile['lv_plt_id']}" response += f"&mdl_have={mdl_have}" response += f"&cstmz_itm_have={cstmz_itm_have}" response += f"&use_pv_mdl_eqp={int(profile['use_pv_mdl_eqp'])}" response += f"&use_mdl_pri={int(profile['use_mdl_pri'])}" response += f"&use_pv_skn_eqp={int(profile['use_pv_skn_eqp'])}" response += f"&use_pv_btn_se_eqp={int(profile['use_pv_btn_se_eqp'])}" response += f"&use_pv_sld_se_eqp={int(profile['use_pv_sld_se_eqp'])}" response += f"&use_pv_chn_sld_se_eqp={int(profile['use_pv_chn_sld_se_eqp'])}" response += f"&use_pv_sldr_tch_se_eqp={int(profile['use_pv_sldr_tch_se_eqp'])}" response += f"&vcld_pts={profile['lv_efct_id']}" response += f"&nxt_pv_id={profile['nxt_pv_id']}" response += f"&nxt_dffclty={profile['nxt_dffclty']}" response += f"&nxt_edtn={profile['nxt_edtn']}" response += f"&dsp_clr_brdr={profile['dsp_clr_brdr']}" response += f"&dsp_intrm_rnk={profile['dsp_intrm_rnk']}" response += f"&dsp_clr_sts={profile['dsp_clr_sts']}" response += f"&rgo_sts={profile['rgo_sts']}" # To be fully fixed if "my_qst_id" not in profile: response += f"&my_qst_id=-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1" response += f"&my_qst_sts=0,0,0,0,0,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1" else: response += f"&my_qst_id={profile['my_qst_id']}" response += f"&my_qst_sts={profile['my_qst_sts']}" response += f"&my_qst_prgrs=0,0,0,0,0,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1" response += f"&my_qst_et=2022-06-19%2010%3A28%3A52.0,2022-06-19%2010%3A28%3A52.0,2022-06-19%2010%3A28%3A52.0,2100-01-01%2008%3A59%3A59.0,2100-01-01%2008%3A59%3A59.0,xxx,xxx,xxx,xxx,xxx,xxx,xxx,xxx,xxx,xxx,xxx,xxx,xxx,xxx,xxx,xxx,xxx,xxx,xxx,xxx" response += f"&clr_sts=0,0,0,0,0,0,0,0,56,52,35,6,6,3,1,0,0,0,0,0" # Store stuff to add to rework response += f"&mdl_eqp_tm={self.time_lut}" mdl_eqp_ary = "-999,-999,-999" c_itm_eqp_ary = "-999,-999,-999,-999,-999,-999,-999,-999,-999,-999,-999,-999" ms_itm_flg_ary = "1,1,1,1,1,1,1,1,1,1,1,1" # get the common_modules, customize_items and customize_item_flags # from the profile shop if profile_shop: mdl_eqp_ary = profile_shop["mdl_eqp_ary"] c_itm_eqp_ary = profile_shop["c_itm_eqp_ary"] ms_itm_flg_ary = profile_shop["ms_itm_flg_ary"] response += f"&mdl_eqp_ary={mdl_eqp_ary}" response += f"&c_itm_eqp_ary={c_itm_eqp_ary}" response += f"&ms_itm_flg_ary={ms_itm_flg_ary}" return ( response ) def handle_pd_unlock_request(self, data: Dict) -> Dict: return ( f'' ) def handle_spend_credit_request(self, data: Dict) -> Dict: profile = self.data.profile.get_profile(data["pd_id"], self.version) if profile is None: return response = "" response += "&cmpgn_rslt=-1,-1,x,-1,-1,x,x,-1,x,-1,-1,x,-1,-1,x,x,-1,x,-1,-1,x,-1,-1,x,x,-1,x,-1,-1,x,-1,-1,x,x,-1,x,-1,-1,x,-1,-1,x,x,-1,x,-1,-1,x,-1,-1,x,x,-1,x" response += "&cmpgn_rslt_num=0" response += f"&vcld_pts={profile['vcld_pts']}" response += f"&lv_str={profile['lv_str']}" response += f"&lv_efct_id={profile['lv_efct_id']}" response += f"&lv_plt_id={profile['lv_plt_id']}" return ( response ) def _get_pv_pd_result(self, song: int, pd_db_song: Dict, pd_db_ranking: Dict, pd_db_customize: Dict, edition: int) -> str: """ Helper function to generate the pv_result string for every song, ranking and edition """ global_ranking = -1 if pd_db_ranking: # make sure there are enough max scores to calculate a ranking if pd_db_ranking["ranking"] != 0: global_ranking = pd_db_ranking["ranking"] # pv_no pv_result = f"{song}," # edition pv_result += f"{edition}," # rslt pv_result += f"{pd_db_song['clr_kind']}," if pd_db_song else "-1," # max_score pv_result += f"{pd_db_song['score']}," if pd_db_song else "-1," # max_atn_pnt pv_result += f"{pd_db_song['atn_pnt']}," if pd_db_song else "-1," # challenge_kind pv_result += f"{pd_db_song['sort_kind']}," if pd_db_song else "0," module_eqp = "-999,-999,-999" customize_eqp = "-999,-999,-999,-999,-999,-999,-999,-999,-999,-999,-999,-999" customize_flag = "-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1" # skin, btn_se, sld_se, chsld_se, sldtch_se pv_settings = "-1,-1,-1,-1,-1" if pd_db_customize: module_eqp = pd_db_customize["mdl_eqp_ary"] customize_eqp = pd_db_customize["c_itm_eqp_ary"] customize_flag = pd_db_customize["ms_itm_flg_ary"] pv_settings = ( f"{pd_db_customize['skin']}," f"{pd_db_customize['btn_se']}," f"{pd_db_customize['sld_se']}," f"{pd_db_customize['chsld_se']}," f"{pd_db_customize['sldtch_se']}" ) pv_result += f"{module_eqp}," pv_result += f"{customize_eqp}," pv_result += f"{customize_flag}," pv_result += f"{pv_settings}," # rvl_pd_id, rvl_score, rvl_attn_pnt, -1, -1 pv_result += "-1,-1,-1,-1,-1," # countrywide_ranking pv_result += f"{global_ranking}," # rgo_purchased pv_result += "1,1,1," # rgo_played pv_result += "0,0,0" return pv_result def handle_get_pv_pd_request(self, data: Dict) -> Dict: song_id = data["pd_pv_id_lst"].split(",") pv = "" for song in song_id: if int(song) > 0: # the request do not send a edition so just perform a query best score and ranking for each edition. # 0=ORIGINAL, 1=EXTRA pd_db_song_0 = self.data.score.get_best_user_score(data["pd_id"], int(song), data["difficulty"], edition=0) pd_db_song_1 = self.data.score.get_best_user_score(data["pd_id"], int(song), data["difficulty"], edition=1) pd_db_ranking_0, pd_db_ranking_1 = None, None if pd_db_song_0: pd_db_ranking_0 = self.data.score.get_global_ranking(data["pd_id"], int(song), data["difficulty"], edition=0) if pd_db_song_1: pd_db_ranking_1 = self.data.score.get_global_ranking(data["pd_id"], int(song), data["difficulty"], edition=1) pd_db_customize = self.data.pv_customize.get_pv_customize(data["pd_id"], int(song)) # generate the pv_result string with the ORIGINAL edition and the EXTRA edition appended pv_result = self._get_pv_pd_result(int(song), pd_db_song_0, pd_db_ranking_0, pd_db_customize, edition=0) pv_result += "," + self._get_pv_pd_result(int(song), pd_db_song_1, pd_db_ranking_1, pd_db_customize, edition=1) self.logger.debug(f"pv_result = {pv_result}") pv += urllib.parse.quote(pv_result) else: pv += urllib.parse.quote(f"{song}***") pv += "," response = "" response += f"&pd_by_pv_id={pv[:-1]}" response += "&pdddt_flg=0" response += f"&pdddt_tm={self.time_lut}" return ( response ) def handle_stage_start_request(self, data: Dict) -> Dict: return ( f'' ) def handle_stage_result_request(self, data: Dict) -> Dict: profile = self.data.profile.get_profile(data["pd_id"], self.version) pd_song_list = data["stg_ply_pv_id"].split(",") pd_song_difficulty = data["stg_difficulty"].split(",") pd_song_edition = data["stg_edtn"].split(",") pd_song_max_score = data["stg_score"].split(",") pd_song_max_atn_pnt = data["stg_atn_pnt"].split(",") pd_song_ranking = data["stg_clr_kind"].split(",") pd_song_sort_kind = data["sort_kind"] pd_song_cool_cnt = data["stg_cool_cnt"].split(",") pd_song_fine_cnt = data["stg_fine_cnt"].split(",") pd_song_safe_cnt = data["stg_safe_cnt"].split(",") pd_song_sad_cnt = data["stg_sad_cnt"].split(",") pd_song_worst_cnt = data["stg_wt_wg_cnt"].split(",") pd_song_max_combo = data["stg_max_cmb"].split(",") for index, value in enumerate(pd_song_list): if "-1" not in pd_song_list[index]: profile_pd_db_song = self.data.score.get_best_user_score(data["pd_id"], pd_song_list[index], pd_song_difficulty[index], pd_song_edition[index]) if profile_pd_db_song is None: self.data.score.put_best_score(data["pd_id"], self.version, pd_song_list[index], pd_song_difficulty[index], pd_song_edition[index], pd_song_max_score[index], pd_song_max_atn_pnt[index], pd_song_ranking[index], pd_song_sort_kind, pd_song_cool_cnt[index], pd_song_fine_cnt[index], pd_song_safe_cnt[index], pd_song_sad_cnt[index], pd_song_worst_cnt[index], pd_song_max_combo[index]) self.data.score.put_playlog(data["pd_id"], self.version, pd_song_list[index], pd_song_difficulty[index], pd_song_edition[index], pd_song_max_score[index], pd_song_max_atn_pnt[index], pd_song_ranking[index], pd_song_sort_kind, pd_song_cool_cnt[index], pd_song_fine_cnt[index], pd_song_safe_cnt[index], pd_song_sad_cnt[index], pd_song_worst_cnt[index], pd_song_max_combo[index]) elif int(pd_song_max_score[index]) >= int(profile_pd_db_song["score"]): self.data.score.put_best_score(data["pd_id"], self.version, pd_song_list[index], pd_song_difficulty[index], pd_song_edition[index], pd_song_max_score[index], pd_song_max_atn_pnt[index], pd_song_ranking[index], pd_song_sort_kind, pd_song_cool_cnt[index], pd_song_fine_cnt[index], pd_song_safe_cnt[index], pd_song_sad_cnt[index], pd_song_worst_cnt[index], pd_song_max_combo[index]) self.data.score.put_playlog(data["pd_id"], self.version, pd_song_list[index], pd_song_difficulty[index], pd_song_edition[index], pd_song_max_score[index], pd_song_max_atn_pnt[index], pd_song_ranking[index], pd_song_sort_kind, pd_song_cool_cnt[index], pd_song_fine_cnt[index], pd_song_safe_cnt[index], pd_song_sad_cnt[index], pd_song_worst_cnt[index], pd_song_max_combo[index]) elif int(pd_song_max_score[index]) != int(profile_pd_db_song["score"]): self.data.score.put_playlog(data["pd_id"], self.version, pd_song_list[index], pd_song_difficulty[index], pd_song_edition[index], pd_song_max_score[index], pd_song_max_atn_pnt[index], pd_song_ranking[index], pd_song_sort_kind, pd_song_cool_cnt[index], pd_song_fine_cnt[index], pd_song_safe_cnt[index], pd_song_sad_cnt[index], pd_song_worst_cnt[index], pd_song_max_combo[index]) # Profile saving based on registration list # Calculate new level best_scores = self.data.score.get_best_scores(data["pd_id"]) total_atn_pnt = 0 for best_score in best_scores: total_atn_pnt += best_score["atn_pnt"] new_level = (total_atn_pnt // 13979) + 1 new_level_pnt = round((total_atn_pnt % 13979) / 13979 * 100) response = "&chllng_kind=-1" response += f"&lv_num_old={int(profile['lv_num'])}" response += f"&lv_pnt_old={int(profile['lv_pnt'])}" # update the profile and commit changes to the db self.data.profile.update_profile( profile["user"], lv_num=new_level, lv_pnt=new_level_pnt, vcld_pts=int(data["vcld_pts"]), hp_vol=int(data["hp_vol"]), btn_se_vol=int(data["btn_se_vol"]), sldr_se_vol2=int(data["sldr_se_vol2"]), sort_kind=int(data["sort_kind"]), nxt_pv_id=int(data["ply_pv_id"]), nxt_dffclty=int(data["nxt_dffclty"]), nxt_edtn=int(data["nxt_edtn"]), my_qst_id=data["my_qst_id"], my_qst_sts=data["my_qst_sts"] ) response += f"&lv_num={new_level}" response += f"&lv_str={profile['lv_str']}" response += f"&lv_pnt={new_level_pnt}" response += f"&lv_efct_id={int(profile['lv_efct_id'])}" response += f"&lv_plt_id={int(profile['lv_plt_id'])}" response += f"&vcld_pts={int(data['vcld_pts'])}" response += f"&prsnt_vcld_pts={int(profile['vcld_pts'])}" response += "&cerwd_kind=-1" response += "&cerwd_value=-1" response += "&cerwd_str_0=***" response += "&cerwd_str_1=***" response += "&ttl_str_ary=xxx,xxx,xxx,xxx,xxx" response += "&ttl_plt_id_ary=-1,-1,-1,-1,-1" response += "&ttl_desc_ary=xxx,xxx,xxx,xxx,xxx" response += "&skin_id_ary=xxx,xxx,xxx,xxx,xxx" response += "&skin_name_ary=xxx,xxx,xxx,xxx,xxx" response += "&skin_illust_ary=xxx,xxx,xxx,xxx,xxx" response += "&skin_desc_ary=xxx,xxx,xxx,xxx,xxx" if "my_qst_id" not in profile: response += f"&my_qst_id=-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1" else: response += f"&my_qst_id={profile['my_qst_id']}" response += "&my_qst_r_qid=-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1" response += "&my_qst_r_knd=-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1" response += "&my_qst_r_vl=-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1" response += "&my_qst_r_nflg=-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1" response += "&my_ccd_r_qid=-1,-1,-1,-1,-1" response += "&my_ccd_r_hnd=-1,-1,-1,-1,-1" response += "&my_ccd_r_vp=-1,-1,-1,-1,-1" return ( response ) def handle_end_request(self, data: Dict) -> Dict: profile = self.data.profile.get_profile(data["pd_id"], self.version) self.data.profile.update_profile( profile["user"], my_qst_id=data["my_qst_id"], my_qst_sts=data["my_qst_sts"] ) return (f'') def handle_shop_exit_request(self, data: Dict) -> Dict: self.data.item.put_shop(data["pd_id"], self.version, data["mdl_eqp_cmn_ary"], data["c_itm_eqp_cmn_ary"], data["ms_itm_flg_cmn_ary"]) if int(data["use_pv_mdl_eqp"]) == 1: self.data.pv_customize.put_pv_customize(data["pd_id"], self.version, data["ply_pv_id"], data["mdl_eqp_pv_ary"], data["c_itm_eqp_pv_ary"], data["ms_itm_flg_pv_ary"]) else: self.data.pv_customize.put_pv_customize(data["pd_id"], self.version, data["ply_pv_id"], "-1,-1,-1", "-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1", "1,1,1,1,1,1,1,1,1,1,1,1") response = "&shp_rslt=1" return ( response ) def handle_card_procedure_request(self, data: Dict) -> str: profile = self.data.profile.get_profile(data["aime_id"], self.version) if profile is None: return "&cd_adm_result=0" response = "&cd_adm_result=1" response += "&chg_name_price=100" response += "&accept_idx=100" response += f"&pd_id={profile['user']}" response += f"&player_name={profile['player_name']}" response += f"&lv_num={profile['lv_num']}" response += f"&lv_pnt={profile['lv_pnt']}" response += f"&lv_str={profile['lv_str']}" response += f"&lv_efct_id={profile['lv_efct_id']}" response += f"&lv_plt_id={profile['lv_plt_id']}" response += f"&vcld_pts={profile['vcld_pts']}" response += f"&passwd_stat={profile['passwd_stat']}" return response def handle_change_name_request(self, data: Dict) -> str: profile = self.data.profile.get_profile(data["pd_id"], self.version) # make sure user has enough Vocaloid Points if profile["vcld_pts"] < int(data["chg_name_price"]): return "&cd_adm_result=0" # update the vocaloid points and player name new_vcld_pts = profile["vcld_pts"] - int(data["chg_name_price"]) self.data.profile.update_profile( profile["user"], player_name=data["player_name"], vcld_pts=new_vcld_pts ) response = "&cd_adm_result=1" response += "&accept_idx=100" response += f"&pd_id={profile['user']}" response += f"&player_name={data['player_name']}" return response def handle_change_passwd_request(self, data: Dict) -> str: profile = self.data.profile.get_profile(data["pd_id"], self.version) # TODO: return correct error number instead of 0 if (data["passwd"] != profile["passwd"]): return "&cd_adm_result=0" # set password to true and update the saved password self.data.profile.update_profile( profile["user"], passwd_stat=1, passwd=data["new_passwd"] ) response = "&cd_adm_result=1" response += "&accept_idx=100" response += f"&pd_id={profile['user']}" return response