forked from Hay1tsme/artemis
127 lines
3.7 KiB
Python
127 lines
3.7 KiB
Python
from urllib import parse
|
|
from datetime import datetime
|
|
from typing import Union
|
|
|
|
|
|
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
|
|
"""
|
|
|
|
def __init__(self, message: str) -> None:
|
|
self.message = message
|
|
super().__init__(self.message)
|
|
|
|
|
|
class BaseBinaryRequest:
|
|
cmd: str
|
|
req_id: str
|
|
|
|
def __init__(self, raw: bytes) -> None:
|
|
self.raw = raw
|
|
self.raw_dict = dict(parse.parse_qsl(raw))
|
|
|
|
if "cmd" not in self.raw_dict:
|
|
raise DivaRequestParseException(f"cmd not in request data {self.raw_dict}")
|
|
|
|
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:
|
|
setattr(self, k, v)
|
|
|
|
|
|
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
|
|
self.raw_dict = dict(parse.parse_qsl(raw))
|
|
|
|
if "cmd" not in self.raw_dict:
|
|
raise DivaRequestParseException(f"cmd not in request data {self.raw_dict}")
|
|
|
|
if "req_id" not in self.raw_dict:
|
|
raise DivaRequestParseException(
|
|
f"req_id not in request data {self.raw_dict}"
|
|
)
|
|
|
|
if "place_id" not in self.raw_dict:
|
|
raise DivaRequestParseException(
|
|
f"place_id not in request data {self.raw_dict}"
|
|
)
|
|
|
|
if "start_up_mode" not in self.raw_dict:
|
|
raise DivaRequestParseException(
|
|
f"start_up_mode not in request data {self.raw_dict}"
|
|
)
|
|
|
|
if "cmm_dly_mod" not in self.raw_dict:
|
|
raise DivaRequestParseException(
|
|
f"cmm_dly_mod not in request data {self.raw_dict}"
|
|
)
|
|
|
|
if "cmm_dly_sec" not in self.raw_dict:
|
|
raise DivaRequestParseException(
|
|
f"cmm_dly_sec not in request data {self.raw_dict}"
|
|
)
|
|
|
|
if "cmm_err_mod" not in self.raw_dict:
|
|
raise DivaRequestParseException(
|
|
f"cmm_err_mod not in request data {self.raw_dict}"
|
|
)
|
|
|
|
if "region_code" not in self.raw_dict:
|
|
raise DivaRequestParseException(
|
|
f"region_code not in request data {self.raw_dict}"
|
|
)
|
|
|
|
if "time_stamp" not in self.raw_dict:
|
|
raise DivaRequestParseException(
|
|
f"time_stamp not in request data {self.raw_dict}"
|
|
)
|
|
|
|
for k, v in self.raw_dict:
|
|
setattr(self, k, v)
|
|
|
|
self.place_id = int(self.place_id)
|
|
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")
|
|
|
|
|
|
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:
|
|
all_vars = vars(self)
|
|
all_vars.pop("cmd")
|
|
all_vars.pop("req_id")
|
|
all_vars.pop("stat")
|
|
|
|
ret = f"cmd={self.cmd}&req_id={self.req_id}&stat={self.stat}"
|
|
additional_params = parse.urlencode(
|
|
all_vars, doseq=True, quote_via=parse.quote, safe=","
|
|
)
|
|
|
|
if additional_params != "":
|
|
ret += f"&{additional_params}"
|
|
|
|
return ret
|