max-library

Асинхронная Python-библиотека (SDK) для работы с корпоративным мессенджером MAX
(web.oneme.ru / ws-api.oneme.ru)

⚡ О библиотеке

max-library — это современная, полностью асинхронная клиентская библиотека на Python для взаимодействия с мессенджером MAX через WebSocket API.

Библиотека создана для построения надежных сервисов, интеграций и ботов. Она решает классические проблемы блокировок и обрывов соединений благодаря разделению чтения входящих пакетов, RPC-диспетчеризации по sequence ID и автоматическому поддержанию соединения.

🌟 Ключевые возможности

• 🚀 100% AsyncIO: Построена на базе websockets и aiohttp, никакого блокирующего ввода-вывода.
• 🔄 Автоматический Heartbeat и Reconnect: Встроенный фоновый пинг-понг и прозрачное восстановление WebSocket-сессии при сетевых сбоях.
• ⚡ Надежная RPC-диспетчеризация: Запросы сопоставляются с ответами через уникальные seq-номера и asyncio.Future, исключая коллизии параллельных вызовов.
• 🛡️ Строгая типизация (Pydantic v2): Все входящие и исходящие события, сообщения, пользователи, чаты и вложения типизированы и валидируются моделями Pydantic.
• 🎯 Декларативные фильтры событий: Удобная система фильтрации входящих сообщений (filters.text, filters.chat_id, filters.has_attachments, поддержка логических операторов &, |, ~).
• 📎 Поддержка файлов и медиа: Разрешение прямых ссылок на скачивание любых файлов и документов из MAX через внутренний протокол (Opcode 88).
• 💬 Реакции и ссылки: Поддержка отправки реакций-эмодзи и парсинг пересланных сообщений / ответов (FORWARD, REPLY).

📦 Установка

pip install max-library

🚀 Быстрый старт

1. Простой эхо-бот

import asyncio
from max_library import MaxClient, filters
from max_library.models import Message

client = MaxClient(token="ВАШ_MAX_TOKEN")

@client.on_connect
async def handle_connect():
    print("Успешное подключение к мессенджеру MAX!")

# Обработчик только текстовых сообщений
@client.on_message(filters.text)
async def handle_text_message(message: Message):
    print(f"Получено сообщение от {message.sender.name}: {message.text}")
    
    if message.text == "/ping":
        await client.send_message(
            chat_id=message.chat_id,
            text="pong!"
        )

async def main():
    async with client:
        # Держим процесс активным
        await asyncio.Event().wait()

if __name__ == "__main__":
    asyncio.run(main())

💡 Примеры использования

Фильтрация сообщений

Вы можете комбинировать фильтры логическими операторами (& - И, | - ИЛИ, ~ - НЕ):

from max_library import MaxClient, filters

client = MaxClient(token="YOUR_TOKEN")

# Ловим только сообщения с картинками в конкретном чате
photo_in_target_chat = filters.chat_id(123456789) & filters.has_attachments("PHOTO")

@client.on_message(photo_in_target_chat)
async def handle_photo(message):
    print(f"Фотография получена в целевом чате! Вложений: {len(message.attachments)}")

# Ловим сообщения с файлами или ссылками
files_or_links = filters.has_attachments("FILE") | filters.has_attachments("LINK")

@client.on_message(files_or_links)
async def handle_files(message):
    print("Получен файл или ссылка")

Получение прямой ссылки на скачивание файла

Файлы и документы в MAX приходят с идентификатором file_id. Библиотека позволяет получить прямую CDN-ссылку на скачивание:

@client.on_message(filters.has_attachments("FILE"))
async def handle_incoming_file(message):
    for attachment in message.attachments:
        if attachment.file_id:
            # Запрашиваем URL для скачивания через RPC (Opcode 88)
            download_url = await client.get_file_download_url(
                chat_id=message.chat_id,
                message_id=message.id,
                file_id=attachment.file_id
            )
            print(f"Файл {attachment.name}: {download_url}")

Отправка реакций

@client.on_message()
async def add_reaction_to_msg(message):
    # Поставить реакцию ❤️ на сообщение
    await client.add_reaction(
        chat_id=message.chat_id,
        message_id=message.id,
        emoji="❤️"
    )

Получение списка чатов и контактов

async with MaxClient(token="YOUR_TOKEN") as client:
    # Список контактов
    contacts = await client.get_contacts()
    for contact in contacts:
        print(f"Контакт: {contact.name} (ID: {contact.user_id})")

    # Список чатов
    chats = await client.get_chats()
    for chat in chats:
        print(f"Чат: {chat.title} (ID: {chat.id})")

📖 Документация API

Подробное описание архитектуры, методов MaxClient, моделей данных Pydantic и кодов исключений доступно в документации API.

🛠️ Пример в Production

Библиотека max-library лежит в основе высоконадежного бота пересылки: 👉 FromMaxToTelegram — асинхронный Telegram-бот, пересылающий текст, медиагруппы (альбомы) и файлы из чатов MAX в Telegram-форумы и каналы с умным кэшем пользователей и ротацией логов.

🤝 Разработка и тестирование

Клонирование репозитория и запуск тестов:

git clone https://github.com/jirnobruh/max-library.git
cd max-library
python -m venv .venv
# Активируйте виртуальное окружение
pip install -e ".[dev]"
pytest

📄 Лицензия

Проект распространяется под лицензией MIT. Автор: jirnobruh.