diff --git a/koreanbots/__init__.py b/koreanbots/__init__.py index 7fdee5b..d7265cc 100644 --- a/koreanbots/__init__.py +++ b/koreanbots/__init__.py @@ -1,12 +1,5 @@ from typing import Literal, NamedTuple -from .client import Koreanbots as Koreanbots -from .errors import * -from .http import KoreanbotsRequester as KoreanbotsRequester -from .model import KoreanbotsBot as KoreanbotsBot -from .model import KoreanbotsServer as KoreanbotsServer -from .model import KoreanbotsUser as KoreanbotsUser - class VersionInfo(NamedTuple): major: int @@ -16,7 +9,7 @@ class VersionInfo(NamedTuple): serial: int -version_info = VersionInfo(3, 1, 0, "final", 0) +version_info = VersionInfo(major=4, minor=0, micro=0, releaselevel="final", serial=0) __version__ = f"{version_info.major}.{version_info.minor}.{version_info.micro}" diff --git a/koreanbots/client.py b/koreanbots/client.py index 0841bd3..3603890 100644 --- a/koreanbots/client.py +++ b/koreanbots/client.py @@ -1,296 +1,118 @@ -from logging import getLogger -from typing import Optional -from warnings import warn - -import aiohttp - -from koreanbots.decorator import strict_literal -from koreanbots.errors import KoreanbotsException -from koreanbots.http import KoreanbotsRequester -from koreanbots.model import ( - KoreanbotsBotResponse, - KoreanbotsResponse, - KoreanbotsServerResponse, - KoreanbotsUserResponse, - KoreanbotsVoteResponse, +from aiohttp import ClientSession + +from koreanbots.domain.entities import ( + Bot, + KoreanbotsDataResponse, + KoreanbotsMessageResponse, + Server, + User, + Vote, ) -from koreanbots.typing import VoteType, WidgetStyle, WidgetType - -log = getLogger(__name__) +from koreanbots.request import KoreanbotsRequester class Koreanbots(KoreanbotsRequester): - """ - KoreanbotsRequester를 감싸는 클라이언트 클래스입니다. - - :param api_key: - API key를 지정합니다. 만약 필요한 경우 이 키를 지정하세요. - :type api_key: - Optional[str] - - :param session: - aiohttp.ClientSession의 클래스입니다. 만약 필요한 경우 이 인수를 지정하세요. 지정하지 않으면 생성합니다. - :type session: - Optional[aiohttp.ClientSession] - """ - - def __init__( - self, - api_key: Optional[str] = None, - session: Optional[aiohttp.ClientSession] = None, - ) -> None: + def __init__(self, api_key: str, session: ClientSession | None = None) -> None: super().__init__(api_key, session) - async def post_guild_count(self, bot_id: int, **kwargs: Optional[int]) -> None: - """ - 길드 개수를 서버에 전송합니다. - - :param bot_id: - 요청할 bot의 ID를 지정합니다. - :type bot_id: - int - """ - await super().post_update_bot_info(bot_id, **kwargs) - - async def get_user_info( - self, user_id: int - ) -> KoreanbotsResponse[KoreanbotsUserResponse]: - """ - 유저 정보를 가져옵니다. - - :param user_id: - 요청할 유저의 ID를 지정합니다. - :type user_id: - int - :return: - 유저 정보를 담고 있는 KoreanbotsUser클래스입니다. - :rtype: - KoreanbotsUser - """ - data = await super().get_user_info(user_id) - - code = data["code"] - version = data["version"] - data = data["data"] - - return KoreanbotsResponse( - code=code, version=version, data=KoreanbotsUserResponse.from_dict(data) + async def get_bot_info(self, bot_id: int) -> KoreanbotsDataResponse[Bot]: + res = await self.request_bot_info(bot_id) + return KoreanbotsDataResponse.from_bot( + code=res["code"], + version=res["version"], + data=res["data"], ) - async def get_bot_info( - self, bot_id: int - ) -> KoreanbotsResponse[KoreanbotsBotResponse]: - """ - 봇 정보를 가져옵니다. - - :param bot_id: - 요청할 봇의 ID를 지정합니다. - :type bot_id: - int - - :return: - 봇 정보를 담고 있는 KoreanbotsBot클래스입니다. - :rtype: - KoreanbotsBot - """ - data = await super().get_bot_info(bot_id) - - code = data["code"] - version = data["version"] - data = data["data"] - - return KoreanbotsResponse( - code=code, version=version, data=KoreanbotsBotResponse.from_dict(data) + async def search_bot( + self, query: str, page: int = 1 + ) -> KoreanbotsDataResponse[list[Bot]]: + res = await self.request_search_bot(query, page) + return KoreanbotsDataResponse.from_list_bot( + code=res["code"], + version=res["version"], + data=res["data"], ) - async def get_server_info( - self, server_id: int - ) -> KoreanbotsResponse[KoreanbotsServerResponse]: - """ - 서버 정보를 가져옵니다. - - :param server_id: - 요청할 서버의 ID를 지정합니다. - :type server_id: - int - - :return: - 봇 정보를 담고 있는 KoreanbotsServer클래스입니다. - :rtype: - KoreanbotsServer - """ - - data = await super().get_server_info(server_id) - - code = data["code"] - version = data["version"] - data = data["data"] - - return KoreanbotsResponse( - code=code, version=version, data=KoreanbotsServerResponse.from_dict(data) + async def get_heart_ranking_list( + self, page: int = 1 + ) -> KoreanbotsDataResponse[list[Bot]]: + res = await self.request_bot_heart_ranking_list(page) + return KoreanbotsDataResponse.from_list_bot( + code=res["code"], + version=res["version"], + data=res["data"], ) - @strict_literal(["widget_type", "style"]) - async def get_widget( - self, - widget_type: WidgetType, - bot_id: int, - style: WidgetStyle = "flat", - scale: float = 1.0, - icon: bool = False, - ) -> str: - """ - 주어진 bot_id로 widget의 url을 반환합니다. - - :param widget_type: - 요청할 widget의 타입을 지정합니다. - :type widget_type: - WidgetType - - :param bot_id: - 요청할 bot의 ID를 지정합니다. - :type bot_id: - int - - :param style: - 요청할 widget의 형식을 지정합니다. 기본값은 flat로 설정되어 있습니다. - :type style: - WidgetStyle, optional - - :param scale: - 요청할 widget의 크기를 지정합니다. 반드시 0.5이상이어야 합니다. 기본값은 1.0입니다. - :type scale: - float, optional - - :param icon: - 요청할 widget의 아이콘을 표시할지를 지정합니다. 기본값은 False입니다. - :type icon: - bool, optional - - :return: - 위젯 url을 반환합니다. - :rtype: str - """ - return await self.get_bot_widget_url(widget_type, bot_id, style, scale, icon) - - async def get_bot_vote( - self, user_id: int, bot_id: int - ) -> KoreanbotsResponse[KoreanbotsVoteResponse]: - """ - user_id를 통해 주어진 bot_id에 대한 투표 여부를 반환합니다. - - :param user_id: - 요청할 user의 ID를 지정합니다. - :type user_id: - int - - :param bot_id: - 요청할 봇의 ID를 지정합니다. - :type bot_id: - int - - :return: - 투표여부를 담고 있는 KoreanbotsVote클래스입니다. - :rtype: - KoreanbotsVote - """ - data = await super().get_bot_vote(user_id, bot_id) - - code = data["code"] - version = data["version"] - data = data["data"] - - return KoreanbotsResponse( - code=code, version=version, data=KoreanbotsVoteResponse.from_dict(data) + async def get_new_bot_list(self) -> KoreanbotsDataResponse[list[Bot]]: + res = await self.request_new_bot_list() + return KoreanbotsDataResponse.from_list_bot( + code=res["code"], + version=res["version"], + data=res["data"], ) - async def get_server_vote( - self, user_id: int, server_id: int - ) -> KoreanbotsResponse[KoreanbotsVoteResponse]: - """ - user_id를 통해 주어진 server_id에 대한 투표 여부를 반환합니다. - - :param user_id: - 요청할 user의 ID를 지정합니다. - :type user_id: - int - - :param server_id: - 요청할 봇의 ID를 지정합니다. - :type server_id: - int - - :return: - 투표여부를 담고 있는 KoreanbotsVote클래스입니다. - :rtype: - KoreanbotsVote - """ - data = await super().get_server_vote(user_id, server_id) - - code = data["code"] - version = data["version"] - data = data["data"] - - return KoreanbotsResponse( - code=code, version=version, data=KoreanbotsVoteResponse.from_dict(data) + async def get_user_is_voted_bot( + self, bot_id: int, user_id: int + ) -> KoreanbotsDataResponse[Vote]: + res = await self.request_user_is_voted_bot(bot_id, user_id) + return KoreanbotsDataResponse.from_vote( + code=res["code"], + version=res["version"], + data=res["data"], ) - # deprecated since 3.0.0 - - async def guildcount(self, bot_id: int, **kwargs: Optional[int]) -> None: - warn( - "guildcount 메서드는 post_guild_count로 변경되었습니다.", DeprecationWarning + async def update_bot_info( + self, bot_id: int, servers: int, shards: int + ) -> KoreanbotsMessageResponse: + res = await self.request_update_bot_info(bot_id, servers, shards) + return KoreanbotsMessageResponse( + code=res["code"], + version=res["version"], + message=res["message"], ) - return await self.post_guild_count(bot_id, **kwargs) - - async def userinfo( - self, user_id: int - ) -> KoreanbotsResponse[KoreanbotsUserResponse]: - warn("userinfo 메서드는 get_user_info로 변경되었습니다.", DeprecationWarning) - - return await self.get_user_info(user_id) - - async def botinfo(self, bot_id: int) -> KoreanbotsResponse[KoreanbotsBotResponse]: - warn("botinfo 메서드는 get_bot_info로 변경되었습니다.", DeprecationWarning) + async def get_server_info(self, server_id: int) -> KoreanbotsDataResponse[Server]: + res = await self.request_server_info(server_id) + return KoreanbotsDataResponse.from_server( + code=res["code"], + version=res["version"], + data=res["data"], + ) - return await self.get_bot_info(bot_id) + async def search_server( + self, query: str, page: int = 1 + ) -> KoreanbotsDataResponse[list[Server]]: + res = await self.request_search_server(query, page) + return KoreanbotsDataResponse.from_list_server( + code=res["code"], + version=res["version"], + data=res["data"], + ) - async def serverinfo( + async def get_server_administrator( self, server_id: int - ) -> KoreanbotsResponse[KoreanbotsServerResponse]: - warn( - "serverinfo 메서드는 get_server_info로 변경되었습니다.", DeprecationWarning + ) -> KoreanbotsDataResponse[User]: + res = await self.request_server_administrator(server_id) + return KoreanbotsDataResponse.from_user( + code=res["code"], + version=res["version"], + data=res["data"], ) - return await self.get_server_info(server_id) - - @strict_literal(["widget_type", "style"]) - async def widget( - self, - widget_type: WidgetType, - bot_id: int, - style: WidgetStyle = "flat", - scale: float = 1.0, - icon: bool = False, - ) -> str: - warn("widget 메서드는 get_widget으로 변경되었습니다.", DeprecationWarning) - - return await self.get_widget(widget_type, bot_id, style, scale, icon) - - async def is_voted_bot( - self, user_id: int, bot_id: int - ) -> KoreanbotsResponse[KoreanbotsVoteResponse]: - warn("is_voted_bot 메서드는 get_bot_vote로 변경되었습니다.", DeprecationWarning) - - return await self.get_bot_vote(user_id, bot_id) - - async def is_voted_server( - self, user_id: int, server_id: int - ) -> KoreanbotsResponse[KoreanbotsVoteResponse]: - warn( - "is_voted_server 메서드는 get_server_vote로 변경되었습니다.", - DeprecationWarning, + async def get_user_is_voted_server( + self, server_id: int, user_id: int + ) -> KoreanbotsDataResponse[Vote]: + res = await self.request_user_is_voted_server(server_id, user_id) + return KoreanbotsDataResponse.from_vote( + code=res["code"], + version=res["version"], + data=res["data"], ) - return await self.get_server_vote(user_id, server_id) + async def get_user_info(self, user_id: int) -> KoreanbotsDataResponse[User]: + res = await self.request_user_info(user_id) + return KoreanbotsDataResponse.from_user( + code=res["code"], + version=res["version"], + data=res["data"], + ) diff --git a/koreanbots/decorator.py b/koreanbots/decorator.py deleted file mode 100644 index 6941065..0000000 --- a/koreanbots/decorator.py +++ /dev/null @@ -1,42 +0,0 @@ -import functools -import inspect -from typing import Any, Callable, List, Literal, cast, get_args - -from koreanbots.typing import CORO - - -def strict_literal(argument_names: List[str]) -> Callable[[CORO], CORO]: - def decorator(f: CORO) -> CORO: - @functools.wraps(f) - async def decorated_function(*args: Any, **kwargs: Any) -> Any: - # First get about func args - full_arg_spec = inspect.getfullargspec(f) - for argument_name in argument_names: - # Get annotation - arg_annoration = full_arg_spec.annotations[argument_name] - # Check annotation is Lireral - if arg_annoration.__origin__ is Literal: - # Literal -> list - literal_list = list(get_args(arg_annoration)) - # Get index - arg_index = full_arg_spec.args.index(argument_name) - # Handle arguments - if arg_index < len(args) and args[arg_index] not in literal_list: - raise ValueError( - f"Arguments do not match. Expected: {literal_list}" - ) - # Handle keyword arguments - elif ( - kwargs.get(argument_name) - and kwargs[argument_name] not in literal_list - ): - if kwargs[argument_name] not in literal_list: - raise ValueError( - f"Arguments do not match. Expected: {literal_list}" - ) - - return await f(*args, **kwargs) - - return cast(CORO, decorated_function) - - return decorator diff --git a/koreanbots/integrations/__init__.py b/koreanbots/domain/__init__.py similarity index 100% rename from koreanbots/integrations/__init__.py rename to koreanbots/domain/__init__.py diff --git a/koreanbots/domain/abc.py b/koreanbots/domain/abc.py new file mode 100644 index 0000000..21fc043 --- /dev/null +++ b/koreanbots/domain/abc.py @@ -0,0 +1,17 @@ +from abc import ABC +from dataclasses import dataclass + + +from koreanbots.domain.deserializer import Deserializer +from koreanbots.domain.serializer import Serializer + + + +@dataclass +class KoreanbotsEntity(ABC): + pass + + +@dataclass +class SerializableEntity(KoreanbotsEntity, Serializer, Deserializer): + pass diff --git a/koreanbots/domain/bot.py b/koreanbots/domain/bot.py new file mode 100644 index 0000000..153c071 --- /dev/null +++ b/koreanbots/domain/bot.py @@ -0,0 +1,63 @@ +from dataclasses import dataclass +from typing import Literal + +from koreanbots.domain.abc import SerializableEntity + +Category = Literal[ + "관리", + "뮤직", + "전적", + "게임", + "도박", + "로깅", + "빗금 명령어", + "웹 대시보드", + "밈", + "레벨링", + "유틸리티", + "대화", + "NSFW", + "검색", + "학교", + "코로나19", + "번역", + "오버워치", + "리그 오브 레전드", + "배틀그라운드", + "마인크래프트", +] + +Status = Literal["online", "idle", "dnd", "streaming", "offline"] + +State = Literal["ok", "reported", "blocked", "private", "archived"] + + +@dataclass +class AbstractBot(SerializableEntity): + id: str + name: str + tag: str + avatar: str | None + flags: int + lib: str + prefix: str + votes: int + servers: int | None + shards: int | None + intro: str + desc: str + web: str | None + git: str | None + url: str | None + discord: str | None + category: list[Category] + vanity: str | None + bg: str | None + banner: str | None + status: Status | None + state: State + + +@dataclass +class BotWithOwnerID(AbstractBot): + owner: str diff --git a/koreanbots/domain/deserializer.py b/koreanbots/domain/deserializer.py new file mode 100644 index 0000000..2082c5a --- /dev/null +++ b/koreanbots/domain/deserializer.py @@ -0,0 +1,39 @@ +from inspect import isclass +from typing import Any, Mapping, Self, Union, get_args, get_origin, get_type_hints + + +class Deserializer: + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> Self: + converted_data: dict[str, Any] = {} + type_hints = get_type_hints(cls) + + for key, value in data.items(): + type_hint = type_hints.get(key) + + # Check for unexpected keys + if type_hint is None: + raise ValueError(f"Unexpected key: {key}") + + # Handle Generics + if origin_type := get_origin(type_hint): + arg_type = get_args(type_hint)[0] + # Handle optional fields + if origin_type is Union and type(None) in get_args(type_hint): + value = arg_type(value) + # Handle list fields + elif origin_type is list: + if isclass(arg_type) and issubclass(arg_type, Deserializer): + if value is None: + value = [] + else: + value = [arg_type.from_dict(v) for v in value] + elif arg_type is int: + value = [int(v) for v in value] + elif isclass(type_hint) and issubclass(type_hint, Deserializer): + value = type_hint.from_dict(value) + else: + value = type_hint(value) + converted_data[key] = value + + return cls(**converted_data) diff --git a/koreanbots/domain/entities.py b/koreanbots/domain/entities.py new file mode 100644 index 0000000..ebdc90f --- /dev/null +++ b/koreanbots/domain/entities.py @@ -0,0 +1,117 @@ +from dataclasses import dataclass +from typing import Any, Generic, TypeVar + +from koreanbots.domain.abc import SerializableEntity +from koreanbots.domain.bot import AbstractBot, BotWithOwnerID +from koreanbots.domain.server import AbstractServer, ServerWithOwnerID +from koreanbots.domain.user import AbstractUser + +T = TypeVar("T") + + +@dataclass +class User(AbstractUser): + bots: list[BotWithOwnerID] + servers: list[ServerWithOwnerID] + + +@dataclass +class Server(AbstractServer): + owners: User + + +@dataclass +class Bot(AbstractBot): + owners: list[User] + + +@dataclass +class Vote(SerializableEntity): + voted: bool + lastVote: int + + +@dataclass +class KoreanbotsResponse: + code: int + version: int + + +@dataclass +class KoreanbotsMessageResponse(KoreanbotsResponse): + message: str + + +@dataclass +class KoreanbotsDataResponse(KoreanbotsResponse, Generic[T]): + data: T + + @classmethod + def from_bot( + cls, code: int, version: int, data: dict[str, Any] + ) -> "KoreanbotsDataResponse[Bot]": + return KoreanbotsDataResponse( + code=code, + version=version, + data=Bot.from_dict(data), + ) + + @classmethod + def from_list_bot( + cls, code: int, version: int, data: list[dict[str, Any]] + ) -> "KoreanbotsDataResponse[list[Bot]]": + return KoreanbotsDataResponse( + code=code, + version=version, + data=[Bot.from_dict(item) for item in data], + ) + + @classmethod + def from_user( + cls, code: int, version: int, data: dict[str, Any] + ) -> "KoreanbotsDataResponse[User]": + return KoreanbotsDataResponse( + code=code, + version=version, + data=User.from_dict(data), + ) + + @classmethod + def from_list_user( + cls, code: int, version: int, data: list[dict[str, Any]] + ) -> "KoreanbotsDataResponse[list[User]]": + return KoreanbotsDataResponse( + code=code, + version=version, + data=[User.from_dict(item) for item in data], + ) + + @classmethod + def from_server( + cls, code: int, version: int, data: dict[str, Any] + ) -> "KoreanbotsDataResponse[Server]": + return KoreanbotsDataResponse( + code=code, + version=version, + data=Server.from_dict(data), + ) + + @classmethod + def from_list_server( + cls, code: int, version: int, data: list[dict[str, Any]] + ) -> "KoreanbotsDataResponse[list[Server]]": + return KoreanbotsDataResponse( + code=code, + version=version, + data=[Server.from_dict(item) for item in data], + ) + + @classmethod + def from_vote( + cls, code: int, version: int, data: dict[str, Any] + ) -> "KoreanbotsDataResponse[Vote]": + return KoreanbotsDataResponse( + code=code, + version=version, + data=Vote.from_dict(data), + ) diff --git a/koreanbots/domain/serializer.py b/koreanbots/domain/serializer.py new file mode 100644 index 0000000..0ba0104 --- /dev/null +++ b/koreanbots/domain/serializer.py @@ -0,0 +1,7 @@ +from dataclasses import asdict, dataclass + + +@dataclass +class Serializer: + def to_dict(self): + return asdict(self) diff --git a/koreanbots/domain/server.py b/koreanbots/domain/server.py new file mode 100644 index 0000000..8fd6983 --- /dev/null +++ b/koreanbots/domain/server.py @@ -0,0 +1,53 @@ +from dataclasses import dataclass +from typing import Literal + +from koreanbots.domain.abc import SerializableEntity + +Category = Literal[ + "커뮤니티", + "IT & 과학", + "봇", + "친목", + "음악", + "교육", + "연애", + "게임", + "오버워치", + "리그 오브 레전드", + "배틀그라운드", + "마인크래프트", +] + +State = Literal["ok", "reported", "blocked", "unreachable"] + + +@dataclass +class Emoji(SerializableEntity): + id: str + name: str + url: str + + +@dataclass +class AbstractServer(SerializableEntity): + id: str + name: str + icon: str | None + flags: int + votes: int + members: int + boostTier: int + intro: str + desc: str + category: list[Category] + invite: str + emojis: list[Emoji] + state: State + vanity: str | None + bg: str | None + banner: str | None + + +@dataclass +class ServerWithOwnerID(AbstractServer): + owner: str diff --git a/koreanbots/domain/user.py b/koreanbots/domain/user.py new file mode 100644 index 0000000..d66f0e1 --- /dev/null +++ b/koreanbots/domain/user.py @@ -0,0 +1,12 @@ +from dataclasses import dataclass + +from koreanbots.domain.abc import SerializableEntity + + +@dataclass +class AbstractUser(SerializableEntity): + id: str + username: str + tag: str + github: str | None + flags: int diff --git a/koreanbots/errors.py b/koreanbots/errors.py deleted file mode 100644 index d8242d0..0000000 --- a/koreanbots/errors.py +++ /dev/null @@ -1,39 +0,0 @@ -from typing import Any, Dict, Type, Union - - -class KoreanbotsException(Exception): - pass - - -class AuthorizeError(KoreanbotsException): - pass - - -class HTTPException(KoreanbotsException): - def __init__(self, code: Any, message: Union[Any, Dict[str, Any]]): - self.status = code - if isinstance(message, dict): - self.status = message.get("code", self.status) - self.error = message.get("message", "KoreanbotsException") - else: - self.error = message - super().__init__(f"{self.status} {self.error}") - - -class BadRequest(HTTPException): - pass - - -class Forbidden(HTTPException): - pass - - -class NotFound(HTTPException): - pass - - -ERROR_MAPPING: Dict[int, Type[HTTPException]] = { - 400: BadRequest, - 403: Forbidden, - 404: NotFound, -} diff --git a/koreanbots/exception.py b/koreanbots/exception.py new file mode 100644 index 0000000..a1af628 --- /dev/null +++ b/koreanbots/exception.py @@ -0,0 +1,3 @@ +class KoreanbotsException(Exception): + + pass diff --git a/koreanbots/http.py b/koreanbots/http.py deleted file mode 100644 index 34cc853..0000000 --- a/koreanbots/http.py +++ /dev/null @@ -1,302 +0,0 @@ -from asyncio import sleep -from asyncio.events import get_event_loop -from asyncio.locks import Event -from datetime import datetime -from functools import wraps -from logging import getLogger -from typing import Any, Literal, Optional, cast - -import aiohttp - -from .decorator import strict_literal -from .errors import ERROR_MAPPING, AuthorizeError, HTTPException -from .typing import CORO, WidgetStyle, WidgetType - -BASE = "https://koreanbots.dev/api/" -VERSION = "v2" - -KOREANBOTS_URL = BASE + VERSION - -log = getLogger(__name__) - - -def required(f: CORO) -> CORO: - @wraps(f) - async def decorator_function( - self: "KoreanbotsRequester", *args: Any, **kwargs: Any - ) -> Any: - if not self.api_key: - raise AuthorizeError("This endpoint required koreanbots token.") - - return await f(self, *args, **kwargs) - - return cast(CORO, decorator_function) - - -class KoreanbotsRequester: - """ - Koreanbots의 API를 요청하는 클래스입니다. - - :param api_key: - KoreanBots의 토큰입니다. 기본값은 None 입니다. - :type api_key: - Optional[str], optional - - :param session: - aiohttp.ClientSession의 클래스입니다. 전달되지 않으면 생성합니다. 기본값은 None 입니다. - :type session: - Optional[aiohttp.ClientSession], optional - """ - - def __init__( - self, - api_key: Optional[str] = None, - session: Optional[aiohttp.ClientSession] = None, - ) -> None: - self.session = session - self.api_key = api_key - self._global_limit = Event() - self._global_limit.set() - - # How to close the session if discord.Client is not specified. - def __del__(self) -> None: - if self.session: - if not self.session.closed: - loop = get_event_loop() - if loop.is_running(): - loop.create_task(self.session.close()) - else: - loop.run_until_complete(self.session.close()) - @required - async def request( - self, - method: Literal["GET", "POST"], - endpoint: str, - **kwargs: Any, - ) -> Any: - """ - Koreanbots의 url을 기반으로 요청합니다. - 레이트리밋을 핸들합니다. - - :param method: - HTTP 메소드입니다. GET, POST만 사용할 수 있습니다. - :type method: - Literal["GET", "POST"] - :param endpoint: - 요청을 실행할 API 페이지의 주소입니다. - :type endpoint: - str - - :raises NotFound: - 요청할 수 없는 페이지입니다. - :raises BadRequest: - 잘못된 요청입니다. - :raises Forbidden: - 요청을 할 권한이 없습니다. - - :raises HTTPException: - 응답에 오류가 있습니다. - - :return: - 요청 결과를 반환합니다. - :rtype: - Dict[str, Any] - """ - - if not self.session: - self.session = aiohttp.ClientSession(headers={"Authorization": self.api_key}) - - if not self._global_limit.is_set(): - await self._global_limit.wait() - - for _ in range(5): - async with self.session.request( - method, KOREANBOTS_URL + endpoint, **kwargs - ) as response: - remain_limit = response.headers.get("x-ratelimit-remaining") - if ( - remain_limit is not None - and int(remain_limit) == 0 - or response.status == 429 - ): - reset_limit_timestamp = int(response.headers["x-ratelimit-reset"]) - reset_limit = datetime.fromtimestamp(reset_limit_timestamp) - retry_after = reset_limit - datetime.now() - self._global_limit.clear() - await sleep(retry_after.total_seconds()) - self._global_limit.set() - continue - - if response.status != 200: - if ERROR_MAPPING.get(response.status): - raise ERROR_MAPPING[response.status]( - response.status, await response.json() - ) - else: - raise HTTPException(response.status, await response.json()) - return await response.json() - - assert None - - async def get_bot_info(self, bot_id: int) -> Any: - """ - 주어진 bot_id로 bot의 정보를 반환합니다. - - :param bot_id: - 요청할 bot의 ID를 지정합니다. - :type bot_id: - int - - :return: - 요청 결과를 반환합니다. - :rtype: - Dict[str, Any] - """ - return await self.request("GET", f"/bots/{bot_id}") - - - async def post_update_bot_info(self, bot_id: int, **kwargs: Optional[int]) -> Any: - """ - 주어진 bot_id로 bot의 정보를 갱신합니다. - - :param bot_id: - 요청할 bot의 ID를 지정합니다. - :type bot_id: - int - - :param kwargs: - 갱신할 정보를 지정합니다. - 'servers' 인자와 'shards' 인자 이외의 값이 들어갈경우 무시합니다. - :type kwargs: - int - - :raises AuthorizeError: - api_key가 없거나 유효하지 않은 경우。 - - :return: - 요청 결과를 반환합니다. - :rtype: - Dict[str, Any] - """ - - return await self.request( - "POST", - f"/bots/{bot_id}/stats", - json={x: kwargs[x] for x in kwargs if x in ["servers", "shards"]}, - ) - - @strict_literal(["widget_type", "style"]) - async def get_bot_widget_url( - self, - widget_type: WidgetType, - bot_id: int, - style: WidgetStyle = "flat", - scale: float = 1.0, - icon: bool = False, - ) -> str: - """ - 주어진 bot_id로 widget의 url을 반환합니다. - - :param widget_type: - 요청할 widget의 타입을 지정합니다. - :type widget_type: - WidgetType - - :param bot_id: - 요청할 bot의 ID를 지정합니다. - :type bot_id: - int - - :param style: - 요청할 widget의 형식을 지정합니다. 기본값은 flat로 설정되어 있습니다. - :type style: - WidgetStyle, optional - - :param scale: - 요청할 widget의 크기를 지정합니다. 반드시 0.5이상이어야 합니다. 기본값은 1.0입니다. - :type scale: - float, optional - - :param icon: - 요청할 widget의 아이콘을 표시할지를 지정합니다. 기본값은 False입니다. - :type icon: - bool, optional - - :return: - 위젯 url을 반환합니다. - :rtype: str - """ - if scale < 0.5: - raise ValueError(f"scale must be greater than to 0.5, not {scale}") - - return ( - KOREANBOTS_URL - + f"/widget/bots/{widget_type}/{bot_id}.svg?style={style}&scale={scale}&icon={icon}" - ) - - async def get_user_info(self, user_id: int) -> Any: - """ - 주어진 user_id로 user의 정보를 반환합니다. - - :param user_id: - 요청할 user의 ID를 지정합니다. - :type user_id: - int - """ - return await self.request("GET", f"/users/{user_id}") - - - async def get_bot_vote(self, user_id: int, bot_id: int) -> Any: - """ - 주어진 bot_id로 user_id를 통해 해당 user의 투표 여부를 반환합니다. - - :param user_id: - 요청할 user의 ID를 지정합니다. - :type user_id: - int - - :param bot_id: - 요청할 bot의 ID를 지정합니다. - :type bot_id: - int - - """ - return await self.request( - "GET", - f"/bots/{bot_id}/vote", - params={"userID": user_id}, - ) - - async def get_server_info(self, server_id: int) -> Any: - """ - 주어진 server_id로 server의 정보를 반환합니다. - - :param server_id: - 요청할 server의 ID를 지정합니다. - :type server_id: - int - - """ - return await self.request("GET", f"/servers/{server_id}") - - - async def get_server_vote(self, user_id: int, server_id: int) -> Any: - """ - 주어진 server_id로 user_id를 통해 해당 user의 투표 여부를 반환합니다. - - :param user_id: - 요청할 user의 ID를 지정합니다. - :type user_id: - int - - :param server_id로: - 요청할 server의 ID를 지정합니다. - :type server_id: - int - - """ - return await self.request( - "GET", - f"/servers/{server_id}/vote", - params={"userID": user_id}, - ) diff --git a/koreanbots/integrations/dico.py b/koreanbots/integrations/dico.py deleted file mode 100644 index 3c4f0ff..0000000 --- a/koreanbots/integrations/dico.py +++ /dev/null @@ -1,105 +0,0 @@ -import logging -from asyncio.events import get_event_loop -from asyncio.tasks import sleep -from typing import Optional - -from aiohttp import ClientSession - -try: - from dico import Client # type: ignore -except ImportError: - pass - -from koreanbots.client import Koreanbots - -log = logging.getLogger(__name__) - - -class DicoKoreanbots(Koreanbots): - """ - KoreanbotsRequester를 감싸는 클라이언트 클래스입니다. - dico 전용입니다. - - :param client: - dico.Client의 클래스입니다. - :type client: - dico.Client - - :param api_key: - API key를 지정합니다. - :type api_key: - str - - :param session: - aiohttp.ClientSession의 클래스입니다. 만약 필요한 경우 이 인수를 지정하세요. 지정하지 않으면 생성합니다. - :type session: - Optional[aiohttp.ClientSession] - - :param run_task: - 봇 정보를 갱신하는 작업을 자동으로 실행합니다. 만약 아니라면 지정하지 않습니다. - :type run_task: - bool - - :param include_shard_count: - 샤드 개수를 포함할지 지정합니다. 만약 아니라면 지정하지 않습니다. - :type include_shard_count: - bool - """ - - def __init__( - self, - client: "Client", - api_key: str, - session: Optional[ClientSession] = None, - run_task: bool = False, - include_shard_count: bool = False, - ): - self.client = client - - if client: - original_close = getattr(client, "close") - - async def close() -> None: - if self.session is not None and not self.session.closed: - await self.session.close() - await original_close() - - setattr(client, "close", close) - - self.include_shard_count = include_shard_count - super().__init__(api_key, session) - - if run_task: - get_event_loop().create_task(self.tasks_send_guildcount()) - - async def tasks_send_guildcount(self) -> None: - """ - 길드 개수를 서버에 전송하는 태스크 입니다. - - :raises RuntimeError: - 클라이언트를 찾을 수 없습니다. - """ - - if not self.client: - raise RuntimeError("Client Not Found") - - await self.client.wait_ready() - - while not self.client.websocket_closed: - if not self.client.application_id: - continue - - kwargs = {"servers": self.client.guild_count} - if self.include_shard_count: - if self.client.shard_count: - kwargs.update({"shards": self.client.shard_count}) - log.info("Initiating guild count update...") - try: - await self.post_guild_count(int(self.client.application_id), **kwargs) - except: - log.exception("Guild count update failed due to an error.") - else: - log.info( - "Guild count updated successfully. Waiting 30 minutes for the next update." - ) - await sleep(1800) diff --git a/koreanbots/integrations/discord.py b/koreanbots/integrations/discord.py deleted file mode 100644 index 3b36e5b..0000000 --- a/koreanbots/integrations/discord.py +++ /dev/null @@ -1,159 +0,0 @@ -from asyncio.tasks import Task, sleep -from logging import getLogger -from typing import ( - TYPE_CHECKING, - Any, - Callable, - Coroutine, - Optional, - TypeVar, - Union, - cast, -) - -from aiohttp import ClientSession - -from koreanbots.client import Koreanbots - -if TYPE_CHECKING: - import nextcord - from discord import Client as DiscordpyClient - from disnake.client import Client as DisnakeClient - -T = TypeVar("T") -Coro = Coroutine[Any, Any, T] -CoroT = TypeVar("CoroT", bound=Callable[..., Coro[Any]]) - -log = getLogger(__name__) - - -class DiscordpyKoreanbots(Koreanbots): - """ - Koreanbots를 감싸는 클라이언트 클래스입니다. - discord.py 및 해당 라이브러리의 포크 전용입니다. - - :param client: - discord.Client의 클래스입니다. 만약 필요한 경우 이 인수를 지정하세요. - :type client: - Optional[DpyABC] - - :param api_key: - API key를 지정합니다. 만약 필요한 경우 이 키를 지정하세요. - :type api_key: - Optional[str] - - :param session: - aiohttp.ClientSession의 클래스입니다. 만약 필요한 경우 이 인수를 지정하세요. 지정하지 않으면 생성합니다. - :type session: - Optional[aiohttp.ClientSession] - - :param run_task: - 봇 정보를 갱신하는 작업을 자동으로 실행합니다. 만약 아니라면 지정하지 않습니다. - :type run_task: - bool - - :param include_shard_count: - 샤드 개수를 포함할지 지정합니다. 만약 아니라면 지정하지 않습니다. - :type include_shard_count: - bool - """ - - def __init__( - self, - client: Union["DiscordpyClient", "nextcord.Client", "DisnakeClient"], - api_key: str, - session: Optional[ClientSession] = None, - run_task: bool = False, - include_shard_count: bool = False, - ): - self.client = client - - # Patch discord.py client.close() method to handle session.close() - original_close = getattr(client, "close") - - async def close() -> None: - if self.session is not None and not self.session.closed: - await self.session.close() - await original_close() - - setattr(client, "close", close) - - self.include_shard_count = include_shard_count - super().__init__(api_key, session) - - if run_task: - client_ready = getattr(client, "on_ready", None) - client_event = getattr(client, "event") - self.guildcount_sender: Optional[Task[None]] = None - - # Set default on_ready handler to start send_guildcount task. - if client_ready is not None: - - async def on_ready() -> None: - self.run_post_guild_count_task() - await client_ready() # call previously registered on_ready handler. - - else: - - async def on_ready() -> None: - self.run_post_guild_count_task() - - def event(coro: CoroT, /) -> CoroT: - if coro.__name__ == "on_ready": - - async def on_ready() -> None: - self.run_post_guild_count_task() - await coro() - - return cast(CoroT, client_event(on_ready)) - - return cast(CoroT, client_event(coro)) - - client.event(on_ready) - setattr(client, "event", event) - - @property - def is_running(self) -> bool: - return self.guildcount_sender is not None and not self.guildcount_sender.done() - - def run_post_guild_count_task(self) -> None: - """ - tasks_send_guildcount를 호출하는 함수입니다. - 사용자가 on_ready 이벤트 핸들러를 정의할 때, 이 함수를 호출해 길드 개수를 지속적으로 갱신할 수 있습니다. - """ - if not self.is_running: - self.guildcount_sender = self.client.loop.create_task( - self.tasks_send_guildcount() - ) - - async def tasks_send_guildcount(self) -> None: - """ - 길드 개수를 서버에 전송하는 태스크 입니다. - - :raises RuntimeError: - 클라이언트를 찾을 수 없습니다. - """ - - if not self.client: - raise RuntimeError("Client Not Found") - - await self.client.wait_until_ready() - - while not self.client.is_closed(): - if not self.client.user: - continue - - kwargs = {"servers": len(self.client.guilds)} - if self.include_shard_count: - if self.client.shard_count: - kwargs.update({"shards": self.client.shard_count}) - log.info("Initiating guild count update...") - try: - await self.post_guild_count(int(self.client.user.id), **kwargs) - except: - log.exception("Guild count update failed due to an error.") - else: - log.info( - "Guild count updated successfully. Waiting 30 minutes for the next update." - ) - await sleep(1800) diff --git a/koreanbots/model.py b/koreanbots/model.py deleted file mode 100644 index f0234fb..0000000 --- a/koreanbots/model.py +++ /dev/null @@ -1,357 +0,0 @@ -from abc import ABC -from dataclasses import dataclass, field -from typing import Any, Dict, Generic, List, Optional, TypeVar - -from .typing import Category, State, Status - - -class KoreanbotsResponseABC(ABC): ... - - -T = TypeVar("T", bound=KoreanbotsResponseABC) - - -@dataclass(frozen=True) -class KoreanbotsResponse(Generic[T]): - code: int - """상태 코드""" - version: int - """버전""" - data: T - - -@dataclass(eq=True, frozen=True) -class KoreanbotsBot(KoreanbotsResponseABC): - """ - 봇의 정보를 가져왔을때 반환되는 인스턴스입니다. - """ - - id: Optional[str] = field(repr=True, compare=True, default=None) - """아이디""" - name: Optional[str] = field(repr=True, compare=False, default=None) - """이름""" - tag: Optional[str] = field(repr=False, compare=False, default=None) - """태그""" - avatar: Optional[str] = field(repr=False, compare=False, default=None) - """아바타""" - flags: int = field(repr=False, compare=False, default=0) - """플래그""" - lib: Optional[str] = field(repr=False, compare=False, default=None) - """라이브러리""" - prefix: Optional[str] = field(repr=False, compare=False, default=None) - """프리픽스""" - votes: int = field(repr=False, compare=False, default=0) - """투표 수""" - servers: int = field(repr=False, compare=False, default=0) - """서버 수""" - shards: int = field(repr=False, compare=False, default=0) - """샤드 수""" - intro: Optional[str] = field(repr=False, compare=False, default=None) - """소개 문구""" - desc: Optional[str] = field(repr=False, compare=False, default=None) - """설명 문구""" - web: Optional[str] = field(repr=False, compare=False, default=None) - """웹사이트 주소""" - git: Optional[str] = field(repr=False, compare=False, default=None) - """깃 주소""" - url: Optional[str] = field(repr=False, compare=False, default=None) - """주소""" - discord: Optional[str] = field(repr=False, compare=False, default=None) - """디스코드 주소""" - category: Optional[Category] = field(repr=False, compare=False, default=None) - """카테고리""" - vanity: Optional[str] = field(repr=False, compare=False, default=None) - """가상 주소""" - bg: Optional[str] = field(repr=False, compare=False, default=None) - """배경 이미지 주소""" - banner: Optional[str] = field(repr=False, compare=False, default=None) - """배너 이미지 주소""" - status: Optional[Status] = field(repr=False, compare=False, default=None) - """상태""" - state: Optional[State] = field(repr=False, compare=False, default=None) - """Koreanbots에서의 상태""" - - -@dataclass(eq=True, frozen=True) -class KoreanbotsUser(KoreanbotsResponseABC): - """ - 유저 정보를 가져왔을때 반환되는 클래스입니다. - """ - - id: int = field(repr=True, compare=True, default=0) - """아이디""" - username: str = field(repr=True, compare=False, default="") - """유저 이름""" - globalName: str = field(repr=True, compare=False, default="") - """유저 글로벌 이름""" - tag: str = field(repr=False, compare=False, default="") - """태그""" - github: Optional[str] = field(repr=False, compare=False, default=None) - """Github 주소""" - flags: int = field(repr=False, compare=False, default=0) - """플래그""" - - -@dataclass(eq=True, frozen=True) -class KoreanbotsServer(KoreanbotsResponseABC): - """ - 서버 정보를 가져왔을때 반환되는 클래스입니다. - """ - - id: int = field(repr=True, compare=True, default=0) - """ID""" - name: str = field(repr=True, compare=False, default="") - """서버 이름""" - flags: int = field(repr=False, compare=False, default=0) - """플래그""" - intro: Optional[str] = field(repr=False, compare=False, default=None) - """소개문구""" - desc: Optional[str] = field(repr=False, compare=False, default=None) - """설명문구""" - votes: int = field(repr=True, compare=False, default=0) - """투표수""" - category: Optional[Category] = field(repr=False, compare=False, default=None) - """카테고리""" - invite: str = field(repr=False, compare=False, default="") - """초대링크""" - state: Optional[State] = field(repr=False, compare=False, default=None) - """Koreanbots에서의 상태""" - vanity: Optional[str] = field(repr=False, compare=False, default=None) - """서버의 가상 주소""" - bg: Optional[str] = field(repr=False, compare=False, default=None) - """배경 이미지 주소""" - banner: Optional[str] = field(repr=False, compare=False, default=None) - """배너 이미지 주소""" - icon: Optional[str] = field(repr=False, compare=False, default=None) - """아이콘""" - members: int = field(repr=False, compare=False, default=0) - """멤버 수""" - emojis: List["Emoji"] = field(repr=False, compare=False, default_factory=list) - """Emoji 인스턴스를 담고 있는 리스트""" - boostTier: int = field(repr=False, compare=False, default=0) - """부스트 레벨""" - - @classmethod - def from_dict(cls, data: Dict[str, Any]) -> "KoreanbotsServer": - return cls( - id=data["id"], - name=data["name"], - flags=data["flags"], - intro=data.get("intro"), - desc=data.get("desc"), - votes=data["votes"], - category=data.get("category"), - invite=data["invite"], - state=data.get("state"), - vanity=data.get("vanity"), - bg=data.get("bg"), - banner=data.get("banner"), - icon=data.get("icon"), - members=data["members"], - emojis=[ - Emoji(id=e["id"], name=e["name"], url=e["url"]) for e in data["emojis"] - ], - boostTier=data["boostTier"], - ) - - -@dataclass(eq=True, frozen=True) -class CircularKoreanbotsBot(KoreanbotsBot): - owners: List[str] = field(repr=False, compare=False, default_factory=list) - - -@dataclass(eq=True, frozen=True) -class CircularKoreanbotsUser(KoreanbotsUser): - bots: List[str] = field(repr=False, compare=False, default_factory=list) - servers: List[str] = field(repr=False, compare=False, default_factory=list) - - -@dataclass(eq=True, frozen=True) -class CircularKoreanbotsServer(KoreanbotsServer): - owner: str = field(repr=False, compare=False, default="") - - -@dataclass(eq=True, frozen=True) -class KoreanbotsUserResponse(KoreanbotsUser): - bots: List["CircularKoreanbotsBot"] = field( - repr=False, compare=False, default_factory=list - ) - servers: List[CircularKoreanbotsServer] = field( - repr=False, compare=False, default_factory=list - ) - - @classmethod - def from_dict(cls, data: Dict[str, Any]) -> "KoreanbotsUserResponse": - return cls( - id=data["id"], - username=data["username"], - globalName=data["globalName"], - tag=data["tag"], - github=data.get("github"), - flags=data["flags"], - servers=[ - CircularKoreanbotsServer( - id=s["id"], - name=s["name"], - flags=s["flags"], - intro=s.get("intro"), - desc=s.get("desc"), - votes=s["votes"], - category=s.get("category"), - invite=s["invite"], - state=s.get("state"), - vanity=s.get("vanity"), - bg=s.get("bg"), - banner=s.get("banner"), - icon=s.get("icon"), - members=s["members"], - emojis=[ - Emoji(id=e["id"], name=e["name"], url=e["url"]) - for e in s["emojis"] - ], - boostTier=s["boostTier"], - owner=s["owner"], - ) - for s in data["servers"] - ], - bots=[ - CircularKoreanbotsBot( - id=b["id"], - name=b["name"], - tag=b["tag"], - avatar=b["avatar"], - flags=b["flags"], - lib=b["lib"], - prefix=b["prefix"], - votes=b["votes"], - servers=b["servers"], - shards=b["shards"], - intro=b.get("intro"), - desc=b.get("desc"), - web=b.get("web"), - git=b.get("git"), - url=b.get("url"), - discord=b.get("discord"), - category=b.get("category"), - vanity=b.get("vanity"), - bg=b.get("bg"), - banner=b.get("banner"), - status=b.get("status"), - state=b.get("state"), - owners=b["owners"], - ) - for b in data["bots"] - ], - ) - - -@dataclass(eq=True, frozen=True) -class KoreanbotsBotResponse(KoreanbotsBot): - owners: List["CircularKoreanbotsUser"] = field( - repr=False, compare=False, default_factory=list - ) - - @classmethod - def from_dict(cls, data: Dict[str, Any]) -> "KoreanbotsBotResponse": - return cls( - id=data["id"], - name=data["name"], - tag=data["tag"], - avatar=data["avatar"], - flags=data["flags"], - lib=data["lib"], - prefix=data["prefix"], - votes=data["votes"], - servers=data["servers"], - shards=data["shards"], - intro=data.get("intro"), - desc=data.get("desc"), - web=data.get("web"), - git=data.get("git"), - url=data.get("url"), - discord=data.get("discord"), - category=data.get("category"), - vanity=data.get("vanity"), - bg=data.get("bg"), - banner=data.get("banner"), - status=data.get("status"), - state=data.get("state"), - owners=[ - CircularKoreanbotsUser( - id=o["id"], - username=o["username"], - globalName=o["globalName"], - tag=o["tag"], - github=o.get("github"), - flags=o["flags"], - bots=o["bots"], - ) - for o in data["owners"] - ], - ) - - -@dataclass(eq=True, frozen=True) -class KoreanbotsServerResponse(KoreanbotsServer): - owner: "CircularKoreanbotsUser" = field( - repr=False, compare=False, default=CircularKoreanbotsUser() - ) - - @classmethod - def from_dict(cls, data: Dict[str, Any]) -> "KoreanbotsServerResponse": - return cls( - id=data["id"], - name=data["name"], - flags=data["flags"], - intro=data.get("intro"), - desc=data.get("desc"), - votes=data["votes"], - category=data.get("category"), - invite=data["invite"], - state=data.get("state"), - vanity=data.get("vanity"), - bg=data.get("bg"), - banner=data.get("banner"), - icon=data.get("icon"), - members=data["members"], - emojis=[ - Emoji(id=e["id"], name=e["name"], url=e["url"]) for e in data["emojis"] - ], - boostTier=data["boostTier"], - owner=CircularKoreanbotsUser( - id=data["owner"]["id"], - username=data["owner"]["username"], - globalName=data["owner"]["globalName"], - tag=data["owner"]["tag"], - github=data["owner"].get("github"), - flags=data["owner"]["flags"], - bots=data["owner"]["bots"], - servers=data["owner"]["servers"], - ), - ) - - -@dataclass(eq=True, frozen=True) -class Emoji: - """ - 이모지 정보를 가져왔을때 반환되는 클래스입니다. - """ - - id: int = field(repr=True, compare=True, default=0) - """ID""" - name: str = field(repr=True, compare=False, default="") - """이모지 이름""" - url: str = field(repr=False, compare=False, default="") - """이모지 url""" - - -@dataclass(eq=True, frozen=True) -class KoreanbotsVoteResponse(KoreanbotsResponseABC): - voted: bool = field(repr=True, compare=True, default=False) - """투표 여부""" - last_vote: int = field(repr=True, compare=True, default=0) - """마지막으로 투표한 일자""" - - @classmethod - def from_dict(cls, data: Dict[str, Any]) -> "KoreanbotsVoteResponse": - return cls(voted=data["voted"], last_vote=data["lastVote"]) diff --git a/koreanbots/py.typed b/koreanbots/py.typed deleted file mode 100644 index 8b13789..0000000 --- a/koreanbots/py.typed +++ /dev/null @@ -1 +0,0 @@ - diff --git a/koreanbots/request.py b/koreanbots/request.py new file mode 100644 index 0000000..6fe9637 --- /dev/null +++ b/koreanbots/request.py @@ -0,0 +1,185 @@ +import asyncio +import logging +import time +from dataclasses import dataclass +from typing import Any, Literal + +from aiohttp import ClientResponse, ClientSession +from yarl import URL + +from koreanbots import __version__ +from koreanbots.exception import KoreanbotsException + +logger = logging.getLogger(__name__) + + +@dataclass +class RateLimitInfo: + """레이트리밋 정보를 담는 데이터클래스""" + + limit: int + remaining: int + reset: int + is_global: bool + + @classmethod + def from_headers(cls, headers: dict[str, str]) -> "RateLimitInfo": + return cls( + limit=int(headers["x-ratelimit-limit"]), + remaining=int(headers["x-ratelimit-remaining"]), + reset=int(headers["x-ratelimit-reset"]), + is_global=headers["x-ratelimit-global"].lower() == "true", + ) + + def get_wait_time(self) -> float: + current_time = int(time.time()) + if self.reset > current_time: + return max(0, self.reset - current_time) + else: + return max(0, self.reset) + + +class KoreanbotsRequester: + BASE = "https://koreanbots.dev/api/" + VERSION = "v2" + + KOREANBOTS_URL = URL(BASE + VERSION) + + def __init__( + self, + api_key: str, + session: ClientSession | None = None, + ) -> None: + self.api_key = api_key + self.session = session + # Set the Authorization header if not already set + if self.session is not None: + if self.session.headers.get("Authorization") is None: + self.session.headers["Authorization"] = self.api_key + self.session.headers["User-Agent"] = f"Koreanbots py-sdk/{__version__}" + self.session.headers["Content-Type"] = "application/json" + + self._lock = asyncio.Lock() + self._global_rate_limit_reset: float = 0.0 + + async def _handle_rate_limit( + self, + response: ClientResponse, + method: Literal["GET", "POST"], + path: str, + params: dict[str, Any] | None, + ) -> dict[str, Any]: + """레이트리밋을 처리하는 내부 메서드""" + rate_limit = RateLimitInfo.from_headers(dict(response.headers)) + + if rate_limit.is_global: + async with self._lock: + wait_time = rate_limit.get_wait_time() + self._global_rate_limit_reset = time.time() + wait_time + + logger.warning( + f"Waiting {wait_time:.2f} seconds for global rate limit reset" + ) + + await asyncio.sleep(wait_time) + else: + wait_time = rate_limit.get_wait_time() + + logger.warning( + f"Waiting {wait_time:.2f} seconds for route-specific rate limit reset" + ) + + await asyncio.sleep(wait_time) + + return await self.request(method, path, params) + + async def _check_global_rate_limit(self) -> None: + current_time = time.time() + if current_time < self._global_rate_limit_reset: + wait_time = self._global_rate_limit_reset - current_time + + logger.warning( + f"Preemptively waiting {wait_time:.2f} seconds for global rate limit" + ) + + await asyncio.sleep(wait_time) + + async def request( + self, + method: Literal["GET", "POST"], + path: str, + params: dict[str, Any] | None = None, + ) -> dict[str, Any]: + # Create a new session if one doesn't exist + if self.session is None: + self.session = ClientSession( + headers={ + "Authorization": self.api_key, + "Content-Type": "application/json", + "User-Agent": f"Koreanbots py-sdk/{__version__}", + } + ) + + await self._check_global_rate_limit() + + url = self.KOREANBOTS_URL.with_path(path) + + # Append query parameters for GET requests + if method == "GET" and params is not None: + url = url.with_query(params) + # Clear the params for the GET request + params = None + + async with self.session.request(method, url, json=params) as response: + if response.status == 429: + return await self._handle_rate_limit(response, method, path, params) + + if response.status != 200: + raise KoreanbotsException(f"HTTP Error: {response.status}") + + return await response.json() + + async def request_bot_info(self, bot_id: int) -> dict[str, Any]: + return await self.request("GET", f"/bots/{bot_id}") + + async def request_search_bot(self, query: str, page: int = 1) -> dict[str, Any]: + return await self.request("GET", "/bots/search", {"query": query, "page": page}) + + async def request_bot_heart_ranking_list(self, page: int = 1) -> dict[str, Any]: + return await self.request("GET", "/list/bots/votes", {"page": page}) + + async def request_new_bot_list(self) -> dict[str, Any]: + return await self.request("GET", "/list/bots/new") + + async def request_user_is_voted_bot( + self, bot_id: int, user_id: int + ) -> dict[str, Any]: + return await self.request("GET", f"/bots/{bot_id}/vote", {"userID": user_id}) + + async def request_update_bot_info( + self, bot_id: int, servers: int, shards: int + ) -> dict[str, Any]: + return await self.request( + "POST", f"/bots/{bot_id}/stats", {"servers": servers, "shards": shards} + ) + + async def request_server_info(self, server_id: int) -> dict[str, Any]: + return await self.request("GET", f"/servers/{server_id}") + + async def request_search_server(self, query: str, page: int = 1) -> dict[str, Any]: + return await self.request( + "GET", "/servers/search", {"query": query, "page": page} + ) + + async def request_server_administrator(self, server_id: int) -> dict[str, Any]: + return await self.request("GET", f"/servers/{server_id}/owners") + + async def request_user_is_voted_server( + self, server_id: int, user_id: int + ) -> dict[str, Any]: + return await self.request( + "GET", f"/servers/{server_id}/vote", {"userID": user_id} + ) + + async def request_user_info(self, user_id: int) -> dict[str, Any]: + return await self.request("GET", f"/users/{user_id}") diff --git a/koreanbots/typing.py b/koreanbots/typing.py deleted file mode 100644 index 84ad4a7..0000000 --- a/koreanbots/typing.py +++ /dev/null @@ -1,49 +0,0 @@ -from typing import Any, Callable, Coroutine, Literal, TypeVar - -CORO = TypeVar("CORO", bound=Callable[..., Coroutine[Any, Any, Any]]) - -WidgetType = Literal["votes", "servers", "status"] - -WidgetStyle = Literal["classic", "flat"] - -VoteType = Literal["bot", "server"] - -Category = Literal[ - "관리", - "뮤직", - "전적", - "게임", - "도박", - "로깅", - "빗금 명령어", - "웹 대시보드", - "밈", - "레벨링", - "유틸리티", - "대화", - "NSFW", - "검색", - "학교", - "코로나19", - "번역", - "오버워치", - "리그 오브 레전드", - "배틀그라운드", - "마인크래프트", -] - -Status = Literal[ - "online", - "idle", - "dnd", - "streaming", - "offline", -] - -State = Literal[ - "ok", - "reported", - "blocked", - "private", - "archived", -] diff --git a/poetry.toml b/poetry.toml new file mode 100644 index 0000000..7c2fca5 --- /dev/null +++ b/poetry.toml @@ -0,0 +1,3 @@ +[virtualenvs] +path = ".venv" +in-project = true \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..c5bfc77 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,35 @@ +[project] +name = "py-sdk" +version = "4.0.0" +description = "Official SDK for Koreanbots." +authors = [ + {name = "Koreanbots"} +] +license = {text = "MIT"} +readme = "README.md" +requires-python = ">=3.11" +dependencies = [ + "aiohttp (>=3.12.15,<4.0.0)" +] +classifiers = [ + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", +] + +[project.urls] +homepage = "https://koreanbots.dev/" +source = "https://github.com/koreanbots/py-sdk" +tracker = "https://github.com/koreanbots/py-sdk/issues" + +[tool.poetry] +packages = [ + {include = "koreanbots"}, +] + + + +[build-system] +requires = ["poetry-core>=2.0.0,<3.0.0"] +build-backend = "poetry.core.masonry.api" diff --git a/setup.py b/setup.py deleted file mode 100644 index 11888d0..0000000 --- a/setup.py +++ /dev/null @@ -1,47 +0,0 @@ -# -*- coding: utf-8 -*- - -import os - -from setuptools import setup - -import koreanbots - -version = koreanbots.__version__ - -path = os.path.dirname(os.path.realpath(__file__)).replace("\\", "/") - -requirements = [] -with open(f"{path}/requirements.txt", encoding="UTF8") as f: - requirements = f.read().splitlines() - -if not version: - raise RuntimeError("version is not defined") - -readme = "" -with open(f"{path}/README.md", encoding="UTF8") as f: - readme = f.read() - -setup( - name="koreanbots", - author="Koreanbots", - url="https://github.com/koreanbots/py-sdk", - project_urls={ - "Homepage": "https://koreanbots.dev/", - "Source": "https://github.com/koreanbots/py-sdk", - "Tracker": "https://github.com/koreanbots/py-sdk/issues", - }, - version=version, - packages=["koreanbots", "koreanbots.integrations"], - license="MIT", - description="Official SDK for Koreanbots.", - long_description=readme, - long_description_content_type="text/markdown", - include_package_data=True, - install_requires=requirements, - python_requires=">=3.8", - package_data={"koreanbots": ["py.typed"]}, - classifiers=[ - "Programming Language :: Python :: 3.8", - "Programming Language :: Python :: 3.9", - ], -)