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

diva: fix start, spend_credit, and get_pv_pd requests

Browse Source
This commit is contained in:
Hay1tsme 2023-10-04 23:58:26 -04:00
parent f836b5dd21
commit 5e03193819
5 changed files with 101 additions and 98 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

106
titles/diva/base.py
View File

@ -431,6 +431,15 @@ class DivaBase:
return
resp = StartResponse(req.cmd, req.req_id, req.pd_id, profile['player_name'])
profile_dict = profile._asdict()
profile_dict.pop("id")
profile_dict.pop("user")
profile_dict.pop("version")
for k, v in profile_dict.items():
if hasattr(resp, k):
setattr(resp, k, v)
# generate the mdl_have string if "unlock_all_modules" is disabled
if not self.game_config.mods.unlock_all_modules:
@ -443,65 +452,12 @@ class DivaBase:
resp.cstmz_itm_have = self.data.customize.get_customize_items_have_string(
req.pd_id, self.version
)
"""
resp.hp_vol = profile['hp_vol']
resp.btn_se_vol = profile['btn_se_vol']
resp.btn_se_vol2 = profile['btn_se_vol2']
resp.sldr_se_vol2 = profile['sldr_se_vol2']
resp.sort_kind = profile['sort_kind']
resp.lv_num = profile['lv_num']
resp.lv_pnt = profile['lv_pnt']
resp.lv_efct_id = profile['lv_efct_id']
resp.lv_plt_id = profile['lv_plt_id']
resp.use_pv_mdl_eqp = int(profile['use_pv_mdl_eqp'])
resp.use_mdl_pri = int(profile['use_mdl_pri'])
resp.use_pv_skn_eqp = int(profile['use_pv_skn_eqp'])
resp.use_pv_btn_se_eqp = int(profile['use_pv_btn_se_eqp'])
resp.use_pv_sld_se_eqp = int(profile['use_pv_sld_se_eqp'])
resp.use_pv_chn_sld_se_eqp = int(profile['use_pv_chn_sld_se_eqp'])
resp.use_pv_sldr_tch_se_eqp = int(profile['use_pv_sldr_tch_se_eqp'])
resp.vcld_pts = profile['lv_efct_id']
resp.nxt_pv_id = profile['nxt_pv_id']
resp.nxt_dffclty = profile['nxt_dffclty']
resp.nxt_edtn = profile['nxt_edtn']
resp.dsp_clr_brdr = profile['dsp_clr_brdr']
resp.dsp_intrm_rnk = profile['dsp_intrm_rnk']
resp.dsp_clr_sts = profile['dsp_clr_sts']
resp.rgo_sts = profile['rgo_sts']
"""
profile_dict = profile._asdict()
profile_dict.pop("id")
profile_dict.pop("user")
profile_dict.pop("version")
for k, v in profile_dict.items():
if hasattr(resp, k):
setattr(resp, k, v)
response = "&ps_result=1"
# Contest progress
response += f"&cv_cid=-1,-1,-1,-1"
response += f"&cv_sc=-1,-1,-1,-1"
response += f"&cv_bv=-1,-1,-1,-1"
response += f"&cv_bv=-1,-1,-1,-1"
response += f"&cv_bf=-1,-1,-1,-1"
# Contest now playing id, return -1 if no current playing contest
response += f"&cnp_cid={profile['cnp_cid']}"
response += f"&cnp_val={profile['cnp_val']}"
# border can be 0=bronzem 1=silver, 2=gold
response += f"&cnp_rr={profile['cnp_rr']}"
# only show contest specifier if it is not empty
response += f"&cnp_sp={profile['cnp_sp']}" if profile["cnp_sp"] != "" else ""
# To be fully fixed
if "my_qst_id" in profile:
resp.my_qst_id = profile['my_qst_id']
resp.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"
# define a helper class to store all counts for clear, great,
# excellent and perfect
class ClearSet:
@ -557,12 +513,8 @@ class DivaBase:
clear_status = ",".join(map(str, clear_list))
response += f"&clr_sts={clear_status}"
resp.clr_sts = clear_status
# Store stuff to add to rework
response += f"&mdl_eqp_tm={self.time_lut}"
# get the common_modules, customize_items and customize_item_flags
# from the profile shop
if profile_shop:
@ -576,20 +528,18 @@ class DivaBase:
pass
def handle_spend_credit_request(self, data: bytes) -> str:
profile = self.data.profile.get_profile(data["pd_id"], self.version)
req = SpendCreditRequest(data)
profile = self.data.profile.get_profile(req.pd_id, self.version)
if profile is None:
return
response = ""
resp = SpendCreditResponse(req.cmd, req.req_id)
resp.vcld_pts = profile['vcld_pts']
resp.lv_str = profile['lv_str']
resp.lv_efct_id = profile['lv_efct_id']
resp.lv_plt_id = profile['lv_plt_id']
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
return resp.make()
def _get_pv_pd_result(
self,
@ -653,31 +603,31 @@ class DivaBase:
return pv_result
def task_generateScoreData(self, data: Dict, pd_by_pv_id, song):
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(
data["pd_id"], int(song), data["difficulty"], edition=0
pd_id, int(song), difficulty, edition=0
)
pd_db_song_1 = self.data.score.get_best_user_score(
data["pd_id"], int(song), data["difficulty"], edition=1
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(
data["pd_id"], int(song), data["difficulty"], edition=0
pd_id, int(song), 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_id, int(song), difficulty, edition=1
)
pd_db_customize = self.data.pv_customize.get_pv_customize(
data["pd_id"], int(song)
pd_id, int(song)
)
# generate the pv_result string with the ORIGINAL edition and the EXTRA edition appended
@ -695,14 +645,14 @@ class DivaBase:
pd_by_pv_id.append(",")
def handle_get_pv_pd_request(self, data: Dict) -> Dict:
song_id = data["pd_pv_id_lst"].split(",")
req = GetPvPdRequest(data)
pv = ""
threads = []
pd_by_pv_id = []
for song in song_id:
thread_ScoreData = Thread(target=self.task_generateScoreData(data, pd_by_pv_id, song))
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))
threads.append(thread_ScoreData)
for x in threads:
@ -714,12 +664,14 @@ class DivaBase:
for x in pd_by_pv_id:
pv += x
resp = GetPvPdResponse(req.cmd, req.req_id)
resp.pd_by_pv_id = pv[:-1]
response = ""
response += f"&pd_by_pv_id={pv[:-1]}"
response += "&pdddt_flg=0"
response += f"&pdddt_tm={self.time_lut}"
return response
return resp.make()
def handle_stage_start_request(self, data: bytes) -> str:
pass

5
titles/diva/handlers/__init__.py
View File

@ -1,2 +1,3 @@
from titles.diva.handlers.base import *
from titles.diva.handlers.user import *
from .base import *
from .user import *
from .pv import *

41
titles/diva/handlers/base.py
View File

@ -45,14 +45,6 @@ class BaseBinaryRequest:
class BaseRequest:
cmd: str
req_id: str
game_id: str
r_rev: str
kc_serial: str
b_serial: str
country_code: str
def __init__(self, raw: Union[str, bytes]) -> None:
self.raw = raw
try:
@ -103,10 +95,18 @@ class BaseRequest:
f"time_stamp not in request data {self.raw_dict}"
)
self.cmd: str
self.req_id: str
self.game_id: str
self.r_rev: str
self.kc_serial: str
self.b_serial: str
self.country_code: str
for k, v in self.raw_dict.items():
setattr(self, k, v)
self.place_id = int(self.place_id)
self.place_id = int(self.place_id, 16)
self.start_up_mode = int(self.start_up_mode)
self.cmm_dly_mod = int(self.cmm_dly_mod)
self.cmm_dly_sec = int(self.cmm_dly_sec)
@ -162,3 +162,26 @@ class AttendResponse(BaseResponse):
ret += "&" + parse.urlencode(ret_dict, safe=",")
return ret
class SpendCreditRequest(BaseRequest):
def __init__(self, raw: str | bytes) -> None:
super().__init__(raw)
try:
self.pd_id = int(self.pd_id)
self.my_qst_id = [int(x) for x in self.my_qst_id.split(",")]
self.my_qst_sts = [int(x) for x in self.my_qst_sts.split(",")]
self.crdt_typ = int(self.crdt_typ)
self.cmpgn_id = [int(x) for x in self.cmpgn_id.split(",")]
self.cmpgn_pb = [int(x) for x in self.cmpgn_pb.split(",")]
except AttributeError as e:
raise DivaRequestParseException(f"StartRequest: {e}")
class SpendCreditResponse(BaseResponse):
def __init__(self, cmd_id: str, req_id: int) -> None:
super().__init__(cmd_id, req_id)
self.cmpgn_rslt = ",".join(["-1,-1,x,-1,-1,x,x,-1,x"] * 6)
self.cmpgn_rslt_num = 0
self.vcld_pts = 0
self.lv_str = ""
self.lv_efct_id = 0
self.lv_plt_id = 0

29
titles/diva/handlers/pv.py Normal file
View File

@ -0,0 +1,29 @@
from typing import Union, List
from titles.diva.handlers.base import (
BaseRequest,
BaseResponse,
DivaRequestParseException,
)
from datetime import datetime
from urllib import parse
from ..const import DivaConstants
class GetPvPdRequest(BaseRequest):
def __init__(self, raw: str | bytes) -> None:
super().__init__(raw)
try:
self.pd_id = int(self.pd_id)
self.accept_idx = int(self.accept_idx)
self.start_idx = int(self.start_idx)
self.difficulty = int(self.difficulty)
self.pd_pv_id_lst: List[int] = [int(x) for x in self.pd_pv_id_lst.split(',')]
except AttributeError as e:
raise DivaRequestParseException(f"GetPvPdRequest: {e}")
class GetPvPdResponse(BaseResponse):
def __init__(self, cmd_id: str, req_id: int) -> None:
super().__init__(cmd_id, req_id)
self.pd_by_pv_id = ""
self.pdddt_flg = 0
self.pdddt_tm = parse.quote(datetime.now().strftime(DivaConstants.LUT_TIME_FMT))

18
titles/diva/handlers/user.py
View File

@ -8,15 +8,15 @@ from urllib import parse
from ..const import DivaConstants
class PreStartRequest(BaseRequest):
pmm: str
idm: str
mmgameid: str
mmuid: str
a_code: str
aime_id: str
aime_a_code: str
def __init__(self, raw: str) -> None:
self.pmm: str
self.idm: str
self.mmgameid: str
self.mmuid: str
self.a_code: str
self.aime_id: str
self.aime_a_code: str
super().__init__(raw)
try:
self.key_obj_type = int(self.key_obj_type)
@ -27,8 +27,6 @@ class PreStartRequest(BaseRequest):
class PreStartResponse(BaseResponse):
def __init__(self, cmd_id: str, req_id: int, pd_id: int) -> None:
super().__init__(cmd_id, req_id)
self.ps_result = 1