[+] BaseParseFactory

[+] AuthLoginFactory
[^] Endpoint
[+] _ApiMethodGenerator
This commit is contained in:
Maxim Khomutov 2025-04-11 19:00:10 +03:00
parent 0682bdb42b
commit 723cf98475
15 changed files with 334 additions and 158 deletions

View File

@ -3,9 +3,8 @@
from .__meta__ import *
from .api import AnixartAPI
from .auth import AnixartAccount, AnixartAccountGuest, AnixartAccountSaved
from .endpoints import *
from . import auth
from . import enums
from . import exceptions

View File

@ -1,16 +1,41 @@
# -*- coding: utf-8 -*-
import requests
from .__meta__ import __version__, __build__
from .auth import AnixartAccount, AnixartAccountGuest
from .enums import AnixartApiErrors
from .endpoints_map import endpoints_map
from .exceptions import AnixartAPIRequestError, AnixartAPIError
from .exceptions import AnixartInitError
debug = True
class _ApiMethodGenerator:
def __init__(self, start_path, _callback: callable):
self._path = start_path
self._callback = _callback
def __setattr__(self, __name, __value):
raise TypeError(f"cannot set '{__name}' attribute of immutable type 'ApiMethodGenerator'")
def __getattr__(self, item):
self._path += f".{item}"
return self
def __call__(self, *args, **kwargs):
if debug:
print(f"[D] __call__ -> {self._path} args={args} kwargs={kwargs}")
return self._callback(self._path, *args, **kwargs)
def __str__(self):
return f"ApiMethodGenerator(path={self._path!r})"
def __repr__(self):
return f"<{self}>"
class AnixartAPI:
API_URL = "https://api.anixart.tv/"
API_URL = "https://api.anixart.tv"
def __init__(self, account: AnixartAccount = None):
if account is None:
@ -60,12 +85,7 @@ class AnixartAPI:
if debug:
print(response)
if response['code'] != 0:
code = response['code']
if code in AnixartApiErrors:
e = AnixartAPIError(f"AnixartAPI send error: {AnixartApiErrors(code).name}")
e.message = AnixartApiErrors(code).name
else:
e = AnixartAPIError(f"AnixartAPI send unknown error, code: {response['code']}")
e = AnixartAPIError(f"AnixartAPI send unknown error, code: {response['code']}")
e.code = response['code']
raise e
@ -93,7 +113,7 @@ class AnixartAPI:
res = self._session.get(self.API_URL + method, params=kwargs)
return self.__parse_response(res)
def execute(self, http_method, endpoint, **kwargs):
def _execute(self, http_method, endpoint, **kwargs):
http_method = http_method.upper()
if http_method == "GET":
return self._get(endpoint, **kwargs)
@ -102,11 +122,29 @@ class AnixartAPI:
else:
raise AnixartAPIRequestError("Allow only GET and POST requests.")
def __execute_endpoint_from_map(self, key, *args, **kwargs):
endpoint = endpoints_map.get(key)
if endpoint is None:
raise AnixartAPIError("Invalid endpoint.")
return self._execute_endpoint(endpoint, *args, **kwargs)
def _execute_endpoint(self, endpoint, *args, **kwargs):
req_settings, headers = endpoint.build_request(*args, **kwargs)
self._session.headers.update(headers)
if endpoint.is_post:
res = self._session.post(**req_settings)
else:
res = self._session.get(**req_settings)
return self.__parse_response(res)
def get(self, endpoint, *args, **kwargs):
return self.execute("GET", endpoint.format(*args), **kwargs)
return self._execute("GET", endpoint.format(*args), **kwargs)
def post(self, endpoint, *args, **kwargs):
return self.execute("POST", endpoint.format(*args), **kwargs)
return self._execute("POST", endpoint.format(*args), **kwargs)
def __getattr__(self, item):
return _ApiMethodGenerator(item, self.__execute_endpoint_from_map)
def __str__(self):
return f'AnixartAPI(account={self.__account!r})'

View File

@ -1 +1 @@
from .account import AnixartAccount, AnixartAccountSaved, AnixartAccountGuest
from .accounts import AnixartAccount, AnixartAccountSaved, AnixartAccountGuest

23
anixart/auth/endpoints.py Normal file
View File

@ -0,0 +1,23 @@
from anixart.utils import endpoint
from .factories import AuthLoginFactory
# ----------- # AUTH # ----------- #
#
# POST
# SING_UP = None
# SING_IN = "/auth/signIn"
#
# !! Not Checked !!
# POST
# AUTH_SING_IN_WITH_GOOGLE = "/auth/google" # {googleIdToken} or {login, email, googleIdToken}
# AUTH_SING_IN_WITH_VK = "/auth/vk" # {vkAccessToken}
class AnixartAuthEndpoints:
"""Anixart API authentication endpoints."""
singup = None # Удалено дабы исключить автореги (по просьбе аниксарта)
login = endpoint("/auth/signIn", "POST", AuthLoginFactory, {}, {"login": str, "password": str})
# login_google = None # endpoint("/auth/google", "POST", {}, {"googleIdToken": str})
# login_vk = None # endpoint("/auth/vk", "POST", {}, {"vkAccessToken": str})

View File

@ -1,11 +0,0 @@
from enum import IntEnum
class AnixartAuthErrors(IntEnum):
INCORRECT_LOGIN = 2
INCORRECT_PASSWORD = 3
def errors_handler(error):
"""Handle errors and return a JSON response."""
pass

41
anixart/auth/factories.py Normal file
View File

@ -0,0 +1,41 @@
from dataclasses import dataclass
from anixart.exceptions import AnixartAuthError
from anixart.utils import BaseParseFactory
@dataclass
class _AuthLoginFactoryProfileToken:
# { "id": 1, "token": "hash?" }
id: int # token id?
token: str # token str
@classmethod
def from_dict(cls, data: dict):
return cls(
id=data.get("id", -1),
token=data.get("token")
)
@dataclass
class AuthLoginFactory(BaseParseFactory):
# { "code": 0, "profile": {...}, "profileToken": {_AuthLoginFactoryProfileToken} }
profile: dict # TODO - create a ProfileFactory
profileToken: _AuthLoginFactoryProfileToken
_errors = {
2: "Incorrect login",
3: "Incorrect password",
}
_exception = AnixartAuthError
@classmethod
def from_dict(cls, data: dict):
return cls(
code=data["code"],
profile=data.get("profile", {}),
profileToken=_AuthLoginFactoryProfileToken.from_dict(data.get("profileToken", {})),
_errors=cls._errors,
_exception=cls._exception
)

View File

@ -1,7 +0,0 @@
from dataclasses import dataclass
@dataclass
class ProfileToken:
id: int
token: str

9
anixart/endpoints_map.py Normal file
View File

@ -0,0 +1,9 @@
from .auth.factories import AuthLoginFactory
endpoints_map = {
# Auth
"auth.login": AuthLoginFactory
# Profile
}

View File

@ -1,15 +1,9 @@
from enum import IntEnum
class AnixartApiErrors(IntEnum):
""" Error codes for AnixartApi authentication."""
INCORRECT_LOGIN = 2
INCORRECT_PASSWORD = 3
class AnixartComment(IntEnum):
DISLIKE = 1
LIKE = 2
class AnixartProfileVotedSort(IntEnum):
LAST_FIRST = 1
OLD_FIRST = 2
@ -19,7 +13,6 @@ class AnixartProfileVotedSort(IntEnum):
STAR_2 = 6
STAR_1 = 7
class AnixartLists(IntEnum):
WATCHING = 1
IN_PLANS = 2

View File

@ -1,13 +1,13 @@
# -*- coding: utf-8 -*-
class AnixartBasError(Exception): ...
class AnixartBaseError(Exception): ...
# Init errors
class AnixartInitError(AnixartBasError, TypeError): ...
class AnixartInitError(AnixartBaseError, TypeError): ...
# API errors
class AnixartAPIError(AnixartBasError):
class AnixartAPIError(AnixartBaseError):
message = "unknown error"
code = 0

View File

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

View File

@ -0,0 +1,80 @@
from anixart.utils import endpoint
# # ----------- # PROFILE # ----------- #
# # TODO PROFILE: SETTINGS, SETTINGS_RELEASE, SETTINGS_RELEASE_FIRST,
# # SETTINGS_COMMENTS, SETTINGS_COLLECTION, EDIT_AVATAR, SETTINGS_RELEASE_LIST,
# # SETTINGS_RELEASE_TYPE
#
# # GET
# PROFILE = "/profile/{}" # + profile id (Токен нужен только что бы был is_my_profile)
# PROFILE_NICK_HISTORY = "/profile/login/history/all/{}/{}" # profile id / page (Токен не нужен)
#
# PROFILE_BLACKLIST = "/profile/blocklist/all/{}" # page
# PROFILE_BLACKLIST_ADD = "/profile/blocklist/add/{}" # profile id
# PROFILE_BLACKLIST_REMOVE = "/profile/blocklist/remove/{}" # profile id
#
# FRIENDS = "/profile/friend/all/{}/{}" # profile id / page
# FRIENDS_RQ_IN = "/profile/friend/requests/in/{}" # page
# FRIENDS_RQ_OUT = "/profile/friend/requests/out/{}" # page
# FRIENDS_RQ_IN_LAST = "/profile/friend/requests/in/last"
# FRIENDS_RQ_OUT_LAST = "/profile/friend/requests/out/last"
# FRIENDS_SEND = "/profile/friend/request/send/{}" # profile id
# FRIENDS_REMOVE = "/profile/friend/request/remove/{}" # profile id
#
# VOTE_VOTED = "/profile/vote/release/voted/{}/{}" # profile id / page
# # Да, ребята из аниксарта не знают английский; ↓
# # noinspection SpellCheckingInspection
# VOTE_UNVENTED = "/profile/vote/release/unvoted/{}" # page
#
# LISTS = "/profile/list/all/{}/{}/{}" # profile id / list id / page
#
# SETTINGS_NOTIFICATION = "/profile/preference/notification/my"
# SETTINGS_NOTIFICATION_RELEASE = "/profile/preference/notification/episode/edit"
# SETTINGS_NOTIFICATION_RELEASE_FIRST = "/profile/preference/notification/episode/first/edit"
# SETTINGS_NOTIFICATION_COMMENTS = "/profile/preference/notification/comment/edit"
# SETTINGS_NOTIFICATION_COLLECTION = "/profile/preference/notification/my/collection/comment/edit"
#
# CHANGE_PASSWORD = "/profile/preference/password/change"
#
# # POST
# EDIT_STATUS = "/profile/preference/status/edit"
# EDIT_SOCIAL = "/profile/preference/social/edit"
# EDIT_AVATAR = "/profile/preference/avatar/edit"
#
# # {"profileStatusNotificationPreferences":[0 - favorite, + all in AnixList]}
# SETTINGS_NOTIFICATION_RELEASE_LIST = "/profile/preference/notification/status/edit"
# # {"profileTypeNotificationPreferences":[type ids]}
# SETTINGS_NOTIFICATION_RELEASE_TYPE = "/profile/preference/notification/type/edit"
#
# # Not Checked
# # GET
# PROFILE_SOCIAL = "/profile/social/{}" # profile id
#
# FRIENDS_RECOMMENDATION = "/profile/friend/recommendations"
# FRIENDS_RQ_HIDE = "profile/friend/request/hide/{}" # profile id
#
# SETTINGS_PROFILE = "/profile/preference/my"
# SETTINGS_PROFILE_CHANGE_EMAIL = "/profile/preference/email/change" # {current_password, current, new}
# SETTINGS_PROFILE_CHANGE_EMAIL_CONFIRM = "/profile/preference/email/change/confirm" # {current}
#
# # /profile/preference/social
# SETTINGS_PROFILE_STATUS_DELETE = "/profile/preference/status/delete"
#
# # POST
# PROFILE_PROCESS = "/profile/process/{}" # profile id
#
# SETTINGS_PROFILE_CHANGE_LOGIN = "/profile/preference/email/login/confirm" # {login}
# SETTINGS_PROFILE_CHANGE_LOGIN_INFO = "/profile/preference/email/login/info" # {login}
#
# SETTINGS_PROFILE_BIND_GOOGLE = "/profile/preference/google/bind" # {idToken, }
# SETTINGS_PROFILE_UNBIND_GOOGLE = "/profile/preference/google/unbind"
# SETTINGS_PROFILE_BIND_VK = "/profile/preference/google/bind" # {accessToken, }
# SETTINGS_PROFILE_UNBIND_VK = "/profile/preference/google/unbind"
#
# SETTINGS_PROFILE_PRIVACY_COUNTS = "/profile/preference/privacy/counts/edit"
# SETTINGS_PROFILE_PRIVACY_FRIENDS_REQUESTS = "/profile/preference/privacy/friendRequests/edit"
# SETTINGS_PROFILE_PRIVACY_SOCIAL = "/profile/preference/privacy/social/edit"
# SETTINGS_PROFILE_PRIVACY_STATS = "/profile/preference/privacy/stats/edit"
class AnixartProfileEndpoints:
profile = endpoint("/profile/{id}", "GET", None, {"id": int}, {})

View File

@ -1,11 +0,0 @@
from enum import IntEnum
class AnixartProfileErrors(IntEnum):
""" Error codes for AnixartApi authentication."""
PROFILE_NOT_FOUND = 2
def errors_handler(error):
"""Handle errors and return a JSON response."""
pass

View File

@ -116,3 +116,72 @@ class ProfileLoginsHistory:
total_count: int
# total_page_count: int
# current_page: int
# From singup
# "profile": {
# "id": 123,
# "login": "loginstr",
# "avatar": "*.jpg",
# "status": "123",
# "badge": null,
# "history": [],
# "votes": [],
# "roles": [],
# "last_activity_time": 100046000,
# "register_date": 100046000,
# "vk_page": "",
# "tg_page": "",
# "inst_page": "",
# "tt_page": "",
# "discord_page": "",
# "ban_expires": 0,
# "ban_reason": null,
# "privilege_level": 0,
# "watching_count": 1,
# "plan_count": 1,
# "completed_count": 1,
# "hold_on_count": 1,
# "dropped_count": 1,
# "favorite_count": 1,
# "comment_count": 1,
# "collection_count": 0,
# "video_count": 0,
# "friend_count": 1,
# "subscription_count": 0,
# "watched_episode_count": 1,
# "watched_time": 1,
# "is_private": false,
# "is_sponsor": false,
# "is_banned": false,
# "is_perm_banned": false,
# "is_bookmarks_transferred": false,
# "is_sponsor_transferred": false,
# "is_vk_bound": true,
# "is_google_bound": true,
# "is_release_type_notifications_enabled": false,
# "is_episode_notifications_enabled": true,
# "is_first_episode_notification_enabled": true,
# "is_related_release_notifications_enabled": true,
# "is_report_process_notifications_enabled": true,
# "is_comment_notifications_enabled": true,
# "is_my_collection_comment_notifications_enabled": true,
# "is_my_article_comment_notifications_enabled": false,
# "is_verified": false,
# "friends_preview": [],
# "collections_preview": [],
# "release_comments_preview": [],
# "comments_preview": [],
# "release_videos_preview": [],
# "watch_dynamics": [],
# "friend_status": null,
# "rating_score": 0,
# "is_blocked": false,
# "is_me_blocked": false,
# "is_stats_hidden": false,
# "is_counts_hidden": false,
# "is_social_hidden": false,
# "is_friend_requests_disallowed": false,
# "is_online": false,
# "sponsorshipExpires": 0
# }

View File

@ -2,12 +2,41 @@
from dataclasses import dataclass, field
from typing import Literal, Any
from anixart.exceptions import AnixartAPIError
@dataclass
class BaseParseFactory:
code: int
_errors: dict[int, str]
_exception: type[AnixartAPIError]
def raise_if_error(self):
if self.code != 0:
error = self._errors.get(self.code)
if error:
raise error
else:
raise ValueError(f"Unknown error code: {self.code}")
@classmethod
def from_dict(cls, data: dict) -> "BaseParseFactory":
return cls(
code=data.get("code", 0),
_errors={},
_exception=AnixartAPIError,
)
@dataclass
class Endpoint:
path: str
method: Literal["GET", "POST"]
parse_factory: type[BaseParseFactory]
required_args: dict[str, type] = field(default_factory=dict)
required_kwargs: dict[str, type] = field(default_factory=dict)
_json = False
_API_ENDPOINT = "https://api.anixart.tv/"
@ -15,11 +44,23 @@ class Endpoint:
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):
if not isinstance(self.required_kwargs, dict):
raise ValueError("Required arguments must be a dictionary.")
if not all(isinstance(v, type) for v in self.required_args.values()):
if not all(isinstance(v, type) for v in self.required_kwargs.values()):
raise ValueError("All values in required arguments must be types.")
@property
def is_get(self) -> bool:
return self.method == "GET"
@property
def is_post(self) -> bool:
return self.method == "POST"
@property
def is_json(self) -> bool:
return self._json
def _post(self, method: str, **kwargs):
headers = {}
url = self._API_ENDPOINT + method
@ -48,125 +89,37 @@ class Endpoint:
missing_args = [] # (arg, reason)
for arg, arg_type in self.required_args.items():
if arg not in kwargs:
missing_args.append((arg, "missing"))
missing_args.append((arg, "arg missing"))
elif not isinstance(kwargs[arg], arg_type):
missing_args.append((arg, f"invalid type: {type(kwargs[arg])}"))
missing_args.append((arg, f"arg invalid type: {type(kwargs[arg])}"))
for arg, arg_type in self.required_kwargs.items():
if arg not in kwargs:
missing_args.append((arg, "kwarg missing"))
elif not isinstance(kwargs[arg], arg_type):
missing_args.append((arg, f"kwarg 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]]:
def build_request(self, **kwargs) -> None | tuple[dict[str, dict[str, Any] | str], dict[Any, Any]] | 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)
args = {arg: kwargs[arg] for arg in self.required_args}
if self.method == "POST":
return self._post(self.path, **kwargs)
return self._post(self.path.format(**args), **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
return self._get(self.path.format(**args), **kwargs)
# ----------- # AUTH # ----------- #
def endpoint(path: str, method: Literal["GET", "POST"], parse_factory: type[BaseParseFactory], required_args: dict[str, type], required_kwargs: dict[str, type]) -> Endpoint:
return Endpoint(path, method, parse_factory, required_args, required_kwargs)
# POST
SING_UP = None # Удалено дабы исключить автореги
SING_IN = "/auth/signIn"
# Not Checked
# POST
_AUTH_SING_IN_WITH_GOOGLE = "/auth/google" # {googleIdToken} or {login, email, googleIdToken}
_AUTH_SING_IN_WITH_VK = "/auth/vk" # {vkAccessToken}
# ----------- # PROFILE # ----------- #
# TODO PROFILE: SETTINGS, SETTINGS_RELEASE, SETTINGS_RELEASE_FIRST,
# SETTINGS_COMMENTS, SETTINGS_COLLECTION, EDIT_AVATAR, SETTINGS_RELEASE_LIST,
# SETTINGS_RELEASE_TYPE
# GET
PROFILE = "/profile/{}" # + profile id (Токен нужен только что бы был is_my_profile)
PROFILE_NICK_HISTORY = "/profile/login/history/all/{}/{}" # profile id / page (Токен не нужен)
PROFILE_BLACKLIST = "/profile/blocklist/all/{}" # page
PROFILE_BLACKLIST_ADD = "/profile/blocklist/add/{}" # profile id
PROFILE_BLACKLIST_REMOVE = "/profile/blocklist/remove/{}" # profile id
FRIENDS = "/profile/friend/all/{}/{}" # profile id / page
FRIENDS_RQ_IN = "/profile/friend/requests/in/{}" # page
FRIENDS_RQ_OUT = "/profile/friend/requests/out/{}" # page
FRIENDS_RQ_IN_LAST = "/profile/friend/requests/in/last"
FRIENDS_RQ_OUT_LAST = "/profile/friend/requests/out/last"
FRIENDS_SEND = "/profile/friend/request/send/{}" # profile id
FRIENDS_REMOVE = "/profile/friend/request/remove/{}" # profile id
VOTE_VOTED = "/profile/vote/release/voted/{}/{}" # profile id / page
# Да, ребята из аниксарта не знают английский; ↓
# noinspection SpellCheckingInspection
VOTE_UNVENTED = "/profile/vote/release/unvoted/{}" # page
LISTS = "/profile/list/all/{}/{}/{}" # profile id / list id / page
SETTINGS_NOTIFICATION = "/profile/preference/notification/my"
SETTINGS_NOTIFICATION_RELEASE = "/profile/preference/notification/episode/edit"
SETTINGS_NOTIFICATION_RELEASE_FIRST = "/profile/preference/notification/episode/first/edit"
SETTINGS_NOTIFICATION_COMMENTS = "/profile/preference/notification/comment/edit"
SETTINGS_NOTIFICATION_COLLECTION = "/profile/preference/notification/my/collection/comment/edit"
CHANGE_PASSWORD = "/profile/preference/password/change"
# POST
EDIT_STATUS = "/profile/preference/status/edit"
EDIT_SOCIAL = "/profile/preference/social/edit"
EDIT_AVATAR = "/profile/preference/avatar/edit"
# {"profileStatusNotificationPreferences":[0 - favorite, + all in AnixList]}
SETTINGS_NOTIFICATION_RELEASE_LIST = "/profile/preference/notification/status/edit"
# {"profileTypeNotificationPreferences":[type ids]}
SETTINGS_NOTIFICATION_RELEASE_TYPE = "/profile/preference/notification/type/edit"
# Not Checked
# GET
PROFILE_SOCIAL = "/profile/social/{}" # profile id
FRIENDS_RECOMMENDATION = "/profile/friend/recommendations"
FRIENDS_RQ_HIDE = "profile/friend/request/hide/{}" # profile id
SETTINGS_PROFILE = "/profile/preference/my"
SETTINGS_PROFILE_CHANGE_EMAIL = "/profile/preference/email/change" # {current_password, current, new}
SETTINGS_PROFILE_CHANGE_EMAIL_CONFIRM = "/profile/preference/email/change/confirm" # {current}
# /profile/preference/social
SETTINGS_PROFILE_STATUS_DELETE = "/profile/preference/status/delete"
# POST
PROFILE_PROCESS = "/profile/process/{}" # profile id
SETTINGS_PROFILE_CHANGE_LOGIN = "/profile/preference/email/login/confirm" # {login}
SETTINGS_PROFILE_CHANGE_LOGIN_INFO = "/profile/preference/email/login/info" # {login}
SETTINGS_PROFILE_BIND_GOOGLE = "/profile/preference/google/bind" # {idToken, }
SETTINGS_PROFILE_UNBIND_GOOGLE = "/profile/preference/google/unbind"
SETTINGS_PROFILE_BIND_VK = "/profile/preference/google/bind" # {accessToken, }
SETTINGS_PROFILE_UNBIND_VK = "/profile/preference/google/unbind"
SETTINGS_PROFILE_PRIVACY_COUNTS = "/profile/preference/privacy/counts/edit"
SETTINGS_PROFILE_PRIVACY_FRIENDS_REQUESTS = "/profile/preference/privacy/friendRequests/edit"
SETTINGS_PROFILE_PRIVACY_SOCIAL = "/profile/preference/privacy/social/edit"
SETTINGS_PROFILE_PRIVACY_STATS = "/profile/preference/privacy/stats/edit"
# ----------- # COLLECTION # ----------- #