bot.py - master - Riddiccc/MAX_bot

bot.py
350 строк12 KB
Riddiccc
MAX Bot для ProfDefence: добавлены bot.py, requirements.txt, README и DEPLOY.md
10 июл 2026, 10:36
10 июл 2026, 10:36a6a92c8
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
MAX Bot для компании ProfDefence (MAX Bot-визитка)
Репозиторий: https://max.ru/id312304199298_7_bot
Страница компании: https://profdefence.ru/max-development.html

Бот-визитка MAX на Python для MAX Hub
"""

import os
import sys
import json
import aiohttp
import asyncio
from typing import Optional, Dict, Any, List


class MAXBot:
    """
    Класс MAX Bot для взаимодействия с MAX Hub API
    """
    
    def __init__(self, bot_token: str = None, base_url: str = "https://api.maxhub.ru"):
        """
        Инициализация бота
        
        Args:
            bot_token: Токен бота, полученный при регистрации в MAX
            base_url: Базовый URL API MAX Hub
        """
        self.bot_token = bot_token or os.getenv("BOT_TOKEN")
        if not self.bot_token:
            raise ValueError("BOT_TOKEN не указан. Установите переменную окружения BOT_TOKEN")
        
        self.base_url = base_url
        self.session: Optional[aiohttp.ClientSession] = None
        
    async def __aenter__(self):
        """Async context manager entry"""
        self.session = aiohttp.ClientSession()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Async context manager exit"""
        if self.session:
            await self.session.close()
    
    async def _make_request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
        """
        Выполнение HTTP-запроса к MAX Hub API
        
        Args:
            method: HTTP метод (GET, POST)
            endpoint: API endpoint
            **kwargs: Дополнительные параметры запроса
            
        Returns:
            Dict с результатом запроса
        """
        if not self.session:
            raise RuntimeError("Сессия не инициализирована. Используйте async with")
            
        url = f"{self.base_url}{endpoint}"
        headers = {
            "Authorization": f"Bearer {self.bot_token}",
            "Content-Type": "application/json"
        }
        
        async with self.session.request(method, url, headers=headers, **kwargs) as response:
            return await response.json()
    
    async def set_my_commands(self, commands: List[Dict[str, str]]) -> bool:
        """
        Установка списка команд бота
        
        Args:
            commands: Список команд в формате [{'name': str, 'description': str}]
            
        Returns:
            bool: Успешность операции
        """
        payload = {"commands": commands}
        result = await self._make_request(
            "POST",
            "/botCommands/setMyCommands",
            json=payload
        )
        return result.get("success", False)
    
    async def send_message(self, chat_id: str, text: str, **kwargs) -> Dict[str, Any]:
        """
        Отправка сообщения пользователю
        
        Args:
            chat_id: ID чата
            text: Текст сообщения
            **kwargs: Дополнительные параметры (reply_markup, parse_mode и т.д.)
            
        Returns:
            Dict с результатом отправки
        """
        payload = {
            "chatId": chat_id,
            "text": text,
            **kwargs
        }
        return await self._make_request(
            "POST",
            "/bot/messages/sendText",
            json=payload
        )
    
    async def get_updates(self, offset: int = None, limit: int = 100, timeout: int = 30) -> List[Dict[str, Any]]:
        """
        Получение обновлений от MAX Hub
        
        Args:
            offset: ID последнего полученного обновления + 1
            limit: Максимальное количество обновлений
            timeout: Таймаут в секундах
            
        Returns:
            Список обновлений
        """
        params = {
            "offset": offset,
            "limit": limit,
            "timeout": timeout
        }
        return await self._make_request("GET", "/bot/updates", params=params)
    
    async def get_me(self) -> Dict[str, Any]:
        """
        Получение информации о боте
        
        Returns:
            Dict с информацией о боте
        """
        return await self._make_request("GET", "/bot/getMe")
    
    async def start_polling(self, on_message=None, on_command=None):
        """
        Запуск polling для получения обновлений
        
        Args:
            on_message: Callback функция для обработки сообщений
            on_command: Callback функция для обработки команд
        """
        print(f"Бот запущен. Имя: {await self.get_me().get('firstName', 'Unknown')}")
        
        offset = None
        try:
            while True:
                updates = await self.get_updates(offset=offset)
                
                for update in updates:
                    offset = update.get("updateId") + 1
                    
                    # Обработка сообщений
                    if on_message:
                        await on_message(update)
                        
                    # Обработка команд
                    if on_command and "message" in update:
                        text = update["message"].get("text", "")
                        if text.startswith("/"):
                            await on_command(update, text)
                            
                await asyncio.sleep(1)
                
        except KeyboardInterrupt:
            print("\nБот остановлен пользователем")
        except Exception as e:
            print(f"Ошибка при получении обновлений: {e}")
            raise


class Bot:
    """
    Основной класс бота для MAX Hub
    """
    
    def __init__(self, token: str = None):
        """
        Инициализация бота
        
        Args:
            token: Токен бота
        """
        self.token = token or os.getenv("BOT_TOKEN")
        if not self.token:
            raise ValueError("BOT_TOKEN не указан")
            
        self.bot = MAXBot(self.token)
        self._handlers = {
            "message": [],
            "command": {}
        }
        
    async def start(self):
        """Запуск бота"""
        async with self.bot:
            await self._setup_commands()
            await self.bot.start_polling(
                on_message=self._handle_message,
                on_command=self._handle_command
            )
    
    async def _setup_commands(self):
        """Настройка команд бота"""
        commands = [
            {
                "name": "hello",
                "description": "Поприветствовать бота"
            },
            {
                "name": "start",
                "description": "Начать работу с ботом"
            },
            {
                "name": "help",
                "description": "Показать справку"
            },
            {
                "name": "company",
                "description": "Информация о компании ProfDefence"
            },
            {
                "name": "contact",
                "description": "Контактная информация"
            }
        ]
        success = await self.bot.set_my_commands(commands)
        print(f"Команды установлены: {'Успешно' if success else 'Не удалось'}")
    
    async def _handle_message(self, update: Dict[str, Any]):
        """Обработка входящих сообщений"""
        # Здесь можно добавить общую логику обработки сообщений
        pass
    
    async def _handle_command(self, update: Dict[str, Any], command: str):
        """Обработка команд"""
        command_name = command.split("@")[0].lower()
        
        if command_name in self._handlers["command"]:
            await self._handlers["command"][command_name](update)
    
    def command(self, command_name: str):
        """
        Декоратор для обработки команд
        
        Args:
            command_name: Имя команды (без /)
        """
        def decorator(func):
            self._handlers["command"][command_name] = func
            return func
        return decorator
    
    def message(self):
        """Декоратор для обработки сообщений"""
        def decorator(func):
            self._handlers["message"].append(func)
            return func
        return decorator


def main():
    """Основная точка входа"""
    import asyncio
    
    bot_token = os.getenv("BOT_TOKEN")
    if not bot_token:
        print("Ошибка: BOT_TOKEN не установлен. Установите переменную окружения.")
        print("Пример: export BOT_TOKEN='your_bot_token'")
        sys.exit(1)
    
    bot = Bot(bot_token)
    
    @bot.command("hello")
    async def hello_command(update):
        """Обработчик команды /hello"""
        chat_id = update["message"]["chat"]["id"]
        await bot.bot.send_message(chat_id, "Привет! ✨\nЯ бот(MAX) для компании ProfDefence.")
    
    @bot.command("start")
    async def start_command(update):
        """Обработчик команды /start"""
        chat_id = update["message"]["chat"]["id"]
        message = (
            "👋 Добро пожаловать в MAX Bot!\n"
            "Я представляю компанию ProfDefence.\n\n"
            "Доступные команды:\n"
            "/hello - Поприветствовать бота\n"
            "/help - Показать справку\n"
            "/company - Информация о компании\n"
            "/contact - Контактная информация"
        )
        await bot.bot.send_message(chat_id, message)
    
    @bot.command("help")
    async def help_command(update):
        """Обработчик команды /help"""
        chat_id = update["message"]["chat"]["id"]
        message = (
            "📖 Справка:\n\n"
            "Этот бот представляет компанию ProfDefence в MAX Hub.\n\n"
            "Доступные команды:\n"
            "/start - Начать работу\n"
            "/hello - Поприветствовать бота\n"
            "/company - О компании\n"
            "/contact - Контакты"
        )
        await bot.bot.send_message(chat_id, message)
    
    @bot.command("company")
    async def company_command(update):
        """Обработчик команды /company"""
        chat_id = update["message"]["chat"]["id"]
        message = (
            "🏢 Компания ProfDefence\n"
            "Сайт: https://profdefence.ru\n\n"
            "MAX Bot-визитка:\n"
            "https://max.ru/id312304199298_7_bot\n\n"
            "MAX Development:\n"
            "https://profdefence.ru/max-development.html"
        )
        await bot.bot.send_message(chat_id, message)
    
    @bot.command("contact")
    async def contact_command(update):
        """Обработчик команды /contact"""
        chat_id = update["message"]["chat"]["id"]
        message = (
            "📞 Контактная информация:\n\n"
            "Сайт: https://profdefence.ru\n"
            "MAX: https://max.ru/id312304199298_7_bot\n\n"
            "Мы готовы ответить на ваши вопросы!"
        )
        await bot.bot.send_message(chat_id, message)
    
    try:
        asyncio.run(bot.start())
    except KeyboardInterrupt:
        print("\nБот остановлен")


if __name__ == "__main__":
    main()