bot.py - master - albm/Bot2

bot.py
754 строки30 KB
albm1978
продолжение
11 июл 2025, 14:04
11 июл 2025, 14:04a6a5324
import json
import os
import logging
import signal
import asyncio
from datetime import datetime
from typing import List, Dict
from aiogram import Bot, Dispatcher, types, F
from aiogram.filters import Command
from aiogram.utils.keyboard import ReplyKeyboardBuilder
from aiogram.types import (
    FSInputFile,
    KeyboardButton,
    ReplyKeyboardMarkup,
    ReplyKeyboardRemove
)
from dotenv import load_dotenv

# ==================== НАСТРОЙКА ЛОГИРОВАНИЯ ====================
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("app/bot.log", encoding='utf-8'),  # Явно указываем кодировку UTF-8
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# ==================== ЗАГРУЗКА ПЕРЕМЕННЫХ ОКРУЖЕНИЯ ====================
load_dotenv()

# ==================== КОНСТАНТЫ И НАСТРОЙКИ ====================
DATA_DIR = os.getenv('DATA_DIR', '/app/data')
DOCUMENTS_DIR = os.getenv('DOCUMENTS_DIR', os.path.join(DATA_DIR, 'documents'))
CONFIG_PATH = os.getenv('CONFIG_PATH', os.path.join(DATA_DIR, 'config.json'))
VISITORS_PATH = os.getenv('VISITORS_PATH', os.path.join(DATA_DIR, 'visitors.json'))

# Для разработки в Windows можно переопределить пути
if os.name == 'nt':
    BASE_DIR = os.path.dirname(os.path.abspath(__file__))
    DATA_DIR = os.path.join(BASE_DIR, 'app', 'data')
    DOCUMENTS_DIR = os.path.join(DATA_DIR, 'documents')
    CONFIG_PATH = os.path.join(DATA_DIR, 'config.json')
    VISITORS_PATH = os.path.join(DATA_DIR, 'visitors.json')


# Создание всех необходимых директорий
os.makedirs(DOCUMENTS_DIR, exist_ok=True)
os.makedirs(os.path.dirname(CONFIG_PATH), exist_ok=True)
os.makedirs(os.path.dirname(VISITORS_PATH), exist_ok=True)


# ==================== КЛАССЫ ДЛЯ РАБОТЫ С КОНФИГУРАЦИЕЙ ====================
class Button:
    """Класс для представления кнопки в меню"""

    def __init__(self, text: str, callback: str, action: str = None, menu: str = None):
        self.text = text
        self.callback = callback
        self.action = action
        self.menu = menu


class Menu:
    """Класс для представления меню"""

    def __init__(self, name: str, text: str, buttons: List[List[Button]]):
        self.name = name
        self.text = text
        self.buttons = buttons


class BotConfig:
    """Класс для управления конфигурацией бота"""
    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance.load()
        return cls._instance

    def __init__(self):
        self.visitors = {}
        self.admin_ids = []
        self.text_responses = {}
        self.documents = {}
        self.menus = {}
        self.load_visitors()

    def load(self):
        """Загрузка конфигурации из файла"""
        try:
            if os.path.exists(CONFIG_PATH):
                with open(CONFIG_PATH, 'r', encoding='utf-8') as f:
                    config = json.load(f)
                    self._validate(config)
                    self._parse_config(config)
                    self._add_admin_panel()
            else:
                logger.warning("Конфигурационный файл не найден, создаем новый")
                self._create_default_config()

            # Проверка наличия главного меню
            if "main" not in self.menus:
                logger.error("Отсутствует меню 'main' в конфигурации!")
                self._create_default_main_menu()

        except Exception as e:
            logger.critical(f"Ошибка загрузки конфига: {e}")
            raise

    def _create_default_main_menu(self):
        """Создает главное меню по умолчанию"""
        self.menus["main"] = Menu(
            name="main",
            text="Главное меню",
            buttons=[
                [Button("ℹ️ Информация", "info", "show_text", "info")]
            ]
        )
        self.text_responses["info"] = "Это информационное сообщение"
        self.save()

    def _create_default_config(self):
        """Создание конфигурации по умолчанию"""
        self.admin_ids = []
        self.text_responses = {}
        self.documents = {}
        self.menus = {
            "main": Menu(
                name="main",
                text="Главное меню",
                buttons=[
                    [Button("ℹ️ Информация", "info", "show_text", "info")],
                    [Button("📂 Документы", "documents", "open_menu", "documents_menu")],
                    [Button("⚙️ Админ", "admin_menu", "open_menu", "admin_panel")]
                ]
            )
        }
        self.save()

    def save(self):
        """Сохранение конфигурации в файл"""
        try:
            config = {
                "admin_ids": self.admin_ids,
                "menus": {},
                "text_responses": self.text_responses,
                "documents": self.documents
            }

            for menu_name, menu in self.menus.items():
                if menu_name in ["admin_panel", "visitors_db_submenu", "config_submenu", "documents_submenu"]:
                    continue

                config["menus"][menu_name] = {
                    "text": menu.text,
                    "buttons": {
                        "rows": [
                            [
                                {
                                    "text": button.text,
                                    "callback": button.callback,
                                    "action": button.action,
                                    "menu": button.menu
                                } for button in row
                            ] for row in menu.buttons
                        ]
                    }
                }

            with open(CONFIG_PATH, 'w', encoding='utf-8') as f:
                json.dump(config, f, ensure_ascii=False, indent=4)

            self.save_visitors()
            logger.info("Конфигурация сохранена")
        except Exception as e:
            logger.error(f"Ошибка сохранения конфига: {e}")
            raise

    def save_visitors(self):
        """Сохранение базы посетителей"""
        try:
            with open(VISITORS_PATH, 'w', encoding='utf-8') as f:
                json.dump(self.visitors, f, ensure_ascii=False, indent=4)
            return True
        except Exception as e:
            logger.error(f"Ошибка сохранения базы посетителей: {e}")
            return False

    def load_visitors(self):
        """Загрузка базы посетителей"""
        try:
            if os.path.exists(VISITORS_PATH):
                with open(VISITORS_PATH, 'r', encoding='utf-8') as f:
                    self.visitors = json.load(f)
                return True
        except Exception as e:
            logger.error(f"Ошибка загрузки базы посетителей: {e}")
        return False

    def add_document(self, file_name: str, description: str) -> str:
        """Добавление документа в систему"""
        doc_key = f"doc_{len(self.documents) + 1}"
        self.documents[doc_key] = {
            "description": description,
            "file": file_name
        }
        self._update_documents_menu()
        self.save()
        return doc_key

    def delete_document(self, doc_key: str) -> bool:
        """Удаление документа из системы"""
        if doc_key not in self.documents:
            return False

        try:
            # Удаление файла документа
            doc_info = self.documents[doc_key]
            file_path = os.path.join(DOCUMENTS_DIR, doc_info['file'])
            if os.path.exists(file_path):
                os.remove(file_path)

            # Удаление записи о документе
            del self.documents[doc_key]

            # Обновление меню
            self._update_documents_menu()
            self.save()
            return True
        except Exception as e:
            logger.error(f"Ошибка при удалении документа {doc_key}: {e}")
            return False

    def _update_documents_menu(self):
        """Обновление меню документов"""
        if "documents_menu" in self.menus:
            # Сохраняем кнопку "Назад"
            back_button = None
            for row in self.menus["documents_menu"].buttons:
                for button in row:
                    if button.callback == "back_to_main":
                        back_button = button
                        break

            # Создаем новые кнопки для документов
            self.menus["documents_menu"].buttons = [
                [Button(
                    text=f"📄 {doc_info['description']}",
                    callback=doc_key,
                    action="download_document"
                )] for doc_key, doc_info in self.documents.items()
            ]

            # Добавляем кнопку "Назад" если она была
            if back_button:
                self.menus["documents_menu"].buttons.append([back_button])

    def add_visitor(self, user_id: int, username: str, first_name: str, last_name: str):
        """Добавление информации о посетителе"""
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        if user_id not in self.visitors:
            self.visitors[user_id] = {
                "username": username,
                "first_name": first_name,
                "last_name": last_name,
                "first_visit": now,
                "last_visit": now,
                "visit_count": 1
            }
        else:
            self.visitors[user_id]["last_visit"] = now
            self.visitors[user_id]["visit_count"] += 1
        self.save_visitors()

    def get_visitors_stats(self) -> str:
        """Получение статистики посетителей"""
        total = len(self.visitors)
        active_today = sum(
            1 for v in self.visitors.values()
            if v["last_visit"].startswith(datetime.now().strftime("%Y-%m-%d")))
        return (f"📊 Статистика:\n\n👥 Всего: {total}\n🟢 Сегодня: {active_today}")

    def get_visitors_list(self) -> str:
        """Получение списка посетителей"""
        if not self.visitors:
            return "Список пуст"
        return "\n".join(
            f"{i}. {data['first_name']} {data['last_name']} (@{data['username']})\n"
            f"   🕒 Первый визит: {data['first_visit']}\n"
            f"   🔄 Последний: {data['last_visit']}\n"
            f"   🔢 Визитов: {data['visit_count']}\n"
            for i, (_, data) in enumerate(self.visitors.items(), 1)
        )

    def get_documents_list(self) -> str:
        """Получение списка документов"""
        if not self.documents:
            return "Документов нет"
        return "\n".join(
            f"{i}. {doc_info['description']}\n"
            f"   🔑 Ключ: {doc_key}\n"
            f"   📄 Файл: {doc_info['file']}\n"
            for i, (doc_key, doc_info) in enumerate(self.documents.items(), 1)
        )

    def _validate(self, config: dict):
        """Валидация структуры конфига"""
        required = ["admin_ids", "menus", "text_responses", "documents"]
        if not all(k in config for k in required):
            raise ValueError(f"Отсутствуют обязательные ключи: {required}")

    def _parse_config(self, config: dict):
        """Парсинг конфигурации"""
        logger.debug(f"Загружаемая конфигурация: {config}")
        self.admin_ids = config.get("admin_ids", [])
        self.text_responses = config.get("text_responses", {})
        self.documents = config.get("documents", {})

        logger.debug(f"Загружено admin_ids: {self.admin_ids}")
        logger.debug(f"Загружено text_responses: {self.text_responses.keys()}")
        logger.debug(f"Загружено documents: {self.documents.keys()}")

        self.menus = {}
        for menu_name, menu_data in config.get("menus", {}).items():
            logger.debug(f"Загрузка меню: {menu_name}")
            buttons = [
                [
                    Button(
                        text=button_data["text"],
                        callback=button_data["callback"],
                        action=button_data.get("action"),
                        menu=button_data.get("menu")
                    ) for button_data in row
                ] for row in menu_data["buttons"]["rows"]
            ]
            self.menus[menu_name] = Menu(menu_name, menu_data["text"], buttons)

        logger.debug(f"Загружено меню: {self.menus.keys()}")

    def _add_admin_panel(self):
        """Добавление админ-панели"""
        if not self.admin_ids:
            return

        admin_panel = Menu(
            "admin_panel", "⚙️ Админ-панель", [
                [Button("📊 Статистика", "visitors_stats")],
                [Button("👥 Список", "visitors_list")],
                [Button("💾 База", "visitors_db_menu", "open_menu", "visitors_db_submenu")],
                [Button("🛠 Конфиг", "config_menu", "open_menu", "config_submenu")],
                [Button("📂 Документы", "documents_menu", "open_menu", "documents_submenu")],
                [Button("🔙 Назад", "back_to_main", "open_menu", "main")]
            ]
        )

        self.menus.update({
            "admin_panel": admin_panel,
            "visitors_db_submenu": Menu(
                "visitors_db_submenu", "💾 База посетителей", [
                    [Button("📥 Выгрузить", "export_visitors")],
                    [Button("📤 Загрузить", "import_visitors")],
                    [Button("🔙 Назад", "back_to_admin", "open_menu", "admin_panel")]
                ]
            ),
            "config_submenu": Menu(
                "config_submenu", "🛠 Конфигурация", [
                    [Button("📥 Скачать", "download_config")],
                    [Button("📤 Загрузить", "upload_config")],
                    [Button("🔙 Назад", "back_to_admin", "open_menu", "admin_panel")]
                ]
            ),
            "documents_submenu": Menu(
                "documents_submenu", "📂 Документы", [
                    [Button("📥 Добавить", "upload_file")],
                    [Button("🗑️ Удалить", "delete_file")],
                    [Button("📋 Список", "list_documents")],
                    [Button("🔙 Назад", "back_to_admin", "open_menu", "admin_panel")]
                ]
            )
        })


# ==================== ИНИЦИАЛИЗАЦИЯ БОТА ====================
try:
    bot = Bot(token=os.getenv('BOT_TOKEN'))
    dp = Dispatcher()
    config_manager = BotConfig()
except Exception as e:
    logger.critical(f"Ошибка инициализации: {e}")
    exit(1)


# ==================== ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ ====================
def create_reply_keyboard(buttons: List[List[Button]]) -> ReplyKeyboardMarkup:
    """Создание reply-клавиатуры"""
    builder = ReplyKeyboardBuilder()
    for row in buttons:
        for button in row:
            builder.add(KeyboardButton(text=button.text))
    builder.adjust(1)
    return builder.as_markup(resize_keyboard=True)


async def show_menu(chat_id: int, menu_name: str):
    """Отображение меню"""
    menu = config_manager.menus.get(menu_name)
    if not menu:
        logger.error(f"Меню {menu_name} не найдено")
        return False

    try:
        await bot.send_message(
            chat_id=chat_id,
            text=menu.text,
            reply_markup=create_reply_keyboard(menu.buttons),
            parse_mode="HTML"
        )
        return True
    except Exception as e:
        logger.error(f"Ошибка отображения меню {menu_name}: {e}")
        return False


async def send_document(chat_id: int, doc_key: str):
    """Отправка документа"""
    if doc_key not in config_manager.documents:
        await bot.send_message(chat_id, "Документ не найден")
        return False

    doc_info = config_manager.documents[doc_key]
    doc_path = os.path.join(DOCUMENTS_DIR, doc_info['file'])

    if not os.path.exists(doc_path):
        await bot.send_message(chat_id, f"Документ недоступен: {doc_info['description']}")
        return False

    try:
        await bot.send_document(
            chat_id=chat_id,
            document=FSInputFile(doc_path),
            caption=doc_info['description']
        )
        return True
    except Exception as e:
        logger.error(f"Ошибка отправки документа {doc_key}: {e}")
        return False


# ==================== ОБРАБОТЧИКИ КОМАНД ====================
@dp.message(Command("start"))
async def cmd_start(message: types.Message):
    """Обработчик команды /start"""
    config_manager.add_visitor(
        message.from_user.id,
        message.from_user.username or "",
        message.from_user.first_name or "",
        message.from_user.last_name or ""
    )
    await show_menu(message.chat.id, "main")


@dp.message(Command("menu"))
async def cmd_menu(message: types.Message):
    """Обработчик команды /menu"""
    await show_menu(message.chat.id, "main")


@dp.message(Command("hide"))
async def cmd_hide(message: types.Message):
    """Обработчик команды /hide"""
    await message.answer(
        "Клавиатура скрыта. Используйте /menu для её отображения.",
        reply_markup=ReplyKeyboardRemove()
    )


# ==================== ОБРАБОТЧИКИ ДОКУМЕНТОВ ====================
@dp.message(F.document & (F.document.file_name == "visitors.json"))
async def handle_visitors_import(message: types.Message):
    """Импорт базы посетителей"""
    if message.from_user.id not in config_manager.admin_ids:
        await message.answer("🚫 Доступ запрещен")
        return

    try:
        file = await bot.get_file(message.document.file_id)
        await bot.download_file(file.file_path, destination=VISITORS_PATH)

        if config_manager.load_visitors():
            await message.answer("✅ База успешно загружена")
            logger.info(f"База обновлена админом {message.from_user.id}")
        else:
            await message.answer("❌ Ошибка загрузки")
    except Exception as e:
        logger.error(f"Ошибка импорта базы: {e}")
        await message.answer(f"❌ Ошибка: {e}")


@dp.message(F.document & (F.document.file_name == "config.json"))
async def handle_config_import(message: types.Message):
    """Импорт конфигурации"""
    if message.from_user.id not in config_manager.admin_ids:
        await message.answer("🚫 Доступ запрещен")
        return

    temp_path = f"{CONFIG_PATH}.temp"
    try:
        file = await bot.get_file(message.document.file_id)
        await bot.download_file(file.file_path, destination=temp_path)

        # Валидация перед применением
        with open(temp_path, 'r', encoding='utf-8') as f:
            new_config = json.load(f)
            required = ["admin_ids", "menus", "text_responses", "documents"]
            if not all(k in new_config for k in required):
                raise ValueError("Некорректная структура конфига")

        # Применение новой конфигурации
        os.replace(temp_path, CONFIG_PATH)
        config_manager.load()
        await message.answer("✅ Конфигурация обновлена")
        logger.info(f"Конфиг обновлен админом {message.from_user.id}")
    except Exception as e:
        logger.error(f"Ошибка импорта конфига: {e}")
        await message.answer(f"❌ Ошибка: {e}")
        if os.path.exists(temp_path):
            os.remove(temp_path)


@dp.message(F.document)
async def handle_document_upload(message: types.Message):
    """Загрузка документа"""
    if message.from_user.id not in config_manager.admin_ids:
        await message.answer("🚫 Доступ запрещен")
        return

    file = message.document
    file_name = file.file_name
    file_path = os.path.join(DOCUMENTS_DIR, file_name)

    try:
        # Скачивание файла
        file_info = await bot.get_file(file.file_id)
        await bot.download_file(file_info.file_path, destination=file_path)

        # Добавление в конфиг
        description = os.path.splitext(file_name)[0]
        doc_key = config_manager.add_document(file_name, description)

        await message.answer(
            f"✅ Файл {file_name} загружен\n"
            f"Описание: {description}\n"
            f"Ключ: {doc_key}"
        )
        logger.info(f"Документ {file_name} добавлен админом {message.from_user.id}")
    except Exception as e:
        logger.error(f"Ошибка загрузки файла: {e}")
        await message.answer(f"❌ Ошибка: {e}")


# ==================== ОБРАБОТЧИКИ СООБЩЕНИЙ ====================
@dp.message(F.text.startswith("doc_"))
async def handle_document_delete(message: types.Message):
    """Удаление документа по ключу"""
    if message.from_user.id not in config_manager.admin_ids:
        await message.answer("🚫 Доступ запрещен")
        return

    doc_key = message.text
    if doc_key in config_manager.documents:
        doc_info = config_manager.documents[doc_key]
        if config_manager.delete_document(doc_key):
            await message.answer(f"✅ Документ '{doc_info['description']}' удален")
            logger.info(f"Документ {doc_key} удален админом {message.from_user.id}")
        else:
            await message.answer(f"❌ Ошибка удаления '{doc_info['description']}'")
    else:
        await message.answer("❌ Документ не найден")


@dp.message()
async def handle_text_message(message: types.Message):
    """Обработчик текстовых сообщений"""
    # Сохранение информации о пользователе
    config_manager.add_visitor(
        message.from_user.id,
        message.from_user.username or "",
        message.from_user.first_name or "",
        message.from_user.last_name or ""
    )

    # Поиск обработчика для текста сообщения
    button = None
    for menu in config_manager.menus.values():
        for row in menu.buttons:
            for btn in row:
                if btn.text == message.text:
                    button = btn
                    break
            if button:
                break
        if button:
            break

    if not button:
        return

    # Проверка прав для админских команд
    admin_commands = ["download_config", "upload_config", "upload_file",
                      "delete_file", "visitors_stats", "visitors_list",
                      "export_visitors", "import_visitors", "list_documents"]
    if button.callback in admin_commands and message.from_user.id not in config_manager.admin_ids:
        await message.answer("🚫 Доступ запрещен")
        return

    # Обработка действий
    if button.action == "open_menu" and button.menu:
        await show_menu(message.chat.id, button.menu)
    elif button.action == "show_text" and button.callback in config_manager.text_responses:
        await message.answer(config_manager.text_responses[button.callback], parse_mode="HTML")
    elif button.action == "download_document":
        await send_document(message.chat.id, button.callback)
    elif button.callback == "download_config":
        await bot.send_document(message.from_user.id, FSInputFile(CONFIG_PATH), caption="📁 Конфигурация")
    elif button.callback == "upload_config":
        await message.answer("📎 Отправьте config.json для обновления конфигурации")
    elif button.callback == "upload_file":
        await message.answer("📎 Отправьте файл для загрузки")
    elif button.callback == "visitors_stats":
        await message.answer(config_manager.get_visitors_stats())
    elif button.callback == "visitors_list":
        await message.answer(config_manager.get_visitors_list())
    elif button.callback == "list_documents":
        await message.answer(config_manager.get_documents_list())
    elif button.callback == "export_visitors":
        if config_manager.save_visitors():
            await bot.send_document(message.from_user.id, FSInputFile(VISITORS_PATH), caption="📁 База посетителей")
        else:
            await message.answer("❌ Ошибка выгрузки")
    elif button.callback == "import_visitors":
        await message.answer("📎 Отправьте visitors.json для импорта базы")
    elif button.callback == "delete_file":
        await message.answer(
            "Введите ключ документа для удаления (начинается с doc_)\n\n"
            f"Доступные документы:\n{config_manager.get_documents_list()}"
        )


# ==================== ЗАПУСК БОТА ====================
async def main():
    try:
        # Проверка и создание файлов
        if not os.path.exists(VISITORS_PATH):
            with open(VISITORS_PATH, 'w') as f:
                json.dump({}, f)

        if not os.path.exists(CONFIG_PATH):
            default_config = {
                "admin_ids": [],
                "menus": {
                    "main": {
                        "text": "Главное меню",
                        "buttons": {
                            "rows": [
                                [{
                                    "text": "ℹ️ Информация",
                                    "callback": "info",
                                    "action": "show_text",
                                    "menu": "info"
                                }]
                            ]
                        }
                    }
                },
                "text_responses": {
                    "info": "Это информационное сообщение"
                },
                "documents": {}
            }
            with open(CONFIG_PATH, 'w') as f:
                json.dump(default_config, f, ensure_ascii=False, indent=4)
    except Exception as e:
        logger.critical(f"Ошибка инициализации: {e}")
        raise



    # Загрузка данных
    try:
        config_manager.load()
        config_manager.load_visitors()
        logger.info("Данные загружены")
    except Exception as e:
        logger.critical(f"Ошибка загрузки данных: {e}")
        raise

    # Настройка обработки сигналов (только для Unix-систем)
    if os.name != 'nt':  # Если не Windows
        loop = asyncio.get_running_loop()
        stop_event = asyncio.Event()

        def signal_handler():
            logger.info("Получен сигнал завершения...")
            stop_event.set()

        for sig in (signal.SIGTERM, signal.SIGINT):
            loop.add_signal_handler(sig, signal_handler)

    # Запуск поллинга
    try:
        polling_task = asyncio.create_task(dp.start_polling(bot, skip_updates=True))
        logger.info("Бот запущен и готов к работе")

        # Для Windows используем бесконечное ожидание с обработкой Ctrl+C
        if os.name == 'nt':
            try:
                while True:
                    await asyncio.sleep(1)
            except KeyboardInterrupt:
                logger.info("Получен сигнал Ctrl+C...")
        else:
            # Для Unix-систем ожидаем stop_event
            await stop_event.wait()

        logger.info("Завершение работы...")

        # Остановка поллинга
        polling_task.cancel()
        try:
            await polling_task
        except asyncio.CancelledError:
            logger.info("Поллинг остановлен")

    except Exception as e:
        logger.critical(f"Ошибка в работе бота: {e}", exc_info=True)

    finally:
        # Финализация
        try:
            logger.info("Сохранение данных...")
            config_manager.save()
            config_manager.save_visitors()

            await bot.session.close()
            logger.info("Бот успешно остановлен")
        except Exception as e:
            logger.error(f"Ошибка при завершении: {e}")

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