artemis/titles/diva/handlers/base.py

187 lines
6.1 KiB
Python
Raw Normal View History

2023-02-25 23:50:35 +00:00
from urllib import parse
2023-02-26 04:40:50 +00:00
from datetime import datetime
from typing import Union, Dict
2023-02-25 23:50:35 +00:00
def lazy_http_form_parse(src: Union[str, bytes]) -> Dict[bytes, bytes]:
out = {}
if type(src) == str:
src = src.encode()
for param in src.split(b"&"):
kvp = param.split(b"=")
out[parse.unquote(kvp[0])] = parse.unquote(kvp[1])
return out
2023-03-09 16:50:11 +00:00
2023-02-25 23:50:35 +00:00
class DivaRequestParseException(Exception):
"""
Exception raised when there is a fault in parsing a diva request,
either due to a malformed request, or missing required items
"""
2023-03-09 16:50:11 +00:00
2023-02-25 23:50:35 +00:00
def __init__(self, message: str) -> None:
self.message = message
super().__init__(self.message)
2023-03-09 16:50:11 +00:00
2023-02-26 04:40:50 +00:00
class BaseBinaryRequest:
cmd: str
req_id: str
2023-03-09 16:50:11 +00:00
2023-02-26 04:40:50 +00:00
def __init__(self, raw: bytes) -> None:
self.raw = raw
self.raw_dict = dict(parse.parse_qsl(raw))
2023-03-09 16:50:11 +00:00
if "cmd" not in self.raw_dict:
2023-02-26 04:40:50 +00:00
raise DivaRequestParseException(f"cmd not in request data {self.raw_dict}")
2023-03-09 16:50:11 +00:00
if "req_id" not in self.raw_dict:
raise DivaRequestParseException(
f"req_id not in request data {self.raw_dict}"
)
for k, v in self.raw_dict.items():
2023-02-26 04:40:50 +00:00
setattr(self, k, v)
2023-03-09 16:50:11 +00:00
2023-02-25 23:50:35 +00:00
class BaseRequest:
2023-02-26 04:40:50 +00:00
def __init__(self, raw: Union[str, bytes]) -> None:
self.raw = raw
try:
self.raw_dict: Dict[str, str] = lazy_http_form_parse(raw)
except UnicodeDecodeError as e:
raise DivaRequestParseException(f"Could not decode data {raw} - {e}")
2023-02-26 04:40:50 +00:00
if "cmd" not in self.raw_dict:
2023-03-09 16:50:11 +00:00
raise DivaRequestParseException(f"cmd not in request data {self.raw_dict}")
2023-02-26 04:40:50 +00:00
if "req_id" not in self.raw_dict:
2023-03-09 16:50:11 +00:00
raise DivaRequestParseException(
f"req_id not in request data {self.raw_dict}"
)
if "place_id" not in self.raw_dict:
2023-03-09 16:50:11 +00:00
raise DivaRequestParseException(
f"place_id not in request data {self.raw_dict}"
)
if "start_up_mode" not in self.raw_dict:
2023-03-09 16:50:11 +00:00
raise DivaRequestParseException(
f"start_up_mode not in request data {self.raw_dict}"
)
if "cmm_dly_mod" not in self.raw_dict:
2023-03-09 16:50:11 +00:00
raise DivaRequestParseException(
f"cmm_dly_mod not in request data {self.raw_dict}"
)
if "cmm_dly_sec" not in self.raw_dict:
2023-03-09 16:50:11 +00:00
raise DivaRequestParseException(
f"cmm_dly_sec not in request data {self.raw_dict}"
)
if "cmm_err_mod" not in self.raw_dict:
2023-03-09 16:50:11 +00:00
raise DivaRequestParseException(
f"cmm_err_mod not in request data {self.raw_dict}"
)
if "region_code" not in self.raw_dict:
2023-03-09 16:50:11 +00:00
raise DivaRequestParseException(
f"region_code not in request data {self.raw_dict}"
)
if "time_stamp" not in self.raw_dict:
2023-03-09 16:50:11 +00:00
raise DivaRequestParseException(
f"time_stamp not in request data {self.raw_dict}"
)
2023-02-26 04:40:50 +00:00
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)
2023-03-09 16:50:11 +00:00
self.place_id = int(self.place_id, 16)
2023-02-26 04:40:50 +00:00
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)
self.cmm_err_mod = int(self.cmm_err_mod)
self.region_code = int(self.region_code)
# datetime.now().astimezone().replace(microsecond=0).isoformat()
self.time_stamp = datetime.strptime(self.time_stamp, "%Y-%m-%dT%H:%M:%S%z")
2023-02-25 23:50:35 +00:00
2023-03-09 16:50:11 +00:00
2023-02-25 23:50:35 +00:00
class BaseResponse:
def __init__(self, cmd_id: str, req_id: int) -> None:
self.cmd = cmd_id
self.req_id = req_id
self.stat = "ok"
def make(self) -> str:
ret = ""
for k, v in vars(self).items():
2023-10-05 03:25:10 +00:00
if type(v) == bool:
v = int(v)
ret += f"{k}={v}&"
if ret[-1] == "&":
ret = ret[:-1]
return ret
2023-03-09 16:50:11 +00:00
class GameInitRequest(BaseRequest):
def __init__(self, raw: Union[str, bytes]) -> None:
super().__init__(raw)
2023-03-09 16:50:11 +00:00
class AttendRequest(BaseRequest):
def __init__(self, raw: Union[str, bytes]) -> None:
super().__init__(raw)
self.power_on = int(self.power_on)
self.is_bb = bool(int(self.power_on))
2023-02-25 23:50:35 +00:00
class AttendResponse(BaseResponse):
def __init__(self, cmd_id: str, req_id: int) -> None:
super().__init__(cmd_id, req_id)
self.atnd_prm1 = [1] * 100
self.atnd_prm2 = [1] * 100
self.atnd_prm3 = [1] * 100
self.atnd_lut = datetime.now()
def make(self) -> str:
ret = super().make()
ret_dict = {
"atnd_prm1": ','.join([str(i) for i in self.atnd_prm1]),
"atnd_prm2": ','.join([str(i) for i in self.atnd_prm2]),
"atnd_prm3": ','.join([str(i) for i in self.atnd_prm3]),
"atnd_lut": parse.quote(self.atnd_lut.strftime('%Y-%m-%d %H:%M:%S:16.0'))
}
ret += "&" + parse.urlencode(ret_dict, safe=",")
2023-02-25 23:50:35 +00:00
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