from collections.abc import Awaitable, Callable
from typing import Any
from pymax.filters import BaseFilter
from pymax.protocols import ClientProtocol
from pymax.types import Chat, Message, ReactionInfo
[docs]
class HandlerMixin(ClientProtocol):
[docs]
def on_message(
self, filter: BaseFilter[Message] | None = None
) -> Callable[
[Callable[[Any], Any | Awaitable[Any]]],
Callable[[Any], Any | Awaitable[Any]],
]:
"""
Декоратор для регистрации обработчика входящих сообщений.
Позволяет установить функцию-обработчик для всех входящих сообщений
или только для сообщений, соответствующих заданному фильтру.
:param filter: Фильтр для обработки сообщений. По умолчанию None.
:type filter: BaseFilter[Message] | None
:return: Декоратор для функции-обработчика.
:rtype: Callable
Example::
@client.on_message(Filter.text("hello"))
async def handle_hello(message: Message):
await client.send_message(
chat_id=message.chat_id,
text="Hello!"
)
"""
def decorator(
handler: Callable[[Any], Any | Awaitable[Any]],
) -> Callable[[Any], Any | Awaitable[Any]]:
self._on_message_handlers.append((handler, filter))
self.logger.debug(f"on_message handler set: {handler}, filter: {filter}")
return handler
return decorator
[docs]
def on_message_edit(
self, filter: BaseFilter[Message] | None = None
) -> Callable[
[Callable[[Any], Any | Awaitable[Any]]],
Callable[[Any], Any | Awaitable[Any]],
]:
"""
Декоратор для установки обработчика отредактированных сообщений.
:param filter: Фильтр для обработки сообщений. По умолчанию None.
:type filter: BaseFilter[Message] | None
:return: Декоратор для функции-обработчика.
:rtype: Callable
"""
def decorator(
handler: Callable[[Any], Any | Awaitable[Any]],
) -> Callable[[Any], Any | Awaitable[Any]]:
self._on_message_edit_handlers.append((handler, filter))
self.logger.debug(f"on_message_edit handler set: {handler}, filter: {filter}")
return handler
return decorator
[docs]
def on_message_delete(
self, filter: BaseFilter[Message] | None = None
) -> Callable[
[Callable[[Any], Any | Awaitable[Any]]],
Callable[[Any], Any | Awaitable[Any]],
]:
"""
Декоратор для установки обработчика удаленных сообщений.
:param filter: Фильтр для обработки сообщений. По умолчанию None.
:type filter: BaseFilter[Message] | None
:return: Декоратор для функции-обработчика.
:rtype: Callable
"""
def decorator(
handler: Callable[[Any], Any | Awaitable[Any]],
) -> Callable[[Any], Any | Awaitable[Any]]:
self._on_message_delete_handlers.append((handler, filter))
self.logger.debug(f"on_message_delete handler set: {handler}, filter: {filter}")
return handler
return decorator
[docs]
def on_reaction_change(
self,
handler: Callable[[str, int, ReactionInfo], Any | Awaitable[Any]],
) -> Callable[[str, int, ReactionInfo], Any | Awaitable[Any]]:
"""
Устанавливает обработчик изменения реакций на сообщения.
:param handler: Функция или coroutine с аргументами (message_id: str, chat_id: int, reaction_info: ReactionInfo).
:type handler: Callable[[str, int, ReactionInfo], Any | Awaitable[Any]]
:return: Установленный обработчик.
:rtype: Callable[[str, int, ReactionInfo], Any | Awaitable[Any]]
"""
self._on_reaction_change_handlers.append(handler)
self.logger.debug("on_reaction_change handler set: %r", handler)
return handler
[docs]
def on_chat_update(
self, handler: Callable[[Chat], Any | Awaitable[Any]]
) -> Callable[[Chat], Any | Awaitable[Any]]:
"""
Устанавливает обработчик обновления информации о чате.
:param handler: Функция или coroutine с аргументом (chat: Chat).
:type handler: Callable[[Chat], Any | Awaitable[Any]]
:return: Установленный обработчик.
:rtype: Callable[[Chat], Any | Awaitable[Any]]
"""
self._on_chat_update_handlers.append(handler)
self.logger.debug("on_chat_update handler set: %r", handler)
return handler
[docs]
def on_raw_receive(
self, handler: Callable[[dict[str, Any]], Any | Awaitable[Any]]
) -> Callable[[dict[str, Any]], Any | Awaitable[Any]]:
"""
Устанавливает обработчик для получения необработанных данных от сервера.
:param handler: Функция или coroutine с аргументом (data: dict).
:type handler: Callable[[dict[str, Any]], Any | Awaitable[Any]]
:return: Установленный обработчик.
:rtype: Callable[[dict[str, Any]], Any | Awaitable[Any]]
"""
self._on_raw_receive_handlers.append(handler)
self.logger.debug("on_raw_receive handler set: %r", handler)
return handler
[docs]
def on_start(
self, handler: Callable[[], Any | Awaitable[Any]]
) -> Callable[[], Any | Awaitable[Any]]:
"""
Устанавливает обработчик, вызываемый при старте клиента.
:param handler: Функция или coroutine без аргументов.
:type handler: Callable[[], Any | Awaitable[Any]]
:return: Установленный обработчик.
:rtype: Callable[[], Any | Awaitable[Any]]
"""
self._on_start_handler = handler
self.logger.debug("on_start handler set: %r", handler)
return handler
[docs]
def task(self, seconds: float, minutes: float = 0, hours: float = 0):
"""
Декоратор для планирования периодической задачи.
:param seconds: Интервал выполнения в секундах.
:type seconds: float
:param minutes: Интервал выполнения в минутах. По умолчанию 0.
:type minutes: float
:param hours: Интервал выполнения в часах. По умолчанию 0.
:type hours: float
:return: Декоратор для функции-обработчика.
:rtype: Callable[[], Any | Awaitable[Any]]
Example::
@client.task(seconds=10)
async def task():
await client.send_message(chat_id=123, text="Hello!")
"""
def decorator(
handler: Callable[[], Any | Awaitable[Any]],
) -> Callable[[], Any | Awaitable[Any]]:
self._scheduled_tasks.append((handler, seconds + minutes * 60 + hours * 3600))
self.logger.debug(
f"task scheduled: {handler}, interval: {seconds + minutes * 60 + hours * 3600}s"
)
return handler
return decorator
[docs]
def add_message_handler(
self,
handler: Callable[[Message], Any | Awaitable[Any]],
filter: BaseFilter[Message] | None = None,
) -> Callable[[Message], Any | Awaitable[Any]]:
"""
Добавляет обработчик входящих сообщений.
:param handler: Обработчик.
:type handler: Callable[[Message], Any | Awaitable[Any]]
:param filter: Фильтр. По умолчанию None.
:type filter: BaseFilter[Message] | None
:return: Обработчик.
:rtype: Callable[[Message], Any | Awaitable[Any]]
"""
self.logger.debug("add_message_handler (alias) used")
self._on_message_handlers.append((handler, filter))
return handler
[docs]
def add_on_start_handler(
self, handler: Callable[[], Any | Awaitable[Any]]
) -> Callable[[], Any | Awaitable[Any]]:
"""
Добавляет обработчик, вызываемый при старте клиента.
:param handler: Функция или coroutine без аргументов.
:type handler: Callable[[], Any | Awaitable[Any]]
:return: Установленный обработчик.
:rtype: Callable[[], Any | Awaitable[Any]]
"""
self.logger.debug("add_on_start_handler (alias) used")
self._on_start_handler = handler
return handler
[docs]
def add_reaction_change_handler(
self,
handler: Callable[[str, int, ReactionInfo], Any | Awaitable[Any]],
) -> Callable[[str, int, ReactionInfo], Any | Awaitable[Any]]:
"""
Добавляет обработчик изменения реакций на сообщения.
:param handler: Функция или coroutine с аргументами (message_id: str, chat_id: int, reaction_info: ReactionInfo).
:type handler: Callable[[str, int, ReactionInfo], Any | Awaitable[Any]]
:return: Установленный обработчик.
:rtype: Callable[[str, int, ReactionInfo], Any | Awaitable[Any]]
"""
self.logger.debug("add_reaction_change_handler (alias) used")
self._on_reaction_change_handlers.append(
handler,
)
return handler
[docs]
def add_chat_update_handler(
self, handler: Callable[[Chat], Any | Awaitable[Any]]
) -> Callable[[Chat], Any | Awaitable[Any]]:
"""
Добавляет обработчик обновления информации о чате.
:param handler: Функция или coroutine с аргументом (chat: Chat).
:type handler: Callable[[Chat], Any | Awaitable[Any]]
:return: Установленный обработчик.
:rtype: Callable[[Chat], Any | Awaitable[Any]]
"""
self.logger.debug("add_chat_update_handler (alias) used")
self._on_chat_update_handlers.append(handler)
return handler
[docs]
def add_raw_receive_handler(
self, handler: Callable[[dict[str, Any]], Any | Awaitable[Any]]
) -> Callable[[dict[str, Any]], Any | Awaitable[Any]]:
"""
Добавляет обработчик для получения необработанных данных от сервера.
:param handler: Функция или coroutine с аргументом (data: dict).
:type handler: Callable[[dict[str, Any]], Any | Awaitable[Any]]
:return: Установленный обработчик.
:rtype: Callable[[dict[str, Any]], Any | Awaitable[Any]]
"""
self.logger.debug("add_raw_receive_handler (alias) used")
self._on_raw_receive_handlers.append(handler)
return handler
[docs]
def add_scheduled_task(
self,
handler: Callable[[], Any | Awaitable[Any]],
interval: float,
) -> Callable[[], Any | Awaitable[Any]]:
"""
Добавляет периодическую задачу.
:param handler: Функция или coroutine без аргументов.
:type handler: Callable[[], Any | Awaitable[Any]]
:param interval: Интервал выполнения в секундах.
:type interval: float
:return: Установленный обработчик.
:rtype: Callable[[], Any | Awaitable[Any]]
"""
self.logger.debug("add_scheduled_task (alias) used")
self._scheduled_tasks.append((handler, interval))
return handler