mirror of
https://github.com/SantaSpeen/anixart.git
synced 2026-04-20 23:30:35 +00:00
[>] {api/, request_handler} > api.py
This commit is contained in:
113
anixart/api.py
Normal file
113
anixart/api.py
Normal file
@@ -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}>"
|
||||
@@ -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" % ("<anixart.api:%-4i>", "%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.")
|
||||
@@ -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 <anixart.api.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 <anixart.api.AnixUserAccount>` object
|
||||
:return: :class:`AnixAPIRequests <anixart.api.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: ...
|
||||
@@ -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" % ("<Anixart.request_handler:%-3i>", "%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
|
||||
Reference in New Issue
Block a user