Compare commits

...

172 Commits

Author SHA1 Message Date
e3c17ab60f Переход на Google Style Docstring 2025-08-06 02:38:34 +03:00
af4c9dc487 Переход на Google Style Docstring 2025-08-06 02:35:33 +03:00
1400f72cd0 Добавлен HandlerException, BaseException -> Exception 2025-08-06 02:34:48 +03:00
e0569de1c5 Добавлен пример на CallbackPayload 2025-08-05 00:54:00 +03:00
ff4575fe84 Добавлен фильтр-обработка CallbackPayload 2025-08-05 00:53:45 +03:00
e922132319 Поправлено преобразование вложений для ответа на callback 2025-08-05 00:53:17 +03:00
b59d97da8a 0.9.5 2025-08-03 14:02:59 +03:00
fe68e41b7a Поправлен импорт GetMessage 2025-08-03 13:58:14 +03:00
50980dfc77 Импорт BaseFilter 2025-08-03 13:57:31 +03:00
036c92d072 Переделана основа для фильтров Command, CommandStart 2025-08-03 13:57:19 +03:00
cb2226eee5 Добавлен BaseFilter 2025-08-03 13:56:56 +03:00
3855f93862 Правки по mypy 2025-08-03 13:56:12 +03:00
01e9cdd2fd Добавлен пример с BaseFilter 2025-08-03 13:55:53 +03:00
338d9c4089 Переделано добавление Middleware 2025-08-03 13:55:31 +03:00
3fa34079ae Доработано преобразование вложений в Pydantic модели 2025-08-02 23:22:36 +03:00
5bc5fb45c8 Мелкие правки 2025-08-02 23:21:57 +03:00
95313ad3dc Переработан метод для получения одного сообщения 2025-07-31 23:38:47 +03:00
42690d24ee Добавлена обработка события dialog_removed 2025-07-31 23:30:30 +03:00
6ad3df5829 0.9.4 2025-07-30 13:29:44 +03:00
622d3a3eb3 Изменена версия aiohttp на 3.12.14 2025-07-30 13:16:42 +03:00
67de8aae1f Правки в тексте 2025-07-30 13:16:14 +03:00
1b0b118239 0.9.3 2025-07-30 09:25:08 +03:00
8d62d0d20a Обновлены примеры использования Middleware 2025-07-30 09:25:00 +03:00
a7173b4371 Добавлены доп. настройки в пример запуска 2025-07-30 08:57:58 +03:00
b25b0c0a5e Переделано оформление 2025-07-30 08:57:41 +03:00
af6b1f1a74 Добавлен PhotoAttachmentRequestPayload для легкого импорта 2025-07-29 22:31:23 +03:00
b1f8fb91cb Поправлена аннотация photo в __init__ ChangeInfo 2025-07-29 22:30:57 +03:00
9c0567d858 Поправлена аннотация photo в change_info 2025-07-29 22:30:10 +03:00
9ab960ebe4 Добавлен метод close_session() 2025-07-29 14:17:35 +03:00
a653ed6792 Изменен фильтр на State. Вместо == теперь in. Работает если в хендлере указано несколько фильтров на State 2025-07-28 15:13:52 +03:00
103535d3ba Объект подписки на Webhook 2025-07-28 01:17:40 +03:00
82dc8d255a Объект ответа подписки на Webhook 2025-07-28 01:16:20 +03:00
35b60a1b44 Объект ответа отписки от Webhook 2025-07-28 01:14:41 +03:00
e07aeef726 Объект ответа подписки на Webhook 2025-07-28 01:14:34 +03:00
03a60014d4 Объект ответа полученных подписок Webhook 2025-07-28 01:14:25 +03:00
b69713f996 Класс на отписку от Webhook 2025-07-28 01:14:07 +03:00
9010c697f4 Класс для подписки на Webhook 2025-07-28 01:13:56 +03:00
b994717f64 Класс для получения подписок Webhook 2025-07-28 01:13:46 +03:00
44dcf39f42 Добавлен API путь /subscriptions 2025-07-28 01:13:12 +03:00
0bd59ddb4a Добавлена проверка на подписки Webhook при start_polling 2025-07-28 01:12:55 +03:00
f497760f5b Добавлены методы: get_subscriptions, subscribe_webhook, unsubscribe_webhook, delete_webhook. Добавлен auto_check_subscriptions к Bot. 2025-07-28 01:12:11 +03:00
5c64c3d040 Правки информации по user_removed 2025-07-27 16:59:12 +03:00
c8dd896691 Переработана система Middleware 2025-07-27 16:58:54 +03:00
e0e7bb53b6 0.9.2 Опциональные fastapi uvicorn 2025-07-27 14:20:08 +03:00
adbea8e55f Поправлены докстринги у Dispatcher и Router 2025-07-27 14:19:42 +03:00
3ff9907a58 0.9.2 fastapi, uvicorn опциональны 2025-07-27 02:15:03 +03:00
6f86d15de4 Добавлен filename к InputMediaBuffer 2025-07-27 02:12:58 +03:00
7ea24fe2af Изменен пример запуска 2025-07-27 02:12:30 +03:00
39fb0c5823 Добавлены типы событий: bot_stopped, dialog_cleared, dialog_muted, dialog_unmuted 2025-07-27 02:12:22 +03:00
af84301e4f Добавлены типы событий: bot_stopped, dialog_cleared, dialog_muted, dialog_unmuted. Правки с auto_requests 2025-07-27 02:12:16 +03:00
ec432fe8ce Добавлены типы событий: bot_stopped, dialog_cleared, dialog_muted, dialog_unmuted 2025-07-27 02:11:48 +03:00
1bfd93f2ea chat_title обязателен 2025-07-27 02:11:30 +03:00
7925087ac7 Добавлено в User @property full_name 2025-07-27 02:11:04 +03:00
7ed540683c Добавлены типы событий: bot_stopped, dialog_cleared, dialog_muted, dialog_unmuted 2025-07-27 02:10:34 +03:00
30350c8521 Откат преобразования Recipient, одного из полей может не быть 2025-07-27 02:10:25 +03:00
54683256ce Добавлены типы событий: bot_stopped, dialog_cleared, dialog_muted, dialog_unmuted 2025-07-27 02:09:50 +03:00
be7f98976e Поправлена логика задержки отправки сообщения с InputMedia, InputMediaBuffer 2025-07-27 02:09:34 +03:00
54c073ab76 Добавлены типы событий: bot_stopped, dialog_cleared, dialog_muted, dialog_unmuted 2025-07-27 02:09:04 +03:00
29b319768b Добавлены типы событий: bot_stopped, dialog_cleared, dialog_muted, dialog_unmuted. Изменена система запуска вебхуков 2025-07-27 02:08:45 +03:00
5e98e540ea Изменена система запуска вебхука, fastapi с uvicorn опциональны 2025-07-27 02:08:27 +03:00
1df293f44d Маленькие правки 2025-07-27 02:07:22 +03:00
c667b82a6c Добавлены типы событий: bot_stopped, dialog_cleared, dialog_muted, dialog_unmuted, message_chat_created 2025-07-27 02:07:06 +03:00
62523c1eb2 Правки по преобразованию 2025-07-25 00:54:57 +03:00
b0b7040206 Добавлен аттрибут is_channel 2025-07-25 00:54:35 +03:00
29d3d7c042 Рефактор для получения обновлений 2025-07-25 00:53:35 +03:00
fd048e8544 ignore 2025-07-25 00:53:08 +03:00
32c0ca7647 Правки по ruff + mypy 2025-07-25 00:52:58 +03:00
02b4e2d39a Создан отдельный файл для MemoryContext 2025-07-25 00:52:38 +03:00
354c296fed Правки по ruff + mypy 2025-07-25 00:52:16 +03:00
8f93cf36e4 Удален импорт UpdateUnion 2025-07-25 00:49:20 +03:00
e1064761e4 Поправлены импорты 2025-07-25 00:48:53 +03:00
2420e4232e Добавлена обработка Error 2025-07-24 19:38:08 +03:00
fd9986b02e Доработана загрузка медиа 2025-07-24 19:37:43 +03:00
48b480ff9e Поправлена загрузка медиа 2025-07-24 19:37:30 +03:00
69e6274f42 Перенесены константы из send_message.py и добавлена AFTER_MEDIA_INPUT_DELAY 2025-07-24 19:37:11 +03:00
0daa9d508d Поправлен "спам" общей ошибка таймаута 2025-07-24 19:35:59 +03:00
0ae0758bc1 Добавлен after_input_media_delay в Bot, поправлены типы возвращаемых данных 2025-07-24 19:33:33 +03:00
7d2826c4b5 Правки по mypy 2025-07-24 18:54:41 +03:00
2cd3d64bb8 Небольшая оптимизация 2025-07-24 01:50:25 +03:00
1fb2fbd654 Добавлен файл для будущих исключений MAX 2025-07-24 01:50:06 +03:00
7534ae60a8 Убран лишний импорт 2025-07-24 01:48:03 +03:00
5ad37b8adf Заменен [] на Field(default_factory=list) для списков в pydantic-моделях, чтобы избежать shared mutable default 2025-07-24 01:30:28 +03:00
d731e5e905 0.9.1 2025-07-20 22:56:53 +03:00
7155f974a2 Добавлен README.md для вебхуков 2025-07-20 22:53:19 +03:00
cd85a1b7fb Добавлен README.md для примеров 2025-07-20 22:53:07 +03:00
bdc51a32c5 Изменен блок "Примеры" 2025-07-20 22:52:43 +03:00
c88b5228b7 Добавлена документация по вебхукам 2025-07-20 22:45:08 +03:00
611114ec09 Удален заголовок 2025-07-20 22:44:57 +03:00
52d69904c6 Пример низкоуровневого вебхука 2025-07-20 21:56:54 +03:00
7e9163adc9 0.9.0 2025-07-20 21:48:11 +03:00
6a0406f476 Изменены аннотации attachments 2025-07-20 21:47:31 +03:00
9241917bb1 Изменены аннотации attachments 2025-07-20 21:47:27 +03:00
0354fbc5fd Переработан вебхук 2025-07-20 21:47:22 +03:00
f600decf2b Изменены аннотации attachments 2025-07-20 21:46:47 +03:00
dd6c8ff9ea Пример низклуровневого вебхука 2025-07-20 21:46:14 +03:00
c844d2b2e6 Пример высокоуровневого вебхука 2025-07-20 21:46:06 +03:00
7336b2ebb9 Поправлен Command 2025-07-20 13:08:26 +03:00
85e4c086d4 Клавиатуры и кнопки 2025-07-19 17:29:35 +03:00
964fba7c32 GitIgnore 2025-07-19 17:08:26 +03:00
78976c5393 Лицензия 2025-07-19 17:06:59 +03:00
e8b7c71d25 Поправлены Wiki 2025-07-19 17:03:14 +03:00
aab87e16b2 Поправлен RequestGeoLocationButton 2025-07-19 17:02:59 +03:00
df383665dc Доработан RequestContact 2025-07-19 17:02:47 +03:00
bd06b33343 Добавлен OpenAppButton 2025-07-19 17:02:37 +03:00
7b70d1de18 Добавлен MessageButton 2025-07-19 17:02:21 +03:00
5f2c908da4 Поправлен LinkButton 2025-07-19 17:02:05 +03:00
1abbc16cc8 Изменены импорты 2025-07-19 17:01:50 +03:00
7b61ceaa58 Добавлен .pack() в ButtonsPayload для удобства 2025-07-19 17:01:42 +03:00
37f7907398 Добавлен CommandStart 2025-07-19 17:01:08 +03:00
9dab5f97fb Изменены импорты 2025-07-19 17:00:45 +03:00
0a3d1ca327 Добавлен CommandStart 2025-07-19 17:00:29 +03:00
93043835d1 Добавлены типы кнопок: message, open_app 2025-07-19 17:00:19 +03:00
3548d0558f Добавлено присваивание @property .me для bot 2025-07-19 16:59:57 +03:00
5ae4de6816 Добавлен @property .me (присваивается при запуске бота) 2025-07-19 16:59:25 +03:00
d77288ea07 Пример со всеми кнопками для клавиатуры 2025-07-19 16:58:42 +03:00
30cf778504 0.8.9 2025-07-19 13:43:34 +03:00
dd1bdb5e37 Добавлена загрузка файлов из буфера, InputMediaBuffer 2025-07-19 13:41:03 +03:00
b20a46de24 Добавлена загрузка файлов из буфера, InputMediaBuffer 2025-07-19 13:39:07 +03:00
de05e7931a 0.8.8 2025-07-17 02:05:48 +03:00
a8727c71e9 Обновлен блок "Документация" 2025-07-15 18:32:54 +03:00
b6c11cd28a Добавлена вики 2025-07-15 18:29:54 +03:00
12f64f0805 Поправлены докстринги 2025-07-15 12:11:00 +03:00
3df4dd21b4 Поправлены докстринги 2025-07-15 12:09:31 +03:00
b60d8571d3 backoff 2025-07-04 15:22:58 +03:00
c9a334a615 Небольшая оптимизация 2025-07-04 13:44:00 +03:00
222ca919fc Добавлена заглушка 2025-07-01 21:31:46 +03:00
111b83bf84 0.8.7 2025-07-01 21:05:43 +03:00
98f4f8717b Поправлен parse_mode 2025-07-01 21:00:46 +03:00
2a351fdbd7 Манифест 2025-07-01 21:00:14 +03:00
3a3c581914 Добавлен first_name при старте бота 2025-06-30 17:11:11 +03:00
57d9035afb 0.8.6 2025-06-30 17:07:18 +03:00
5f2beb7e07 Поправлена модель для обработки Геолокации 2025-06-30 17:06:48 +03:00
e8ad92c3c3 Немного TYPE_CHECKING 2025-06-30 12:23:43 +03:00
1be05dea83 Немного TYPE_CHECKING 2025-06-30 12:23:28 +03:00
c1df0c5338 Поправлена версия 2025-06-29 15:29:55 +03:00
b5947e5f47 Немного оптимизировано и поправлен список зависимостей 2025-06-29 15:27:19 +03:00
c481e3e931 Изменена версия 2025-06-23 09:51:46 +03:00
63af421777 Поправлены импорты 2025-06-23 09:51:20 +03:00
512eb9a4af Добавлен Middleware 2025-06-23 09:49:24 +03:00
8aa9c65fcc disable_notifications изменен на notify 2025-06-21 21:58:10 +03:00
8f5fc9f398 Добавил настройку для Bot() : auto_requests 2025-06-21 20:12:19 +03:00
de684aa200 Добавлены настройки для Bot() по умолчанию: parse_mode, disable_notifications 2025-06-21 19:58:38 +03:00
7b8aa3d092 Добавлен метод .forward() для Message 2025-06-21 19:35:36 +03:00
d35e15941f Добавлен метод .reply() к Message 2025-06-21 19:30:27 +03:00
294c05ef91 Правки... 2025-06-21 02:23:57 +03:00
bca4c3fd6c Правки 2025-06-21 02:20:47 +03:00
2ba1ba23b6 Поправил версию 2025-06-21 02:17:39 +03:00
7ea18c3187 Поправил ссылки на примеры 2025-06-21 02:16:14 +03:00
390dc94279 Обновил версию 2025-06-21 02:14:33 +03:00
9d0eeb9f89 Обновил ридми 2025-06-21 02:11:06 +03:00
ee58238261 Добавлены примеры и обновлена модель Update 2025-06-21 02:05:33 +03:00
ab52abc474 0.7 2025-06-20 17:59:33 +03:00
688b1502d1 fix 2025-06-20 17:58:35 +03:00
07bd5d090f !! 2025-06-20 17:51:47 +03:00
98ab03670a !! 2025-06-20 17:50:46 +03:00
d12d3fc15f . 2025-06-20 17:48:54 +03:00
e08e620b92 Добавлены докстринги 2025-06-20 17:23:51 +03:00
1e4d22a2bc - 2025-06-20 03:33:10 +03:00
bfa247e082 - 2025-06-20 03:00:41 +03:00
68748d0899 . 2025-06-20 02:58:55 +03:00
6560fe011d Пару плюшек + обновил ридми 2025-06-20 02:57:45 +03:00
1374d863f0 Добавлен types.InputMedia для простой загрузки медиафайлов 2025-06-20 02:24:14 +03:00
85f58913c3 Добавлен метод set_my_commands у bot 2025-06-19 23:28:41 +03:00
9591780152 a 2025-06-19 03:19:24 +03:00
29fd18fe91 fix 2025-06-19 03:08:42 +03:00
285c7758ea fix 2025-06-19 03:05:28 +03:00
e5c4cfa573 fix 2025-06-19 03:03:40 +03:00
4de32ca476 Добавлены: билдер инлайн клавиатур, FSM like aiogram, start_polling и логгирование 2025-06-19 02:40:38 +03:00
9a39dce1a6 добавлен блок chats из документации и start_polling к Dispatcher 2025-06-18 17:00:03 +03:00
ff19f99704 upd 2025-06-17 23:20:49 +03:00
eff34b42c2 upd 2025-06-17 23:20:36 +03:00
b2283ab538 upd 2025-06-17 23:14:25 +03:00
171 changed files with 8538 additions and 541 deletions

2
.gitignore vendored
View File

@@ -1,4 +1,6 @@
# Byte-compiled / optimized / DLL files
test
test.py
__pycache__/
*.py[cod]
*$py.class

21
LICENSE Normal file
View File

@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2025 Denis
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

2
MANIFEST.in Normal file
View File

@@ -0,0 +1,2 @@
include LICENSE.md
include README.md

133
README.md
View File

@@ -1,16 +1,123 @@
# maxapi
<p align="center">
<a href="https://github.com/love-apples/maxapi"><img src="https://s.iimg.su/s/29/DCvw4dx2HZgFdTcpqGAs6xdnJnvD44r9zLga2GGe.png" alt="MaxAPI"></a>
</p>
#### Библиотека (like aiogram) для взаимодействия с социальной сетью MAX по Webhook (или подписке бота)
Информация на данный момент:
* Проект не готов, ведется активная разработка
* Планируется:
+ Сокращение импортов в ваших хендлерах (громадные импорты в example.py)
+ Сокращение "построения" клавиатур
+ Разработка контекста бота
+ Разработка Longpoll метода
+ Доработка базовой составляющей проекта
+ и так далее...
<p align="center">
<a href='https://github.com/love-apples/maxapi/wiki'>Документация</a>
<a href='https://github.com/love-apples/maxapi/tree/main/examples'>Примеры</a>
<a href='https://max.ru/join/IPAok63C3vFqbWTFdutMUtjmrAkGqO56YeAN7iyDfc8'>MAX Чат</a>
<a href='https://t.me/maxapi_github'>TG Чат</a>
</p>
### Контакты
[Группа MAX](https://max.ru/join/IPAok63C3vFqbWTFdutMUtjmrAkGqO56YeAN7iyDfc8)
<p align="center">
<a href='https://pypi.org/project/maxapi/'>
<img src='https://img.shields.io/pypi/v/maxapi.svg' alt='PyPI version'>
</a>
<a href='https://pypi.org/project/maxapi/'>
<img src='https://img.shields.io/pypi/pyversions/maxapi.svg' alt='Python Version'>
</a>
<a href='https://love-apples/maxapi/blob/main/LICENSE'>
<img src='https://img.shields.io/github/license/love-apples/maxapi.svg' alt='License'>
</a>
</p>
## 📦 Установка из PyPi
Стабильная версия
```bash
pip install maxapi
```
## 🐱‍👤 Установка из GitHub
Свежая версия, возможны баги. Рекомендуется только для ознакомления с новыми коммитами.
```bash
pip install git+https://github.com/love-apples/maxapi.git
```
## 🚀 Быстрый старт
Если вы тестируете бота в чате - не забудьте дать ему права администратора!
### Запуск Polling
Если у бота установлены подписки на Webhook - события не будут приходить при методе `start_polling`. При таком случае удалите подписки на Webhook через [MasterBot](https://web.max.ru/masterbot) или через `await bot.delete_webhook()` перед `start_polling`.
```python
import asyncio
import logging
from maxapi import Bot, Dispatcher
from maxapi.types import BotStarted, Command, MessageCreated
logging.basicConfig(level=logging.INFO)
bot = Bot(ут_ваш_токен')
dp = Dispatcher()
# Ответ бота при нажатии на кнопку "Начать"
@dp.bot_started()
async def bot_started(event: BotStarted):
await event.bot.send_message(
chat_id=event.chat_id,
text='Привет! Отправь мне /start'
)
# Ответ бота на команду /start
@dp.message_created(Command('start'))
async def hello(event: MessageCreated):
await event.message.answer(f"Пример чат-бота для MAX 💙")
async def main():
await dp.start_polling(bot)
if __name__ == '__main__':
asyncio.run(main())
```
### Запуск Webhook
Перед запуском бота через Webhook, вам нужно установить дополнительные зависимости (fastapi, uvicorn). Можно это сделать через команду:
```bash
pip install maxapi[webhook]
```
Указан пример простого запуска, для более низкого уровня можете рассмотреть [этот пример](https://github.com/love-apples/maxapi/blob/main/examples/webhook/low_level.py).
```python
import asyncio
import logging
from maxapi import Bot, Dispatcher
from maxapi.types import BotStarted, Command, MessageCreated
logging.basicConfig(level=logging.INFO)
bot = Bot(ут_ваш_токен')
dp = Dispatcher()
# Команда /start боту
@dp.message_created(Command('start'))
async def hello(event: MessageCreated):
await event.message.answer(f"Привет из вебхука!")
async def main():
await dp.handle_webhook(
bot=bot,
host='localhost',
port=8080,
log_level='critical' # Можно убрать для подробного логгирования
)
if __name__ == '__main__':
asyncio.run(main())
```

View File

@@ -1,50 +0,0 @@
from maxapi.bot import Bot
from maxapi.dispatcher import Dispatcher
from maxapi.types.updates.message_created import MessageCreated
from maxapi.types.updates.message_callback import MessageCallback
from maxapi.types.attachments.attachment import ButtonsPayload
from maxapi.types.attachments.buttons.callback_button import CallbackButton
from maxapi.types.attachments.attachment import Attachment
from maxapi.enums.attachment import AttachmentType
from maxapi.enums.button_type import ButtonType
from maxapi.enums.intent import Intent
from maxapi.filters import F
bot = Bot('токен')
dp = Dispatcher()
# Отвечает только на текст "Привет"
@dp.message_created(F.message.body.text == 'Привет')
async def hello(obj: MessageCreated):
await obj.message.answer('Привет 👋')
# Отвечает только на текст "Клавиатура"
@dp.message_created(F.message.body.text == 'Клавиатура')
async def hello(obj: MessageCreated):
button_1 = CallbackButton(type=ButtonType.CALLBACK, text='Кнопка 1', payload='1', intent=Intent.DEFAULT)
button_2 = CallbackButton(type=ButtonType.CALLBACK, text='Кнопка 2', payload='2', intent=Intent.DEFAULT)
keyboard = ButtonsPayload(buttons=[[button_1], [button_2]])
attachments = [Attachment(type=AttachmentType.INLINE_KEYBOARD, payload=keyboard)]
await obj.message.answer('Привет 👋', attachments=attachments)
# Ответчает на коллбек с начинкой "1"
@dp.message_callback(F.callback.payload == '1')
async def _(obj: MessageCallback):
await obj.message.answer('Вы нажали на кнопку 1 🤩')
# Ответчает на коллбек с начинкой "2"
@dp.message_callback(F.callback.payload == '2')
async def _(obj: MessageCallback):
await obj.message.answer('Вы нажали на кнопку 2 🥳')
# Отвечает на любое текстовое сообщение
@dp.message_created(F.message.body.text)
async def hello(obj: MessageCreated):
await obj.message.answer(f'Повторяю за вами: {obj.message.body.text}')
dp.handle_webhook(bot)

13
examples/README.md Normal file
View File

@@ -0,0 +1,13 @@
## ⭐️ Примеры
- [Эхо бот](https://github.com/love-apples/maxapi/blob/main/examples/echo/main.py)
- [Обработчик доступных событий](https://github.com/love-apples/maxapi/blob/main/examples/events/main.py)
- [Обработчики с MagicFilter](https://github.com/love-apples/maxapi/blob/main/examples/magic_filters/main.py)
- [Демонстрация роутинга, InputMedia и механика контекста](https://github.com/love-apples/maxapi/tree/main/examples/router_with_input_media) (audio.mp3 для команды /media)
- [Получение ID](https://github.com/love-apples/maxapi/tree/main/examples/get_ids/main.py)
- [Миддлварь в хендлерах](https://github.com/love-apples/maxapi/tree/main/examples/middleware_in_handlers/main.py)
- [Вебхуки](https://github.com/love-apples/maxapi/tree/main/examples/webhook)
- [Клавиатуры](https://github.com/love-apples/maxapi/tree/main/examples/keyboard/main.py)
- [Миддлварь в роутерах](https://github.com/love-apples/maxapi/tree/main/examples/middleware_for_router/main.py)
- [Свой фильтр на BaseFilter](https://github.com/love-apples/maxapi/tree/main/examples/base_filter/main.py)
- [Фильтр callback payload](https://github.com/love-apples/maxapi/tree/main/examples/callback_payload/main.py)

View File

@@ -0,0 +1,38 @@
import asyncio
import logging
from maxapi import Bot, Dispatcher
from maxapi.types import MessageCreated, CommandStart, UpdateUnion
from maxapi.filters import BaseFilter
logging.basicConfig(level=logging.INFO)
bot = Bot(token='тут_ваш_токен')
dp = Dispatcher()
class FilterChat(BaseFilter):
"""
Фильтр, который срабатывает только в чате с названием `Test`
"""
async def __call__(self, event: UpdateUnion):
if not event.chat:
return False
return event.chat == 'Test'
@dp.message_created(CommandStart(), FilterChat())
async def custom_data(event: MessageCreated):
await event.message.answer('Привет!')
async def main():
await dp.start_polling(bot)
if __name__ == '__main__':
asyncio.run(main())

View File

@@ -0,0 +1,61 @@
import asyncio
import logging
from maxapi import Bot, Dispatcher, F
from maxapi.filters.callback_payload import CallbackPayload
from maxapi.filters.command import CommandStart
from maxapi.types import (
CallbackButton,
MessageCreated,
MessageCallback,
)
from maxapi.utils.inline_keyboard import InlineKeyboardBuilder
logging.basicConfig(level=logging.INFO)
bot = Bot('тут_ваш_токен')
dp = Dispatcher()
class MyPayload(CallbackPayload, prefix='mypayload'):
foo: str
action: str
class AnotherPayload(CallbackPayload, prefix='another'):
bar: str
value: int
@dp.message_created(CommandStart())
async def show_keyboard(event: MessageCreated):
kb = InlineKeyboardBuilder()
kb.row(
CallbackButton(
text='Первая кнопка',
payload=MyPayload(foo='123', action='edit').pack(),
),
CallbackButton(
text='Вторая кнопка',
payload=AnotherPayload(bar='abc', value=42).pack(),
),
)
await event.message.answer('Нажми кнопку!', attachments=[kb.as_markup()])
@dp.message_callback(MyPayload.filter(F.foo == '123'))
async def on_first_callback(event: MessageCallback, payload: MyPayload):
await event.answer(new_text=f'Первая кнопка: foo={payload.foo}, action={payload.action}')
@dp.message_callback(AnotherPayload.filter())
async def on_second_callback(event: MessageCallback, payload: AnotherPayload):
await event.answer(new_text=f'Вторая кнопка: bar={payload.bar}, value={payload.value}')
async def main():
await dp.start_polling(bot)
if __name__ == '__main__':
asyncio.run(main())

24
examples/echo/main.py Normal file
View File

@@ -0,0 +1,24 @@
import asyncio
import logging
from maxapi import Bot, Dispatcher
from maxapi.filters import F
from maxapi.types import MessageCreated
logging.basicConfig(level=logging.INFO)
bot = Bot('тут_ваш_токен')
dp = Dispatcher()
@dp.message_created(F.message.body.text)
async def echo(event: MessageCreated):
await event.message.answer(f"Повторяю за вами: {event.message.body.text}")
async def main():
await dp.start_polling(bot)
if __name__ == '__main__':
asyncio.run(main())

180
examples/events/main.py Normal file
View File

@@ -0,0 +1,180 @@
import asyncio
import logging
from maxapi import Bot, Dispatcher
from maxapi.types import (
BotStarted,
Command,
MessageCreated,
CallbackButton,
MessageCallback,
BotAdded,
ChatTitleChanged,
MessageEdited,
MessageRemoved,
UserAdded,
UserRemoved,
BotStopped,
DialogCleared,
DialogMuted,
DialogUnmuted,
ChatButton,
MessageChatCreated
)
from maxapi.utils.inline_keyboard import InlineKeyboardBuilder
logging.basicConfig(level=logging.INFO)
bot = Bot('тут_ваш_токен')
dp = Dispatcher()
@dp.message_created(Command('start'))
async def hello(event: MessageCreated):
builder = InlineKeyboardBuilder()
builder.row(
CallbackButton(
text='Кнопка 1',
payload='btn_1'
),
CallbackButton(
text='Кнопка 2',
payload='btn_2',
)
)
builder.add(
ChatButton(
text='Создать чат',
chat_title='Тест чат'
)
)
await event.message.answer(
text='Привет!',
attachments=[
builder.as_markup(),
] # Для MAX клавиатура это вложение,
) # поэтому она в attachments
@dp.bot_added()
async def bot_added(event: BotAdded):
if not event.chat:
logging.info('Не удалось получить chat, возможно отключен auto_requests!')
return
await bot.send_message(
chat_id=event.chat.id,
text=f'Привет чат {event.chat.title}!'
)
@dp.message_removed()
async def message_removed(event: MessageRemoved):
await bot.send_message(
chat_id=event.chat_id,
text='Я всё видел!'
)
@dp.bot_started()
async def bot_started(event: BotStarted):
await bot.send_message(
chat_id=event.chat_id,
text='Привет! Отправь мне /start'
)
@dp.chat_title_changed()
async def chat_title_changed(event: ChatTitleChanged):
await bot.send_message(
chat_id=event.chat_id,
text=f'Крутое новое название "{event.title}"!'
)
@dp.message_callback()
async def message_callback(event: MessageCallback):
await event.answer(
new_text=f'Вы нажали на кнопку {event.callback.payload}!'
)
@dp.message_edited()
async def message_edited(event: MessageEdited):
await event.message.answer(
text='Вы отредактировали сообщение!'
)
@dp.user_removed()
async def user_removed(event: UserRemoved):
if not event.from_user:
return await bot.send_message(
chat_id=event.chat_id,
text=f'Неизвестный кикнул {event.user.first_name} 😢'
)
await bot.send_message(
chat_id=event.chat_id,
text=f'{event.from_user.first_name} кикнул {event.user.first_name} 😢'
)
@dp.user_added()
async def user_added(event: UserAdded):
if not event.chat:
return await bot.send_message(
chat_id=event.chat_id,
text=f'Чат приветствует вас, {event.user.first_name}!'
)
await bot.send_message(
chat_id=event.chat_id,
text=f'Чат "{event.chat.title}" приветствует вас, {event.user.first_name}!'
)
@dp.bot_stopped()
async def bot_stopped(event: BotStopped):
logging.info(event.from_user.full_name, 'остановил бота') # type: ignore
@dp.dialog_cleared()
async def dialog_cleared(event: DialogCleared):
logging.info(event.from_user.full_name, 'очистил историю чата с ботом') # type: ignore
@dp.dialog_muted()
async def dialog_muted(event: DialogMuted):
logging.info(event.from_user.full_name, 'отключил оповещения от чата бота до ', event.muted_until_datetime) # type: ignore
@dp.dialog_unmuted()
async def dialog_unmuted(event: DialogUnmuted):
logging.info(event.from_user.full_name, 'включил оповещения от чата бота') # type: ignore
@dp.dialog_unmuted()
async def dialog_removed(event: DialogUnmuted):
logging.info(event.from_user.full_name, 'удалил диалог с ботом') # type: ignore
@dp.message_chat_created()
async def message_chat_created(event: MessageChatCreated):
await bot.send_message(
chat_id=event.chat.chat_id,
text=f'Чат создан! Ссылка: {event.chat.link}'
)
async def main():
await dp.start_polling(bot)
if __name__ == '__main__':
asyncio.run(main())

39
examples/get_ids/main.py Normal file
View File

@@ -0,0 +1,39 @@
import asyncio
import logging
from maxapi import Bot, Dispatcher, F
from maxapi.enums.parse_mode import ParseMode
from maxapi.types import MessageCreated
logging.basicConfig(level=logging.INFO)
bot = Bot('тут_ваш_токен')
dp = Dispatcher()
@dp.message_created(F.message.link.type == 'forward')
async def get_ids_from_forward(event: MessageCreated):
text = (
'Информация о пересланном сообщении:\n\n'
f'Из чата: <b>{event.message.link.chat_id}</b>\n'
f'От пользователя: <b>{event.message.link.sender.user_id}</b>'
)
await event.message.reply(text)
@dp.message_created()
async def get_ids(event: MessageCreated):
text = (
f'Ваш ID: <b>{event.from_user.user_id}</b>\n'
f'ID этого чата: <b>{event.chat.chat_id}</b>'
)
await event.message.answer(text, parse_mode=ParseMode.HTML)
async def main():
await dp.start_polling(bot)
if __name__ == '__main__':
asyncio.run(main())

153
examples/keyboard/main.py Normal file
View File

@@ -0,0 +1,153 @@
import asyncio
import logging
from maxapi import Bot, Dispatcher
# Кнопки
from maxapi.types import (
ChatButton,
LinkButton,
CallbackButton,
RequestGeoLocationButton,
MessageButton,
ButtonsPayload, # Для постройки клавиатуры без InlineKeyboardBuilder
RequestContactButton,
OpenAppButton,
)
from maxapi.types import (
MessageCreated,
MessageCallback,
MessageChatCreated,
CommandStart,
Command
)
from maxapi.utils.inline_keyboard import InlineKeyboardBuilder
logging.basicConfig(level=logging.INFO)
bot = Bot('тут_ваш_токен')
dp = Dispatcher()
@dp.message_created(CommandStart())
async def echo(event: MessageCreated):
await event.message.answer(
(
'Привет! Мои команды:\n\n'
'/builder - Клавиатура из InlineKeyboardBuilder\n'
'/pyaload - Клавиатура из pydantic моделей\n'
)
)
@dp.message_created(Command('builder'))
async def builder(event: MessageCreated):
builder = InlineKeyboardBuilder()
builder.row(
ChatButton(
text="Создать чат",
chat_title='Test',
chat_description='Test desc'
),
LinkButton(
text="Документация MAX",
url="https://dev.max.ru/docs"
),
)
builder.row(
RequestGeoLocationButton(text="Геолокация"),
MessageButton(text="Сообщение"),
)
builder.row(
RequestContactButton(text="Контакт"),
OpenAppButton(
text="Приложение",
web_app=event.bot.me.username,
contact_id=event.bot.me.user_id
),
)
builder.row(
CallbackButton(
text='Callback',
payload='test',
)
)
await event.message.answer(
text='Клавиатура из InlineKeyboardBuilder',
attachments=[
builder.as_markup()
])
@dp.message_created(Command('payload'))
async def payload(event: MessageCreated):
buttons = [
[
# кнопку типа "chat" убрали из документации,
# возможны баги
ChatButton(
text="Создать чат",
chat_title='Test',
chat_description='Test desc'
),
LinkButton(
text="Документация MAX",
url="https://dev.max.ru/docs"
),
],
[
RequestGeoLocationButton(text="Геолокация"),
MessageButton(text="Сообщение"),
],
[
RequestContactButton(text="Контакт"),
OpenAppButton(
text="Приложение",
web_app=event.bot.me.username,
contact_id=event.bot.me.user_id
),
],
[
CallbackButton(
text='Callback',
payload='test',
)
]
]
buttons_payload = ButtonsPayload(buttons=buttons).pack()
await event.message.answer(
text='Клавиатура из pydantic моделей',
attachments=[
buttons_payload
])
@dp.message_chat_created()
async def message_chat_created(obj: MessageChatCreated):
await obj.bot.send_message(
chat_id=obj.chat.chat_id,
text=f'Чат создан! Ссылка: {obj.chat.link}'
)
@dp.message_callback()
async def message_callback(callback: MessageCallback):
await callback.message.answer('Вы нажали на Callback!')
async def main():
await dp.start_polling(bot)
if __name__ == '__main__':
asyncio.run(main())

View File

@@ -0,0 +1,48 @@
import asyncio
import logging
from maxapi import Bot, Dispatcher, F
from maxapi.types import MessageCreated
logging.basicConfig(level=logging.INFO)
bot = Bot('тут_ваш_токен')
dp = Dispatcher()
@dp.message_created(F.message.body.text == 'привет')
async def on_hello(event: MessageCreated):
await event.message.answer('Привет!')
@dp.message_created(F.message.body.text.lower().contains('помощь'))
async def on_help(event: MessageCreated):
await event.message.answer('Чем могу помочь?')
@dp.message_created(F.message.body.text.regexp(r'^\d{4}$'))
async def on_code(event: MessageCreated):
await event.message.answer('Принят 4-значный код')
@dp.message_created(F.message.body.attachments)
async def on_attachment(event: MessageCreated):
await event.message.answer('Получено вложение')
@dp.message_created(F.message.body.text.len() > 20)
async def on_long_text(event: MessageCreated):
await event.message.answer('Слишком длинное сообщение')
@dp.message_created(F.message.body.text.len() > 0)
async def on_non_empty(event: MessageCreated):
await event.message.answer('Вы что-то написали.')
async def main():
await dp.start_polling(bot)
if __name__ == '__main__':
asyncio.run(main())

View File

@@ -0,0 +1,41 @@
import asyncio
import logging
from typing import Any, Awaitable, Callable, Dict
from maxapi import Bot, Dispatcher
from maxapi.types import MessageCreated, Command, UpdateUnion
from maxapi.filters.middleware import BaseMiddleware
logging.basicConfig(level=logging.INFO)
bot = Bot(token='тут_ваш_токен')
dp = Dispatcher()
class CustomDataForRouterMiddleware(BaseMiddleware):
async def __call__(
self,
handler: Callable[[Any, Dict[str, Any]], Awaitable[Any]],
event_object: UpdateUnion,
data: Dict[str, Any],
) -> Any:
data['custom_data'] = f'Это ID того кто вызвал команду: {event_object.from_user.user_id}'
result = await handler(event_object, data)
return result
@dp.message_created(Command('custom_data'))
async def custom_data(event: MessageCreated, custom_data: str):
await event.message.answer(custom_data)
async def main():
dp.middleware(CustomDataForRouterMiddleware())
await dp.start_polling(bot)
if __name__ == '__main__':
asyncio.run(main())

View File

@@ -0,0 +1,62 @@
import asyncio
import logging
from typing import Any, Awaitable, Callable, Dict
from maxapi import Bot, Dispatcher
from maxapi.filters.middleware import BaseMiddleware
from maxapi.types import MessageCreated, Command, UpdateUnion
logging.basicConfig(level=logging.INFO)
bot = Bot(token='тут_ваш_токен')
dp = Dispatcher()
class CheckChatTitleMiddleware(BaseMiddleware):
async def __call__(
self,
handler: Callable[[Any, Dict[str, Any]], Awaitable[Any]],
event_object: UpdateUnion,
data: Dict[str, Any],
) -> Any:
if event_object.chat.title == 'MAXApi':
return await handler(event_object, data)
class CustomDataMiddleware(BaseMiddleware):
async def __call__(
self,
handler: Callable[[Any, Dict[str, Any]], Awaitable[Any]],
event_object: UpdateUnion,
data: Dict[str, Any],
) -> Any:
data['custom_data'] = f'Это ID того кто вызвал команду: {event_object.from_user.user_id}'
await handler(event_object, data)
@dp.message_created(Command('start'), CheckChatTitleMiddleware())
async def start(event: MessageCreated):
await event.message.answer('Это сообщение было отправлено, так как ваш чат называется "MAXApi"!')
@dp.message_created(Command('custom_data'), CustomDataMiddleware())
async def custom_data(event: MessageCreated, custom_data: str):
await event.message.answer(custom_data)
@dp.message_created(Command('many_middlewares'), CheckChatTitleMiddleware(), CustomDataMiddleware())
async def many_middlewares(event: MessageCreated, custom_data: str):
await event.message.answer('Это сообщение было отправлено, так как ваш чат называется "MAXApi"!')
await event.message.answer(custom_data)
async def main():
await dp.start_polling(bot)
if __name__ == '__main__':
asyncio.run(main())

Binary file not shown.

View File

@@ -0,0 +1,162 @@
import asyncio
import logging
from maxapi import Bot, Dispatcher, F
from maxapi.context import MemoryContext, State, StatesGroup
from maxapi.types import BotStarted, Command, MessageCreated, CallbackButton, MessageCallback, BotCommand
from maxapi.utils.inline_keyboard import InlineKeyboardBuilder
from router import router
logging.basicConfig(level=logging.INFO)
bot = Bot('тут_ваш_токен')
dp = Dispatcher()
dp.include_routers(router)
start_text = '''Пример чат-бота для MAX 💙
Мои команды:
/clear очищает ваш контекст
/state или /context показывают ваше контекстное состояние
/data показывает вашу контекстную память
'''
class Form(StatesGroup):
name = State()
age = State()
@dp.on_started()
async def _():
logging.info('Бот стартовал!')
@dp.bot_started()
async def bot_started(event: BotStarted):
await event.bot.send_message(
chat_id=event.chat_id,
text='Привет! Отправь мне /start'
)
@dp.message_created(Command('clear'))
async def hello(event: MessageCreated, context: MemoryContext):
await context.clear()
await event.message.answer(f"Ваш контекст был очищен!")
@dp.message_created(Command('data'))
async def hello(event: MessageCreated, context: MemoryContext):
data = await context.get_data()
await event.message.answer(f"Ваша контекстная память: {str(data)}")
@dp.message_created(Command('context'))
@dp.message_created(Command('state'))
async def hello(event: MessageCreated, context: MemoryContext):
data = await context.get_state()
await event.message.answer(f"Ваше контекстное состояние: {str(data)}")
@dp.message_created(Command('start'))
async def hello(event: MessageCreated):
builder = InlineKeyboardBuilder()
builder.row(
CallbackButton(
text='Ввести свое имя',
payload='btn_1'
),
CallbackButton(
text='Ввести свой возраст',
payload='btn_2'
)
)
builder.row(
CallbackButton(
text='Не хочу',
payload='btn_3'
)
)
await event.message.answer(
text=start_text,
attachments=[
builder.as_markup(),
] # Для MAX клавиатура это вложение,
) # поэтому она в списке вложений
@dp.message_callback(F.callback.payload == 'btn_1')
async def hello(event: MessageCallback, context: MemoryContext):
await context.set_state(Form.name)
await event.message.delete()
await event.message.answer(f'Отправьте свое имя:')
@dp.message_callback(F.callback.payload == 'btn_2')
async def hello(event: MessageCallback, context: MemoryContext):
await context.set_state(Form.age)
await event.message.delete()
await event.message.answer(f'Отправьте ваш возраст:')
@dp.message_callback(F.callback.payload == 'btn_3')
async def hello(event: MessageCallback, context: MemoryContext):
await event.message.delete()
await event.message.answer(f'Ну ладно 🥲')
@dp.message_created(F.message.body.text, Form.name)
async def hello(event: MessageCreated, context: MemoryContext):
await context.update_data(name=event.message.body.text)
data = await context.get_data()
await event.message.answer(f"Приятно познакомиться, {data['name'].title()}!")
@dp.message_created(F.message.body.text, Form.age)
async def hello(event: MessageCreated, context: MemoryContext):
await context.update_data(age=event.message.body.text)
await event.message.answer(f"Ого! А мне всего пару недель 😁")
async def main():
await bot.set_my_commands(
BotCommand(
name='/start',
description='Перезапустить бота'
),
BotCommand(
name='/clear',
description='Очищает ваш контекст'
),
BotCommand(
name='/state',
description='Показывают ваше контекстное состояние'
),
BotCommand(
name='/data',
description='Показывает вашу контекстную память'
),
BotCommand(
name='/context',
description='Показывают ваше контекстное состояние'
)
)
await dp.start_polling(bot)
# await dp.handle_webhook(
# bot=bot,
# host='localhost',
# port=8080
# )
if __name__ == '__main__':
asyncio.run(main())

View File

@@ -0,0 +1,24 @@
from maxapi import F, Router
from maxapi.types import Command, MessageCreated
from maxapi.types import InputMedia
router = Router()
file = __file__.split('\\')[-1]
@router.message_created(Command('router'))
async def hello(obj: MessageCreated):
await obj.message.answer(f"Пишу тебе из роута {file}")
# новая команда для примера, /media,
# пример использования: /media image.png (медиафайл берется указанному пути)
@router.message_created(Command('media'))
async def hello(event: MessageCreated):
await event.message.answer(
attachments=[
InputMedia(
path=event.message.body.text.replace('/media ', '')
)
]
)

View File

@@ -0,0 +1,4 @@
## Вебхуки
- [Высокоуровневый](https://github.com/love-apples/maxapi/tree/main/examples/webhook/high_level.py)
- [Низкоуровневый](https://github.com/love-apples/maxapi/tree/main/examples/webhook/low_level.py)

View File

@@ -0,0 +1,28 @@
import asyncio
import logging
from maxapi import Bot, Dispatcher
from maxapi.types import MessageCreated
logging.basicConfig(level=logging.INFO)
bot = Bot('тут_ваш_токен')
dp = Dispatcher()
@dp.message_created()
async def handle_message(event: MessageCreated):
await event.message.answer('Бот работает через вебхук!')
async def main():
await dp.handle_webhook(
bot=bot,
host='localhost',
port=8080,
log_level='critical' # Можно убрать для подробного логгирования
)
if __name__ == '__main__':
asyncio.run(main())

View File

@@ -0,0 +1,63 @@
import asyncio
import logging
try:
from fastapi import Request
from fastapi.responses import JSONResponse
except ImportError:
raise ImportError(
'\n\t Не установлен fastapi!'
'\n\t Выполните команду для установки fastapi: '
'\n\t pip install fastapi>=0.68.0'
'\n\t Или сразу все зависимости для работы вебхука:'
'\n\t pip install maxapi[webhook]'
)
from maxapi import Bot, Dispatcher
from maxapi.methods.types.getted_updates import process_update_webhook
from maxapi.types import MessageCreated
logging.basicConfig(level=logging.INFO)
bot = Bot('тут_ваш_токен')
dp = Dispatcher()
@dp.message_created()
async def handle_message(event: MessageCreated):
await event.message.answer('Бот работает через вебхук!')
# Регистрация обработчика
# для вебхука
@dp.webhook_post('/')
async def _(request: Request):
# Сериализация полученного запроса
event_json = await request.json()
# Десериализация полученного запроса
# в pydantic
event_object = await process_update_webhook(
event_json=event_json,
bot=bot
)
# ...свой код
print(f'Информация из вебхука: {event_json}')
# ...свой код
# Окончательная обработка запроса
await dp.handle(event_object)
# Ответ вебхука
return JSONResponse(content={'ok': True}, status_code=200)
async def main():
# Запуск сервера
await dp.init_serve(bot, log_level='critical')
if __name__ == '__main__':
asyncio.run(main())

10
maxapi/__init__.py Normal file
View File

@@ -0,0 +1,10 @@
from .bot import Bot
from .dispatcher import Dispatcher, Router
from .filters import F
__all__ = [
'Bot',
'Dispatcher',
'F',
'Router'
]

File diff suppressed because it is too large Load Diff

27
maxapi/client/default.py Normal file
View File

@@ -0,0 +1,27 @@
from aiohttp import ClientTimeout
class DefaultConnectionProperties:
'''
Класс для хранения параметров соединения по умолчанию для aiohttp-клиента.
Args:
timeout (int): Таймаут всего соединения в секундах (по умолчанию 5 * 30).
sock_connect (int): Таймаут установки TCP-соединения в секундах (по умолчанию 30).
**kwargs: Дополнительные параметры, которые будут сохранены как есть.
Attributes:
timeout (ClientTimeout): Экземпляр aiohttp.ClientTimeout с заданными параметрами.
kwargs (dict): Дополнительные параметры.
'''
def __init__(self, timeout: int = 5 * 30, sock_connect: int = 30, **kwargs):
'''
Инициализация параметров соединения.
Args:
timeout (int): Таймаут всего соединения в секундах.
sock_connect (int): Таймаут установки TCP-соединения в секундах.
**kwargs: Дополнительные параметры.
'''
self.timeout = ClientTimeout(total=timeout, sock_connect=sock_connect)
self.kwargs = kwargs

View File

View File

@@ -1,36 +1,219 @@
import aiohttp
from __future__ import annotations
import os
import mimetypes
from typing import TYPE_CHECKING, Any, Optional
import aiofiles
import puremagic
from pydantic import BaseModel
from aiohttp import ClientSession, ClientConnectionError, FormData
from ..exceptions.invalid_token import InvalidToken
from ..exceptions.max import MaxConnection
from ..types.errors import Error
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..enums.upload_type import UploadType
from ..loggers import logger_bot
if TYPE_CHECKING:
from ..bot import Bot
class BaseConnection:
"""
Базовый класс для всех методов API.
Содержит общую логику выполнения запроса (сериализация, отправка HTTP-запроса, обработка ответа).
"""
API_URL = 'https://botapi.max.ru'
RETRY_DELAY = 2
ATTEMPTS_COUNT = 5
AFTER_MEDIA_INPUT_DELAY = 2.0
def __init__(self) -> None:
"""
Инициализация BaseConnection.
Атрибуты:
bot (Optional[Bot]): Экземпляр бота.
session (Optional[ClientSession]): aiohttp-сессия.
after_input_media_delay (float): Задержка после ввода медиа.
"""
self.bot: Optional[Bot] = None
self.session: Optional[ClientSession] = None
self.after_input_media_delay: float = self.AFTER_MEDIA_INPUT_DELAY
async def request(
self,
method: HTTPMethod,
path: ApiPath,
model: BaseModel,
is_return_raw: bool = False,
**kwargs
):
async with aiohttp.ClientSession(self.API_URL) as s:
r = await s.request(
method=method.value,
url=path.value,
**kwargs
self,
method: HTTPMethod,
path: ApiPath | str,
model: BaseModel | Any = None,
is_return_raw: bool = False,
**kwargs
):
"""
Выполняет HTTP-запрос к API.
Args:
method (HTTPMethod): HTTP-метод (GET, POST и т.д.).
path (ApiPath | str): Путь до конечной точки.
model (BaseModel | Any, optional): Pydantic-модель для десериализации ответа, если is_return_raw=False.
is_return_raw (bool, optional): Если True — вернуть сырой ответ, иначе — результат десериализации.
**kwargs: Дополнительные параметры (query, headers, json).
Returns:
model | dict | Error: Объект модели, dict или ошибка.
Raises:
RuntimeError: Если бот не инициализирован.
MaxConnection: Ошибка соединения.
InvalidToken: Ошибка авторизации (401).
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
if not self.bot.session:
self.bot.session = ClientSession(
base_url=self.bot.API_URL,
timeout=self.bot.default_connection.timeout,
**self.bot.default_connection.kwargs
)
if not r.ok:
raw = await r.text()
return Error(code=r.status, text=raw)
try:
r = await self.bot.session.request(
method=method.value,
url=path.value if isinstance(path, ApiPath) else path,
**kwargs
)
except ClientConnectionError as e:
raise MaxConnection(f'Ошибка при отправке запроса: {e}')
if r.status == 401:
await self.bot.session.close()
raise InvalidToken('Неверный токен!')
if not r.ok:
raw = await r.json()
error = Error(code=r.status, raw=raw)
logger_bot.error(error)
return error
if is_return_raw: return raw
raw = await r.json()
return model(**raw)
if is_return_raw:
return raw
model = model(**raw) # type: ignore
if hasattr(model, 'message'):
attr = getattr(model, 'message')
if hasattr(attr, 'bot'):
attr.bot = self.bot
if hasattr(model, 'bot'):
model.bot = self.bot
return model
async def upload_file(
self,
url: str,
path: str,
type: UploadType
):
"""
Загружает файл на сервер.
Args:
url (str): URL загрузки.
path (str): Путь к файлу.
type (UploadType): Тип файла.
Returns:
str: Сырой .text() ответ от сервера.
"""
async with aiofiles.open(path, 'rb') as f:
file_data = await f.read()
basename = os.path.basename(path)
_, ext = os.path.splitext(basename)
form = FormData()
form.add_field(
name='data',
value=file_data,
filename=basename,
content_type=f"{type.value}/{ext.lstrip('.')}"
)
async with ClientSession() as session:
response = await session.post(
url=url,
data=form
)
return await response.text()
async def upload_file_buffer(
self,
filename: str,
url: str,
buffer: bytes,
type: UploadType
):
"""
Загружает файл из буфера.
Args:
filename (str): Имя файла.
url (str): URL загрузки.
buffer (bytes): Буфер данных.
type (UploadType): Тип файла.
Returns:
str: Сырой .text() ответ от сервера.
"""
try:
matches = puremagic.magic_string(buffer[:4096])
if matches:
mime_type = matches[0][1]
ext = mimetypes.guess_extension(mime_type) or ''
else:
mime_type = f"{type.value}/*"
ext = ''
except Exception:
mime_type = f"{type.value}/*"
ext = ''
basename = f'{filename}{ext}'
form = FormData()
form.add_field(
name='data',
value=buffer,
filename=basename,
content_type=mime_type
)
async with ClientSession() as session:
response = await session.post(
url=url,
data=form
)
return await response.text()

View File

@@ -0,0 +1,9 @@
from ..context.state_machine import State, StatesGroup
from .context import MemoryContext
__all__ = [
'State',
'StatesGroup',
'MemoryContext'
]

93
maxapi/context/context.py Normal file
View File

@@ -0,0 +1,93 @@
import asyncio
from typing import Any, Dict, Optional, Union
from ..context.state_machine import State
class MemoryContext:
"""
Контекст хранения данных пользователя с блокировками.
Args:
chat_id (int): Идентификатор чата
user_id (int): Идентификатор пользователя
"""
def __init__(self, chat_id: int, user_id: int):
self.chat_id = chat_id
self.user_id = user_id
self._context: Dict[str, Any] = {}
self._state: State | str | None = None
self._lock = asyncio.Lock()
async def get_data(self) -> dict[str, Any]:
"""
Возвращает текущий контекст данных.
Returns:
Словарь с данными контекста
"""
async with self._lock:
return self._context
async def set_data(self, data: dict[str, Any]):
"""
Полностью заменяет контекст данных.
Args:
data: Новый словарь контекста
"""
async with self._lock:
self._context = data
async def update_data(self, **kwargs):
"""
Обновляет контекст данных новыми значениями.
Args:
**kwargs: Пары ключ-значение для обновления
"""
async with self._lock:
self._context.update(kwargs)
async def set_state(self, state: Optional[Union[State, str]] = None):
"""
Устанавливает новое состояние.
Args:
state: Новое состояние или None для сброса
"""
async with self._lock:
self._state = state
async def get_state(self):
"""
Возвращает текущее состояние.
Returns:
Текущее состояние или None
"""
async with self._lock:
return self._state
async def clear(self):
"""
Очищает контекст и сбрасывает состояние.
"""
async with self._lock:
self._state = None
self._context = {}

View File

@@ -0,0 +1,41 @@
from typing import List
class State:
"""
Представляет отдельное состояние в FSM-группе.
При использовании внутри StatesGroup, автоматически присваивает уникальное имя в формате 'ИмяКласса:имя_поля'.
"""
def __init__(self):
self.name = None
def __set_name__(self, owner, attr_name):
self.name = f'{owner.__name__}:{attr_name}'
def __str__(self):
return self.name
class StatesGroup:
"""
Базовый класс для описания группы состояний FSM.
Атрибуты должны быть экземплярами State. Метод `states()` возвращает список всех состояний в виде строк.
"""
@classmethod
def states(cls) -> List[str]:
"""
Получить список всех состояний в формате 'ИмяКласса:имя_состояния'.
Returns:
Список строковых представлений состояний
"""
return [str(getattr(cls, attr)) for attr in dir(cls)
if isinstance(getattr(cls, attr), State)]

View File

@@ -1,59 +1,95 @@
from typing import Callable, List
from __future__ import annotations
import uvicorn
import asyncio
from fastapi import FastAPI, Request
from magic_filter import MagicFilter
import functools
from typing import Any, Awaitable, Callable, Dict, List, TYPE_CHECKING, Literal, Optional
from asyncio.exceptions import TimeoutError as AsyncioTimeoutError
from .filters import filter_m
from .types.updates import Update
from aiohttp import ClientConnectorError
from maxapi.exceptions.dispatcher import HandlerException
from .filters.filter import BaseFilter
from .filters.middleware import BaseMiddleware
from .filters.handler import Handler
from .context import MemoryContext
from .types.updates import UpdateUnion
from .types.errors import Error
from .methods.types.getted_updates import process_update_request, process_update_webhook
from .filters import filter_attrs
from .bot import Bot
from .enums.update import UpdateType
from .types.updates.bot_added import BotAdded
from .types.updates.bot_removed import BotRemoved
from .types.updates.bot_started import BotStarted
from .types.updates.chat_title_changed import ChatTitleChanged
from .types.updates.message_callback import MessageCallback
from .types.updates.message_chat_created import MessageChatCreated
from .types.updates.message_created import MessageCreated
from .types.updates.message_edited import MessageEdited
from .types.updates.message_removed import MessageRemoved
from .types.updates.user_added import UserAdded
from .types.updates.user_removed import UserRemoved
from .loggers import logger
from .loggers import logger_dp
class Handler:
try:
from fastapi import FastAPI, Request # type: ignore
from fastapi.responses import JSONResponse # type: ignore
FASTAPI_INSTALLED = True
except ImportError:
FASTAPI_INSTALLED = False
try:
from uvicorn import Config, Server # type: ignore
UVICORN_INSTALLED = True
except ImportError:
UVICORN_INSTALLED = False
def __init__(
self,
*args,
func_event: Callable,
update_type: UpdateType,
**kwargs
):
self.func_event = func_event
self.update_type = update_type
self.filters = []
if TYPE_CHECKING:
from magic_filter import MagicFilter
for arg in args:
if isinstance(arg, MagicFilter):
arg: MagicFilter = arg
self.filters.append(arg)
CONNECTION_RETRY_DELAY = 30
GET_UPDATES_RETRY_DELAY = 5
class Dispatcher:
def __init__(self):
self.event_handlers = []
self.bot = None
"""
Основной класс для обработки событий бота.
Обеспечивает запуск поллинга и вебхука, маршрутизацию событий,
применение middleware, фильтров и вызов соответствующих обработчиков.
"""
def __init__(self, router_id: str | None = None) -> None:
"""
Инициализация диспетчера.
Args:
router_id (str | None): Идентификатор роутера для логов.
"""
self.router_id = router_id
self.event_handlers: List[Handler] = []
self.contexts: List[MemoryContext] = []
self.routers: List[Router | Dispatcher] = []
self.filters: List[MagicFilter] = []
self.base_filters: List[BaseFilter] = []
self.middlewares: List[BaseMiddleware] = []
self.bot: Optional[Bot] = None
self.webhook_app: Optional[FastAPI] = None
self.on_started_func: Optional[Callable] = None
self.polling = False
self.message_created = Event(update_type=UpdateType.MESSAGE_CREATED, router=self)
self.bot_added = Event(update_type=UpdateType.BOT_ADDED, router=self)
self.bot_removed = Event(update_type=UpdateType.BOT_REMOVED, router=self)
self.bot_started = Event(update_type=UpdateType.BOT_STARTED, router=self)
self.bot_stopped = Event(update_type=UpdateType.BOT_STOPPED, router=self)
self.dialog_cleared = Event(update_type=UpdateType.DIALOG_CLEARED, router=self)
self.dialog_muted = Event(update_type=UpdateType.DIALOG_MUTED, router=self)
self.dialog_unmuted = Event(update_type=UpdateType.DIALOG_UNMUTED, router=self)
self.dialog_removed = Event(update_type=UpdateType.DIALOG_REMOVED, router=self)
self.chat_title_changed = Event(update_type=UpdateType.CHAT_TITLE_CHANGED, router=self)
self.message_callback = Event(update_type=UpdateType.MESSAGE_CALLBACK, router=self)
self.message_chat_created = Event(update_type=UpdateType.MESSAGE_CHAT_CREATED, router=self)
@@ -61,91 +97,497 @@ class Dispatcher:
self.message_removed = Event(update_type=UpdateType.MESSAGE_REMOVED, router=self)
self.user_added = Event(update_type=UpdateType.USER_ADDED, router=self)
self.user_removed = Event(update_type=UpdateType.USER_REMOVED, router=self)
self.on_started = Event(update_type=UpdateType.ON_STARTED, router=self)
def webhook_post(self, path: str):
def decorator(func):
if self.webhook_app is None:
try:
from fastapi import FastAPI # type: ignore
except ImportError:
raise ImportError(
'\n\t Не установлен fastapi!'
'\n\t Выполните команду для установки fastapi: '
'\n\t pip install fastapi>=0.68.0'
'\n\t Или сразу все зависимости для работы вебхука:'
'\n\t pip install maxapi[webhook]'
)
self.webhook_app = FastAPI()
return self.webhook_app.post(path)(func)
return decorator
async def check_me(self):
"""
Проверяет и логирует информацию о боте.
"""
me = await self.bot.get_me()
self.bot._me = me
logger_dp.info(f'Бот: @{me.username} first_name={me.first_name} id={me.user_id}')
def build_middleware_chain(
self,
middlewares: list[BaseMiddleware],
handler: Callable[[Any, dict[str, Any]], Awaitable[Any]]
) -> Callable[[Any, dict[str, Any]], Awaitable[Any]]:
"""
Формирует цепочку вызова middleware вокруг хендлера.
Args:
middlewares (list[BaseMiddleware]): Список middleware.
handler (Callable): Финальный обработчик.
Returns:
Callable: Обёрнутый обработчик.
"""
for mw in reversed(middlewares):
handler = functools.partial(mw, handler)
return handler
def include_routers(self, *routers: 'Router'):
for router in routers:
for event in router.event_handlers:
self.event_handlers.append(event)
"""
Добавляет указанные роутеры в диспетчер.
def handle_webhook(self, bot: Bot, host: str = '0.0.0.0', port: int = 8080):
Args:
*routers (Router): Роутеры для добавления.
"""
self.routers += [r for r in routers]
def outer_middleware(self, middleware: BaseMiddleware) -> None:
"""
Добавляет Middleware на первое место в списке.
Args:
middleware (BaseMiddleware): Middleware.
"""
self.middlewares.insert(0, middleware)
def middleware(self, middleware: BaseMiddleware) -> None:
"""
Добавляет Middleware в конец списка.
Args:
middleware (BaseMiddleware): Middleware.
"""
self.middlewares.append(middleware)
def filter(self, base_filter: BaseFilter) -> None:
"""
Добавляет фильтр в список.
Args:
base_filter (BaseFilter): Фильтр.
"""
self.base_filters.append(base_filter)
async def __ready(self, bot: Bot):
"""
Подготавливает диспетчер: сохраняет бота, регистрирует обработчики, вызывает on_started.
Args:
bot (Bot): Экземпляр бота.
"""
self.bot = bot
if self.polling and self.bot.auto_check_subscriptions:
response = await self.bot.get_subscriptions()
if response.subscriptions:
logger_subscriptions_text = ', '.join([s.url for s in response.subscriptions])
logger_dp.warning('БОТ ИГНОРИРУЕТ POLLING! Обнаружены установленные подписки: %s', logger_subscriptions_text)
await self.check_me()
self.routers += [self]
handlers_count = sum(len(router.event_handlers) for router in self.routers)
app = FastAPI()
logger_dp.info(f'{handlers_count} событий на обработку')
@app.post("/")
async def _(request: Request):
try:
event_json = await request.json()
event = Update(**event_json)
if self.on_started_func:
await self.on_started_func()
def __get_memory_context(self, chat_id: int, user_id: int):
"""
Возвращает существующий или создаёт новый MemoryContext по chat_id и user_id.
event_object = None
match event.update_type:
case UpdateType.BOT_ADDED:
event_object = BotAdded(**event_json)
case UpdateType.BOT_REMOVED:
event_object = BotRemoved(**event_json)
case UpdateType.BOT_STARTED:
event_object = BotStarted(**event_json)
case UpdateType.CHAT_TITLE_CHANGED:
event_object = ChatTitleChanged(**event_json)
case UpdateType.MESSAGE_CALLBACK:
event_object = MessageCallback(**event_json)
event_object.message.bot = self.bot
case UpdateType.MESSAGE_CHAT_CREATED:
event_object = MessageChatCreated(**event_json)
case UpdateType.MESSAGE_CREATED:
event_object = MessageCreated(**event_json)
event_object.message.bot = self.bot
case UpdateType.MESSAGE_EDITED:
event_object = MessageEdited(**event_json)
case UpdateType.MESSAGE_REMOVED:
event_object = MessageRemoved(**event_json)
case UpdateType.USER_ADDED:
event_object = UserAdded(**event_json)
case UpdateType.USER_REMOVED:
event_object = UserRemoved(**event_json)
Args:
chat_id (int): Идентификатор чата.
user_id (int): Идентификатор пользователя.
handlers: List[Handler] = self.event_handlers
for handler in handlers:
Returns:
MemoryContext: Контекст.
"""
if not handler.update_type == event.update_type:
for ctx in self.contexts:
if ctx.chat_id == chat_id and ctx.user_id == user_id:
return ctx
new_ctx = MemoryContext(chat_id, user_id)
self.contexts.append(new_ctx)
return new_ctx
async def call_handler(
self,
handler: Callable[[Any, dict[str, Any]], Awaitable[Any]],
event_object: UpdateType,
data: Dict[str, Any]
):
"""
Вызывает хендлер с нужными аргументами.
Args:
handler: Handler.
event_object: Объект события.
data: Данные для хендлера.
Returns:
None
"""
func_args = handler.func_event.__annotations__.keys()
kwargs_filtered = {k: v for k, v in data.items() if k in func_args}
if kwargs_filtered:
await handler.func_event(event_object, **kwargs_filtered)
else:
await handler.func_event(event_object)
async def process_base_filters(
self,
event: UpdateUnion,
filters: List[BaseFilter]
) -> Optional[Dict[str, Any]] | Literal[False]:
"""
Асинхронно применяет фильтры к событию.
Args:
event (UpdateUnion): Событие.
filters (List[BaseFilter]): Список фильтров.
Returns:
Optional[Dict[str, Any]] | Literal[False]: Словарь с результатом или False.
"""
data = {}
for _filter in filters:
result = await _filter(event)
if isinstance(result, dict):
data.update(result)
elif not result:
return result
return data
async def handle(self, event_object: UpdateUnion):
"""
Основной обработчик события. Применяет фильтры, middleware и вызывает нужный handler.
Args:
event_object (UpdateUnion): Событие.
"""
try:
ids = event_object.get_ids()
memory_context = self.__get_memory_context(*ids)
current_state = await memory_context.get_state()
kwargs = {'context': memory_context}
router_id = None
process_info = f'{event_object.update_type} | chat_id: {ids[0]}, user_id: {ids[1]}'
is_handled = False
for index, router in enumerate(self.routers):
if is_handled:
break
router_id = router.router_id or index
if router.filters:
if not filter_attrs(event_object, *router.filters):
continue
result_router_filter = await self.process_base_filters(
event=event_object,
filters=router.base_filters
)
if isinstance(result_router_filter, dict):
kwargs.update(result_router_filter)
elif not result_router_filter:
continue
for handler in router.event_handlers:
if not handler.update_type == event_object.update_type:
continue
if handler.filters:
if not filter_m(event_object, *handler.filters):
if not filter_attrs(event_object, *handler.filters):
continue
if handler.states:
if current_state not in handler.states:
continue
func_args = handler.func_event.__annotations__.keys()
if handler.base_filters:
result_filter = await self.process_base_filters(
event=event_object,
filters=handler.base_filters
)
if isinstance(result_filter, dict):
kwargs.update(result_filter)
elif not result_filter:
continue
if isinstance(router, Router):
full_middlewares = self.middlewares + router.middlewares + handler.middlewares
elif isinstance(router, Dispatcher):
full_middlewares = self.middlewares + handler.middlewares
handler_chain = self.build_middleware_chain(
full_middlewares,
functools.partial(self.call_handler, handler)
)
kwargs_filtered = {k: v for k, v in kwargs.items() if k in func_args}
try:
await handler_chain(event_object, kwargs_filtered)
except:
raise HandlerException(
handler_title=handler.func_event.__name__,
memory_context={
'data': await memory_context.get_data(),
'state': current_state
}
)
await handler.func_event(event_object)
logger_dp.info(f'Обработано: router_id: {router_id} | {process_info}')
is_handled = True
break
return True
except Exception as e:
print(e)
...
logger.info(f'{len(self.event_handlers)} event handlers started')
uvicorn.run(app, host=host, port=port, log_level='critical')
if not is_handled:
logger_dp.info(f'Проигнорировано: router_id: {router_id} | {process_info}')
except Exception as e:
logger_dp.error(f"Ошибка при обработке события: router_id: {router_id} | {process_info} | {e} ")
async def start_polling(self, bot: Bot):
"""
Запускает цикл получения обновлений (long polling).
Args:
bot (Bot): Экземпляр бота.
"""
self.polling = True
await self.__ready(bot)
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
while self.polling:
try:
events: Dict = await self.bot.get_updates()
except AsyncioTimeoutError:
continue
try:
if isinstance(events, Error):
logger_dp.info(f'Ошибка при получении обновлений: {events}, жду {GET_UPDATES_RETRY_DELAY} секунд')
await asyncio.sleep(GET_UPDATES_RETRY_DELAY)
continue
self.bot.marker_updates = events.get('marker')
processed_events = await process_update_request(
events=events,
bot=self.bot
)
for event in processed_events:
await self.handle(event)
except ClientConnectorError:
logger_dp.error(f'Ошибка подключения, жду {CONNECTION_RETRY_DELAY} секунд')
await asyncio.sleep(CONNECTION_RETRY_DELAY)
except Exception as e:
logger_dp.error(f'Общая ошибка при обработке событий: {e.__class__} - {e}')
async def handle_webhook(self, bot: Bot, host: str = 'localhost', port: int = 8080, **kwargs):
"""
Запускает FastAPI-приложение для приёма обновлений через вебхук.
Args:
bot (Bot): Экземпляр бота.
host (str): Хост сервера.
port (int): Порт сервера.
"""
if not FASTAPI_INSTALLED:
raise ImportError(
'\n\t Не установлен fastapi!'
'\n\t Выполните команду для установки fastapi: '
'\n\t pip install fastapi>=0.68.0'
'\n\t Или сразу все зависимости для работы вебхука:'
'\n\t pip install maxapi[webhook]'
)
elif not UVICORN_INSTALLED:
raise ImportError(
'\n\t Не установлен uvicorn!'
'\n\t Выполните команду для установки uvicorn: '
'\n\t pip install uvicorn>=0.15.0'
'\n\t Или сразу все зависимости для работы вебхука:'
'\n\t pip install maxapi[webhook]'
)
@self.webhook_post('/')
async def _(request: Request):
event_json = await request.json()
event_object = await process_update_webhook(
event_json=event_json,
bot=bot
)
await self.handle(event_object)
return JSONResponse(content={'ok': True}, status_code=200)
await self.init_serve(
bot=bot,
host=host,
port=port,
**kwargs
)
async def init_serve(self, bot: Bot, host: str = 'localhost', port: int = 8080, **kwargs):
"""
Запускает сервер для обработки вебхуков.
Args:
bot (Bot): Экземпляр бота.
host (str): Хост.
port (int): Порт.
"""
if not UVICORN_INSTALLED:
raise ImportError(
'\n\t Не установлен uvicorn!'
'\n\t Выполните команду для установки uvicorn: '
'\n\t pip install uvicorn>=0.15.0'
'\n\t Или сразу все зависимости для работы вебхука:'
'\n\t pip install maxapi[webhook]'
)
if self.webhook_app is None:
raise RuntimeError('webhook_app не инициализирован')
config = Config(app=self.webhook_app, host=host, port=port, **kwargs)
server = Server(config)
await self.__ready(bot)
await server.serve()
class Router(Dispatcher):
def __init__(self):
super().__init__()
"""
Роутер для группировки обработчиков событий.
"""
def __init__(self, router_id: str | None = None):
"""
Инициализация роутера.
Args:
router_id (str | None): Идентификатор роутера для логов.
"""
super().__init__(router_id)
class Event:
"""
Декоратор для регистрации обработчиков событий.
"""
def __init__(self, update_type: UpdateType, router: Dispatcher | Router):
"""
Инициализирует событие-декоратор.
Args:
update_type (UpdateType): Тип события.
router (Dispatcher | Router): Экземпляр роутера или диспетчера.
"""
self.update_type = update_type
self.router = router
def __call__(self, *args, **kwargs):
"""
Регистрирует функцию как обработчик события.
Returns:
Callable: Исходная функция.
"""
def decorator(func_event: Callable):
self.router.event_handlers.append(
Handler(
func_event=func_event,
update_type=self.update_type,
*args, **kwargs
if self.update_type == UpdateType.ON_STARTED:
self.router.on_started_func = func_event
else:
self.router.event_handlers.append(
Handler(
func_event=func_event,
update_type=self.update_type,
*args, **kwargs
)
)
)
return func_event
return decorator

0
maxapi/enums/__init__.py Normal file
View File

View File

@@ -1,7 +1,23 @@
from enum import Enum
class ApiPath(str, Enum):
"""
Перечисление всех доступных API-эндпоинтов.
Используется для унифицированного указания путей при отправке запросов.
"""
ME = '/me'
CHATS = '/chats'
MESSAGES = '/messages'
UPDATES = '/updates'
UPDATES = '/updates'
VIDEOS = '/videos'
ANSWERS = '/answers'
ACTIONS = '/actions'
PIN = '/pin'
MEMBERS = '/members'
ADMINS = '/admins'
UPLOADS = '/uploads'
SUBSCRIPTIONS = '/subscriptions'

View File

@@ -1,6 +1,14 @@
from enum import Enum
class AttachmentType(str, Enum):
"""
Типы вложений, поддерживаемые в сообщениях.
Используется для указания типа содержимого при отправке или обработке вложений.
"""
IMAGE = 'image'
VIDEO = 'video'
AUDIO = 'audio'
@@ -8,4 +16,5 @@ class AttachmentType(str, Enum):
STICKER = 'sticker'
CONTACT = 'contact'
INLINE_KEYBOARD = 'inline_keyboard'
LOCATION = 'location'
LOCATION = 'location'
SHARE = 'share'

View File

@@ -1,9 +1,18 @@
from enum import Enum
class ButtonType(Enum):
class ButtonType(str, Enum):
"""
Типы кнопок, доступных в интерфейсе бота.
Определяют поведение при нажатии на кнопку в сообщении.
"""
REQUEST_CONTACT = 'request_contact'
CALLBACK = 'callback'
LINK = 'link'
REQUEST_GEO_LOCATION = 'request_geo_location'
CHAT = 'chat'
MESSAGE = 'message'
OPEN_APP = 'open_app'

View File

@@ -0,0 +1,17 @@
from enum import Enum
class ChatPermission(str, Enum):
"""
Права доступа пользователя в чате.
Используются для управления разрешениями при добавлении участников или изменении настроек чата.
"""
READ_ALL_MESSAGES = 'read_all_messages'
ADD_REMOVE_MEMBERS = 'add_remove_members'
ADD_ADMINS = 'add_admins'
CHANGE_CHAT_INFO = 'change_chat_info'
PIN_MESSAGE = 'pin_message'
WRITE = 'write'

View File

@@ -0,0 +1,16 @@
from enum import Enum
class ChatStatus(str, Enum):
"""
Статус чата относительно пользователя или системы.
Используется для отображения текущего состояния чата или определения доступных действий.
"""
ACTIVE = 'active'
REMOVED = 'removed'
LEFT = 'left'
CLOSED = 'closed'
SUSPENDED = 'suspended'

View File

@@ -2,5 +2,12 @@ from enum import Enum
class ChatType(str, Enum):
"""
Тип чата.
Используется для различения личных и групповых чатов.
"""
DIALOG = 'dialog'
CHAT = 'chat'

View File

@@ -2,7 +2,15 @@ from enum import Enum
class HTTPMethod(str, Enum):
"""
HTTP-методы, поддерживаемые клиентом API.
Используются при выполнении запросов к серверу.
"""
POST = 'POST'
GET = 'GET'
PATCH = 'PATCH'
PUT = 'PUT'
DELETE = 'DELETE'

View File

@@ -1,6 +1,14 @@
from enum import Enum
class Intent(str, Enum):
"""
Тип интента (намерения) кнопки.
Используется для стилизации и логической классификации пользовательских действий.
"""
DEFAULT = 'default'
POSITIVE = 'positive'
NEGATIVE = 'negative'

View File

@@ -2,5 +2,12 @@ from enum import Enum
class MessageLinkType(str, Enum):
"""
Тип связи между сообщениями.
Используется для указания типа привязки: пересылка или ответ.
"""
FORWARD = 'forward'
REPLY = 'reply'

View File

@@ -1,5 +1,13 @@
from enum import Enum
class ParseMode(str, Enum):
"""
Формат разметки текста сообщений.
Используется для указания способа интерпретации стилей (жирный, курсив, ссылки и т.д.).
"""
MARKDOWN = 'markdown'
HTML = 'html'

View File

@@ -0,0 +1,17 @@
from enum import Enum
class SenderAction(str, Enum):
"""
Действия отправителя, отображаемые получателю в интерфейсе.
Используются для имитации активности (например, "печатает...") перед отправкой сообщения или медиа.
"""
TYPING_ON = 'typing_on'
SENDING_PHOTO = 'sending_photo'
SENDING_VIDEO = 'sending_video'
SENDING_AUDIO = 'sending_audio'
SENDING_FILE = 'sending_file'
MARK_SEEN = 'mark_seen'

View File

@@ -2,6 +2,13 @@ from enum import Enum
class TextStyle(Enum):
"""
Стили текста, применяемые в сообщениях.
Используются для форматирования и выделения частей текста в сообщении.
"""
UNDERLINE = 'underline'
STRONG = 'strong'
EMPHASIZED = 'emphasized'

View File

@@ -1,6 +1,14 @@
from enum import Enum
class UpdateType(str, Enum):
"""
Типы обновлений (ивентов) от API.
Используются для обработки различных событий в боте или чате.
"""
MESSAGE_CREATED = 'message_created'
BOT_ADDED = 'bot_added'
BOT_REMOVED = 'bot_removed'
@@ -11,4 +19,12 @@ class UpdateType(str, Enum):
MESSAGE_EDITED = 'message_edited'
MESSAGE_REMOVED = 'message_removed'
USER_ADDED = 'user_added'
USER_REMOVED = 'user_removed'
USER_REMOVED = 'user_removed'
BOT_STOPPED = 'bot_stopped'
DIALOG_CLEARED = 'dialog_cleared'
DIALOG_MUTED = 'dialog_muted'
DIALOG_UNMUTED = 'dialog_unmuted'
DIALOG_REMOVED = 'dialog_removed'
# Для начинки диспатчера
ON_STARTED = 'on_started'

View File

@@ -0,0 +1,15 @@
from enum import Enum
class UploadType(str, Enum):
"""
Типы загружаемых файлов.
Используются для указания категории контента при загрузке на сервер.
"""
IMAGE = 'image'
VIDEO = 'video'
AUDIO = 'audio'
FILE = 'file'

View File

@@ -0,0 +1,17 @@
class HandlerException(Exception):
def __init__(self, handler_title: str, *args, **kwargs):
self.handler_title = handler_title
self.extra = kwargs
message = f'Обработчик: {handler_title!r}'
if args:
message += f', детали: {args}'
if kwargs:
message += f', другое: {kwargs}'
super().__init__(message)

View File

@@ -0,0 +1,4 @@
class NotAvailableForDownload(Exception):
...

View File

@@ -0,0 +1,4 @@
class InvalidToken(Exception):
...

11
maxapi/exceptions/max.py Normal file
View File

@@ -0,0 +1,11 @@
class MaxConnection(Exception):
...
class MaxUploadFileFailed(Exception):
...
class MaxIconParamsException(Exception):
...

View File

@@ -1,53 +1,27 @@
from magic_filter import MagicFilter
from magic_filter.operations.call import CallOperation as mf_call
from magic_filter.operations.function import FunctionOperation as mf_func
from magic_filter.operations.comparator import ComparatorOperation as mf_comparator
from .filter import BaseFilter
F = MagicFilter()
__all__ = [
'BaseFilter'
]
def filter_m(obj, *magic_args):
def filter_attrs(obj: object, *filters: MagicFilter) -> bool:
"""
Применяет один или несколько фильтров MagicFilter к объекту.
Args:
obj (object): Объект, к которому применяются фильтры (например, event или message).
*filters (MagicFilter): Один или несколько выражений MagicFilter.
Returns:
bool: True, если все фильтры возвращают True, иначе False.
"""
try:
for arg in magic_args:
attr_last = None
method_found = False
operations = arg._operations
if isinstance(operations[-1], mf_call):
operations = operations[:len(operations)-2]
method_found = True
elif isinstance(operations[-1], mf_func):
operations = operations[:len(operations)-1]
method_found = True
elif isinstance(operations[-1], mf_comparator):
operations = operations[:len(operations)-1]
for element in operations:
if attr_last is None:
attr_last = getattr(obj, element.name)
else:
attr_last = getattr(attr_last, element.name)
if attr_last is None:
break
if isinstance(arg._operations[-1], mf_comparator):
return attr_last == arg._operations[-1].right
if not method_found:
return bool(attr_last)
if attr_last is None:
return False
if isinstance(arg._operations[-1], mf_func):
func_operation: mf_func = arg._operations[-1]
return func_operation.resolve(attr_last, attr_last)
else:
method = getattr(attr_last, arg._operations[-2].name)
args = arg._operations[-1].args
return method(*args)
except Exception as e:
...
return all(f.resolve(obj) for f in filters)
except Exception:
return False

View File

@@ -0,0 +1,175 @@
from __future__ import annotations
from typing import Any, ClassVar, List, Optional, Type, TYPE_CHECKING
from magic_filter import MagicFilter
from pydantic import BaseModel
from ..types.updates.message_callback import MessageCallback
from ..types.updates import UpdateUnion
from .filter import BaseFilter
PAYLOAD_MAX = 1024
class CallbackPayload(BaseModel):
"""
Базовый класс для сериализации/десериализации callback payload.
Атрибуты:
prefix (str): Префикс для payload (используется при pack/unpack) (по умолчанию название класса).
separator (str): Разделитель между значениями (по умолчанию '|').
"""
if TYPE_CHECKING:
prefix: ClassVar[str]
separator: ClassVar[str]
def __init_subclass__(cls, **kwargs: Any) -> None:
"""
Автоматически проставляет prefix и separator при наследовании.
"""
cls.prefix = kwargs.get('prefix', str(cls.__name__))
cls.separator = kwargs.get('separator', '|')
def pack(self) -> str:
"""
Собирает данные payload в строку для передачи в callback payload.
Raises:
ValueError: Если в значении встречается разделитель или payload слишком длинный.
Returns:
str: Сериализованный payload.
"""
values = [self.prefix]
for name in self.attrs():
value = getattr(self, name)
str_value = '' if value is None else str(value)
if self.separator in str_value:
raise ValueError(
f'Символ разделителя "{self.separator}" не должен встречаться в значении поля {name}'
)
values.append(str_value)
data = self.separator.join(values)
if len(data.encode()) > PAYLOAD_MAX:
raise ValueError(
f'Payload слишком длинный! Максимум: {PAYLOAD_MAX} байт'
)
return data
@classmethod
def unpack(cls, data: str):
"""
Десериализует payload из строки.
Args:
data (str): Строка payload (из callback payload).
Raises:
ValueError: Некорректный prefix или количество аргументов.
Returns:
CallbackPayload: Экземпляр payload с заполненными полями.
"""
parts = data.split(cls.separator)
if not parts[0] == cls.prefix:
raise ValueError('Некорректный prefix')
field_names = cls.attrs()
if not len(parts) - 1 == len(field_names):
raise ValueError(
f'Ожидалось {len(field_names)} аргументов, получено {len(parts) - 1}'
)
kwargs = dict(zip(field_names, parts[1:]))
return cls(**kwargs)
@classmethod
def attrs(cls) -> List[str]:
"""
Возвращает список полей для сериализации/десериализации (исключая prefix и separator).
Returns:
List[str]: Имена полей модели.
"""
return [
k for k in cls.model_fields.keys()
if k not in ('prefix', 'separator')
]
@classmethod
def filter(cls, rule: Optional[MagicFilter] = None) -> PayloadFilter:
"""
Создаёт PayloadFilter для фильтрации callback-ивентов по payload.
Args:
rule (Optional[MagicFilter]): Фильтр на payload.
Returns:
PayloadFilter: Экземпляр фильтра для хэндлера.
"""
return PayloadFilter(model=cls, rule=rule)
class PayloadFilter(BaseFilter):
"""
Фильтр для MessageCallback по payload.
"""
def __init__(self, model: Type[CallbackPayload], rule: Optional[MagicFilter]):
"""
Args:
model (Type[CallbackPayload]): Класс payload для распаковки.
rule (Optional[MagicFilter]): Фильтр (условие) для payload.
"""
self.model = model
self.rule = rule
async def __call__(self, event: UpdateUnion):
"""
Проверяет event на MessageCallback и применяет фильтр к payload.
Args:
event (UpdateUnion): Обновление/событие.
Returns:
dict | bool: dict с payload при совпадении, иначе False.
"""
if not isinstance(event, MessageCallback):
return False
if not event.callback.payload:
return False
try:
payload = self.model.unpack(event.callback.payload)
except Exception:
return False
if not self.rule or self.rule.resolve(payload):
return {'payload': payload}
return False

116
maxapi/filters/command.py Normal file
View File

@@ -0,0 +1,116 @@
from typing import List, Tuple
from ..types.updates import UpdateUnion
from ..filters.filter import BaseFilter
from ..types.updates.message_created import MessageCreated
class Command(BaseFilter):
"""
Фильтр сообщений на соответствие команде.
Args:
commands (str | List[str]): Ожидаемая команда или список команд без префикса.
prefix (str, optional): Префикс команды (по умолчанию '/').
check_case (bool, optional): Учитывать регистр при сравнении (по умолчанию False).
Attributes:
commands (List[str]): Список команд без префикса.
prefix (str): Префикс команды.
check_case (bool): Флаг чувствительности к регистру.
"""
def __init__(self, commands: str | List[str], prefix: str = '/', check_case: bool = False):
"""
Инициализация фильтра команд.
"""
if isinstance(commands, str):
self.commands = [commands]
else:
self.commands = commands
self.prefix = prefix
self.check_case = check_case
if not check_case:
self.commands = [cmd.lower() for cmd in self.commands]
def parse_command(self, text: str) -> Tuple[str, List[str]]:
"""
Извлекает команду из текста.
Args:
text (str): Текст сообщения.
Returns:
Optional[str]: Найденная команда с префиксом, либо None.
"""
args = text.split()
first = args[0]
if not first.startswith(self.prefix):
return '', []
return first[len(self.prefix):], args
async def __call__(self, event: UpdateUnion):
"""
Проверяет, соответствует ли сообщение заданной(ым) команде(ам).
Args:
event (MessageCreated): Событие сообщения.
Returns:
bool: True, если команда совпадает, иначе False.
"""
if not isinstance(event, MessageCreated):
return False
text = event.message.body.text
if not text:
return False
parsed_command, args = self.parse_command(text)
if not parsed_command:
return False
if not self.check_case:
if parsed_command.lower() in [commands.lower() for commands in self.commands]:
return {'args': args}
else:
return False
if parsed_command in self.commands:
return {'args': args}
return False
class CommandStart(Command):
"""
Фильтр для команды /start.
Args:
prefix (str, optional): Префикс команды (по умолчанию '/').
check_case (bool, optional): Учитывать регистр (по умолчанию False).
"""
def __init__(self, prefix = '/', check_case = False):
super().__init__(
'start',
prefix,
check_case
)
async def __call__(self, event):
return await super().__call__(event)

21
maxapi/filters/filter.py Normal file
View File

@@ -0,0 +1,21 @@
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from ..types.updates import UpdateUnion
class BaseFilter:
"""
Базовый класс для фильтров.
Определяет интерфейс фильтрации событий.
Потомки должны переопределять метод __call__.
Methods:
__call__(event): Асинхронная проверка события на соответствие фильтру.
"""
async def __call__(self, event: UpdateUnion) -> bool | dict:
return True

59
maxapi/filters/handler.py Normal file
View File

@@ -0,0 +1,59 @@
from typing import Callable, List, Optional
from magic_filter import MagicFilter
from ..filters.filter import BaseFilter
from ..filters.middleware import BaseMiddleware
from ..context.state_machine import State
from ..enums.update import UpdateType
from ..loggers import logger_dp
class Handler:
"""
Обработчик события.
Связывает функцию-обработчик с типом события, состояниями и фильтрами.
"""
def __init__(
self,
*args,
func_event: Callable,
update_type: UpdateType,
**kwargs
):
"""
Создаёт обработчик события.
Args:
*args: Список фильтров (MagicFilter, State, Command, BaseFilter, BaseMiddleware).
func_event (Callable): Функция-обработчик.
update_type (UpdateType): Тип обновления.
"""
self.func_event: Callable = func_event
self.update_type: UpdateType = update_type
self.filters: Optional[List[MagicFilter]] = []
self.base_filters: Optional[List[BaseFilter]] = []
self.states: Optional[List[State]] = []
self.middlewares: List[BaseMiddleware] = []
for arg in args:
if isinstance(arg, MagicFilter):
self.filters.append(arg)
elif isinstance(arg, State):
self.states.append(arg)
elif isinstance(arg, BaseMiddleware):
self.middlewares.append(arg)
elif isinstance(arg, BaseFilter):
self.base_filters.append(arg)
else:
logger_dp.info(
f'Неизвестный фильтр `{arg}` при регистрации `{func_event.__name__}`'
)

View File

@@ -0,0 +1,30 @@
from typing import Any, Callable, Awaitable
class BaseMiddleware:
"""
Базовый класс для мидлварей.
Используется для обработки события до и после вызова хендлера.
"""
async def __call__(
self,
handler: Callable[[Any, dict[str, Any]], Awaitable[Any]],
event_object: Any,
data: dict[str, Any]
) -> Any:
"""
Вызывает хендлер с переданным событием и данными.
Args:
handler (Callable): Хендлер события.
event_object (Any): Событие.
data (dict): Дополнительные данные.
Returns:
Any: Результат работы хендлера.
"""
return await handler(event_object, data)

View File

@@ -1,4 +1,5 @@
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('bot')
logger_bot = logging.getLogger('bot')
logger_connection = logging.getLogger('connection')
logger_dp = logging.getLogger('dispatcher')

View File

View File

@@ -0,0 +1,65 @@
from typing import TYPE_CHECKING, Any, Dict, List, Optional
from .types.added_admin_chat import AddedListAdminChat
from ..types.users import ChatAdmin
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class AddAdminChat(BaseConnection):
"""
Класс для добавления списка администраторов в чат через API.
Args:
bot (Bot): Экземпляр бота, через который выполняется запрос.
chat_id (int): Идентификатор чата.
admins (List[ChatAdmin]): Список администраторов для добавления.
marker (int, optional): Маркер для пагинации или дополнительных настроек. По умолчанию None.
"""
def __init__(
self,
bot: 'Bot',
chat_id: int,
admins: List[ChatAdmin],
marker: Optional[int] = None
):
self.bot = bot
self.chat_id = chat_id
self.admins = admins
self.marker = marker
async def fetch(self) -> AddedListAdminChat:
"""
Выполняет HTTP POST запрос для добавления администраторов в чат.
Формирует JSON с данными администраторов и отправляет запрос на соответствующий API-эндпоинт.
Returns:
AddedListAdminChat: Результат операции с информацией об успешности.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
json: Dict[str, Any] = {}
json['admins'] = [admin.model_dump() for admin in self.admins]
json['marker'] = self.marker
return await super().request(
method=HTTPMethod.POST,
path=ApiPath.CHATS.value + '/' + str(self.chat_id) + ApiPath.MEMBERS + ApiPath.ADMINS,
model=AddedListAdminChat,
params=self.bot.params,
json=json
)

View File

@@ -0,0 +1,61 @@
from typing import TYPE_CHECKING, Any, Dict, List
from ..methods.types.added_members_chat import AddedMembersChat
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class AddMembersChat(BaseConnection):
"""
Класс для добавления участников в чат через API.
Args:
bot (Bot): Экземпляр бота, через который выполняется запрос.
chat_id (int): Идентификатор целевого чата.
user_ids (List[int]): Список ID пользователей для добавления в чат.
"""
def __init__(
self,
bot: 'Bot',
chat_id: int,
user_ids: List[int],
):
self.bot = bot
self.chat_id = chat_id
self.user_ids = user_ids
async def fetch(self) -> AddedMembersChat:
"""
Отправляет POST-запрос на добавление пользователей в чат.
Формирует JSON с ID пользователей и вызывает базовый метод запроса.
Returns:
AddedMembersChat: Результат операции с информацией об успешности добавления.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
json: Dict[str, Any] = {}
json['user_ids'] = self.user_ids
return await super().request(
method=HTTPMethod.POST,
path=ApiPath.CHATS.value + '/' + str(self.chat_id) + ApiPath.MEMBERS,
model=AddedMembersChat,
params=self.bot.params,
json=json
)

View File

@@ -1,8 +1,9 @@
from typing import Any, Dict, List, TYPE_CHECKING, Optional
from ..types.attachments.image import PhotoAttachmentRequestPayload
from typing import Any, Dict, List, TYPE_CHECKING
from ..types.users import BotCommand, User
from ..types.users import User
from ..types.command import BotCommand
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
@@ -15,13 +16,25 @@ if TYPE_CHECKING:
class ChangeInfo(BaseConnection):
"""
Класс для изменения информации о боте.
Args:
bot (Bot): Объект бота
name (str, optional): Новое имя бота
description (str, optional): Новое описание
commands (List[BotCommand], optional): Список команд
photo (PhotoAttachmentRequestPayload, optional): Данные фото
"""
def __init__(
self,
bot: 'Bot',
name: str = None,
description: str = None,
commands: List[BotCommand] = None,
photo: Dict[str, Any] = None
name: Optional[str] = None,
description: Optional[str] = None,
commands: Optional[List[BotCommand]] = None,
photo: Optional[PhotoAttachmentRequestPayload] = None
):
self.bot = bot
self.name = name
@@ -29,13 +42,27 @@ class ChangeInfo(BaseConnection):
self.commands = commands
self.photo = photo
async def request(self) -> User:
json = {}
async def fetch(self) -> User:
"""Отправляет запрос на изменение информации о боте.
if self.name: json['name'] = self.name
if self.description: json['description'] = self.description
if self.commands: json['commands'] = [command.model_dump() for command in self.commands]
if self.photo: json['photo'] = self.photo
Returns:
User: Объект с обновленными данными бота
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
json: Dict[str, Any] = {}
if self.name:
json['name'] = self.name
if self.description:
json['description'] = self.description
if self.commands:
json['commands'] = [command.model_dump() for command in self.commands]
if self.photo:
json['photo'] = self.photo.model_dump()
return await super().request(
method=HTTPMethod.PATCH,

View File

@@ -0,0 +1,49 @@
from typing import TYPE_CHECKING
from ..methods.types.deleted_bot_from_chat import DeletedBotFromChat
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class DeleteMeFromMessage(BaseConnection):
"""
Класс для удаления бота из участников указанного чата.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
chat_id (int): Идентификатор чата, из которого нужно удалить бота.
"""
def __init__(
self,
bot: 'Bot',
chat_id: int,
):
self.bot = bot
self.chat_id = chat_id
async def fetch(self) -> DeletedBotFromChat:
"""
Отправляет DELETE-запрос для удаления бота из чата.
Returns:
DeletedBotFromChat: Результат операции удаления.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
return await super().request(
method=HTTPMethod.DELETE,
path=ApiPath.CHATS + '/' + str(self.chat_id) + ApiPath.MEMBERS + ApiPath.ME,
model=DeletedBotFromChat,
params=self.bot.params,
)

View File

@@ -0,0 +1,49 @@
from typing import TYPE_CHECKING
from ..methods.types.deleted_chat import DeletedChat
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class DeleteChat(BaseConnection):
"""
Класс для удаления чата через API.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
chat_id (int): Идентификатор чата, который необходимо удалить.
"""
def __init__(
self,
bot: 'Bot',
chat_id: int
):
self.bot = bot
self.chat_id = chat_id
async def fetch(self) -> DeletedChat:
"""
Отправляет DELETE-запрос для удаления указанного чата.
Returns:
DeletedChat: Результат операции удаления чата.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
return await super().request(
method=HTTPMethod.DELETE,
path=ApiPath.CHATS.value + '/' + str(self.chat_id),
model=DeletedChat,
params=self.bot.params
)

View File

@@ -0,0 +1,55 @@
from typing import TYPE_CHECKING
from ..methods.types.deleted_message import DeletedMessage
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class DeleteMessage(BaseConnection):
"""
Класс для удаления сообщения через API.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
message_id (str): Идентификатор сообщения, которое нужно удалить.
"""
def __init__(
self,
bot: 'Bot',
message_id: str,
):
self.bot = bot
self.message_id = message_id
async def fetch(self) -> DeletedMessage:
"""
Выполняет DELETE-запрос для удаления сообщения.
Использует параметр message_id для идентификации сообщения.
Returns:
DeletedMessage: Результат операции удаления сообщения.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
params = self.bot.params.copy()
params['message_id'] = self.message_id
return await super().request(
method=HTTPMethod.DELETE,
path=ApiPath.MESSAGES,
model=DeletedMessage,
params=params,
)

View File

@@ -0,0 +1,50 @@
from typing import TYPE_CHECKING
from ..methods.types.deleted_pin_message import DeletedPinMessage
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class DeletePinMessage(BaseConnection):
"""
Класс для удаления закреплённого сообщения в чате через API.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
chat_id (int): Идентификатор чата, из которого нужно удалить закреплённое сообщение.
"""
def __init__(
self,
bot: 'Bot',
chat_id: int,
):
self.bot = bot
self.chat_id = chat_id
async def fetch(self) -> DeletedPinMessage:
"""
Выполняет DELETE-запрос для удаления закреплённого сообщения.
Returns:
DeletedPinMessage: Результат операции удаления закреплённого сообщения.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
return await super().request(
method=HTTPMethod.DELETE,
path=ApiPath.CHATS + '/' + str(self.chat_id) + ApiPath.PIN,
model=DeletedPinMessage,
params=self.bot.params,
)

View File

@@ -0,0 +1,46 @@
from typing import TYPE_CHECKING
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class DownloadMedia(BaseConnection):
"""
Класс для скачивания медиафайлов.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
media_url (str): Ссылка на медиа.
media_token (str): Токен медиа.
"""
def __init__(
self,
bot: 'Bot',
path: str,
media_url: str,
media_token: str
):
self.bot = bot
self.path = path
self.media_url = media_url
self.media_token = media_token
async def fetch(self) -> int:
"""
Выполняет GET-запрос для скачивания медиафайла
Returns:
int: Код операции.
"""
return await super().download_file(
path=self.path,
url=self.media_url,
token=self.media_token
)

View File

@@ -0,0 +1,99 @@
from logging import getLogger
from typing import TYPE_CHECKING, Any, Dict, Optional
from collections import Counter
from ..exceptions.max import MaxIconParamsException
from ..types.attachments.image import PhotoAttachmentRequestPayload
from ..types.chats import Chat
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
logger = getLogger(__name__)
if TYPE_CHECKING:
from ..bot import Bot
class EditChat(BaseConnection):
"""
Класс для редактирования информации о чате через API.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
chat_id (int): Идентификатор чата для редактирования.
icon (PhotoAttachmentRequestPayload, optional): Новый значок (иконка) чата.
title (str, optional): Новое название чата.
pin (str, optional): Идентификатор закреплённого сообщения.
notify (bool, optional): Включение или отключение уведомлений (по умолчанию True).
"""
def __init__(
self,
bot: 'Bot',
chat_id: int,
icon: Optional[PhotoAttachmentRequestPayload] = None,
title: Optional[str] = None,
pin: Optional[str] = None,
notify: Optional[bool] = None,
):
self.bot = bot
self.chat_id = chat_id
self.icon = icon
self.title = title
self.pin = pin
self.notify = notify
async def fetch(self) -> Chat:
"""
Выполняет PATCH-запрос для обновления параметров чата.
Валидация:
- Проверяется, что в `icon` атрибуты модели взаимоисключающие (в модели должно быть ровно 2 поля с None).
- Если условие не выполнено, логируется ошибка и запрос не отправляется.
Returns:
Chat: Обновлённый объект чата.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
json: Dict[str, Any] = {}
if self.icon:
dump = self.icon.model_dump()
counter = Counter(dump.values())
if None not in counter or \
not counter[None] == 2:
raise MaxIconParamsException(
'Все атрибуты модели Icon являются взаимоисключающими | '
'https://dev.max.ru/docs-api/methods/PATCH/chats/-chatId-'
)
json['icon'] = dump
if self.title:
json['title'] = self.title
if self.pin:
json['pin'] = self.pin
if self.notify:
json['notify'] = self.notify
return await super().request(
method=HTTPMethod.PATCH,
path=ApiPath.CHATS.value + '/' + str(self.chat_id),
model=Chat,
params=self.bot.params,
json=json
)

View File

@@ -1,16 +1,21 @@
from __future__ import annotations
import asyncio
from ..types.errors import Error
from typing import Any, Dict, List, TYPE_CHECKING, Optional
from typing import List, TYPE_CHECKING
from aiomax.enums.parse_mode import ParseMode
from ..utils.message import process_input_media
from .types.edited_message import EditedMessage
from ..types.message import NewMessageLink
from ..types.attachments.attachment import Attachment
from ..types.input_media import InputMedia, InputMediaBuffer
from ..enums.parse_mode import ParseMode
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..types.message import NewMessageLink
from .types.edited_message import EditedMessage
from ..connection.base import BaseConnection
from ..loggers import logger_bot
if TYPE_CHECKING:
@@ -18,15 +23,29 @@ if TYPE_CHECKING:
class EditMessage(BaseConnection):
"""
Класс для редактирования существующего сообщения через API.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
message_id (str): Идентификатор сообщения для редактирования.
text (str, optional): Новый текст сообщения.
attachments (List[Attachment | InputMedia | InputMediaBuffer], optional): Список вложений для сообщения.
link (NewMessageLink, optional): Связь с другим сообщением (ответ или пересылка).
notify (bool, optional): Отправлять ли уведомление о сообщении (по умолчанию True).
parse_mode (ParseMode, optional): Формат разметки текста (markdown, html и т.д.).
"""
def __init__(
self,
bot: 'Bot',
bot: Bot,
message_id: str,
text: str = None,
attachments: List[Attachment] = None,
link: NewMessageLink = None,
notify: bool = True,
parse_mode: ParseMode = None
text: Optional[str] = None,
attachments: Optional[List[Attachment | InputMedia | InputMediaBuffer]] = None,
link: Optional[NewMessageLink] = None,
notify: Optional[bool] = None,
parse_mode: Optional[ParseMode] = None
):
self.bot = bot
self.message_id = message_id
@@ -36,24 +55,69 @@ class EditMessage(BaseConnection):
self.notify = notify
self.parse_mode = parse_mode
async def request(self) -> 'EditedMessage':
async def fetch(self) -> Optional[EditedMessage | Error]:
"""
Выполняет PUT-запрос для обновления сообщения.
Формирует тело запроса на основе переданных параметров и отправляет запрос к API.
Returns:
EditedMessage: Обновлённое сообщение.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
params = self.bot.params.copy()
json = {}
json: Dict[str, Any] = {'attachments': []}
params['message_id'] = self.message_id
if not self.text is None: json['text'] = self.text
if self.attachments: json['attachments'] = \
[att.model_dump() for att in self.attachments]
if not self.link is None: json['link'] = self.link.model_dump()
if not self.notify is None: json['notify'] = self.notify
if not self.parse_mode is None: json['format'] = self.parse_mode.value
if self.text is not None:
json['text'] = self.text
if self.attachments:
for att in self.attachments:
return await super().request(
method=HTTPMethod.PUT,
path=ApiPath.MESSAGES,
model=EditedMessage,
params=params,
json=json
)
if isinstance(att, InputMedia) or isinstance(att, InputMediaBuffer):
input_media = await process_input_media(
base_connection=self,
bot=self.bot,
att=att
)
json['attachments'].append(
input_media.model_dump()
)
else:
json['attachments'].append(att.model_dump())
if self.link is not None:
json['link'] = self.link.model_dump()
if self.notify is not None:
json['notify'] = self.notify
if self.parse_mode is not None:
json['format'] = self.parse_mode.value
await asyncio.sleep(self.bot.after_input_media_delay)
response = None
for attempt in range(self.ATTEMPTS_COUNT):
response = await super().request(
method=HTTPMethod.PUT,
path=ApiPath.MESSAGES,
model=EditedMessage,
params=params,
json=json
)
if isinstance(response, Error):
if response.raw.get('code') == 'attachment.not.ready':
logger_bot.info(f'Ошибка при отправке загруженного медиа, попытка {attempt+1}, жду {self.RETRY_DELAY} секунды')
await asyncio.sleep(self.RETRY_DELAY)
continue
return response
return response

View File

@@ -0,0 +1,50 @@
from typing import TYPE_CHECKING
from ..types.chats import Chat
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class GetChatById(BaseConnection):
"""
Класс для получения информации о чате по его идентификатору.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
id (int): Идентификатор чата.
"""
def __init__(
self,
bot: 'Bot',
id: int
):
self.bot = bot
self.id = id
async def fetch(self) -> Chat:
"""
Выполняет GET-запрос для получения данных чата.
Returns:
Chat: Объект чата с полной информацией.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
return await super().request(
method=HTTPMethod.GET,
path=ApiPath.CHATS.value + '/' + str(self.id),
model=Chat,
params=self.bot.params
)

View File

@@ -0,0 +1,60 @@
from re import findall
from typing import TYPE_CHECKING
from ..types.chats import Chat
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class GetChatByLink(BaseConnection):
"""
Класс для получения информации о чате по ссылке.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
link (str): Ссылка на чат (с содержанием @ или без).
Attributes:
link (list[str]): Список валидных частей ссылки.
PATTERN_LINK (str): Регулярное выражение для парсинга ссылки.
"""
PATTERN_LINK = r'@?[a-zA-Z]+[a-zA-Z0-9-_]*'
def __init__(
self,
bot: 'Bot',
link: str
):
self.bot = bot
self.link = findall(self.PATTERN_LINK, link)
if not self.link:
return
async def fetch(self) -> Chat:
"""
Выполняет GET-запрос для получения данных чата по ссылке.
Returns:
Chat: Объект с информацией о чате.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
return await super().request(
method=HTTPMethod.GET,
path=ApiPath.CHATS.value + '/' + self.link[-1],
model=Chat,
params=self.bot.params
)

View File

@@ -1,11 +1,7 @@
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Optional
from ..types.chats import Chats
from ..types.users import User
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
@@ -17,13 +13,53 @@ if TYPE_CHECKING:
class GetChats(BaseConnection):
def __init__(self, bot: 'Bot'):
self.bot = bot
"""
Класс для получения списка чатов через API.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
count (int, optional): Максимальное количество чатов для получения. По умолчанию 50.
marker (int, optional): Маркер для постраничной навигации. По умолчанию None.
Attributes:
bot (Bot): Экземпляр бота.
count (int): Количество чатов для запроса.
marker (int | None): Маркер для пагинации.
"""
def __init__(
self,
bot: 'Bot',
count: int = 50,
marker: Optional[int] = None
):
self.bot = bot
self.count = count
self.marker = marker
async def fetch(self) -> Chats:
"""
Выполняет GET-запрос для получения списка чатов.
Returns:
Chats: Объект с данными по списку чатов.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
params = self.bot.params.copy()
params['count'] = self.count
if self.marker:
params['marker'] = self.marker
async def request(self) -> Chats:
return await super().request(
method=HTTPMethod.GET,
path=ApiPath.CHATS,
model=Chats,
params=self.bot.params
params=params
)

View File

@@ -0,0 +1,54 @@
from typing import TYPE_CHECKING
from ..methods.types.getted_list_admin_chat import GettedListAdminChat
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class GetListAdminChat(BaseConnection):
"""
Класс для получения списка администраторов чата через API.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
chat_id (int): Идентификатор чата.
Attributes:
bot (Bot): Экземпляр бота.
chat_id (int): Идентификатор чата.
"""
def __init__(
self,
bot: 'Bot',
chat_id: int
):
self.bot = bot
self.chat_id = chat_id
async def fetch(self) -> GettedListAdminChat:
"""
Выполняет GET-запрос для получения списка администраторов указанного чата.
Returns:
GettedListAdminChat: Объект с информацией о администраторах чата.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
return await super().request(
method=HTTPMethod.GET,
path=ApiPath.CHATS.value + '/' + str(self.chat_id) + ApiPath.MEMBERS + ApiPath.ADMINS,
model=GettedListAdminChat,
params=self.bot.params
)

View File

@@ -1,9 +1,5 @@
from typing import TYPE_CHECKING
from ..types.chats import Chats
from ..types.users import User
from ..enums.http_method import HTTPMethod
@@ -17,10 +13,29 @@ if TYPE_CHECKING:
class GetMe(BaseConnection):
"""
Класс для получения информации о боте.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
"""
def __init__(self, bot: 'Bot'):
self.bot = bot
async def request(self) -> Chats:
async def fetch(self) -> User:
"""
Выполняет GET-запрос для получения данных о боте.
Returns:
User: Объект пользователя с полной информацией.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
return await super().request(
method=HTTPMethod.GET,
path=ApiPath.ME,

View File

@@ -0,0 +1,54 @@
from typing import TYPE_CHECKING
from ..types.chats import ChatMember
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class GetMeFromChat(BaseConnection):
"""
Класс для получения информации о текущем боте в конкретном чате.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
chat_id (int): Идентификатор чата.
Attributes:
bot (Bot): Экземпляр бота.
chat_id (int): Идентификатор чата.
"""
def __init__(
self,
bot: 'Bot',
chat_id: int
):
self.bot = bot
self.chat_id = chat_id
async def fetch(self) -> ChatMember:
"""
Выполняет GET-запрос для получения информации о боте в указанном чате.
Returns:
ChatMember: Информация о боте как участнике чата.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
return await super().request(
method=HTTPMethod.GET,
path=ApiPath.CHATS + '/' + str(self.chat_id) + ApiPath.MEMBERS + ApiPath.ME,
model=ChatMember,
params=self.bot.params
)

View File

@@ -0,0 +1,79 @@
from typing import TYPE_CHECKING, List, Optional
from ..methods.types.getted_members_chat import GettedMembersChat
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class GetMembersChat(BaseConnection):
"""
Класс для получения списка участников чата через API.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
chat_id (int): Идентификатор чата.
user_ids (List[str], optional): Список ID пользователей для фильтрации. По умолчанию None.
marker (int, optional): Маркер для пагинации (начальная позиция). По умолчанию None.
count (int, optional): Максимальное количество участников для получения. По умолчанию None.
Attributes:
bot (Bot): Экземпляр бота.
chat_id (int): Идентификатор чата.
user_ids (List[int] | None): Список ID пользователей для фильтра.
marker (int | None): Позиция для пагинации.
count (int | None): Максимальное количество участников.
"""
def __init__(
self,
bot: 'Bot',
chat_id: int,
user_ids: Optional[List[int]] = None,
marker: Optional[int] = None,
count: Optional[int] = None,
):
self.bot = bot
self.chat_id = chat_id
self.user_ids = user_ids
self.marker = marker
self.count = count
async def fetch(self) -> GettedMembersChat:
"""
Выполняет GET-запрос для получения участников чата с опциональной фильтрацией.
Формирует параметры запроса с учётом фильтров и передаёт их базовому методу.
Returns:
GettedMembersChat: Объект с данными по участникам чата.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
params = self.bot.params.copy()
if self.user_ids:
params['user_ids'] = ','.join([str(user_id) for user_id in self.user_ids])
if self.marker:
params['marker'] = self.marker
if self.count:
params['marker'] = self.count
return await super().request(
method=HTTPMethod.GET,
path=ApiPath.CHATS.value + '/' + str(self.chat_id) + ApiPath.MEMBERS,
model=GettedMembersChat,
params=params
)

View File

@@ -0,0 +1,49 @@
from datetime import datetime
from typing import TYPE_CHECKING, List, Optional, Union
from ..types.message import Message
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class GetMessage(BaseConnection):
"""
Класс для получения сообщения.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
message_id (str, optional): ID сообщения (mid), чтобы получить одно сообщение в чате.
"""
def __init__(
self,
bot: 'Bot',
message_id: Optional[str] = None,
):
self.bot = bot
self.message_id = message_id
async def fetch(self) -> Message:
"""
Выполняет GET-запрос для получения сообщения.
Returns:
Message: Объект с полученным сообщением.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
return await super().request(
method=HTTPMethod.GET,
path=ApiPath.MESSAGES + '/' + self.message_id,
model=Message,
params=self.bot.params
)

View File

@@ -1,6 +1,5 @@
from typing import TYPE_CHECKING
from datetime import datetime
from typing import TYPE_CHECKING, List, Optional, Union
from ..types.message import Messages
from ..enums.http_method import HTTPMethod
@@ -13,14 +12,78 @@ if TYPE_CHECKING:
class GetMessages(BaseConnection):
def __init__(self, bot: 'Bot', chat_id: int = None):
"""
Класс для получения сообщений из чата через API.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
chat_id (int): Идентификатор чата.
message_ids (List[str], optional): Список идентификаторов сообщений для выборки. По умолчанию None.
from_time (datetime | int, optional): Временная метка начала выборки сообщений (timestamp или datetime). По умолчанию None.
to_time (datetime | int, optional): Временная метка конца выборки сообщений (timestamp или datetime). По умолчанию None.
count (int, optional): Максимальное количество сообщений для получения. По умолчанию 50.
Attributes:
bot (Bot): Экземпляр бота.
chat_id (int): Идентификатор чата.
message_ids (List[str] | None): Фильтр по идентификаторам сообщений.
from_time (datetime | int | None): Начальная временная метка.
to_time (datetime | int | None): Конечная временная метка.
count (int): Максимальное число сообщений.
"""
def __init__(
self,
bot: 'Bot',
chat_id: Optional[int] = None,
message_ids: Optional[List[str]] = None,
from_time: Optional[Union[datetime, int]] = None,
to_time: Optional[Union[datetime, int]] = None,
count: int = 50,
):
self.bot = bot
self.chat_id = chat_id
self.message_ids = message_ids
self.from_time = from_time
self.to_time = to_time
self.count = count
async def request(self) -> Messages:
async def fetch(self) -> Messages:
"""
Выполняет GET-запрос для получения сообщений с учётом параметров фильтрации.
Преобразует datetime в UNIX timestamp при необходимости.
Returns:
Messages: Объект с полученными сообщениями.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
params = self.bot.params.copy()
if self.chat_id: params['chat_id'] = self.chat_id
if self.chat_id:
params['chat_id'] = self.chat_id
if self.message_ids:
params['message_ids'] = ','.join(self.message_ids)
if self.from_time:
if isinstance(self.from_time, datetime):
params['from_time'] = int(self.from_time.timestamp())
else:
params['from_time'] = self.from_time
if self.to_time:
if isinstance(self.to_time, datetime):
params['to_time'] = int(self.to_time.timestamp())
else:
params['to_time'] = self.to_time
params['count'] = self.count
return await super().request(
method=HTTPMethod.GET,

View File

@@ -0,0 +1,49 @@
from typing import TYPE_CHECKING
from .types.getted_pineed_message import GettedPin
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class GetPinnedMessage(BaseConnection):
"""
Класс для получения закреплённого сообщения в указанном чате.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
chat_id (int): Идентификатор чата.
"""
def __init__(
self,
bot: 'Bot',
chat_id: int,
):
self.bot = bot
self.chat_id = chat_id
async def fetch(self) -> GettedPin:
"""
Выполняет GET-запрос для получения закреплённого сообщения.
Returns:
GettedPin: Объект с информацией о закреплённом сообщении.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
return await super().request(
method=HTTPMethod.GET,
path=ApiPath.CHATS + '/' + str(self.chat_id) + ApiPath.PIN,
model=GettedPin,
params=self.bot.params
)

View File

@@ -0,0 +1,44 @@
from typing import TYPE_CHECKING
from ..methods.types.getted_subscriptions import GettedSubscriptions
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class GetSubscriptions(BaseConnection):
"""
Если ваш бот получает данные через WebHook, этот класс возвращает список всех подписок.
"""
def __init__(
self,
bot: 'Bot',
):
self.bot = bot
async def fetch(self) -> GettedSubscriptions:
"""
Отправляет запрос на получение списка всех подписок.
Returns:
GettedSubscriptions: Объект со списком подписок
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
return await super().request(
method=HTTPMethod.GET,
path=ApiPath.SUBSCRIPTIONS,
model=GettedSubscriptions,
params=self.bot.params
)

View File

@@ -0,0 +1,62 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Dict
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class GetUpdates(BaseConnection):
"""
Класс для получения обновлений (updates) из API.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
limit (int, optional): Максимальное количество обновлений для получения. По умолчанию 100.
Attributes:
bot (Bot): Экземпляр бота.
limit (int): Лимит на количество обновлений.
"""
def __init__(
self,
bot: Bot,
limit: int = 100,
):
self.bot = bot
self.limit = limit
async def fetch(self) -> Dict:
"""
Выполняет GET-запрос для получения обновлений с указанным лимитом.
Возвращает необработанный JSON с обновлениями.
Returns:
UpdateUnion: Объединённый тип данных обновлений.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
params = self.bot.params.copy()
params['limit'] = self.limit
event_json = await super().request(
method=HTTPMethod.GET,
path=ApiPath.UPDATES,
model=None,
params=params,
is_return_raw=True
)
return event_json

View File

@@ -0,0 +1,59 @@
from typing import TYPE_CHECKING
from ..methods.types.getted_upload_url import GettedUploadUrl
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..enums.upload_type import UploadType
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class GetUploadURL(BaseConnection):
"""
Класс для получения URL загрузки файла определённого типа.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
type (UploadType): Тип загружаемого файла (например, image, video и т.д.).
"""
def __init__(
self,
bot: 'Bot',
type: UploadType
):
self.bot = bot
self.type = type
async def fetch(self) -> GettedUploadUrl:
"""
Выполняет POST-запрос для получения URL загрузки файла.
Возвращает объект с данными URL.
Returns:
GettedUploadUrl: Результат с URL для загрузки.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
params = self.bot.params.copy()
params['type'] = self.type.value
return await super().request(
method=HTTPMethod.POST,
path=ApiPath.UPLOADS,
model=GettedUploadUrl,
params=params,
)

View File

@@ -0,0 +1,50 @@
from typing import TYPE_CHECKING
from ..types.attachments.video import Video
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class GetVideo(BaseConnection):
"""
Класс для получения информации о видео по его токену.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
video_token (str): Токен видео для запроса.
"""
def __init__(
self,
bot: 'Bot',
video_token: str
):
self.bot = bot
self.video_token = video_token
async def fetch(self) -> Video:
"""
Выполняет GET-запрос для получения данных видео по токену.
Returns:
Video: Объект с информацией о видео.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
return await super().request(
method=HTTPMethod.GET,
path=ApiPath.VIDEOS.value + '/' + self.video_token,
model=Video,
params=self.bot.params,
)

View File

@@ -0,0 +1,70 @@
from typing import TYPE_CHECKING, Any, Dict, Optional
from .types.pinned_message import PinnedMessage
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class PinMessage(BaseConnection):
"""
Класс для закрепления сообщения в чате.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
chat_id (int): Идентификатор чата, в котором закрепляется сообщение.
message_id (str): Идентификатор сообщения для закрепления.
notify (bool, optional): Отправлять ли уведомление о закреплении (по умолчанию True).
Attributes:
bot (Bot): Экземпляр бота.
chat_id (int): Идентификатор чата.
message_id (str): Идентификатор закрепляемого сообщения.
notify (bool): Флаг отправки уведомления.
"""
def __init__(
self,
bot: 'Bot',
chat_id: int,
message_id: str,
notify: Optional[bool] = None
):
self.bot = bot
self.chat_id = chat_id
self.message_id = message_id
self.notify = notify
async def fetch(self) -> PinnedMessage:
"""
Выполняет PUT-запрос для закрепления сообщения в чате.
Формирует тело запроса с ID сообщения и флагом уведомления.
Returns:
PinnedMessage: Объект с информацией о закреплённом сообщении.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
json: Dict[str, Any] = {}
json['message_id'] = self.message_id
json['notify'] = self.notify
return await super().request(
method=HTTPMethod.PUT,
path=ApiPath.CHATS + '/' + str(self.chat_id) + ApiPath.PIN,
model=PinnedMessage,
params=self.bot.params,
json=json
)

View File

@@ -0,0 +1,59 @@
from typing import TYPE_CHECKING
from .types.removed_admin import RemovedAdmin
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class RemoveAdmin(BaseConnection):
"""
Класс для отмены прав администратора в чате.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
chat_id (int): Идентификатор чата.
user_id (int): Идентификатор пользователя.
Attributes:
bot (Bot): Экземпляр бота.
chat_id (int): Идентификатор чата.
user_id (int): Идентификатор пользователя.
"""
def __init__(
self,
bot: 'Bot',
chat_id: int,
user_id: int
):
self.bot = bot
self.chat_id = chat_id
self.user_id = user_id
async def fetch(self) -> RemovedAdmin:
"""
Выполняет DELETE-запрос для отмены прав администратора в чате.
Returns:
RemovedAdmin: Объект с результатом отмены прав администратора.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
return await super().request(
method=HTTPMethod.DELETE,
path=ApiPath.CHATS + '/' + str(self.chat_id) + \
ApiPath.MEMBERS + ApiPath.ADMINS + '/' + str(self.user_id),
model=RemovedAdmin,
params=self.bot.params,
)

View File

@@ -0,0 +1,71 @@
from typing import TYPE_CHECKING
from .types.removed_member_chat import RemovedMemberChat
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class RemoveMemberChat(BaseConnection):
"""
Класс для удаления участника из чата с опцией блокировки.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
chat_id (int): Идентификатор чата.
user_id (int): Идентификатор пользователя, которого необходимо удалить.
block (bool, optional): Блокировать пользователя после удаления. По умолчанию False.
Attributes:
bot (Bot): Экземпляр бота.
chat_id (int): Идентификатор чата.
user_id (int): Идентификатор пользователя.
block (bool): Флаг блокировки пользователя.
"""
def __init__(
self,
bot: 'Bot',
chat_id: int,
user_id: int,
block: bool = False,
):
self.bot = bot
self.chat_id = chat_id
self.user_id = user_id
self.block = block
async def fetch(self) -> RemovedMemberChat:
"""
Выполняет DELETE-запрос для удаления пользователя из чата.
Параметр `block` определяет, будет ли пользователь заблокирован после удаления.
Returns:
RemovedMemberChat: Результат удаления участника.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
params = self.bot.params.copy()
params['chat_id'] = self.chat_id
params['user_id'] = self.user_id
params['block'] = str(self.block).lower()
return await super().request(
method=HTTPMethod.DELETE,
path=ApiPath.CHATS.value + '/' + str(self.chat_id) + ApiPath.MEMBERS,
model=RemovedMemberChat,
params=params,
)

View File

@@ -0,0 +1,66 @@
from typing import TYPE_CHECKING, Any, Dict, Optional
from ..methods.types.sended_action import SendedAction
from ..enums.sender_action import SenderAction
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class SendAction(BaseConnection):
"""
Класс для отправки действия пользователя (например, индикатора печати) в чат.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
chat_id (int | None): Идентификатор чата. Если None, действие не отправляется.
action (SenderAction, optional): Тип действия. По умолчанию SenderAction.TYPING_ON.
Attributes:
bot (Bot): Экземпляр бота.
chat_id (int | None): Идентификатор чата.
action (SenderAction): Тип действия.
"""
def __init__(
self,
bot: 'Bot',
chat_id: Optional[int] = None,
action: SenderAction = SenderAction.TYPING_ON
):
self.bot = bot
self.chat_id = chat_id
self.action = action
async def fetch(self) -> SendedAction:
"""
Выполняет POST-запрос для отправки действия в указанный чат.
Returns:
SendedAction: Результат выполнения запроса.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
json: Dict[str, Any] = {}
json['action'] = self.action.value
return await super().request(
method=HTTPMethod.POST,
path=ApiPath.CHATS + '/' + str(self.chat_id) + ApiPath.ACTIONS,
model=SendedAction,
params=self.bot.params,
json=json
)

View File

@@ -0,0 +1,78 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any, Dict, Optional
from ..methods.types.sended_callback import SendedCallback
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
from ..types.message import Message
class SendCallback(BaseConnection):
"""
Класс для отправки callback-ответа с опциональным сообщением и уведомлением.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
callback_id (str): Идентификатор callback.
message (Message, optional): Сообщение для отправки в ответе.
notification (str, optional): Текст уведомления.
Attributes:
bot (Bot): Экземпляр бота.
callback_id (str): Идентификатор callback.
message (Message | None): Сообщение для отправки.
notification (str | None): Текст уведомления.
"""
def __init__(
self,
bot: 'Bot',
callback_id: str,
message: Optional[Message] = None,
notification: Optional[str] = None
):
self.bot = bot
self.callback_id = callback_id
self.message = message
self.notification = notification
async def fetch(self) -> SendedCallback:
"""
Выполняет POST-запрос для отправки callback-ответа.
Возвращает результат отправки.
Returns:
SendedCallback: Объект с результатом отправки callback.
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
params = self.bot.params.copy()
params['callback_id'] = self.callback_id
json: Dict[str, Any] = {}
if self.message:
json['message'] = self.message.model_dump()
if self.notification:
json['notification'] = self.notification
return await super().request(
method=HTTPMethod.POST,
path=ApiPath.ANSWERS,
model=SendedCallback,
params=params,
json=json
)

View File

@@ -1,68 +1,135 @@
from typing import List, TYPE_CHECKING
import asyncio
from typing import Any, Dict, List, TYPE_CHECKING, Optional
from ..enums.parse_mode import ParseMode
from ..utils.message import process_input_media
from .types.sended_message import SendedMessage
from ..types.errors import Error
from ..types.message import NewMessageLink
from ..types.input_media import InputMedia, InputMediaBuffer
from ..types.attachments.attachment import Attachment
from ..enums.parse_mode import ParseMode
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from .types.sended_message import SendedMessage
from ..connection.base import BaseConnection
from ..loggers import logger_bot
if TYPE_CHECKING:
from ..bot import Bot
class SendMessage(BaseConnection):
"""
Класс для отправки сообщения в чат или пользователю с поддержкой вложений и форматирования.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
chat_id (int, optional): Идентификатор чата, куда отправлять сообщение.
user_id (int, optional): Идентификатор пользователя, если нужно отправить личное сообщение.
text (str, optional): Текст сообщения.
attachments (List[Attachment | InputMedia | InputMediaBuffer], optional): Список вложений к сообщению.
link (NewMessageLink, optional): Связь с другим сообщением (например, ответ или пересылка).
notify (bool, optional): Отправлять ли уведомление о сообщении. По умолчанию True.
parse_mode (ParseMode, optional): Режим разбора текста (например, Markdown, HTML).
"""
def __init__(
self,
bot: 'Bot',
chat_id: int = None,
user_id: int = None,
disable_link_preview: bool = False,
text: str = None,
attachments: List[Attachment] = None,
link: NewMessageLink = None,
notify: bool = True,
parse_mode: ParseMode = None
chat_id: Optional[int] = None,
user_id: Optional[int] = None,
text: Optional[str] = None,
attachments: Optional[List[Attachment | InputMedia | InputMediaBuffer]] = None,
link: Optional[NewMessageLink] = None,
notify: Optional[bool] = None,
parse_mode: Optional[ParseMode] = None
):
self.bot = bot
self.chat_id = chat_id
self.user_id = user_id
self.disable_link_preview = disable_link_preview
self.text = text
self.attachments = attachments
self.link = link
self.notify = notify
self.parse_mode = parse_mode
async def request(self) -> 'SendedMessage':
async def fetch(self) -> Optional[SendedMessage | Error]:
"""
Отправляет сообщение с вложениями (если есть), с обработкой задержки готовности вложений.
Возвращает результат отправки или ошибку.
Возвращаемое значение:
SendedMessage или Error
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
params = self.bot.params.copy()
json = {}
json: Dict[str, Any] = {'attachments': []}
if self.chat_id: params['chat_id'] = self.chat_id
elif self.user_id: params['user_id'] = self.user_id
if self.chat_id:
params['chat_id'] = self.chat_id
elif self.user_id:
params['user_id'] = self.user_id
json['text'] = self.text
json['disable_link_preview'] = str(self.disable_link_preview).lower()
if self.attachments: json['attachments'] = \
[att.model_dump() for att in self.attachments]
HAS_INPUT_MEDIA = False
if not self.link is None: json['link'] = self.link.model_dump()
if not self.notify is None: json['notify'] = self.notify
if not self.parse_mode is None: json['format'] = self.parse_mode.value
if self.attachments:
for att in self.attachments:
if isinstance(att, (InputMedia, InputMediaBuffer)):
HAS_INPUT_MEDIA = True
input_media = await process_input_media(
base_connection=self,
bot=self.bot,
att=att
)
json['attachments'].append(
input_media.model_dump()
)
else:
json['attachments'].append(att.model_dump())
if self.link is not None:
json['link'] = self.link.model_dump()
json['notify'] = self.notify
if self.parse_mode is not None:
json['format'] = self.parse_mode.value
if HAS_INPUT_MEDIA:
await asyncio.sleep(self.bot.after_input_media_delay)
return await super().request(
method=HTTPMethod.POST,
path=ApiPath.MESSAGES,
model=SendedMessage,
params=params,
json=json
)
response = None
for attempt in range(self.ATTEMPTS_COUNT):
response = await super().request(
method=HTTPMethod.POST,
path=ApiPath.MESSAGES,
model=SendedMessage,
params=params,
json=json
)
if isinstance(response, Error):
if response.raw.get('code') == 'attachment.not.ready':
logger_bot.info(f'Ошибка при отправке загруженного медиа, попытка {attempt+1}, жду {self.RETRY_DELAY} секунды')
await asyncio.sleep(self.RETRY_DELAY)
continue
return response
return response

View File

@@ -0,0 +1,70 @@
from typing import TYPE_CHECKING, Any, Dict, List, Optional
from ..methods.types.subscribed import Subscribed
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..enums.update import UpdateType
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class SubscribeWebhook(BaseConnection):
"""
Подписывает бота на получение обновлений через WebHook.
После вызова этого метода бот будет получать уведомления о новых событиях в чатах на указанный URL.
Ваш сервер должен прослушивать один из следующих портов: `80`, `8080`, `443`, `8443`, `16384`-`32383`.
Args:
bot (Bot): Экземпляр бота для выполнения запроса.
url (str): URL HTTP(S)-эндпойнта вашего бота. Должен начинаться с http(s)://
update_types (Optional[List[str]]): Список типов обновлений, которые ваш бот хочет получать. Для полного списка типов см. объект
secret (str): От 5 до 256 символов. Cекрет, который должен быть отправлен в заголовке X-Max-Bot-Api-Secret в каждом запросе Webhook. Разрешены только символы A-Z, a-z, 0-9, и дефис. Заголовок рекомендован, чтобы запрос поступал из установленного веб-узла
"""
def __init__(
self,
bot: 'Bot',
url: str,
update_types: Optional[List[UpdateType]] = None,
secret: Optional[str] = None
):
self.bot = bot
self.url = url
self.update_types = update_types
self.secret = secret
async def fetch(self) -> Subscribed:
"""
Отправляет запрос на подписку бота на получение обновлений через WebHook
Returns:
Subscribed: Объект с информацией об операции
"""
if self.bot is None:
raise RuntimeError('Bot не инициализирован')
json: Dict[str, Any] = {}
json['url'] = self.url
if self.update_types:
json['update_types'] = self.update_types
if self.secret:
json['secret'] = self.secret
return await super().request(
method=HTTPMethod.POST,
path=ApiPath.SUBSCRIPTIONS,
model=Subscribed,
params=self.bot.params,
json=json
)

View File

View File

@@ -0,0 +1,16 @@
from typing import Optional
from pydantic import BaseModel
class AddedListAdminChat(BaseModel):
"""
Ответ API при добавлении списка администраторов в чат.
Attributes:
success (bool): Статус успешности операции.
message (Optional[str]): Дополнительное сообщение или ошибка.
"""
success: bool
message: Optional[str] = None

View File

@@ -0,0 +1,16 @@
from typing import Optional
from pydantic import BaseModel
class AddedMembersChat(BaseModel):
"""
Ответ API при добавлении списка пользователей в чат.
Attributes:
success (bool): Статус успешности операции.
message (Optional[str]): Дополнительное сообщение или ошибка.
"""
success: bool
message: Optional[str] = None

View File

@@ -0,0 +1,16 @@
from typing import Optional
from pydantic import BaseModel
class DeletedBotFromChat(BaseModel):
"""
Ответ API при удалении бота из чата.
Attributes:
success (bool): Статус успешности операции.
message (Optional[str]): Дополнительное сообщение или ошибка.
"""
success: bool
message: Optional[str] = None

View File

@@ -0,0 +1,16 @@
from typing import Optional
from pydantic import BaseModel
class DeletedChat(BaseModel):
"""
Ответ API при удалении чата (?).
Attributes:
success (bool): Статус успешности операции.
message (Optional[str]): Дополнительное сообщение или ошибка.
"""
success: bool
message: Optional[str] = None

View File

@@ -0,0 +1,16 @@
from typing import Optional
from pydantic import BaseModel
class DeletedMessage(BaseModel):
"""
Ответ API при удалении сообщения.
Attributes:
success (bool): Статус успешности операции.
message (Optional[str]): Дополнительное сообщение или ошибка.
"""
success: bool
message: Optional[str] = None

View File

@@ -0,0 +1,16 @@
from typing import Optional
from pydantic import BaseModel
class DeletedPinMessage(BaseModel):
"""
Ответ API при удалении закрепленного в чате сообщения.
Attributes:
success (bool): Статус успешности операции.
message (Optional[str]): Дополнительное сообщение или ошибка.
"""
success: bool
message: Optional[str] = None

View File

@@ -1,7 +1,16 @@
from typing import Optional
from pydantic import BaseModel
from ...types.message import Message
class EditedMessage(BaseModel):
message: Message
"""
Ответ API при изменении сообщения.
Attributes:
success (bool): Статус успешности операции.
message (Optional[str]): Дополнительное сообщение или ошибка.
"""
success: bool
message: Optional[str] = None

View File

@@ -0,0 +1,18 @@
from typing import List, Optional
from pydantic import BaseModel
from ...types.chats import ChatMember
class GettedListAdminChat(BaseModel):
"""
Ответ API с полученным списком администраторов чата.
Attributes:
members (List[ChatMember]): Список участников с правами администратора.
marker (Optional[int]): Маркер для постраничной навигации (если есть).
"""
members: List[ChatMember]
marker: Optional[int] = None

View File

@@ -0,0 +1,18 @@
from typing import List, Optional
from pydantic import BaseModel
from ...types.chats import ChatMember
class GettedMembersChat(BaseModel):
"""
Ответ API с полученным списком участников чата.
Attributes:
members (List[ChatMember]): Список участников с правами администратора.
marker (Optional[int]): Маркер для постраничной навигации (если есть).
"""
members: List[ChatMember]
marker: Optional[int] = None

View File

@@ -0,0 +1,16 @@
from typing import Optional
from pydantic import BaseModel
from ...types.message import Message
class GettedPin(BaseModel):
"""
Ответ API с информацией о закреплённом сообщении.
Attributes:
message (Optional[Message]): Закреплённое сообщение, если оно есть.
"""
message: Optional[Message] = None

View File

@@ -0,0 +1,16 @@
from typing import List
from pydantic import BaseModel
from ...types.subscription import Subscription
class GettedSubscriptions(BaseModel):
"""
Ответ API с отправленным сообщением.
Attributes:
message (Message): Объект отправленного сообщения.
"""
subscriptions: List[Subscription]

Some files were not shown because too many files have changed in this diff Show More