mirror of
https://github.com/SantaSpeen/anixart.git
synced 2026-05-19 16:10:29 +00:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 0682bdb42b | |||
| 63bf159679 | |||
| 8a944319d6 |
+23
-37
@@ -4,19 +4,17 @@ import requests
|
|||||||
from .__meta__ import __version__, __build__
|
from .__meta__ import __version__, __build__
|
||||||
from .auth import AnixartAccount, AnixartAccountGuest
|
from .auth import AnixartAccount, AnixartAccountGuest
|
||||||
from .enums import AnixartApiErrors
|
from .enums import AnixartApiErrors
|
||||||
from .endpoints import API_URL
|
|
||||||
from .exceptions import AnixartAPIRequestError, AnixartAPIError
|
from .exceptions import AnixartAPIRequestError, AnixartAPIError
|
||||||
from .exceptions import AnixartInitError
|
from .exceptions import AnixartInitError
|
||||||
|
|
||||||
|
debug = True
|
||||||
|
|
||||||
class AnixartAPI:
|
class AnixartAPI:
|
||||||
|
API_URL = "https://api.anixart.tv/"
|
||||||
|
|
||||||
def __init__(self, account: AnixartAccount = None):
|
def __init__(self, account: AnixartAccount = None):
|
||||||
if account is None:
|
if account is None:
|
||||||
account = AnixartAccountGuest()
|
account = AnixartAccountGuest()
|
||||||
if not isinstance(account, AnixartAccount):
|
|
||||||
raise AnixartInitError(f'Use class "AnixartAccount" for user. But {type(account)} given.')
|
|
||||||
|
|
||||||
self.use_account(account)
|
self.use_account(account)
|
||||||
|
|
||||||
def use_account(self, account: AnixartAccount):
|
def use_account(self, account: AnixartAccount):
|
||||||
@@ -37,6 +35,9 @@ class AnixartAPI:
|
|||||||
return self.__account
|
return self.__account
|
||||||
|
|
||||||
def __parse_response(self, res: requests.Response):
|
def __parse_response(self, res: requests.Response):
|
||||||
|
if debug:
|
||||||
|
print(f"[D] -> {res.request.method} body='{res.request.body!s}' url='{res.url!s}'")
|
||||||
|
print(f"[D] <- {res.status_code}, {len(res.text)=}")
|
||||||
if res.status_code != 200:
|
if res.status_code != 200:
|
||||||
e = AnixartAPIRequestError("Bad Request: Invalid request parameters.")
|
e = AnixartAPIRequestError("Bad Request: Invalid request parameters.")
|
||||||
e.message = (
|
e.message = (
|
||||||
@@ -56,7 +57,8 @@ class AnixartAPI:
|
|||||||
except ValueError as e:
|
except ValueError as e:
|
||||||
raise AnixartAPIError("Failed to parse JSON response")
|
raise AnixartAPIError("Failed to parse JSON response")
|
||||||
|
|
||||||
# print(response)
|
if debug:
|
||||||
|
print(response)
|
||||||
if response['code'] != 0:
|
if response['code'] != 0:
|
||||||
code = response['code']
|
code = response['code']
|
||||||
if code in AnixartApiErrors:
|
if code in AnixartApiErrors:
|
||||||
@@ -69,42 +71,26 @@ class AnixartAPI:
|
|||||||
|
|
||||||
return response
|
return response
|
||||||
|
|
||||||
def _post(self, method: str, payload: dict = None, is_json: bool = False, **kwargs):
|
def _post(self, method: str, _json: bool = False, **kwargs):
|
||||||
if payload is None:
|
url = self.API_URL + method
|
||||||
payload = {}
|
kwargs["token"] = kwargs.get("token", self.__token)
|
||||||
url = API_URL + method
|
if kwargs["token"]:
|
||||||
if payload.get("token") is None:
|
url += f"?token={self.__token}"
|
||||||
if self.__token is not None:
|
|
||||||
payload.update({"token": self.__token})
|
req_settings = {"url": url}
|
||||||
url += "?token=" + self.__token
|
if _json:
|
||||||
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["Content-Type"] = "application/json; charset=UTF-8"
|
self._session.headers["Content-Type"] = "application/json; charset=UTF-8"
|
||||||
self._session.headers["Content-Length"] = str(len(payload))
|
req_settings.update({"json": kwargs})
|
||||||
kwargs.update({"json": payload})
|
|
||||||
else:
|
else:
|
||||||
kwargs.update({"data": payload})
|
req_settings.update({"data": kwargs})
|
||||||
res = self._session.post(**kwargs)
|
res = self._session.post(**req_settings)
|
||||||
self._session.headers["Content-Type"] = ""
|
|
||||||
self._session.headers["Content-Length"] = ""
|
|
||||||
return self.__parse_response(res)
|
return self.__parse_response(res)
|
||||||
|
|
||||||
def _get(self, method: str, payload: dict = None, **kwargs):
|
def _get(self, method: str, **kwargs):
|
||||||
if payload is None:
|
if self.__token:
|
||||||
payload = {}
|
kwargs["token"] = kwargs.get("token", self.__token)
|
||||||
if payload.get("token") is None:
|
|
||||||
if self.__token is not None:
|
res = self._session.get(self.API_URL + method, params=kwargs)
|
||||||
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)
|
return self.__parse_response(res)
|
||||||
|
|
||||||
def execute(self, http_method, endpoint, **kwargs):
|
def execute(self, http_method, endpoint, **kwargs):
|
||||||
|
|||||||
@@ -55,8 +55,7 @@ class AnixartAccount:
|
|||||||
|
|
||||||
def login(self):
|
def login(self):
|
||||||
"""Login to Anixart and return the token."""
|
"""Login to Anixart and return the token."""
|
||||||
payload = {"login": self.username, "password": self._password}
|
res = self._api.post(endpoints.SING_IN, login=self.username, password=self._password)
|
||||||
res = self._api.post(endpoints.SING_IN, payload)
|
|
||||||
uid = res["profile"]["id"]
|
uid = res["profile"]["id"]
|
||||||
token = res["profileToken"]["token"]
|
token = res["profileToken"]["token"]
|
||||||
self._id = uid
|
self._id = uid
|
||||||
|
|||||||
+79
-1
@@ -1,7 +1,85 @@
|
|||||||
# -*- coding: utf-8 -*-
|
# -*- coding: utf-8 -*-
|
||||||
|
from dataclasses import dataclass, field
|
||||||
|
from typing import Literal, Any
|
||||||
|
|
||||||
|
|
||||||
API_URL = "https://api.anixart.tv"
|
@dataclass
|
||||||
|
class Endpoint:
|
||||||
|
path: str
|
||||||
|
method: Literal["GET", "POST"]
|
||||||
|
required_args: dict[str, type] = field(default_factory=dict)
|
||||||
|
|
||||||
|
_json = False
|
||||||
|
_API_ENDPOINT = "https://api.anixart.tv/"
|
||||||
|
|
||||||
|
def __post_init__(self):
|
||||||
|
if self.method not in ["GET", "POST"]:
|
||||||
|
raise ValueError("Method must be either GET or POST.")
|
||||||
|
if not isinstance(self.required_args, dict):
|
||||||
|
raise ValueError("Required arguments must be a dictionary.")
|
||||||
|
if not all(isinstance(v, type) for v in self.required_args.values()):
|
||||||
|
raise ValueError("All values in required arguments must be types.")
|
||||||
|
|
||||||
|
def _post(self, method: str, **kwargs):
|
||||||
|
headers = {}
|
||||||
|
url = self._API_ENDPOINT + method
|
||||||
|
if token:=kwargs.get("token"):
|
||||||
|
kwargs["token"] = token
|
||||||
|
url += f"?token={token}"
|
||||||
|
|
||||||
|
req_settings = {"url": url}
|
||||||
|
if self._json:
|
||||||
|
headers["Content-Type"] = "application/json; charset=UTF-8"
|
||||||
|
req_settings.update({"json": kwargs})
|
||||||
|
else:
|
||||||
|
req_settings.update({"data": kwargs})
|
||||||
|
return req_settings, headers
|
||||||
|
|
||||||
|
def _get(self, method: str, **kwargs):
|
||||||
|
headers = {}
|
||||||
|
url = self._API_ENDPOINT + method
|
||||||
|
if token:=kwargs.get("token"):
|
||||||
|
kwargs["token"] = token
|
||||||
|
req_settings = {"url": url, "params": kwargs}
|
||||||
|
return req_settings, headers
|
||||||
|
|
||||||
|
def _check_arguments(self, **kwargs):
|
||||||
|
"""Check if all required arguments are present."""
|
||||||
|
missing_args = [] # (arg, reason)
|
||||||
|
for arg, arg_type in self.required_args.items():
|
||||||
|
if arg not in kwargs:
|
||||||
|
missing_args.append((arg, "missing"))
|
||||||
|
elif not isinstance(kwargs[arg], arg_type):
|
||||||
|
missing_args.append((arg, f"invalid type: {type(kwargs[arg])}"))
|
||||||
|
if missing_args:
|
||||||
|
pretty_args = ", ".join(f"{arg} ({reason})" for arg, reason in missing_args)
|
||||||
|
raise ValueError(f"Missing or invalid arguments: {pretty_args}")
|
||||||
|
|
||||||
|
def build_request(self, **kwargs) -> tuple[dict[str, dict[str, Any] | str], dict[str, str]]:
|
||||||
|
"""
|
||||||
|
Build the request for the endpoint.
|
||||||
|
:param kwargs: Arguments to be passed to the endpoint.
|
||||||
|
:return: A tuple containing the HTTP method, headers, and request settings.
|
||||||
|
"""
|
||||||
|
self._check_arguments(**kwargs)
|
||||||
|
if self.method == "POST":
|
||||||
|
return self._post(self.path, **kwargs)
|
||||||
|
if self.method == "GET":
|
||||||
|
return self._get(self.path, **kwargs)
|
||||||
|
|
||||||
|
def endpoint(path: str, method: Literal["GET", "POST"], required_args: dict[str, type]) -> Endpoint:
|
||||||
|
return Endpoint(path, method, required_args)
|
||||||
|
|
||||||
|
class AnixartAuthEndpoints:
|
||||||
|
"""Anixart API authentication endpoints."""
|
||||||
|
login = endpoint("/auth/signIn", "POST", {"login": str, "password": str})
|
||||||
|
|
||||||
|
class AnixartEndpoints:
|
||||||
|
"""Anixart API endpoints."""
|
||||||
|
|
||||||
|
def __init__(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
# ----------- # AUTH # ----------- #
|
# ----------- # AUTH # ----------- #
|
||||||
|
|
||||||
|
|||||||
@@ -1 +1 @@
|
|||||||
from .objects import Profile, ProfileVote, ProfileRoles, ProfileHistory, ProfileFriendsPreview
|
from .objects import Profile, ProfileFull, ProfileVote, ProfileRoles, ProfileHistory, ProfileFriendsPreview
|
||||||
|
|||||||
@@ -16,10 +16,13 @@ class ProfileRoles:
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class ProfileFriendsPreview:
|
class Profile:
|
||||||
id: int
|
id: int
|
||||||
avatar: str
|
|
||||||
login: str
|
login: str
|
||||||
|
avatar: str
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class ProfileFriendsPreview(Profile):
|
||||||
friend_count: int
|
friend_count: int
|
||||||
friend_status: int
|
friend_status: int
|
||||||
is_sponsor: bool
|
is_sponsor: bool
|
||||||
@@ -32,10 +35,7 @@ class ProfileFriendsPreview:
|
|||||||
badge_url: None
|
badge_url: None
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class Profile:
|
class ProfileFull(Profile):
|
||||||
id: int
|
|
||||||
login: str
|
|
||||||
avatar: str
|
|
||||||
status: str
|
status: str
|
||||||
rating_score: int
|
rating_score: int
|
||||||
history: list[ProfileHistory]
|
history: list[ProfileHistory]
|
||||||
@@ -115,4 +115,4 @@ class ProfileLoginsHistory:
|
|||||||
content: list[_login]
|
content: list[_login]
|
||||||
total_count: int
|
total_count: int
|
||||||
# total_page_count: int
|
# total_page_count: int
|
||||||
# current_page: int
|
# current_page: int
|
||||||
|
|||||||
Reference in New Issue
Block a user