mirror of
https://github.com/SantaSpeen/anixart.git
synced 2026-02-16 02:20:47 +00:00
Compare commits
3 Commits
aee5c542b4
...
0682bdb42b
| Author | SHA1 | Date | |
|---|---|---|---|
| 0682bdb42b | |||
| 63bf159679 | |||
| 8a944319d6 |
@@ -4,19 +4,17 @@ import requests
|
||||
from .__meta__ import __version__, __build__
|
||||
from .auth import AnixartAccount, AnixartAccountGuest
|
||||
from .enums import AnixartApiErrors
|
||||
from .endpoints import API_URL
|
||||
from .exceptions import AnixartAPIRequestError, AnixartAPIError
|
||||
from .exceptions import AnixartInitError
|
||||
|
||||
debug = True
|
||||
|
||||
class AnixartAPI:
|
||||
API_URL = "https://api.anixart.tv/"
|
||||
|
||||
def __init__(self, account: AnixartAccount = None):
|
||||
if account is None:
|
||||
account = AnixartAccountGuest()
|
||||
if not isinstance(account, AnixartAccount):
|
||||
raise AnixartInitError(f'Use class "AnixartAccount" for user. But {type(account)} given.')
|
||||
|
||||
self.use_account(account)
|
||||
|
||||
def use_account(self, account: AnixartAccount):
|
||||
@@ -37,6 +35,9 @@ class AnixartAPI:
|
||||
return self.__account
|
||||
|
||||
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:
|
||||
e = AnixartAPIRequestError("Bad Request: Invalid request parameters.")
|
||||
e.message = (
|
||||
@@ -56,7 +57,8 @@ class AnixartAPI:
|
||||
except ValueError as e:
|
||||
raise AnixartAPIError("Failed to parse JSON response")
|
||||
|
||||
# print(response)
|
||||
if debug:
|
||||
print(response)
|
||||
if response['code'] != 0:
|
||||
code = response['code']
|
||||
if code in AnixartApiErrors:
|
||||
@@ -69,42 +71,26 @@ class AnixartAPI:
|
||||
|
||||
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:
|
||||
def _post(self, method: str, _json: bool = False, **kwargs):
|
||||
url = self.API_URL + method
|
||||
kwargs["token"] = kwargs.get("token", self.__token)
|
||||
if kwargs["token"]:
|
||||
url += f"?token={self.__token}"
|
||||
|
||||
req_settings = {"url": url}
|
||||
if _json:
|
||||
self._session.headers["Content-Type"] = "application/json; charset=UTF-8"
|
||||
self._session.headers["Content-Length"] = str(len(payload))
|
||||
kwargs.update({"json": payload})
|
||||
req_settings.update({"json": kwargs})
|
||||
else:
|
||||
kwargs.update({"data": payload})
|
||||
res = self._session.post(**kwargs)
|
||||
self._session.headers["Content-Type"] = ""
|
||||
self._session.headers["Content-Length"] = ""
|
||||
req_settings.update({"data": kwargs})
|
||||
res = self._session.post(**req_settings)
|
||||
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)
|
||||
def _get(self, method: str, **kwargs):
|
||||
if self.__token:
|
||||
kwargs["token"] = kwargs.get("token", self.__token)
|
||||
|
||||
res = self._session.get(self.API_URL + method, params=kwargs)
|
||||
return self.__parse_response(res)
|
||||
|
||||
def execute(self, http_method, endpoint, **kwargs):
|
||||
|
||||
@@ -55,8 +55,7 @@ class AnixartAccount:
|
||||
|
||||
def login(self):
|
||||
"""Login to Anixart and return the token."""
|
||||
payload = {"login": self.username, "password": self._password}
|
||||
res = self._api.post(endpoints.SING_IN, payload)
|
||||
res = self._api.post(endpoints.SING_IN, login=self.username, password=self._password)
|
||||
uid = res["profile"]["id"]
|
||||
token = res["profileToken"]["token"]
|
||||
self._id = uid
|
||||
|
||||
@@ -1,7 +1,85 @@
|
||||
# -*- 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 # ----------- #
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@dataclass
|
||||
class ProfileFriendsPreview:
|
||||
class Profile:
|
||||
id: int
|
||||
avatar: str
|
||||
login: str
|
||||
avatar: str
|
||||
|
||||
@dataclass
|
||||
class ProfileFriendsPreview(Profile):
|
||||
friend_count: int
|
||||
friend_status: int
|
||||
is_sponsor: bool
|
||||
@@ -32,10 +35,7 @@ class ProfileFriendsPreview:
|
||||
badge_url: None
|
||||
|
||||
@dataclass
|
||||
class Profile:
|
||||
id: int
|
||||
login: str
|
||||
avatar: str
|
||||
class ProfileFull(Profile):
|
||||
status: str
|
||||
rating_score: int
|
||||
history: list[ProfileHistory]
|
||||
@@ -115,4 +115,4 @@ class ProfileLoginsHistory:
|
||||
content: list[_login]
|
||||
total_count: int
|
||||
# total_page_count: int
|
||||
# current_page: int
|
||||
# current_page: int
|
||||
|
||||
Reference in New Issue
Block a user