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

Merge branch 'develop' into diva_handler_classes

Browse Source
This commit is contained in:
Hay1tsme
2024-01-12 12:15:28 -05:00
parent 6b158bf18d 9a7d5bc689
commit c6f2f7e0ef
195 changed files with 6255 additions and 5300 deletions
Download Patch File Download Diff File Expand all files Collapse all files

208
titles/diva/base.py
View File

@ -1,8 +1,7 @@
from datetime import datetime
from typing import Any, List, Dict
import datetime
from typing import Dict
import logging
import json
import urllib
import urllib.parse
from threading import Thread
from core.config import CoreConfig
@ -25,14 +24,13 @@ class DivaBase:
dt = datetime.now()
self.time_lut = parse.quote(dt.strftime("%Y-%m-%d %H:%M:%S:16.0"))
def handle_test_request(self, data: bytes) -> str:
pass
async def handle_test_request(self, data: Dict) -> Dict:
return ""
def handle_game_init_request(self, data: bytes) -> str:
req = GameInitRequest(data)
return None
async def handle_game_init_request(self, data: Dict) -> Dict:
return f""
def handle_attend_request(self, data: bytes) -> str:
async def handle_attend_request(self, data: bytes) -> str:
req = AttendRequest(data)
resp = AttendResponse(req.req_id)
@ -54,7 +52,7 @@ class DivaBase:
return resp.make()
def handle_ping_request(self, data: bytes) -> str:
async def handle_ping_request(self, data: bytes) -> str:
encoded = "&"
params = {
"ping_b_msg": f"Welcome to {self.core_cfg.server.name} network!" if not self.game_config.server.banner_msg else self.game_config.server.banner_msg,
@ -98,7 +96,7 @@ class DivaBase:
return encoded
def handle_pv_list_request(self, data: bytes) -> str:
async 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()
@ -135,10 +133,10 @@ class DivaBase:
return response
def handle_shop_catalog_request(self, data: bytes) -> str:
async def handle_shop_catalog_request(self, data: Dict) -> Dict:
catalog = ""
shopList = self.data.static.get_enabled_shops(self.version)
shopList = await 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()
@ -173,9 +171,9 @@ class DivaBase:
return response
def handle_buy_module_request(self, data: bytes) -> str:
profile = self.data.profile.get_profile(data["pd_id"], self.version)
module = self.data.static.get_enabled_shop(self.version, int(data["mdl_id"]))
async def handle_buy_module_request(self, data: Dict) -> Dict:
profile = await self.data.profile.get_profile(data["pd_id"], self.version)
module = await self.data.static.get_enabled_shop(self.version, int(data["mdl_id"]))
# make sure module is available to purchase
if not module:
@ -187,11 +185,11 @@ class DivaBase:
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"])
await self.data.profile.update_profile(profile["user"], vcld_pts=new_vcld_pts)
await 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)
mdl_have = await self.data.module.get_modules_have_string(data["pd_id"], self.version)
response = "&shp_rslt=1"
response += f"&mdl_id={data['mdl_id']}"
@ -200,10 +198,10 @@ class DivaBase:
return response
def handle_cstmz_itm_ctlg_request(self, data: bytes) -> str:
async def handle_cstmz_itm_ctlg_request(self, data: Dict) -> Dict:
catalog = ""
itemList = self.data.static.get_enabled_items(self.version)
itemList = await 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()
@ -238,9 +236,9 @@ class DivaBase:
return response
def handle_buy_cstmz_itm_request(self, data: bytes) -> str:
profile = self.data.profile.get_profile(data["pd_id"], self.version)
item = self.data.static.get_enabled_item(
async def handle_buy_cstmz_itm_request(self, data: Dict) -> Dict:
profile = await self.data.profile.get_profile(data["pd_id"], self.version)
item = await self.data.static.get_enabled_item(
self.version, int(data["cstmz_itm_id"])
)
@ -255,14 +253,14 @@ class DivaBase:
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)
await self.data.profile.update_profile(profile["user"], vcld_pts=new_vcld_pts)
self.data.customize.put_customize_item(
await 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(
cstmz_itm_have = await self.data.customize.get_customize_items_have_string(
data["pd_id"], self.version
)
@ -273,7 +271,7 @@ class DivaBase:
return response
def handle_festa_info_request(self, data: bytes) -> str:
async def handle_festa_info_request(self, data: Dict) -> Dict:
encoded = "&"
params = {
"fi_id": "1,2",
@ -296,7 +294,7 @@ class DivaBase:
return encoded
def handle_contest_info_request(self, data: bytes) -> str:
async def handle_contest_info_request(self, data: Dict) -> Dict:
response = ""
response += f"&ci_lut={self.time_lut}"
@ -304,10 +302,10 @@ class DivaBase:
return response
def handle_qst_inf_request(self, data: bytes) -> str:
async def handle_qst_inf_request(self, data: Dict) -> Dict:
quest = ""
questList = self.data.static.get_enabled_quests(self.version)
questList = await 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()
@ -354,47 +352,47 @@ class DivaBase:
return response
def handle_nv_ranking_request(self, data: bytes) -> str:
pass
async def handle_nv_ranking_request(self, data: Dict) -> Dict:
return f""
def handle_ps_ranking_request(self, data: bytes) -> str:
pass
async def handle_ps_ranking_request(self, data: Dict) -> Dict:
return f""
def handle_ng_word_request(self, data: bytes) -> str:
pass
async def handle_ng_word_request(self, data: Dict) -> Dict:
return f""
def handle_rmt_wp_list_request(self, data: bytes) -> str:
pass
async def handle_rmt_wp_list_request(self, data: Dict) -> Dict:
return f""
def handle_pv_def_chr_list_request(self, data: bytes) -> str:
pass
async def handle_pv_def_chr_list_request(self, data: Dict) -> Dict:
return f""
def handle_pv_ng_mdl_list_request(self, data: bytes) -> str:
pass
async def handle_pv_ng_mdl_list_request(self, data: Dict) -> Dict:
return f""
def handle_cstmz_itm_ng_mdl_lst_request(self, data: bytes) -> str:
pass
async def handle_cstmz_itm_ng_mdl_lst_request(self, data: Dict) -> Dict:
return f""
def handle_banner_info_request(self, data: bytes) -> str:
pass
async def handle_banner_info_request(self, data: Dict) -> Dict:
return f""
def handle_banner_data_request(self, data: bytes) -> str:
pass
async def handle_banner_data_request(self, data: Dict) -> Dict:
return f""
def handle_cm_ply_info_request(self, data: bytes) -> str:
pass
async def handle_cm_ply_info_request(self, data: Dict) -> Dict:
return f""
def handle_pstd_h_ctrl_request(self, data: bytes) -> str:
pass
async def handle_pstd_h_ctrl_request(self, data: Dict) -> Dict:
return f""
def handle_pstd_item_ng_lst_request(self, data: bytes) -> str:
pass
async def handle_pstd_item_ng_lst_request(self, data: Dict) -> Dict:
return f""
def handle_pre_start_request(self, data: bytes) -> str:
async def handle_pre_start_request(self, data: bytes) -> str:
req = PreStartRequest(data)
resp = PreStartResponse(req.req_id, req.aime_id)
profile = self.data.profile.get_profile(req.aime_id, self.version)
profile_shop = self.data.item.get_shop(req.aime_id, self.version)
profile = await self.data.profile.get_profile(req.aime_id, self.version)
profile_shop = await self.data.item.get_shop(req.aime_id, self.version)
if profile is None:
return f"&ps_result=-3"
@ -413,19 +411,19 @@ class DivaBase:
return resp.make()
def handle_registration_request(self, data: bytes) -> str:
async def handle_registration_request(self, data: bytes) -> str:
req = RegisterRequest(data)
pd_id = self.data.profile.create_profile(
pd_id = await self.data.profile.create_profile(
self.version, req.aime_id, req.player_name
)
if pd_id is None:
return "&cd_adm_result=-1"
return RegisterResponse(req.req_id, req.aime_id).make()
def handle_start_request(self, data: bytes) -> str:
async def handle_start_request(self, data: bytes) -> str:
req = StartRequest(data)
profile = self.data.profile.get_profile(req.pd_id, self.version)
profile_shop = self.data.item.get_shop(req.pd_id, self.version)
profile = await self.data.profile.get_profile(req.pd_id, self.version)
profile_shop = await self.data.item.get_shop(req.pd_id, self.version)
if profile is None:
return
@ -442,13 +440,13 @@ class DivaBase:
# generate the mdl_have string if "unlock_all_modules" is disabled
if not self.game_config.mods.unlock_all_modules:
resp.mdl_have = self.data.module.get_modules_have_string(
resp.mdl_have = await self.data.module.get_modules_have_string(
req.pd_id, self.version
)
# generate the cstmz_itm_have string if "unlock_all_items" is disabled
if not self.game_config.mods.unlock_all_items:
resp.cstmz_itm_have = self.data.customize.get_customize_items_have_string(
resp.cstmz_itm_have = await self.data.customize.get_customize_items_have_string(
req.pd_id, self.version
)
@ -476,7 +474,7 @@ class DivaBase:
}
# get clear status from user scores
pv_records = self.data.score.get_best_scores(req.pd_id)
pv_records = await self.data.score.get_best_scores(req.pd_id)
clear_status = "0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0"
if pv_records is not None:
@ -523,12 +521,12 @@ class DivaBase:
return resp.make()
def handle_pd_unlock_request(self, data: bytes) -> str:
async def handle_pd_unlock_request(self, data: bytes) -> str:
pass
def handle_spend_credit_request(self, data: bytes) -> str:
async def handle_spend_credit_request(self, data: bytes) -> str:
req = SpendCreditRequest(data)
profile = self.data.profile.get_profile(req.pd_id, self.version)
profile = await self.data.profile.get_profile(req.pd_id, self.version)
if profile is None:
return
@ -602,30 +600,30 @@ class DivaBase:
return pv_result
def task_generateScoreData(self, pd_id: int, difficulty: int, pd_by_pv_id: str, song: int):
async def task_generateScoreData(self, pd_id: int, difficulty: int, pd_by_pv_id: str, song: int):
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(
pd_db_song_0 = await self.data.score.get_best_user_score(
pd_id, int(song), difficulty, edition=0
)
pd_db_song_1 = self.data.score.get_best_user_score(
pd_db_song_1 = await self.data.score.get_best_user_score(
pd_id, int(song), 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(
pd_db_ranking_0 = await self.data.score.get_global_ranking(
pd_id, int(song), difficulty, edition=0
)
if pd_db_song_1:
pd_db_ranking_1 = self.data.score.get_global_ranking(
pd_db_ranking_1 = await self.data.score.get_global_ranking(
pd_id, int(song), difficulty, edition=1
)
pd_db_customize = self.data.pv_customize.get_pv_customize(
pd_db_customize = await self.data.pv_customize.get_pv_customize(
pd_id, int(song)
)
@ -643,7 +641,7 @@ class DivaBase:
pd_by_pv_id.append(urllib.parse.quote(f"{song}***"))
pd_by_pv_id.append(",")
def handle_get_pv_pd_request(self, data: Dict) -> Dict:
async def handle_get_pv_pd_request(self, data: Dict) -> Dict:
req = GetPvPdRequest(data)
pv = ""
@ -651,7 +649,7 @@ class DivaBase:
pd_by_pv_id = []
for song in req.pd_pv_id_lst:
thread_ScoreData = Thread(target=self.task_generateScoreData(req.pd_id, req.difficulty, pd_by_pv_id, song))
thread_ScoreData = Thread(target=await self.task_generateScoreData(req.pd_id, req.difficulty, pd_by_pv_id, song))
threads.append(thread_ScoreData)
for x in threads:
@ -672,13 +670,13 @@ class DivaBase:
return resp.make()
def handle_stage_start_request(self, data: bytes) -> str:
pass
async def handle_stage_start_request(self, data: Dict) -> Dict:
return f""
def handle_stage_result_request(self, data: bytes) -> str:
async def handle_stage_result_request(self, data: bytes) -> str:
req = StageResultRequest(data)
resp = StageResultResponse(req.cmd, req.req_id)
profile = self.data.profile.get_profile(req.pd_id, self.version)
profile = await self.data.profile.get_profile(req.pd_id, self.version)
pd_song_list = req.stg_ply_pv_id
pd_song_difficulty = req.stg_difficulty
@ -696,14 +694,14 @@ class DivaBase:
for index, value in enumerate(pd_song_list):
if pd_song_list[index] > 0:
profile_pd_db_song = self.data.score.get_best_user_score(
profile_pd_db_song = await self.data.score.get_best_user_score(
req.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(
await self.data.score.put_best_score(
req.pd_id,
self.version,
pd_song_list[index],
@ -720,7 +718,7 @@ class DivaBase:
pd_song_worst_cnt[index],
pd_song_max_combo[index],
)
self.data.score.put_playlog(
await self.data.score.put_playlog(
req.pd_id,
self.version,
pd_song_list[index],
@ -738,7 +736,7 @@ class DivaBase:
pd_song_max_combo[index],
)
elif int(pd_song_max_score[index]) >= int(profile_pd_db_song["score"]):
self.data.score.put_best_score(
await self.data.score.put_best_score(
req.pd_id,
self.version,
pd_song_list[index],
@ -755,7 +753,7 @@ class DivaBase:
pd_song_worst_cnt[index],
pd_song_max_combo[index],
)
self.data.score.put_playlog(
await self.data.score.put_playlog(
req.pd_id,
self.version,
pd_song_list[index],
@ -773,7 +771,7 @@ class DivaBase:
pd_song_max_combo[index],
)
elif int(pd_song_max_score[index]) != int(profile_pd_db_song["score"]):
self.data.score.put_playlog(
await self.data.score.put_playlog(
req.pd_id,
self.version,
pd_song_list[index],
@ -794,7 +792,7 @@ class DivaBase:
# Profile saving based on registration list
# Calculate new level
best_scores = self.data.score.get_best_scores(req.pd_id)
best_scores = await self.data.score.get_best_scores(data["pd_id"])
total_atn_pnt = 0
for best_score in best_scores:
@ -822,7 +820,7 @@ class DivaBase:
response += f"&lv_pnt_old={int(profile['lv_pnt'])}"
# update the profile and commit changes to the db
self.data.profile.update_profile(
await self.data.profile.update_profile(
profile["user"],
lv_num=new_level,
lv_pnt=new_level_pnt,
@ -870,17 +868,17 @@ class DivaBase:
return resp.make()
def handle_end_request(self, data: bytes) -> str:
async def handle_end_request(self, data: bytes) -> str:
req = EndRequest(data)
profile = self.data.profile.get_profile(req.pd_id, self.version)
profile = await self.data.profile.get_profile(req.pd_id, self.version)
self.data.profile.update_profile(
await self.data.profile.update_profile(
profile["user"], my_qst_id=req.my_qst_id, my_qst_sts=req.my_qst_sts
)
return None
def handle_shop_exit_request(self, data: bytes) -> str:
self.data.item.put_shop(
async def handle_shop_exit_request(self, data: bytes) -> str:
await self.data.item.put_shop(
data["pd_id"],
self.version,
data["mdl_eqp_cmn_ary"],
@ -888,7 +886,7 @@ class DivaBase:
data["ms_itm_flg_cmn_ary"],
)
if int(data["use_pv_mdl_eqp"]) == 1:
self.data.pv_customize.put_pv_customize(
await self.data.pv_customize.put_pv_customize(
data["pd_id"],
self.version,
data["ply_pv_id"],
@ -897,7 +895,7 @@ class DivaBase:
data["ms_itm_flg_pv_ary"],
)
else:
self.data.pv_customize.put_pv_customize(
await self.data.pv_customize.put_pv_customize(
data["pd_id"],
self.version,
data["ply_pv_id"],
@ -909,8 +907,8 @@ class DivaBase:
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)
async def handle_card_procedure_request(self, data: Dict) -> str:
profile = await self.data.profile.get_profile(data["aime_id"], self.version)
if profile is None:
return "&cd_adm_result=0"
@ -929,8 +927,8 @@ class DivaBase:
return response
def handle_change_name_request(self, data: Dict) -> str:
profile = self.data.profile.get_profile(data["pd_id"], self.version)
async def handle_change_name_request(self, data: Dict) -> str:
profile = await 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"]):
@ -938,7 +936,7 @@ class DivaBase:
# update the vocaloid points and player name
new_vcld_pts = profile["vcld_pts"] - int(data["chg_name_price"])
self.data.profile.update_profile(
await self.data.profile.update_profile(
profile["user"], player_name=data["player_name"], vcld_pts=new_vcld_pts
)
@ -949,15 +947,15 @@ class DivaBase:
return response
def handle_change_passwd_request(self, data: Dict) -> str:
profile = self.data.profile.get_profile(data["pd_id"], self.version)
async def handle_change_passwd_request(self, data: Dict) -> str:
profile = await 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(
await self.data.profile.update_profile(
profile["user"], passwd_stat=1, passwd=data["new_passwd"]
)