diff --git a/anixart/api.py b/anixart/api.py new file mode 100644 index 0000000..03e939e --- /dev/null +++ b/anixart/api.py @@ -0,0 +1,113 @@ +# -*- coding: utf-8 -*- +import requests + +from .__meta__ import __version__, __build__ +from .auth import AnixartAccount +from .enums import AnixartApiErrors +from .endpoints import API_URL +from .exceptions import AnixartAPIRequestError, AnixartAPIError +from .exceptions import AnixartInitError + + +class AnixartAPI: + + def __init__(self, account: AnixartAccount): + if not isinstance(account, AnixartAccount): + raise AnixartInitError(f'Use class "AnixartAccount" for user. But {type(account)} given.') + + self.__account = account + + self.__token = account.token + + self._session = account.session + self._session.headers = { + 'User-Agent': f'AnixartPyAPI/{__version__}-{__build__} (Linux; Android 15; AnixartPyAPI Build/{__build__})' + } + + @property + def account(self): + return self.__account + + def __parse_response(self, res: requests.Response): + if res.status_code != 200: + e = AnixartAPIRequestError("Bad Request: Invalid request parameters.") + e.message = ( + f"Bad Request: Invalid request parameters." + f"Request: {res.request.method} {res.url}" + f"Status code: {res.status_code}" + f"Response: {res.text}" + f"Client headers: {self._session.headers}" + f"Client payload: {res.request.body}" + ) + e.code = 400 + raise e + try: + response = res.json() + except ValueError: + raise AnixartAPIError("Failed to parse JSON response") + + if response['code'] != 0: + code = response['code'] + if code in AnixartApiErrors: + e = AnixartAPIError(f"AnixartAPI send error. message={AnixartApiErrors[code]}") + e.message = AnixartApiErrors[code] + else: + e = AnixartAPIError(f"AnixartAPI send unknown error. code={response['code']}") + e.code = response['code'] + raise e + + return response + + def post(self, method: str, payload: dict = None, is_json: bool = False, **kwargs): + if payload is None: + payload = {} + url = API_URL + method + if payload.get("token") is None: + if self.__token is not None: + payload.update({"token": self.__token}) + url += "?token=" + self.__token + else: + token = kwargs.get("token") + if token is not None: + payload.update({"token": token}) + url += "?token=" + token + kwargs = {"url": url} + if is_json: + self._session.headers.update({"Content-Type": "application/json; charset=UTF-8"}) + self._session.headers.update({"Content-Length": str(len(str(payload)))}) + kwargs.update({"json": payload}) + else: + kwargs.update({"data": payload}) + res = self._session.post(**kwargs) + self._session.headers["Content-Type"] = "" + self._session.headers["Content-Length"] = "" + return self.__parse_response(res) + + def get(self, method: str, payload: dict = None, **kwargs): + if payload is None: + payload = {} + if payload.get("token") is None: + if self.__token is not None: + payload.update({"token": self.__token}) + else: + token = kwargs.get("token") + if token is not None: + payload.update({"token": token}) + res = self._session.get(API_URL + method, params=payload) + + return self.__parse_response(res) + + def execute(self, http_method, endpoint, **kwargs): + http_method = http_method.upper() + if http_method == "GET": + return self.get(endpoint, **kwargs) + elif http_method == "POST": + return self.post(endpoint, **kwargs) + else: + raise AnixartAPIRequestError("Allow only GET and POST requests.") + + def __str__(self): + return f'AnixartAPI(account={self.__account!r})' + + def __repr__(self): + return f"<{self}>" diff --git a/anixart/api/api.py b/anixart/api/api.py deleted file mode 100644 index 88fbcf7..0000000 --- a/anixart/api/api.py +++ /dev/null @@ -1,78 +0,0 @@ -import logging - -import requests - -from ..auth import AnixartAuth -from ..exceptions import AnixartInitError, AnixartAPIRequestError -from ..request_handler import AnixartRequestsHandler - -_log_name = "file:%-29s -> %s" % ("", "%s") - - -class AnixartUserAccount: - def __init__(self, login, password, config_file="anixart_data.json", **kwargs): - self.kwargs = kwargs - log_level = logging.CRITICAL - log_format = '[%(name)-43s] %(levelname)-5s: %(message)s' - if kwargs.get("loglevel") is not None: - log_level = kwargs.get("loglevel") - if kwargs.get("logformat") is not None: - log_format = kwargs.get("logformat") - logging.basicConfig(level=log_level, format=log_format) - init_log = logging.getLogger("anixart.api.AnixUserAccount") - init_log.debug(_log_name, 23, "__init__ - INIT") - self.login = login - self.password = password - if not isinstance(login, str) or not isinstance(password, str): - raise AnixartInitError("Use normal auth data. In string.") - self.token = None - self.id = None - self.config_file = config_file - self.session = requests.Session() - init_log.debug(_log_name, 32, f"{str(self)}") - init_log.debug(_log_name, 33, "__init__() - OK") - - def __str__(self): - return f'AnixartUserAccount(login="{self.login}", password="{self.password}", ' \ - f'config_file="{self.config_file}", kwargs="{self.kwargs}")' - - def get_login(self): - return self.login - - def get_password(self): - return self.password - - def get_token(self): - return self.token - - def get_id(self): - return self.id - - -class AnixartAPI: - - def __init__(self, user: AnixartUserAccount): - init_log = logging.getLogger("anixart.api.AnixartAPI") - init_log.debug(_log_name, 56, "__init__ - INIT") - if not isinstance(user, AnixartUserAccount): - init_log.critical('Use anixart.api.AnixartUserAccount for user.') - raise AnixartInitError('Use class "AnixartUserAccount" for user.') - self.auth = AnixartAuth(user) - if user.token is None or user.id is None: - init_log.debug(_log_name, 62, "Singing in..") - self.auth.sing_in() - self.user = user - init_log.debug(_log_name, 65, "__init__ - OK.") - self.http_handler = AnixartRequestsHandler(user.token, user.session) - - def __str__(self): - return f'AnixAPI({self.user})' - - def execute(self, http_method, endpoint, **kwargs): - http_method = http_method.upper() - if http_method == "GET": - return self.http_handler.get(endpoint, **kwargs) - elif http_method == "POST": - return self.http_handler.post(endpoint, **kwargs) - else: - raise AnixartAPIRequestError("Allow only GET and POST requests.") diff --git a/anixart/api/api.pyi b/anixart/api/api.pyi deleted file mode 100644 index 4e82064..0000000 --- a/anixart/api/api.pyi +++ /dev/null @@ -1,65 +0,0 @@ -import requests - -from ..auth import AnixartAuth -from ..request_handler import AnixartRequestsHandler - - -class AnixartUserAccount: - def __init__(self, login, password, config_file="anixart_data.json", **kwargs): - """ - Info: - Anixart login class object. - ~~~~~~~~~~~~~~~~~~~~~~~~~~~ - Usage: - >>> user = AnixartUserAccount("login", "password", config_file="anixart_data.json") - >>> print(user.__login) - Availible params: - ~~~~~~~~~~~~~~~~~ - * login -> Your anixart nick - * password -> Your anixart password - * need_reg -> If you need new account, set True - * mail -> Real email for registration. - * config_file -> Patch to anixart login cache - :param login: Your anixart nick - :param password: Anixart password - :param need_reg: If you need new account, set True - :param email: Real email for registration - :param config_file: Patch to anixart login cache - :type login: str - :type password: str - :type need_reg: bool - :type email: str - :type config_file: str - :return: :class:`AnixUserAccount ` object - """ - self.kwargs: dict = kwargs - self.login: str = login - self.password: str = password - self.config_file: str = config_file - self.token: str = None - self.id: int = None - self.session: requests.Session = requests.Session() - - def __str__(self) -> str: ... - def get_login(self) -> str: ... - def get_password(self) -> str: ... - def get_token(self) -> str: ... - def get_id(self) -> int: ... - -class AnixartAPI: - def __init__(self, user: AnixartUserAccount): - """ - Info: - Anixart API class object. - ~~~~~~~~~~~~~~~~~~~~~~~~~ - Usage: - >>> user = AnixartUserAccount("login", "password", config_file="anixart_data.json") - >>> anix = AnixartAPI(user) - :param user: :class:`AnixUserAccount ` object - :return: :class:`AnixAPIRequests ` object - """ - self.auth = AnixartAuth(user) - self.user = user - self.http_handler = AnixartRequestsHandler(user.token, user.session) - def __str__(self) -> str: ... - def execute(self, http_method: str, endpoint: str) -> requests.Request: ... diff --git a/anixart/request_handler.py b/anixart/request_handler.py deleted file mode 100644 index f28bc7a..0000000 --- a/anixart/request_handler.py +++ /dev/null @@ -1,91 +0,0 @@ -# -*- coding: utf-8 -*- - -import logging - -import requests - -from .__meta__ import __version__, __build__ -from .endpoints import API_URL -from .exceptions import AnixartAPIRequestError, AnixartAPIError -_log_name = "file:%-28s -> %s" % ("", "%s") - - -def _parse_res_code(res, payload, http_method, http_headers): - json = res.json() - error = json.get("error") - code = json.get("code") - if res.status_code >= 400: - raise AnixartAPIRequestError(f"\n\nAnixartPyAPIWrapper: ERROR\n" - f"Send this info to author: https://t.me/SantaSpeen\n" - f"URL: {http_method} {res.url}\n" - f"Status code: {res.status_code}\n" - f"Res headers: {res.headers}\n" - f"Req headers: {http_headers}\n" - f"Server res: {json}\n" - f"Client req: {payload}\n") - if error: - raise AnixartAPIRequestError(f"Internal server error: {error}; Payload: {payload}") - if code: - if code == 0: - return - else: - raise AnixartAPIError(f"AnixartAPI send error code: {code}; Json: {json}") - - -class AnixartRequestsHandler: - - def __init__(self, token: str = None, session: requests.Session = None, - _log_class="Anixart.request_handler.AnixartRequestsHandler"): - self.__log = logging.getLogger(_log_class) - self.__log.debug(_log_name, 44, f"__init__ - INIT from {self}") - if session: - self.__session = session - else: - self.__log.debug(_log_name, 48, "Create new session.") - self.__session = requests.Session() - self.__session.headers = { - 'User-Agent': f'AnixartPyAPIWrapper/{__version__}-{__build__}' - f' (Linux; Android 12; SantaSpeen AnixartPyAPIWrapper Build/{__build__})'} - self.__token = token - - def post(self, method: str, payload: dict = None, is_json: bool = False, **kwargs): - if payload is None: - payload = {} - url = API_URL + method - if payload.get("token") is None: - if self.__token is not None: - payload.update({"token": self.__token}) - url += "?token=" + self.__token - else: - token = kwargs.get("token") - if token is not None: - payload.update({"token": token}) - url += "?token=" + token - kwargs = {"url": url} - if is_json: - self.__session.headers.update({"Content-Type": "application/json; charset=UTF-8"}) - self.__session.headers.update({"Content-Length": str(len(str(payload)))}) - kwargs.update({"json": payload}) - else: - kwargs.update({"data": payload}) - self.__log.debug(_log_name, 79, f"{'json' if is_json else ''} POST {method}; {payload}") - res = self.__session.post(**kwargs) - _parse_res_code(res, payload, "POST", self.__session.headers) - self.__session.headers["Content-Type"] = "" - self.__session.headers["Content-Length"] = "" - return res - - def get(self, method: str, payload: dict = None, **kwargs): - if payload is None: - payload = {} - if payload.get("token") is None: - if self.__token is not None: - payload.update({"token": self.__token}) - else: - token = kwargs.get("token") - if token is not None: - payload.update({"token": token}) - self.__log.debug(_log_name, 101, f"GET {method}; {payload}") - res = self.__session.get(API_URL + method, params=payload) - _parse_res_code(res, payload, "GET", self.__session.headers) - return res