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