love_apples/maxapi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

upd

Browse Source
This commit is contained in:
Денис Семёнов 2025-06-17 23:20:36 +03:00
parent b2283ab538
commit eff34b42c2
65 changed files with 0 additions and 1803 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

160
.gitignore vendored
View File

@ -1,160 +0,0 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
#pdm.lock
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
# in version control.
# https://pdm.fming.dev/#use-with-ide
.pdm.toml
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/

16
README.md
View File

@ -1,16 +0,0 @@
# maxapi
#### Библиотека (like aiogram) для взаимодействия с социальной сетью MAX по Webhook (или подписке бота)
Информация на данный момент:
* Проект не готов, ведется активная разработка
* Планируется:
+ Сокращение импортов в ваших хендлерах (громадные импорты в example.py)
+ Сокращение "построения" клавиатур
+ Разработка контекста бота
+ Разработка Longpoll метода
+ Доработка базовой составляющей проекта
+ и так далее...
### Контакты
[Группа MAX](https://max.ru/join/IPAok63C3vFqbWTFdutMUtjmrAkGqO56YeAN7iyDfc8)

62
example.py
View File

@ -1,62 +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 == 'q')
async def hello(obj: MessageCreated):
msg = await obj.message.answer('Привет 👋')
a = await obj.bot.get_video('f9LHodD0cOJ5BfLGZ81uXgypU1z7PNhJMkmIe_dtEcxfC3V8vxWk65mRJX8MFQ5F9OAs3yDgbUv6DS6X1p7P')
...
# Отвечает только на текст "Клавиатура"
@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):
a = await obj.answer('test')
...
# Ответчает на коллбек с начинкой "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.message_created()
async def hello(obj: MessageCreated):
# await obj.message.answer(f'Повторяю за вами: {obj.message.body.text}')
pass
dp.handle_webhook(bot)

143
maxapi/bot.py
View File

@ -1,143 +0,0 @@
from datetime import datetime
from typing import Any, Dict, List, TYPE_CHECKING
from .methods.send_callback import SendCallback
from .methods.get_video import GetVideo
from .methods.delete_message import DeleteMessage
from .methods.edit_message import EditMessage
from .enums.parse_mode import ParseMode
from .types.attachments.attachment import Attachment
from .types.message import NewMessageLink
from .types.users import BotCommand
from .methods.change_info import ChangeInfo
from .methods.get_me import GetMe
from .methods.get_messages import GetMessages
from .methods.get_chats import GetChats
from .methods.send_message import SendMessage
from .connection.base import BaseConnection
if TYPE_CHECKING:
from .types.message import Message
class Bot(BaseConnection):
def __init__(self, token: str):
super().__init__()
self.bot = self
self.__token = token
self.params = {
'access_token': self.__token
}
async def send_message(
self,
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
):
return await SendMessage(
bot=self,
chat_id=chat_id,
user_id=user_id,
disable_link_preview=disable_link_preview,
text=text,
attachments=attachments,
link=link,
notify=notify,
parse_mode=parse_mode
).request()
async def edit_message(
self,
message_id: str,
text: str = None,
attachments: List[Attachment] = None,
link: NewMessageLink = None,
notify: bool = True,
parse_mode: ParseMode = None
):
return await EditMessage(
bot=self,
message_id=message_id,
text=text,
attachments=attachments,
link=link,
notify=notify,
parse_mode=parse_mode
).request()
async def delete_message(
self,
message_id: str
):
return await DeleteMessage(
bot=self,
message_id=message_id,
).request()
async def get_messages(
self,
chat_id: int = None,
message_ids: List[str] = None,
from_time: datetime | int = None,
to_time: datetime | int = None,
count: int = 50,
):
return await GetMessages(
bot=self,
chat_id=chat_id,
message_ids=message_ids,
from_time=from_time,
to_time=to_time,
count=count
).request()
async def get_message(self, message_id: str):
return await self.get_messages(message_ids=[message_id])
async def get_me(self):
return await GetMe(self).request()
async def change_info(
self,
name: str = None,
description: str = None,
commands: List[BotCommand] = None,
photo: Dict[str, Any] = None
):
return await ChangeInfo(
bot=self,
name=name,
description=description,
commands=commands,
photo=photo
).request()
async def get_chats(self):
return await GetChats(self).request()
async def get_video(self, video_token: str):
return await GetVideo(self, video_token).request()
async def send_callback(
self,
callback_id: str,
message: 'Message' = None,
notification: str = None
):
return await SendCallback(
bot=self,
callback_id=callback_id,
message=message,
notification=notification
).request()

49
maxapi/connection/base.py
View File

@ -1,49 +0,0 @@
import aiohttp
from pydantic import BaseModel
from ..types.errors import Error
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
class BaseConnection:
API_URL = 'https://botapi.max.ru'
def __init__(self):
self.bot = None
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 if isinstance(path, ApiPath) else path,
**kwargs
)
if not r.ok:
raw = await r.text()
return Error(code=r.status, text=raw)
raw = await r.json()
if is_return_raw: return raw
model = model(**raw)
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

153
maxapi/dispatcher.py
View File

@ -1,153 +0,0 @@
from typing import Callable, List
import uvicorn
from fastapi import FastAPI, Request
from magic_filter import MagicFilter
from .filters import filter_m
from .types.updates import Update
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
class Handler:
def __init__(
self,
*args,
func_event: Callable,
update_type: UpdateType,
**kwargs
):
self.func_event = func_event
self.update_type = update_type
self.filters = []
for arg in args:
if isinstance(arg, MagicFilter):
arg: MagicFilter = arg
self.filters.append(arg)
class Dispatcher:
def __init__(self):
self.event_handlers = []
self.bot = None
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.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)
self.message_edited = Event(update_type=UpdateType.MESSAGE_EDITED, router=self)
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)
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 = 'localhost', port: int = 8080):
self.bot = bot
app = FastAPI()
@app.post("/")
async def _(request: Request):
try:
event_json = await request.json()
event = Update(**event_json)
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
event_object.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
event_object.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)
handlers: List[Handler] = self.event_handlers
for handler in handlers:
if not handler.update_type == event.update_type:
continue
if handler.filters:
if not filter_m(event_object, *handler.filters):
continue
await handler.func_event(event_object)
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')
class Router(Dispatcher):
def __init__(self):
super().__init__()
class Event:
def __init__(self, update_type: UpdateType, router: Dispatcher | Router):
self.update_type = update_type
self.router = router
def __call__(self, *args, **kwargs):
def decorator(func_event: Callable):
self.router.event_handlers.append(
Handler(
func_event=func_event,
update_type=self.update_type,
*args, **kwargs
)
)
return func_event
return decorator

9
maxapi/enums/api_path.py
View File

@ -1,9 +0,0 @@
from enum import Enum
class ApiPath(str, Enum):
ME = '/me'
CHATS = '/chats'
MESSAGES = '/messages'
UPDATES = '/updates'
VIDEOS = '/videos'
ANSWERS = '/answers'

11
maxapi/enums/attachment.py
View File

@ -1,11 +0,0 @@
from enum import Enum
class AttachmentType(str, Enum):
IMAGE = 'image'
VIDEO = 'video'
AUDIO = 'audio'
FILE = 'file'
STICKER = 'sticker'
CONTACT = 'contact'
INLINE_KEYBOARD = 'inline_keyboard'
LOCATION = 'location'

9
maxapi/enums/button_type.py
View File

@ -1,9 +0,0 @@
from enum import Enum
class ButtonType(Enum):
REQUEST_CONTACT = 'request_contact'
CALLBACK = 'callback'
LINK = 'link'
REQUEST_GEO_LOCATION = 'request_geo_location'
CHAT = 'chat'

6
maxapi/enums/chat_type.py
View File

@ -1,6 +0,0 @@
from enum import Enum
class ChatType(str, Enum):
DIALOG = 'dialog'
CHAT = 'chat'

9
maxapi/enums/http_method.py
View File

@ -1,9 +0,0 @@
from enum import Enum
class HTTPMethod(str, Enum):
POST = 'POST'
GET = 'GET'
PATCH = 'PATCH'
PUT = 'PUT'
DELETE = 'DELETE'

6
maxapi/enums/intent.py
View File

@ -1,6 +0,0 @@
from enum import Enum
class Intent(str, Enum):
DEFAULT = 'default'
POSITIVE = 'positive'
NEGATIVE = 'negative'

6
maxapi/enums/message_link_type.py
View File

@ -1,6 +0,0 @@
from enum import Enum
class MessageLinkType(str, Enum):
FORWARD = 'forward'
REPLY = 'reply'

5
maxapi/enums/parse_mode.py
View File

@ -1,5 +0,0 @@
from enum import Enum
class ParseMode(str, Enum):
MARKDOWN = 'markdown'
HTML = 'html'

13
maxapi/enums/text_style.py
View File

@ -1,13 +0,0 @@
from enum import Enum
class TextStyle(Enum):
UNDERLINE = 'underline'
STRONG = 'strong'
EMPHASIZED = 'emphasized'
MONOSPACED = 'monospaced'
LINK = 'link'
STRIKETHROUGH = 'strikethrough'
USER_MENTION = 'user_mention'
HEADING = 'heading'
HIGHLIGHTED = 'highlighted'

14
maxapi/enums/update.py
View File

@ -1,14 +0,0 @@
from enum import Enum
class UpdateType(str, Enum):
MESSAGE_CREATED = 'message_created'
BOT_ADDED = 'bot_added'
BOT_REMOVED = 'bot_removed'
BOT_STARTED = 'bot_started'
CHAT_TITLE_CHANGED = 'chat_title_changed'
MESSAGE_CALLBACK = 'message_callback'
MESSAGE_CHAT_CREATED = 'message_chat_created'
MESSAGE_EDITED = 'message_edited'
MESSAGE_REMOVED = 'message_removed'
USER_ADDED = 'user_added'
USER_REMOVED = 'user_removed'

53
maxapi/filters/__init__.py
View File

@ -1,53 +0,0 @@
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
F = MagicFilter()
def filter_m(obj, *magic_args):
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:
...

4
maxapi/loggers.py
View File

@ -1,4 +0,0 @@
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('bot')

46
maxapi/methods/change_info.py
View File

@ -1,46 +0,0 @@
from typing import Any, Dict, List, TYPE_CHECKING
from ..types.users import BotCommand, User
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 ChangeInfo(BaseConnection):
def __init__(
self,
bot: 'Bot',
name: str = None,
description: str = None,
commands: List[BotCommand] = None,
photo: Dict[str, Any] = None
):
self.bot = bot
self.name = name
self.description = description
self.commands = commands
self.photo = photo
async def request(self) -> User:
json = {}
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
return await super().request(
method=HTTPMethod.PATCH,
path=ApiPath.ME,
model=User,
params=self.bot.params,
json=json
)

33
maxapi/methods/delete_message.py
View File

@ -1,33 +0,0 @@
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):
def __init__(
self,
bot: 'Bot',
message_id: str,
):
self.bot = bot
self.message_id = message_id
async def request(self) -> DeletedMessage:
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,
)

55
maxapi/methods/edit_message.py
View File

@ -1,55 +0,0 @@
from typing import List, TYPE_CHECKING
from .types.edited_message import EditedMessage
from ..types.message import NewMessageLink
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 ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class EditMessage(BaseConnection):
def __init__(
self,
bot: 'Bot',
message_id: str,
text: str = None,
attachments: List['Attachment'] = None,
link: 'NewMessageLink' = None,
notify: bool = True,
parse_mode: ParseMode = None
):
self.bot = bot
self.message_id = message_id
self.text = text
self.attachments = attachments
self.link = link
self.notify = notify
self.parse_mode = parse_mode
async def request(self) -> EditedMessage:
params = self.bot.params.copy()
json = {}
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
return await super().request(
method=HTTPMethod.PUT,
path=ApiPath.MESSAGES,
model=EditedMessage,
params=params,
json=json
)

29
maxapi/methods/get_chats.py
View File

@ -1,29 +0,0 @@
from typing import TYPE_CHECKING
from ..types.chats import Chats
from ..types.users import User
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 GetChats(BaseConnection):
def __init__(self, bot: 'Bot'):
self.bot = bot
async def request(self) -> Chats:
return await super().request(
method=HTTPMethod.GET,
path=ApiPath.CHATS,
model=Chats,
params=self.bot.params
)

29
maxapi/methods/get_me.py
View File

@ -1,29 +0,0 @@
from typing import TYPE_CHECKING
from ..types.chats import Chats
from ..types.users import User
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 GetMe(BaseConnection):
def __init__(self, bot: 'Bot'):
self.bot = bot
async def request(self) -> Chats:
return await super().request(
method=HTTPMethod.GET,
path=ApiPath.ME,
model=User,
params=self.bot.params
)

60
maxapi/methods/get_messages.py
View File

@ -1,60 +0,0 @@
from datetime import datetime
from typing import TYPE_CHECKING, List
from ..types.message import Messages
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 GetMessages(BaseConnection):
def __init__(
self,
bot: 'Bot',
chat_id: int,
message_ids: List[str] = None,
from_time: datetime | int = None,
to_time: 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:
params = self.bot.params.copy()
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,
path=ApiPath.MESSAGES,
model=Messages,
params=params
)

34
maxapi/methods/get_video.py
View File

@ -1,34 +0,0 @@
from typing import List, TYPE_CHECKING
from ..types.attachments.video import Video
from .types.edited_message import EditedMessage
from ..types.message import NewMessageLink
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 ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class GetVideo(BaseConnection):
def __init__(
self,
bot: 'Bot',
video_token: str
):
self.bot = bot
self.video_token = video_token
async def request(self) -> Video:
return await super().request(
method=HTTPMethod.GET,
path=ApiPath.VIDEOS.value + '/' + self.video_token,
model=Video,
params=self.bot.params,
)

53
maxapi/methods/send_callback.py
View File

@ -1,53 +0,0 @@
from typing import List, TYPE_CHECKING
from ..methods.types.sended_callback import SendedCallback
from .types.sended_message import SendedMessage
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 ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
from ..types.message import Message
class SendCallback(BaseConnection):
def __init__(
self,
bot: 'Bot',
callback_id: str,
message: 'Message' = None,
notification: str = None
):
self.bot = bot
self.callback_id = callback_id
self.message = message
self.notification = notification
async def request(self) -> SendedCallback:
try:
params = self.bot.params.copy()
params['callback_id'] = self.callback_id
json = {}
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
)
except Exception as e:
print(e)
...

69
maxapi/methods/send_message.py
View File

@ -1,69 +0,0 @@
from typing import List, TYPE_CHECKING
from .types.sended_message import SendedMessage
from ..types.message import NewMessageLink
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 ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class SendMessage(BaseConnection):
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
):
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:
try:
params = self.bot.params.copy()
json = {}
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]
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
return await super().request(
method=HTTPMethod.POST,
path=ApiPath.MESSAGES,
model=SendedMessage,
params=params,
json=json
)
except Exception as e:
print(e)
...

7
maxapi/methods/types/deleted_message.py
View File

@ -1,7 +0,0 @@
from typing import Optional
from pydantic import BaseModel
class DeletedMessage(BaseModel):
success: bool
message: Optional[str] = None

7
maxapi/methods/types/edited_message.py
View File

@ -1,7 +0,0 @@
from typing import Optional
from pydantic import BaseModel
class EditedMessage(BaseModel):
success: bool
message: Optional[str] = None

14
maxapi/methods/types/sended_callback.py
View File

@ -1,14 +0,0 @@
from typing import TYPE_CHECKING, Any, Optional
from pydantic import BaseModel, Field
if TYPE_CHECKING:
from ...bot import Bot
class SendedCallback(BaseModel):
success: bool
message: Optional[str] = None
bot: Optional[Any] = Field(default=None, exclude=True)
if TYPE_CHECKING:
bot: Optional[Bot]

8
maxapi/methods/types/sended_message.py
View File

@ -1,8 +0,0 @@
from typing import Any
from pydantic import BaseModel
from ...types.message import Message
class SendedMessage(BaseModel):
message: Message

0
maxapi/types/__init__.py
View File

59
maxapi/types/attachments/attachment.py
View File

@ -1,59 +0,0 @@
from typing import List, Optional, Union
from pydantic import BaseModel
from ...types.attachments.buttons.chat_button import ChatButton
from ...types.attachments.buttons.request_contact import RequestContact
from ...types.attachments.buttons.request_geo_location_button import RequestGeoLocationButton
from ...types.attachments.buttons.link_button import LinkButton
from ...types.users import User
from ...enums.attachment import AttachmentType
from .buttons.callback_button import CallbackButton
AttachmentUnion = []
class StickerAttachmentPayload(BaseModel):
url: str
code: str
class PhotoAttachmentPayload(BaseModel):
photo_id: int
token: str
url: str
class OtherAttachmentPayload(BaseModel):
url: str
token: Optional[str] = None
class ContactAttachmentPayload(BaseModel):
vcf_info: Optional[str] = None
max_info: Optional[User] = None
class ButtonsPayload(BaseModel):
buttons: List[List[
Union[
LinkButton,
CallbackButton,
RequestGeoLocationButton,
RequestContact,
ChatButton
]
]]
class Attachment(BaseModel):
type: AttachmentType
payload: Optional[Union[
PhotoAttachmentPayload,
OtherAttachmentPayload,
ContactAttachmentPayload,
ButtonsPayload,
StickerAttachmentPayload
]] = None
class Config:
use_enum_values = True

8
maxapi/types/attachments/audio.py
View File

@ -1,8 +0,0 @@
from typing import Literal, Optional
from .attachment import Attachment
class Audio(Attachment):
type: Literal['audio'] = 'audio'
transcription: Optional[str] = None

12
maxapi/types/attachments/buttons/__init__.py
View File

@ -1,12 +0,0 @@
from typing import Literal
from pydantic import BaseModel
from ....enums.button_type import ButtonType
class Button(BaseModel):
type: ButtonType
text: str
class Config:
use_enum_values = True

9
maxapi/types/attachments/buttons/attachment_button.py
View File

@ -1,9 +0,0 @@
from typing import Literal
from pydantic import BaseModel
from ..attachment import ButtonsPayload
class AttachmentButton(BaseModel):
type: Literal['inline_keyboard'] = 'inline_keyboard'
payload: ButtonsPayload

9
maxapi/types/attachments/buttons/callback_button.py
View File

@ -1,9 +0,0 @@
from typing import Optional
from ....enums.intent import Intent
from . import Button
class CallbackButton(Button):
payload: Optional[str] = None
intent: Intent

11
maxapi/types/attachments/buttons/chat_button.py
View File

@ -1,11 +0,0 @@
from typing import Optional
from ....types.attachments.buttons import Button
class ChatButton(Button):
chat_title: Optional[str] = None
chat_description: Optional[str] = None
start_payload: Optional[str] = None
chat_title: Optional[str] = None
uuid: Optional[int] = None

7
maxapi/types/attachments/buttons/link_button.py
View File

@ -1,7 +0,0 @@
from typing import Optional
from ....types.attachments.buttons import Button
class LinkButton(Button):
url: Optional[str] = None

5
maxapi/types/attachments/buttons/request_contact.py
View File

@ -1,5 +0,0 @@
from ....types.attachments.buttons import Button
class RequestContact(Button):
...

5
maxapi/types/attachments/buttons/request_geo_location_button.py
View File

@ -1,5 +0,0 @@
from ....types.attachments.buttons import Button
class RequestGeoLocationButton(Button):
quick: bool = False

7
maxapi/types/attachments/contact.py
View File

@ -1,7 +0,0 @@
from typing import Literal
from .attachment import Attachment
class Contact(Attachment):
type: Literal['contact'] = 'contact'

9
maxapi/types/attachments/file.py
View File

@ -1,9 +0,0 @@
from typing import Literal, Optional
from .attachment import Attachment
class File(Attachment):
type: Literal['file'] = 'file'
filename: Optional[str] = None
size: Optional[int] = None

6
maxapi/types/attachments/image.py
View File

@ -1,6 +0,0 @@
from typing import Literal
from .attachment import Attachment
class Image(Attachment):
type: Literal['image'] = 'image'

9
maxapi/types/attachments/location.py
View File

@ -1,9 +0,0 @@
from typing import Literal, Optional
from .attachment import Attachment
class Location(Attachment):
type: Literal['location'] = 'location'
latitude: Optional[float] = None
longitude: Optional[float] = None

9
maxapi/types/attachments/share.py
View File

@ -1,9 +0,0 @@
from typing import Optional
from .attachment import Attachment
class Share(Attachment):
title: Optional[str] = None
description: Optional[str] = None
image_url: Optional[str] = None

9
maxapi/types/attachments/sticker.py
View File

@ -1,9 +0,0 @@
from typing import Literal, Optional
from .attachment import Attachment
class Sticker(Attachment):
type: Literal['sticker'] = 'sticker'
width: Optional[int] = None
height: Optional[int] = None

35
maxapi/types/attachments/video.py
View File

@ -1,35 +0,0 @@
from typing import TYPE_CHECKING, Any, Literal, Optional
from pydantic import BaseModel, Field
from .attachment import Attachment
if TYPE_CHECKING:
from ...bot import Bot
class VideoUrl(BaseModel):
mp4_1080: Optional[str] = None
mp4_720: Optional[str] = None
mp4_480: Optional[str] = None
mp4_360: Optional[str] = None
mp4_240: Optional[str] = None
mp4_144: Optional[str] = None
hls: Optional[str] = None
class VideoThumbnail(BaseModel):
url: str
class Video(Attachment):
type: Optional[Literal['video']] = 'video'
token: Optional[str] = None
urls: Optional[VideoUrl] = None
thumbnail: VideoThumbnail
width: Optional[int] = None
height: Optional[int] = None
duration: Optional[int] = None
bot: Optional[Any] = Field(default=None, exclude=True)
if TYPE_CHECKING:
bot: Optional['Bot']

13
maxapi/types/callback.py
View File

@ -1,13 +0,0 @@
from typing import List, Optional, Union
from pydantic import BaseModel
from ..types.users import User
from ..types.users import User
class Callback(BaseModel):
timestamp: int
callback_id: str
payload: Optional[str] = None
user: User

46
maxapi/types/chats.py
View File

@ -1,46 +0,0 @@
from pydantic import BaseModel
from typing import List, Optional
from enum import Enum
from datetime import datetime
from ..types.users import User
from ..types.message import Message
class ChatType(str, Enum):
DIALOG = "dialog"
CHAT = "chat"
class ChatStatus(str, Enum):
ACTIVE = "active"
REMOVED = "removed"
LEFT = "left"
CLOSED = "closed"
SUSPENDED = "suspended"
class Icon(BaseModel):
url: str
class Chat(BaseModel):
chat_id: int
type: ChatType
status: ChatStatus
title: Optional[str] = None
icon: Optional[Icon] = None
last_event_time: int
participants_count: int
owner_id: Optional[int] = None
participants: None = None
is_public: bool
link: Optional[str] = None
description: Optional[str] = None
dialog_with_user: Optional[User] = None
messages_count: Optional[int] = None
chat_message_id: Optional[str] = None
pinned_message: Optional[Message] = None
class Config:
arbitrary_types_allowed=True
class Chats(BaseModel):
chats: List[Chat] = []

6
maxapi/types/errors.py
View File

@ -1,6 +0,0 @@
from pydantic import BaseModel
class Error(BaseModel):
code: int
text: str

147
maxapi/types/message.py
View File

@ -1,147 +0,0 @@
from __future__ import annotations
from pydantic import BaseModel, Field
from typing import Any, Optional, List, Union, TYPE_CHECKING
from ..enums.parse_mode import ParseMode
from ..types.attachments.attachment import Attachment
from ..types.attachments.share import Share
from .attachments.buttons.attachment_button import AttachmentButton
from ..enums.text_style import TextStyle
from ..enums.chat_type import ChatType
from ..enums.message_link_type import MessageLinkType
from .attachments.sticker import Sticker
from .attachments.file import File
from .attachments.image import Image
from .attachments.video import Video
from .attachments.audio import Audio
from ..types.users import User
if TYPE_CHECKING:
from ..bot import Bot
class MarkupElement(BaseModel):
type: TextStyle
from_: int = Field(..., alias='from')
length: int
class Config:
populate_by_name = True
class MarkupLink(MarkupElement):
url: Optional[str] = None
class Recipient(BaseModel):
user_id: Optional[int] = None # Для пользователя
chat_id: Optional[int] = None # Для чата
chat_type: ChatType # Тип получателя (диалог или чат)
class MessageBody(BaseModel):
mid: str
seq: int
text: str = None
attachments: Optional[
List[
Union[
AttachmentButton,
Audio,
Video,
File,
Image,
Sticker,
Share
]
]
] = []
markup: Optional[
List[
Union[
MarkupLink, MarkupElement
]
]
] = []
class MessageStat(BaseModel):
views: int
class LinkedMessage(BaseModel):
type: MessageLinkType
sender: User
chat_id: Optional[int] = None
message: MessageBody
class Message(BaseModel):
sender: User
recipient: Recipient
timestamp: int
link: Optional[LinkedMessage] = None
body: Optional[MessageBody] = None
stat: Optional[MessageStat] = None
url: Optional[str] = None
bot: Optional[Any] = Field(default=None, exclude=True)
if TYPE_CHECKING:
bot: Optional[Bot]
async def answer(self,
text: str = None,
disable_link_preview: bool = False,
attachments: List[Attachment] = None,
link: NewMessageLink = None,
notify: bool = True,
parse_mode: ParseMode = None
):
return await self.bot.send_message(
chat_id=self.recipient.chat_id,
user_id=self.recipient.user_id,
text=text,
disable_link_preview=disable_link_preview,
attachments=attachments,
link=link,
notify=notify,
parse_mode=parse_mode
)
async def edit(
self,
text: str = None,
attachments: List[Attachment] = None,
link: NewMessageLink = None,
notify: bool = True,
parse_mode: ParseMode = None
):
return await self.bot.edit_message(
message_id=self.body.mid,
text=text,
attachments=attachments,
link=link,
notify=notify,
parse_mode=parse_mode
)
async def delete(self):
return await self.bot.delete_message(
message_id=self.body.mid,
)
class Messages(BaseModel):
messages: List[Message]
bot: Optional[Any] = Field(default=None, exclude=True)
if TYPE_CHECKING:
bot: Optional[Bot]
class NewMessageLink(BaseModel):
type: MessageLinkType
mid: str

11
maxapi/types/updates/__init__.py
View File

@ -1,11 +0,0 @@
from pydantic import BaseModel
from ...enums.update import UpdateType
class Update(BaseModel):
update_type: UpdateType
timestamp: int
class Config:
arbitrary_types_allowed=True

8
maxapi/types/updates/bot_added.py
View File

@ -1,8 +0,0 @@
from typing import Optional
from . import Update
from ...types.users import User
class BotAdded(Update):
chat_id: Optional[int] = None
user: User

8
maxapi/types/updates/bot_removed.py
View File

@ -1,8 +0,0 @@
from typing import Optional
from . import Update
from ...types.users import User
class BotRemoved(Update):
chat_id: Optional[int] = None
user: User

10
maxapi/types/updates/bot_started.py
View File

@ -1,10 +0,0 @@
from typing import Optional
from . import Update
from ...types.users import User
class BotStarted(Update):
chat_id: Optional[int] = None
user: User
user_locale: Optional[str] = None
payload: Optional[str] = None

9
maxapi/types/updates/chat_title_changed.py
View File

@ -1,9 +0,0 @@
from typing import Optional
from . import Update
from ...types.users import User
class ChatTitleChanged(Update):
chat_id: Optional[int] = None
user: User
title: Optional[str] = None

73
maxapi/types/updates/message_callback.py
View File

@ -1,73 +0,0 @@
from typing import Any, List, Optional, TYPE_CHECKING, Union
from pydantic import BaseModel, Field
from . import Update
from ...types.callback import Callback
from ...types.message import Message
from ...enums.parse_mode import ParseMode
from ...types.message import NewMessageLink
from ...types.attachments.share import Share
from ..attachments.buttons.attachment_button import AttachmentButton
from ..attachments.sticker import Sticker
from ..attachments.file import File
from ..attachments.image import Image
from ..attachments.video import Video
from ..attachments.audio import Audio
if TYPE_CHECKING:
from ...bot import Bot
class MessageForCallback(BaseModel):
text: Optional[str] = None
attachments: Optional[
List[
Union[
AttachmentButton,
Audio,
Video,
File,
Image,
Sticker,
Share
]
]
] = []
link: Optional[NewMessageLink] = None
notify: Optional[bool] = True
format: Optional[ParseMode] = None
class MessageCallback(Update):
message: Message
user_locale: Optional[str] = None
callback: Callback
bot: Optional[Any] = Field(default=None, exclude=True)
if TYPE_CHECKING:
bot: Optional[Bot]
async def answer(
self,
text: str,
link: NewMessageLink = None,
notify: bool = True,
format: ParseMode = None,
notification: str = None
):
message = MessageForCallback()
message.text = text
message.attachments = self.message.body.attachments
message.link = link
message.notify = notify
message.format = format
return await self.bot.send_callback(
callback_id=self.callback.callback_id,
message=message,
notification=notification
)

11
maxapi/types/updates/message_chat_created.py
View File

@ -1,11 +0,0 @@
from typing import Optional
from ...types.chats import Chat
from . import Update
class MessageChatCreated(Update):
chat: Chat
title: Optional[str] = None
message_id: Optional[str] = None
start_payload: Optional[str] = None

19
maxapi/types/updates/message_created.py
View File

@ -1,19 +0,0 @@
from __future__ import annotations
from typing import Any, Optional, TYPE_CHECKING, ForwardRef
from pydantic import Field
from . import Update
from ...types.message import Message
if TYPE_CHECKING:
from ...bot import Bot
class MessageCreated(Update):
message: Message
user_locale: Optional[str] = None
bot: Optional[Any] = Field(default=None, exclude=True)
if TYPE_CHECKING:
bot: Optional[Bot]

6
maxapi/types/updates/message_edited.py
View File

@ -1,6 +0,0 @@
from . import Update
from ...types.message import Message
class MessageEdited(Update):
message: Message

9
maxapi/types/updates/message_removed.py
View File

@ -1,9 +0,0 @@
from typing import Optional
from . import Update
class MessageRemoved(Update):
message_id: Optional[str] = None
chat_id: Optional[int] = None
user_id: Optional[int] = None

10
maxapi/types/updates/user_added.py
View File

@ -1,10 +0,0 @@
from typing import Optional
from . import Update
from ...types.users import User
class UserAdded(Update):
inviter_id: Optional[int] = None
chat_id: Optional[int] = None
user: User

10
maxapi/types/updates/user_removed.py
View File

@ -1,10 +0,0 @@
from typing import Optional
from . import Update
from ...types.users import User
class UserRemoved(Update):
admin_id: Optional[int] = None
chat_id: Optional[int] = None
user: User

26
maxapi/types/users.py
View File

@ -1,26 +0,0 @@
from pydantic import BaseModel
from typing import List, Optional
from datetime import datetime
class BotCommand(BaseModel):
name: str
description: Optional[str] = None
class User(BaseModel):
user_id: int
first_name: str
last_name: Optional[str] = None
username: Optional[str] = None
is_bot: bool
last_activity_time: int
description: Optional[str] = None
avatar_url: Optional[str] = None
full_avatar_url: Optional[str] = None
commands: Optional[List[BotCommand]] = None
class Config:
json_encoders = {
datetime: lambda v: int(v.timestamp() * 1000) # Конвертация datetime в Unix-время (ms)
}