3 Commits

Author SHA1 Message Date
0682bdb42b Endpoint constructor 2025-04-04 18:17:36 +03:00
63bf159679 [~] minor 2025-04-04 18:17:23 +03:00
8a944319d6 [+] Fixes for post 2025-04-04 17:57:52 +03:00
5 changed files with 111 additions and 48 deletions

View File

@@ -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):

View File

@@ -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

View File

@@ -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 # ----------- #

View File

@@ -1 +1 @@
from .objects import Profile, ProfileVote, ProfileRoles, ProfileHistory, ProfileFriendsPreview
from .objects import Profile, ProfileFull, ProfileVote, ProfileRoles, ProfileHistory, ProfileFriendsPreview

View File

@@ -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