mirror of
https://github.com/MrSedan/neuro-reply-bot-reworked.git
synced 2026-01-15 05:59:43 +03:00
Added comments
This commit is contained in:
@@ -1 +1,2 @@
|
||||
"""Main neuroapi package"""
|
||||
from ._neuroapi import neuroapi
|
||||
|
||||
@@ -1,18 +1,31 @@
|
||||
from aiohttp import ClientSession
|
||||
|
||||
from .api_method import ApiMethod
|
||||
|
||||
from neuroapi.types import Admin as AdminType
|
||||
|
||||
from .api_method import ApiMethod
|
||||
|
||||
|
||||
class Admin(ApiMethod):
|
||||
|
||||
"""Class for admin methods"""
|
||||
async def get(self):
|
||||
"""
|
||||
Asynchronous function to retrieve data from the specified API endpoint and return a list of admins.
|
||||
:return List[Admin]
|
||||
"""
|
||||
async with ClientSession() as session:
|
||||
response = await session.get(self.api_url+'/admin/get')
|
||||
return [AdminType.from_dict(admin) for admin in await response.json()]
|
||||
|
||||
async def is_admin(self, id: str):
|
||||
"""
|
||||
Asynchronous function to check if the user with the given ID is an admin.
|
||||
|
||||
Args:
|
||||
id (str): The ID of the user to be checked.
|
||||
|
||||
Returns:
|
||||
bool: True if the user is an admin, False otherwise.
|
||||
"""
|
||||
async with ClientSession() as session:
|
||||
response = await session.get(self.api_url+f'/admin/is-admin/{id}')
|
||||
if await response.text() == 'false':
|
||||
|
||||
@@ -4,4 +4,5 @@ from ..config import GlobalConfig as Config
|
||||
|
||||
|
||||
class ApiMethod(BaseModel):
|
||||
"""Base class for API methods"""
|
||||
api_url: str = Field(Config().api_url)
|
||||
|
||||
@@ -6,13 +6,29 @@ from .api_method import ApiMethod
|
||||
|
||||
|
||||
class BotSettings(ApiMethod):
|
||||
"""Class for bot settings API methods"""
|
||||
async def get(self)-> BotSettingsType:
|
||||
"""
|
||||
Asynchronous function that retrieves bot settings from the API.
|
||||
|
||||
Returns:
|
||||
BotSettings: The bot settings retrieved from the API.
|
||||
"""
|
||||
async with ClientSession() as session:
|
||||
response = await session.get(self.api_url+'/settings')
|
||||
settings = BotSettingsType.from_dict(await response.json())
|
||||
return settings
|
||||
|
||||
async def get_update(self) -> BotSettingsType:
|
||||
"""
|
||||
Asynchronously gets and returns the bot settings from the specified API URL. Clearing server cache.
|
||||
|
||||
Parameters:
|
||||
self: The instance of the class.
|
||||
|
||||
Returns:
|
||||
BotSettings: The bot settings obtained from the API.
|
||||
"""
|
||||
async with ClientSession() as session:
|
||||
response = await session.get(self.api_url+'/settings/active')
|
||||
settings = BotSettingsType.from_dict(await response.json())
|
||||
|
||||
@@ -1 +1,2 @@
|
||||
"""Enums package"""
|
||||
from .get_all import EGetAll
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
from enum import Enum
|
||||
|
||||
|
||||
class EGetAll(Enum):
|
||||
class EGetAll(Enum):
|
||||
all = 'all'
|
||||
will_post = 'will-post'
|
||||
posted = 'posted'
|
||||
|
||||
@@ -9,7 +9,24 @@ from .api_method import ApiMethod
|
||||
|
||||
|
||||
class Image(ApiMethod):
|
||||
"""Class for Image API methods"""
|
||||
async def add(self, from_id: str, file_id: str, has_spoiler: bool | None, message_id: int, text: str, media_group_id: str | None, message_entities: Optional[List[MessageEntity]], message: types.Message):
|
||||
"""
|
||||
An asynchronous function to add an image to post, along with its metadata, to a specific API endpoint. Also, creates a new post.
|
||||
|
||||
Args:
|
||||
from_id (str): The ID of the user who sent the image.
|
||||
file_id (str): The ID of the file containing the image.
|
||||
has_spoiler (bool | None): A boolean indicating whether the image has spoiler content.
|
||||
message_id (int): The ID of the message containing the image.
|
||||
text (str): The text associated with the image.
|
||||
media_group_id (str | None): The ID of the media group containing the image, if applicable.
|
||||
message_entities (Optional[List[MessageEntity]]): A list of message entities associated with the image.
|
||||
message (types.Message): The message object associated with the image.
|
||||
|
||||
Returns:
|
||||
None
|
||||
"""
|
||||
payload = {'from_user_id': from_id, 'file_id': file_id,
|
||||
'has_spoiler': has_spoiler, 'message_id': message_id }
|
||||
if text != '':
|
||||
|
||||
@@ -12,8 +12,20 @@ from .enums import EGetAll
|
||||
|
||||
|
||||
class Post(ApiMethod):
|
||||
|
||||
"""Class for Post API methods"""
|
||||
async def new(self, text: str, from_user_id: str, media_group_id: str = "None", message_entities: Optional[List[MessageEntity]] = None):
|
||||
"""
|
||||
Asynchronously creates a new post with the given text, from_user_id, media_group_id, and message_entities.
|
||||
|
||||
Args:
|
||||
text (str): The text of the post.
|
||||
from_user_id (str): The ID of the user creating the post.
|
||||
media_group_id (str, optional): The media group ID. Defaults to "None".
|
||||
message_entities (List[MessageEntity], optional): List of message entities. Defaults to None.
|
||||
|
||||
Returns:
|
||||
Post: A new post created from the given data.
|
||||
"""
|
||||
payload = {'text': text, 'from_user_id': from_user_id}
|
||||
if media_group_id != 'None':
|
||||
payload['media_group_id'] = media_group_id
|
||||
@@ -32,23 +44,46 @@ class Post(ApiMethod):
|
||||
return neuroTypes.Post.from_dict(data)
|
||||
|
||||
async def __get_all(self, status: EGetAll):
|
||||
"""
|
||||
An asynchronous function to retrieve all items based on the given status using the provided API URL.
|
||||
It takes a status parameter of type EGetAll.
|
||||
The function returns the response obtained from the API call.
|
||||
"""
|
||||
async with ClientSession() as session:
|
||||
response = await session.get(self.api_url+f'/post/get-all/{status.value}')
|
||||
return response
|
||||
|
||||
async def get_all(self):
|
||||
"""
|
||||
Asynchronously retrieves all items and returns a list of Post objects.
|
||||
"""
|
||||
result = await self.__get_all(EGetAll.all)
|
||||
return [neuroTypes.Post.from_dict(post) for post in await result.json()]
|
||||
|
||||
async def get_will_post(self):
|
||||
"""
|
||||
Asynchronously retrieves and returns the will_post data from the API.
|
||||
"""
|
||||
result = await self.__get_all(EGetAll.will_post)
|
||||
return [neuroTypes.Post.from_dict(post) for post in await result.json()]
|
||||
|
||||
async def get_posted(self):
|
||||
"""
|
||||
Asynchronously gets all the posted items and returns a list of Post objects.
|
||||
"""
|
||||
result = await self.__get_all(EGetAll.posted)
|
||||
return [neuroTypes.Post.from_dict(post) for post in await result.json()]
|
||||
|
||||
async def get(self, post_id: str):
|
||||
"""
|
||||
An asynchronous function to retrieve a post by its ID from the API.
|
||||
|
||||
Args:
|
||||
post_id (str): The ID of the post to retrieve.
|
||||
|
||||
Returns:
|
||||
Post: The retrieved post object.
|
||||
"""
|
||||
async with ClientSession() as session:
|
||||
response = await session.get(self.api_url+f'/post/get/{post_id}')
|
||||
data = await response.json()
|
||||
@@ -57,6 +92,15 @@ class Post(ApiMethod):
|
||||
return neuroTypes.Post.from_dict(data)
|
||||
|
||||
async def get_by_order(self, post_order: str):
|
||||
"""
|
||||
Asynchronously gets a post by order from the API.
|
||||
|
||||
Args:
|
||||
post_order (str): The order of the post to retrieve.
|
||||
|
||||
Returns:
|
||||
Post: The post retrieved from the API.
|
||||
"""
|
||||
async with ClientSession() as session:
|
||||
response = await session.get(self.api_url+f'/post/get-post-by-order/{post_order}')
|
||||
data = await response.json()
|
||||
@@ -65,6 +109,15 @@ class Post(ApiMethod):
|
||||
return neuroTypes.Post.from_dict(data)
|
||||
|
||||
async def get_by_media_group_id(self, media_group_id: str):
|
||||
"""
|
||||
Asynchronous function to retrieve data by media group ID.
|
||||
|
||||
Args:
|
||||
media_group_id (str): The media group ID for retrieval.
|
||||
|
||||
Returns:
|
||||
neuroTypes.Post: The retrieved post data.
|
||||
"""
|
||||
async with ClientSession() as session:
|
||||
response = await session.get(self.api_url+f'/post/get-by-media-group-id/{media_group_id}')
|
||||
data = await response.json()
|
||||
@@ -73,6 +126,16 @@ class Post(ApiMethod):
|
||||
return neuroTypes.Post.from_dict(data)
|
||||
|
||||
async def edit_text(self, post_id: str, text: str):
|
||||
"""
|
||||
Asynchronously edits the text of a post.
|
||||
|
||||
Args:
|
||||
post_id (str): The ID of the post to edit.
|
||||
text (str): The new text for the post.
|
||||
|
||||
Returns:
|
||||
Post: The edited post object.
|
||||
"""
|
||||
response = requests.post(
|
||||
self.api_url+f"/post/edit/{post_id}", data={"text": text})
|
||||
data = response.json()
|
||||
@@ -81,6 +144,20 @@ class Post(ApiMethod):
|
||||
return neuroTypes.Post.from_dict(data)
|
||||
|
||||
async def edit_text_by_order_num(self, order: str, text: str, message_entities: Optional[List[MessageEntity]] = None):
|
||||
"""
|
||||
Asynchronously edits text by order number.
|
||||
|
||||
Args:
|
||||
order (str): The order number.
|
||||
text (str): The new text.
|
||||
message_entities (Optional[List[MessageEntity]], optional): A list of message entities. Defaults to None.
|
||||
|
||||
Returns:
|
||||
Post: The edited post.
|
||||
|
||||
Raises:
|
||||
Exception: If the response contains an error status code.
|
||||
"""
|
||||
payload = {"text": text}
|
||||
if message_entities is not None:
|
||||
if message_entities is not None:
|
||||
@@ -98,6 +175,18 @@ class Post(ApiMethod):
|
||||
return neuroTypes.Post.from_dict(data)
|
||||
|
||||
async def get_post_to_post(self):
|
||||
"""
|
||||
Retrieves a post from the API and returns it as a `neuroTypes.Post` object.
|
||||
|
||||
Returns:
|
||||
neuroTypes.Post: The retrieved post.
|
||||
|
||||
Raises:
|
||||
Exception: If the API returns a non-200 status code or if there is an error message in the response.
|
||||
|
||||
Returns:
|
||||
None: If the API returns a 404 status code.
|
||||
"""
|
||||
response = requests.get(self.api_url+f"/post/post")
|
||||
data = response.json()
|
||||
if 'statusCode' in data:
|
||||
@@ -108,11 +197,29 @@ class Post(ApiMethod):
|
||||
return neuroTypes.Post.from_dict(data)
|
||||
|
||||
async def delete_by_order(self, order: str):
|
||||
"""
|
||||
Asynchronously deletes a post by order.
|
||||
|
||||
Args:
|
||||
order (str): The order of the post to be deleted.
|
||||
|
||||
Returns:
|
||||
None
|
||||
"""
|
||||
response = requests.delete(self.api_url+f"/post/delete-post-by-order/{order}")
|
||||
data = response.json()
|
||||
if 'statusCode' in data:
|
||||
raise Exception(data['message'])
|
||||
async def get_deleted_posts(self) -> List[neuroTypes.Post]:
|
||||
"""
|
||||
Asynchronously retrieves a list of deleted posts from the API.
|
||||
|
||||
Parameters:
|
||||
self: The instance of the class.
|
||||
|
||||
Returns:
|
||||
List[Post]: A list of Post objects representing the deleted posts.
|
||||
"""
|
||||
async with ClientSession() as session:
|
||||
response = await session.get(self.api_url+f'/post/get-deleted')
|
||||
data = await response.json()
|
||||
@@ -121,6 +228,15 @@ class Post(ApiMethod):
|
||||
return [neuroTypes.Post.from_dict(post) for post in data]
|
||||
|
||||
async def restore_post(self, order: str):
|
||||
"""
|
||||
Asynchronously restores a post using the given order string.
|
||||
|
||||
Args:
|
||||
order (str): The order string used to identify the post to be restored.
|
||||
|
||||
Returns:
|
||||
Post: A Post object representing the restored post.
|
||||
"""
|
||||
async with ClientSession() as session:
|
||||
response = await session.put(self.api_url+f'/post/restore-post-by-order/{order}')
|
||||
data = await response.json()
|
||||
|
||||
@@ -4,7 +4,21 @@ from .api_method import ApiMethod
|
||||
|
||||
|
||||
class User(ApiMethod):
|
||||
"""User class for API Methods"""
|
||||
async def get(self, id: str, username: str):
|
||||
"""
|
||||
Asynchronous function to retrieve user information by ID and username.
|
||||
|
||||
Args:
|
||||
id (str): The user ID.
|
||||
username (str): The username.
|
||||
|
||||
Raises:
|
||||
Exception: If the API request failing.
|
||||
|
||||
Returns:
|
||||
None
|
||||
"""
|
||||
payload = {'id': id, 'username': username}
|
||||
async with ClientSession() as session:
|
||||
response = await session.post(
|
||||
|
||||
@@ -6,6 +6,7 @@ from ._methods.user import User
|
||||
|
||||
|
||||
class neuroapi:
|
||||
"""Class with all neuroapi methods"""
|
||||
post = Post()
|
||||
admin = Admin()
|
||||
user = User()
|
||||
|
||||
@@ -6,6 +6,7 @@ from pydantic_settings import BaseSettings
|
||||
|
||||
|
||||
class GlobalConfig(BaseSettings):
|
||||
"""Config class"""
|
||||
api_url: str = Field("http://localhost:3000", alias='API_URL')
|
||||
|
||||
# Redis config
|
||||
@@ -39,8 +40,10 @@ class GlobalConfig(BaseSettings):
|
||||
|
||||
@property
|
||||
def redis_url(self):
|
||||
"""Getter method to construct and return the redis URL using the provided redis password, host, port, and database number"""
|
||||
return f'redis://:{self.redis_password}@{self.redis_host}:{self.redis_port}/{self.redis_db}'
|
||||
|
||||
class Config:
|
||||
"""Config file for pydantic settings"""
|
||||
env_file = '.env'
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
"""Types for neuroapi"""
|
||||
from ._admin import Admin
|
||||
from ._bot import NeuroApiBot
|
||||
from ._bot_settings import BotSettings
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
from ._api_model import ApiModel
|
||||
|
||||
|
||||
class Admin(ApiModel):
|
||||
class Admin(ApiModel):
|
||||
"""Represents an admin with fields user_id of type int and user_name of type str"""
|
||||
user_id: int
|
||||
user_name: str
|
||||
|
||||
@@ -6,7 +6,20 @@ from pydantic import BaseModel
|
||||
class ApiModel(BaseModel):
|
||||
@classmethod
|
||||
def from_dict(cls: 'ApiModel', obj: Dict[str, Any]) -> 'ApiModel':
|
||||
"""
|
||||
Create an instance of ApiModel from a dictionary object.
|
||||
|
||||
Args:
|
||||
cls: The class object.
|
||||
obj: A dictionary containing attributes for the ApiModel.
|
||||
|
||||
Returns:
|
||||
ApiModel: An instance of the ApiModel class.
|
||||
"""
|
||||
return cls(**obj)
|
||||
|
||||
def to_dict(self, **kwargs: Any) -> Dict[str, Any]:
|
||||
"""
|
||||
Convert the object to a dictionary
|
||||
"""
|
||||
return self.model_dump(**kwargs)
|
||||
@@ -15,20 +15,59 @@ class NeuroApiBot:
|
||||
_instances = {}
|
||||
|
||||
def __init__(self, token: str, storage: RedisStorage | None = None) -> None:
|
||||
"""
|
||||
Initializes the class with the provided token and optional RedisStorage object.
|
||||
|
||||
Args:
|
||||
token (str): The token for the bot.
|
||||
storage (RedisStorage | None, optional): The RedisStorage object for storing data. Defaults to None.
|
||||
|
||||
Returns:
|
||||
None
|
||||
"""
|
||||
token_data = Token(token=token)
|
||||
self.bot = Bot(token_data.token)
|
||||
self.dp = Dispatcher(storage=storage)
|
||||
|
||||
def __new__(cls, token: str, storage: RedisStorage | None = None) -> 'NeuroApiBot':
|
||||
"""
|
||||
Create a new instance of NeuroApiBot, using the provided token and optional storage. Return bot if its created early with this token
|
||||
|
||||
Args:
|
||||
cls: The class itself.
|
||||
token (str): The token for the instance.
|
||||
storage (RedisStorage | None, optional): The optional storage for the instance. Defaults to None.
|
||||
|
||||
Returns:
|
||||
'NeuroApiBot': The instance of NeuroApiBot.
|
||||
"""
|
||||
token_data = Token(token=token)
|
||||
if token_data.token not in cls._instances:
|
||||
cls._instances[token_data.token] = super(NeuroApiBot, cls).__new__(cls)
|
||||
return cls._instances[token_data.token]
|
||||
|
||||
def include_router(self, *routerClasses: Handler) -> None:
|
||||
"""
|
||||
Include the given router classes in the dispatcher.
|
||||
|
||||
Parameters:
|
||||
*routerClasses (Handler): The router classes to include.
|
||||
|
||||
Returns:
|
||||
None
|
||||
"""
|
||||
for routerClass in routerClasses:
|
||||
assert issubclass(routerClass, Handler)
|
||||
self.dp.include_routers(routerClass(self.bot)())
|
||||
|
||||
async def start(self, skip_updates=True):
|
||||
"""
|
||||
Starts the bot by calling the `start_polling` method of the `dp` object.
|
||||
|
||||
:param skip_updates: A boolean indicating whether to skip updates or not. Defaults to True.
|
||||
:type skip_updates: bool
|
||||
|
||||
:return: None
|
||||
:rtype: None
|
||||
"""
|
||||
await self.dp.start_polling(self.bot, skip_updates=skip_updates)
|
||||
@@ -8,10 +8,18 @@ from ._singleton import Singleton
|
||||
|
||||
|
||||
class BotSettings(ApiModel, Singleton):
|
||||
"""
|
||||
Bot settings model with UUID, message times, channel, and activity status.
|
||||
"""
|
||||
uuid: UUID
|
||||
message_times: List[str] = Field([], alias='messageTimes')
|
||||
channel: str
|
||||
is_active: bool = Field(False, alias='isActive')
|
||||
|
||||
def get_text(self):
|
||||
"""
|
||||
Method to get the text containing channel and message times.
|
||||
|
||||
:return: string - the text containing channel and message times
|
||||
"""
|
||||
return f"Канал: {self.channel}\nВремя: {', '.join(self.message_times)}"
|
||||
@@ -4,6 +4,14 @@ from ._api_model import ApiModel
|
||||
|
||||
|
||||
class Image(ApiModel):
|
||||
"""
|
||||
Represents an image with the following fields:
|
||||
|
||||
- message_id: int
|
||||
- file_id: str
|
||||
- has_spoiler: bool
|
||||
- post_uuid: UUID
|
||||
"""
|
||||
message_id: int
|
||||
file_id: str
|
||||
has_spoiler: bool
|
||||
|
||||
@@ -11,6 +11,19 @@ from ._image import Image
|
||||
|
||||
|
||||
class Post(ApiModel):
|
||||
"""
|
||||
Represents a post with the following fields:
|
||||
|
||||
- uuid: UUID
|
||||
- posted: bool
|
||||
- text: str
|
||||
- media_group_id: int | str
|
||||
- timestamp: datetime
|
||||
- from_user_id: int
|
||||
- images: Optional[List[Image]] = Field(None)
|
||||
- message_entities: Optional[List[MessageEntity]] = Field(None)
|
||||
|
||||
"""
|
||||
uuid: UUID
|
||||
posted: bool
|
||||
text: str
|
||||
@@ -23,6 +36,16 @@ class Post(ApiModel):
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls: 'Post', obj: Dict[str, Any]) -> 'Post':
|
||||
"""
|
||||
Create a Post object from a dictionary.
|
||||
|
||||
Args:
|
||||
cls: The class object.
|
||||
obj: A dictionary containing post data.
|
||||
|
||||
Returns:
|
||||
Post: A Post object created from the dictionary data.
|
||||
"""
|
||||
mes_ent = json.loads(obj.get('message_entities', '[]'))
|
||||
media_group_id_data = obj.get('media_group_id')
|
||||
media_group_id = media_group_id_data if media_group_id_data is not None else 'None'
|
||||
@@ -31,6 +54,11 @@ class Post(ApiModel):
|
||||
return cls(**obj)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
"""
|
||||
Convert the object to a dictionary representation.
|
||||
|
||||
:return: dict - A dictionary representation of the object.
|
||||
"""
|
||||
obj = super().to_dict()
|
||||
obj['message_entities'] = json.dumps(obj['message_entities'])
|
||||
obj['media_group_id'] = str(obj['media_group_id'])
|
||||
|
||||
@@ -2,13 +2,32 @@ from typing import Self
|
||||
|
||||
|
||||
class Singleton:
|
||||
"""
|
||||
Returns the instance of the Singleton class.
|
||||
|
||||
:return: Self - The instance of the Singleton class.
|
||||
"""
|
||||
_instances = {}
|
||||
|
||||
def __new__(cls, *args, **kwargs) -> Self:
|
||||
"""
|
||||
Create a new instance of the class if it doesn't exist, and return the existing instance if it does.
|
||||
|
||||
Parameters:
|
||||
cls: The class.
|
||||
*args: Variable length argument list.
|
||||
**kwargs: Arbitrary keyword arguments.
|
||||
|
||||
Returns:
|
||||
Self: The instance of the class.
|
||||
"""
|
||||
if cls not in cls._instances:
|
||||
cls._instances[cls] = super(Singleton, cls).__new__(cls)
|
||||
return cls._instances[cls]
|
||||
|
||||
@classmethod
|
||||
def get_instance(cls: Self):
|
||||
"""
|
||||
Return the instance belonging to the class.
|
||||
"""
|
||||
return cls._instances[cls]
|
||||
@@ -7,6 +7,12 @@ from ._api_model import ApiModel
|
||||
|
||||
|
||||
class User(ApiModel):
|
||||
"""
|
||||
User model with fields:
|
||||
- id: int
|
||||
- username: str
|
||||
- banned: bool | None
|
||||
"""
|
||||
id: int
|
||||
username: str = Field(..., alias='user_name')
|
||||
banned: Optional[bool] = Field(None)
|
||||
|
||||
Reference in New Issue
Block a user