/
Befun
/
FamileControll
Обзор
Документация
Войти
/
Befun
/
FamileControll
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
bot.py
1 607 строк
69 KB
BeFun
Fixed bugs
27 фев 2026, 21:43
27 фев 2026, 21:43
e6d5448
Код
Авторство
О чём код?
""" Telegram Bot для системы родительского контроля Красивый интерфейс с inline кнопками """ import asyncio import logging from datetime import datetime, timedelta from typing import Optional, List, Dict from aiogram import Bot, Dispatcher, types, F from aiogram.filters import Command, CommandStart from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton, CallbackQuery from aiogram.utils.keyboard import InlineKeyboardBuilder from aiogram.exceptions import TelegramAPIError from aiogram.fsm.context import FSMContext from aiogram.fsm.state import State, StatesGroup from config import load_config, save_config from database import Database from blocker import AppBlocker, DownloadBlocker, PCController from process_monitor import ProcessMonitor, AppScanner from sleep_mode import SleepMode from updater import Updater logger = logging.getLogger(__name__) # Состояния для FSM (конечный автомат) class AddGameStates(StatesGroup): waiting_for_name = State() waiting_for_path = State() class AddBrowserStates(StatesGroup): waiting_for_name = State() waiting_for_path = State() class AddExtensionStates(StatesGroup): waiting_for_extension = State() class AddFamilyStates(StatesGroup): waiting_for_id = State() class BlockTimeStates(StatesGroup): waiting_for_time = State() class ParentalControlBot: """Основной класс Telegram бота""" def __init__(self, token: str, admin_id: int = None): self.bot = Bot(token=token) self.dp = Dispatcher() self.admin_id = admin_id # Инициализация компонентов self.db = Database() self.process_monitor = ProcessMonitor(self.db) self.app_scanner = AppScanner(self.db) self.blocker = AppBlocker(self.db, self.process_monitor) self.download_blocker = DownloadBlocker(self.db) self.pc_controller = PCController(self.db) self.sleep_mode = SleepMode(self.db, self.blocker) self.updater = Updater() # Установка callback для уведомлений self.sleep_mode.set_notification_callback(self._send_notification) # Регистрация обработчиков self._register_handlers() # Задачи self._monitor_task = None self._sleep_task = None # Ожидающие пользователи (для добавления в семью) self._pending_users: Dict[int, dict] = {} async def _send_notification(self, message: str, notification_type: str): """Отправка уведомления администратору""" if self.admin_id: try: await self.bot.send_message( self.admin_id, message, parse_mode='HTML' ) except TelegramAPIError as e: logger.error(f"Ошибка отправки уведомления: {e}") def _register_handlers(self): """Регистрация обработчиков бота""" @self.dp.message(CommandStart()) async def cmd_start(message: types.Message): await self._handle_start(message) @self.dp.message(Command("menu")) async def cmd_menu(message: types.Message): await self._show_main_menu(message) @self.dp.message(Command("scan")) async def cmd_scan(message: types.Message): await self._handle_scan(message) @self.dp.message(Command("block")) async def cmd_block(message: types.Message): await self._handle_block_command(message) @self.dp.message(Command("unblock")) async def cmd_unblock(message: types.Message): await self._handle_unblock_command(message) @self.dp.message(Command("allow")) async def cmd_allow(message: types.Message): await self._handle_allow_command(message) @self.dp.message(Command("sleep")) async def cmd_sleep(message: types.Message): await self._handle_sleep_command(message) @self.dp.message(Command("family")) async def cmd_family(message: types.Message): await self._handle_family_command(message) @self.dp.message(Command("history")) async def cmd_history(message: types.Message): await self._handle_history(message) @self.dp.message(Command("status")) async def cmd_status(message: types.Message): await self._handle_status(message) @self.dp.message(Command("shutdown")) async def cmd_shutdown(message: types.Message): await self._handle_shutdown(message) @self.dp.message(Command("help")) async def cmd_help(message: types.Message): await self._show_help(message) @self.dp.message(Command("update")) async def cmd_update(message: types.Message): await self._handle_update(message) # Обработка состояний FSM @self.dp.message(AddGameStates.waiting_for_name) async def process_game_name(message: types.Message, state: FSMContext): await self._process_add_game_name(message, state) @self.dp.message(AddGameStates.waiting_for_path) async def process_game_path(message: types.Message, state: FSMContext): await self._process_add_game_path(message, state) @self.dp.message(AddBrowserStates.waiting_for_name) async def process_browser_name(message: types.Message, state: FSMContext): await self._process_add_browser_name(message, state) @self.dp.message(AddBrowserStates.waiting_for_path) async def process_browser_path(message: types.Message, state: FSMContext): await self._process_add_browser_path(message, state) @self.dp.message(AddExtensionStates.waiting_for_extension) async def process_extension(message: types.Message, state: FSMContext): await self._process_add_extension(message, state) @self.dp.message(AddFamilyStates.waiting_for_id) async def process_family_id(message: types.Message, state: FSMContext): await self._process_add_family_id(message, state) # Callback handlers @self.dp.callback_query(F.data.startswith("menu_")) async def callback_menu(callback: CallbackQuery): await self._handle_menu_callback(callback) @self.dp.callback_query(F.data.startswith("block_")) async def callback_block(callback: CallbackQuery): await self._handle_block_callback(callback) @self.dp.callback_query(F.data.startswith("unblock_")) async def callback_unblock(callback: CallbackQuery): await self._handle_unblock_callback(callback) @self.dp.callback_query(F.data.startswith("allow_")) async def callback_allow(callback: CallbackQuery): await self._handle_allow_callback(callback) @self.dp.callback_query(F.data.startswith("sleep_")) async def callback_sleep(callback: CallbackQuery): await self._handle_sleep_callback(callback) @self.dp.callback_query(F.data.startswith("family_")) async def callback_family(callback: CallbackQuery, state: FSMContext): await self._handle_family_callback(callback, state) @self.dp.callback_query(F.data.startswith("pc_")) async def callback_pc(callback: CallbackQuery): await self._handle_pc_callback(callback) @self.dp.callback_query(F.data.startswith("download_")) async def callback_download(callback: CallbackQuery, state: FSMContext): await self._handle_download_callback(callback, state) @self.dp.callback_query(F.data.startswith("settings_")) async def callback_settings(callback: CallbackQuery): await self._handle_settings_callback(callback) @self.dp.callback_query(F.data.startswith("addgame_")) async def callback_addgame(callback: CallbackQuery, state: FSMContext): await self._handle_addgame_callback(callback, state) @self.dp.callback_query(F.data.startswith("addbrowser_")) async def callback_addbrowser(callback: CallbackQuery, state: FSMContext): await self._handle_addbrowser_callback(callback, state) # ==================== Проверка прав ==================== async def _check_access(self, user_id: int) -> bool: """Проверка доступа пользователя""" user = self.db.get_user(user_id) if not user: return False return user['is_active'] == 1 async def _is_admin(self, user_id: int) -> bool: """Проверка прав администратора""" return self.db.is_admin(user_id) async def _require_admin(self, message: types.Message) -> bool: """Требование прав администратора""" if not await self._is_admin(message.from_user.id): await message.answer("⛔ Эта команда доступна только администратору!") return False return True # ==================== Команды ==================== async def _handle_start(self, message: types.Message): """Обработка команды /start""" user_id = message.from_user.id user = self.db.get_user(user_id) if user: # Пользователь уже есть в базе await self._show_main_menu(message) return all_users = self.db.get_all_users() if not all_users: # Первый пользователь - администратор self.db.add_user( user_id, username=message.from_user.username, first_name=message.from_user.first_name, last_name=message.from_user.last_name, role='admin' ) self.admin_id = user_id await message.answer( "👋 <b>Добро пожаловать!</b>\n\n" "Вы зарегистрированы как <b>администратор</b> системы родительского контроля.\n\n" "Используйте /menu для доступа к управлению.", parse_mode='HTML' ) return # Новый пользователь - сохраняем в ожидающие self._pending_users[user_id] = { 'username': message.from_user.username, 'first_name': message.from_user.first_name, 'last_name': message.from_user.last_name } # Уведомляем админа admin = self.db.get_admin() if admin: name = message.from_user.first_name or message.from_user.username or f"ID: {user_id}" try: await self.bot.send_message( admin['telegram_id'], f"🔔 <b>Новый пользователь хочет присоединиться!</b>\n\n" f"👤 Имя: <b>{name}</b>\n" f"🆔 ID: <code>{user_id}</code>\n\n" f"Используйте /family для управления.", parse_mode='HTML' ) except: pass await message.answer( "👋 <b>Система родительского контроля</b>\n\n" "Ваш запрос отправлен администратору.\n" "Ожидайте подтверждения.", parse_mode='HTML' ) async def _show_main_menu(self, message: types.Message): """Показ главного меню""" user_id = message.from_user.id is_admin = await self._is_admin(user_id) keyboard = self._build_main_menu_keyboard(is_admin) text = "🏠 <b>Главное меню</b>\n\n" text += f"👤 Ваш статус: <b>{'Администратор' if is_admin else 'Член семьи'}</b>\n\n" text += "Выберите действие:" await message.answer(text, reply_markup=keyboard, parse_mode='HTML') def _build_main_menu_keyboard(self, is_admin: bool) -> InlineKeyboardMarkup: """Построение клавиатуры главного меню""" builder = InlineKeyboardBuilder() # Основные действия builder.row( InlineKeyboardButton(text="🎮 Игры", callback_data="menu_games"), InlineKeyboardButton(text="🌐 Браузеры", callback_data="menu_browsers") ) builder.row( InlineKeyboardButton(text="😴 Режим сна", callback_data="menu_sleep"), InlineKeyboardButton(text="📊 Статус", callback_data="menu_status") ) builder.row( InlineKeyboardButton(text="📜 История", callback_data="menu_history"), InlineKeyboardButton(text="📥 Скачивания", callback_data="menu_downloads") ) if is_admin: builder.row( InlineKeyboardButton(text="👨👩👧👦 Семья", callback_data="menu_family"), InlineKeyboardButton(text="💻 ПК", callback_data="menu_pc") ) builder.row( InlineKeyboardButton(text="🔄 Обновления", callback_data="menu_updates"), InlineKeyboardButton(text="⚙️ Настройки", callback_data="menu_settings") ) return builder.as_markup() async def _handle_menu_callback(self, callback: CallbackQuery): """Обработка callback главного меню""" data = callback.data.replace("menu_", "") user_id = callback.from_user.id # Обработка кнопки "Назад" if data == "back": is_admin = await self._is_admin(user_id) keyboard = self._build_main_menu_keyboard(is_admin) text = "🏠 <b>Главное меню</b>\n\n" text += f"👤 Ваш статус: <b>{'Администратор' if is_admin else 'Член семьи'}</b>\n\n" text += "Выберите действие:" await callback.message.edit_text(text, reply_markup=keyboard, parse_mode='HTML') return # Обработка остальных пунктов меню if data == "games": await self._show_games_menu(callback) elif data == "browsers": await self._show_browsers_menu(callback) elif data == "sleep": await self._show_sleep_menu(callback) elif data == "status": await self._show_status(callback) elif data == "history": await self._show_history_menu(callback) elif data == "downloads": await self._show_downloads_menu(callback) elif data == "family": if await self._is_admin(user_id): await self._show_family_menu(callback) else: await callback.answer("⛔ Нет доступа", show_alert=True) elif data == "pc": if await self._is_admin(user_id): await self._show_pc_menu(callback) else: await callback.answer("⛔ Нет доступа", show_alert=True) elif data == "updates": if await self._is_admin(user_id): await self._show_updates_menu(callback) else: await callback.answer("⛔ Нет доступа", show_alert=True) elif data == "settings": if await self._is_admin(user_id): await self._show_settings_menu(callback) else: await callback.answer("⛔ Нет доступа", show_alert=True) else: await callback.answer("❌ Неизвестная команда") # ==================== Меню игр ==================== async def _show_games_menu(self, callback: CallbackQuery): """Показ меню игр""" games = self.db.get_discovered_apps(app_type='game') blocked = self.db.get_blocked_apps() blocked_names = {app['name'] for app in blocked if app.get('app_type') == 'game'} text = "🎮 <b>Управление играми</b>\n\n" if not games: text += "❌ Игры не найдены.\n" text += "Используйте сканирование или добавьте вручную." else: text += f"Найдено игр: <b>{len(games)}</b>\n" text += f"Заблокировано: <b>{len(blocked_names)}</b>\n\n" text += "Выберите действие:" builder = InlineKeyboardBuilder() # Кнопки управления builder.row( InlineKeyboardButton(text="🔍 Сканировать", callback_data="block_scan_games"), InlineKeyboardButton(text="➕ Добавить", callback_data="addgame_start") ) builder.row( InlineKeyboardButton(text="📋 Список", callback_data="block_list_games") ) if games: builder.row( InlineKeyboardButton(text="🚫 Заблокировать все", callback_data="block_all_games"), InlineKeyboardButton(text="✅ Разблокировать все", callback_data="unblock_all_games") ) builder.row( InlineKeyboardButton(text="◀️ Назад", callback_data="menu_back") ) await callback.message.edit_text(text, reply_markup=builder.as_markup(), parse_mode='HTML') async def _show_browsers_menu(self, callback: CallbackQuery): """Показ меню браузеров""" browsers = self.db.get_discovered_apps(app_type='browser') blocked = self.db.get_blocked_apps() blocked_names = {app['name'] for app in blocked if app.get('app_type') == 'browser'} text = "🌐 <b>Управление браузерами</b>\n\n" if not browsers: text += "❌ Браузеры не найдены.\n" text += "Используйте сканирование или добавьте вручную." else: text += f"Найдено браузеров: <b>{len(browsers)}</b>\n" text += f"Заблокировано: <b>{len(blocked_names)}</b>\n\n" text += "Выберите действие:" builder = InlineKeyboardBuilder() builder.row( InlineKeyboardButton(text="🔍 Сканировать", callback_data="block_scan_browsers"), InlineKeyboardButton(text="➕ Добавить", callback_data="addbrowser_start") ) builder.row( InlineKeyboardButton(text="📋 Список", callback_data="block_list_browsers") ) if browsers: builder.row( InlineKeyboardButton(text="🚫 Заблокировать все", callback_data="block_all_browsers"), InlineKeyboardButton(text="✅ Разблокировать все", callback_data="unblock_all_browsers") ) builder.row( InlineKeyboardButton(text="◀️ Назад", callback_data="menu_back") ) await callback.message.edit_text(text, reply_markup=builder.as_markup(), parse_mode='HTML') # ==================== Добавление игр/браузеров ==================== async def _handle_addgame_callback(self, callback: CallbackQuery, state: FSMContext = None): """Обработка добавления игры""" data = callback.data.replace("addgame_", "") user_id = callback.from_user.id if not await self._is_admin(user_id): await callback.answer("⛔ Нет доступа", show_alert=True) return if data == "start": await callback.message.edit_text( "➕ <b>Добавление игры</b>\n\n" "Введите название игры:", parse_mode='HTML' ) await state.set_state(AddGameStates.waiting_for_name) await callback.answer() async def _process_add_game_name(self, message: types.Message, state: FSMContext): """Обработка названия игры""" await state.update_data(game_name=message.text) await message.answer( f"✅ Название: <b>{message.text}</b>\n\n" f"Введите путь к исполняемому файлу (.exe) или отправьте '-' чтобы пропустить:", parse_mode='HTML' ) await state.set_state(AddGameStates.waiting_for_path) async def _process_add_game_path(self, message: types.Message, state: FSMContext): """Обработка пути к игре""" data = await state.get_data() game_name = data['game_name'] game_path = message.text if message.text != '-' else None # Добавляем в базу self.db.add_discovered_app(game_name, game_path, 'game', 'game') # Логируем self.db.log_activity( telegram_id=message.from_user.id, action='add_game', target=game_name, details=f"Path: {game_path}" ) await state.clear() await message.answer( f"✅ Игра <b>{game_name}</b> добавлена!", parse_mode='HTML' ) async def _process_add_browser_name(self, message: types.Message, state: FSMContext): """Обработка названия браузера""" await state.update_data(browser_name=message.text) await message.answer( f"✅ Название: <b>{message.text}</b>\n\n" f"Введите путь к исполняемому файлу (.exe) или отправьте '-' чтобы пропустить:", parse_mode='HTML' ) await state.set_state(AddBrowserStates.waiting_for_path) async def _process_add_browser_path(self, message: types.Message, state: FSMContext): """Обработка пути к браузеру""" data = await state.get_data() browser_name = data['browser_name'] browser_path = message.text if message.text != '-' else None # Добавляем в базу self.db.add_discovered_app(browser_name, browser_path, 'browser', 'browser') # Логируем self.db.log_activity( telegram_id=message.from_user.id, action='add_browser', target=browser_name, details=f"Path: {browser_path}" ) await state.clear() await message.answer( f"✅ Браузер <b>{browser_name}</b> добавлен!", parse_mode='HTML' ) # ==================== Меню режима сна ==================== async def _show_sleep_menu(self, callback: CallbackQuery): """Показ меню режима сна""" status = self.sleep_mode.get_status() text = "😴 <b>Режим сна</b>\n\n" if status['is_active']: schedule = status['schedule'] text += f"✅ Режим сна активен\n" text += f"🕐 Время сна: <b>{schedule['start_time']}</b> - <b>{schedule['end_time']}</b>\n" if status['is_sleep_time']: text += "\n🌙 <b>Сейчас время сна!</b>\n" text += "Все игры и браузеры заблокированы." else: text += "\n☀️ Сейчас активное время." else: text += "❌ Режим сна не настроен.\n" text += "Настройте время сна для автоматической блокировки." builder = InlineKeyboardBuilder() if status['is_active']: builder.row( InlineKeyboardButton(text="⚙️ Изменить", callback_data="sleep_configure"), InlineKeyboardButton(text="❌ Отключить", callback_data="sleep_disable") ) else: builder.row( InlineKeyboardButton(text="⚙️ Настроить", callback_data="sleep_configure") ) # Быстрые настройки builder.row( InlineKeyboardButton(text="🌙 22:00 - 07:00", callback_data="sleep_set_22:00_07:00"), InlineKeyboardButton(text="🌙 21:00 - 08:00", callback_data="sleep_set_21:00_08:00") ) builder.row( InlineKeyboardButton(text="◀️ Назад", callback_data="menu_back") ) await callback.message.edit_text(text, reply_markup=builder.as_markup(), parse_mode='HTML') async def _handle_sleep_callback(self, callback: CallbackQuery): """Обработка callback режима сна""" data = callback.data.replace("sleep_", "") user_id = callback.from_user.id if data == "configure": await callback.message.edit_text( "⚙️ <b>Настройка режима сна</b>\n\n" "Используйте команду:\n" "<code>/sleep HH:MM HH:MM</code>\n\n" "Пример: <code>/sleep 22:00 07:00</code>", parse_mode='HTML' ) elif data == "disable": self.sleep_mode.disable(user_id) await callback.answer("✅ Режим сна отключён") await self._show_sleep_menu(callback) elif data.startswith("set_"): times = data.replace("set_", "").split("_") if len(times) == 2: if self.sleep_mode.configure(times[0], times[1], user_id): await callback.answer(f"✅ Режим сна установлен: {times[0]} - {times[1]}") else: await callback.answer("❌ Ошибка настройки") await self._show_sleep_menu(callback) async def _handle_sleep_command(self, message: types.Message): """Обработка команды /sleep""" args = message.text.split()[1:] if len(args) == 2: start_time, end_time = args if self.sleep_mode.configure(start_time, end_time, message.from_user.id): await message.answer( f"✅ <b>Режим сна настроен</b>\n\n" f"🕐 Время сна: <b>{start_time}</b> - <b>{end_time}</b>", parse_mode='HTML' ) else: await message.answer("❌ Ошибка настройки. Используйте формат: /sleep 22:00 07:00") else: status = self.sleep_mode.get_status() text = "😴 <b>Режим сна</b>\n\n" text += f"Использование: <code>/sleep HH:MM HH:MM</code>\n" text += f"Пример: <code>/sleep 22:00 07:00</code>\n\n" if status['is_active']: schedule = status['schedule'] text += f"Текущее расписание: <b>{schedule['start_time']}</b> - <b>{schedule['end_time']}</b>" await message.answer(text, parse_mode='HTML') # ==================== Меню семьи ==================== async def _show_family_menu(self, callback: CallbackQuery): """Показ меню семьи (только для админа)""" users = self.db.get_all_users() admin = self.db.get_admin() text = "👨👩👧👦 <b>Управление семьёй</b>\n\n" if not users: text += "❌ Пользователи не найдены." else: text += f"Всего пользователей: <b>{len(users)}</b>\n\n" for user in users: role = "👑 Админ" if user['role'] == 'admin' else "👤 Член семьи" name = user['first_name'] or user['username'] or f"ID: {user['telegram_id']}" text += f"{role}: <b>{name}</b>\n" # Показываем ожидающих пользователей if self._pending_users: text += "\n⏳ <b>Ожидающие подтверждения:</b>\n" for uid, info in self._pending_users.items(): name = info['first_name'] or info['username'] or f"ID: {uid}" text += f"• {name} (ID: <code>{uid}</code>)\n" builder = InlineKeyboardBuilder() # Кнопки для одобрения ожидающих for uid, info in self._pending_users.items(): name = info['first_name'] or info['username'] or f"ID: {uid}" builder.row( InlineKeyboardButton( text=f"✅ Принять {name}", callback_data=f"family_approve_{uid}" ) ) # Кнопки для удаления членов семьи (кроме админа) for user in users: if user['role'] != 'admin': name = user['first_name'] or user['username'] or f"ID: {user['telegram_id']}" builder.row( InlineKeyboardButton( text=f"❌ Удалить {name}", callback_data=f"family_remove_{user['telegram_id']}" ) ) builder.row( InlineKeyboardButton(text="➕ Добавить по ID", callback_data="family_addbyid") ) builder.row( InlineKeyboardButton(text="◀️ Назад", callback_data="menu_back") ) await callback.message.edit_text(text, reply_markup=builder.as_markup(), parse_mode='HTML') async def _handle_family_callback(self, callback: CallbackQuery, state: FSMContext = None): """Обработка callback семьи""" data = callback.data.replace("family_", "") user_id = callback.from_user.id if not await self._is_admin(user_id): await callback.answer("⛔ Нет доступа", show_alert=True) return if data == "addbyid": await callback.message.edit_text( "➕ <b>Добавление члена семьи по ID</b>\n\n" "Введите Telegram ID пользователя:", parse_mode='HTML' ) await state.set_state(AddFamilyStates.waiting_for_id) return if data.startswith("approve_"): target_id = int(data.replace("approve_", "")) if target_id in self._pending_users: info = self._pending_users[target_id] self.db.add_user( target_id, username=info['username'], first_name=info['first_name'], last_name=info['last_name'], role='family' ) del self._pending_users[target_id] # Уведомляем нового пользователя try: await self.bot.send_message( target_id, "✅ <b>Вы добавлены в семью!</b>\n\n" "Теперь вы можете управлять системой родительского контроля.\n" "Используйте /menu для начала работы.", parse_mode='HTML' ) except: pass await callback.answer("✅ Пользователь добавлен") await self._show_family_menu(callback) else: await callback.answer("❌ Пользователь не найден") elif data.startswith("remove_"): target_id = int(data.replace("remove_", "")) if self.db.remove_user(target_id): await callback.answer("✅ Пользователь удалён") else: await callback.answer("❌ Нельзя удалить администратора") await self._show_family_menu(callback) async def _process_add_family_id(self, message: types.Message, state: FSMContext): """Обработка добавления члена семьи по ID""" try: target_id = int(message.text.strip()) # Проверяем, не существует ли уже if self.db.get_user(target_id): await message.answer("❌ Пользователь уже добавлен.") await state.clear() return # Добавляем пользователя self.db.add_user(target_id, role='family') # Пробуем уведомить try: await self.bot.send_message( target_id, "✅ <b>Вы добавлены в семью!</b>\n\n" "Теперь вы можете управлять системой родительского контроля.\n" "Используйте /menu для начала работы.", parse_mode='HTML' ) except: pass await message.answer( f"✅ Пользователь с ID <code>{target_id}</code> добавлен!", parse_mode='HTML' ) except ValueError: await message.answer("❌ Неверный формат ID. Введите числовой ID.") return await state.clear() async def _handle_family_command(self, message: types.Message): """Обработка команды /family""" if not await self._require_admin(message): return users = self.db.get_all_users() text = "👨👩👧👦 <b>Члены семьи</b>\n\n" for user in users: role = "👑 Админ" if user['role'] == 'admin' else "👤 Член семьи" name = user['first_name'] or user['username'] or f"ID: {user['telegram_id']}" text += f"{role}: <b>{name}</b>\n" if user['last_activity']: text += f" Последняя активность: {user['last_activity']}\n" await message.answer(text, parse_mode='HTML') # ==================== Меню ПК ==================== async def _show_pc_menu(self, callback: CallbackQuery): """Показ меню управления ПК""" text = "💻 <b>Управление компьютером</b>\n\n" text += "Выберите действие:" builder = InlineKeyboardBuilder() builder.row( InlineKeyboardButton(text="🔴 Выключить", callback_data="pc_shutdown"), InlineKeyboardButton(text="🔄 Перезагрузить", callback_data="pc_restart") ) builder.row( InlineKeyboardButton(text="🔒 Заблокировать", callback_data="pc_lock"), InlineKeyboardButton(text="❌ Отменить выключение", callback_data="pc_cancel") ) builder.row( InlineKeyboardButton(text="◀️ Назад", callback_data="menu_back") ) await callback.message.edit_text(text, reply_markup=builder.as_markup(), parse_mode='HTML') async def _handle_pc_callback(self, callback: CallbackQuery): """Обработка callback управления ПК""" data = callback.data.replace("pc_", "") user_id = callback.from_user.id if not await self._is_admin(user_id): await callback.answer("⛔ Нет доступа", show_alert=True) return if data == "shutdown": self.pc_controller.shutdown(user_id, delay_seconds=60) await callback.message.edit_text( "🔴 <b>Выключение ПК</b>\n\n" "Компьютер выключится через 60 секунд.\n" "Используйте 'Отменить выключение' для отмены.", parse_mode='HTML' ) elif data == "restart": self.pc_controller.restart(user_id, delay_seconds=60) await callback.message.edit_text( "🔄 <b>Перезагрузка ПК</b>\n\n" "Компьютер перезагрузится через 60 секунд.", parse_mode='HTML' ) elif data == "lock": self.pc_controller.lock_screen(user_id) await callback.answer("🔒 Экран заблокирован") elif data == "cancel": self.pc_controller.cancel_shutdown(user_id) await callback.answer("✅ Выключение отменено") async def _handle_shutdown(self, message: types.Message): """Обработка команды /shutdown""" if not await self._require_admin(message): return args = message.text.split()[1:] delay = int(args[0]) * 60 if args else 60 self.pc_controller.shutdown(message.from_user.id, delay_seconds=delay) await message.answer( f"🔴 <b>Выключение ПК</b>\n\n" f"Компьютер выключится через {delay} секунд.", parse_mode='HTML' ) # ==================== История ==================== async def _show_history_menu(self, callback: CallbackQuery): """Показ меню истории""" user_id = callback.from_user.id is_admin = await self._is_admin(user_id) # Обычные пользователи видят только свою историю telegram_id = None if is_admin else user_id history = self.db.get_history(limit=20, telegram_id=telegram_id) text = "📜 <b>История действий</b>\n\n" if not history: text += "История пуста." else: for entry in history: action_icons = { 'block_app': '🚫', 'unblock_app': '✅', 'allow_app': '✔️', 'disallow_app': '❌', 'add_game': '🎮', 'add_browser': '🌐', 'add_extension': '📥', 'sleep_mode_activated': '🌙', 'sleep_mode_deactivated': '☀️', 'shutdown_pc': '🔴', 'restart_pc': '🔄', 'lock_screen': '🔒' } icon = action_icons.get(entry['action'], '📌') time_str = entry['timestamp'][:16] if entry['timestamp'] else '' name = entry.get('first_name') or entry.get('username') or 'Unknown' text += f"{icon} <b>{entry['action']}</b>\n" text += f" {entry['target'] or ''} | {name} | {time_str}\n" builder = InlineKeyboardBuilder() builder.row(InlineKeyboardButton(text="◀️ Назад", callback_data="menu_back")) await callback.message.edit_text(text, reply_markup=builder.as_markup(), parse_mode='HTML') async def _handle_history(self, message: types.Message): """Обработка команды /history""" user_id = message.from_user.id is_admin = await self._is_admin(user_id) telegram_id = None if is_admin else user_id history = self.db.get_history(limit=50, telegram_id=telegram_id) text = "📜 <b>История действий</b>\n\n" for entry in history[:30]: icon = '📌' time_str = entry['timestamp'][:16] if entry['timestamp'] else '' text += f"{icon} {entry['action']}: {entry['target'] or ''} ({time_str})\n" await message.answer(text, parse_mode='HTML') # ==================== Статус ==================== async def _show_status(self, callback: CallbackQuery): """Показ статуса системы""" blocked = self.db.get_blocked_apps() allowed = self.db.get_allowed_apps() sleep_status = self.sleep_mode.get_status() running = self.process_monitor.get_monitored_processes() text = "📊 <b>Статус системы</b>\n\n" text += f"🚫 Заблокировано приложений: <b>{len(blocked)}</b>\n" text += f"✔️ Разрешено приложений: <b>{len(allowed)}</b>\n" text += f"🔄 Запущено игр/браузеров: <b>{len(running)}</b>\n\n" text += f"😴 Режим сна: " if sleep_status['is_active']: schedule = sleep_status['schedule'] text += f"<b>активен</b> ({schedule['start_time']}-{schedule['end_time']})\n" else: text += "<b>отключён</b>\n" if running: text += "\n<b>Запущенные процессы:</b>\n" for proc in running[:10]: status = "🚫" if self.db.is_app_blocked(proc['name'], proc['exe']) else "✅" text += f"{status} {proc['name']}\n" builder = InlineKeyboardBuilder() builder.row(InlineKeyboardButton(text="🔄 Обновить", callback_data="menu_status")) builder.row(InlineKeyboardButton(text="◀️ Назад", callback_data="menu_back")) await callback.message.edit_text(text, reply_markup=builder.as_markup(), parse_mode='HTML') async def _handle_status(self, message: types.Message): """Обработка команды /status""" blocked = self.db.get_blocked_apps() sleep_status = self.sleep_mode.get_status() text = "📊 <b>Статус системы</b>\n\n" text += f"🚫 Заблокировано: <b>{len(blocked)}</b>\n" if sleep_status['is_active']: schedule = sleep_status['schedule'] text += f"😴 Режим сна: <b>{schedule['start_time']}-{schedule['end_time']}</b>\n" await message.answer(text, parse_mode='HTML') # ==================== Блокировка ==================== async def _handle_block_command(self, message: types.Message): """Обработка команды /block""" args = message.text.split()[1:] if not args: await message.answer( "🚫 <b>Блокировка приложения</b>\n\n" "Использование:\n" "<code>/block название [время]</code>\n\n" "Примеры:\n" "<code>/block minecraft</code> - навсегда\n" "<code>/block chrome 60</code> - на 60 минут\n" "<code>/block steam 120</code> - на 2 часа", parse_mode='HTML' ) return app_name = args[0] duration = int(args[1]) if len(args) > 1 else None # Определяем тип приложения app_type = 'app' games = self.db.get_discovered_apps(app_type='game') browsers = self.db.get_discovered_apps(app_type='browser') for g in games: if app_name.lower() in g['name'].lower(): app_type = 'game' break for b in browsers: if app_name.lower() in b['name'].lower(): app_type = 'browser' break if self.blocker.block_app( app_name=app_name, blocked_by=message.from_user.id, duration_minutes=duration, app_type=app_type ): text = f"✅ <b>{app_name}</b> заблокирован" if duration: text += f" на {duration} минут" await message.answer(text, parse_mode='HTML') else: await message.answer("❌ Ошибка блокировки") async def _handle_unblock_command(self, message: types.Message): """Обработка команды /unblock""" args = message.text.split()[1:] if not args: await message.answer( "✅ <b>Разблокировка приложения</b>\n\n" "Использование:\n" "<code>/unblock название</code>\n\n" "Пример:\n" "<code>/unblock minecraft</code>", parse_mode='HTML' ) return app_name = args[0] if self.blocker.unblock_app(app_name, message.from_user.id): await message.answer(f"✅ <b>{app_name}</b> разблокирован", parse_mode='HTML') else: await message.answer("❌ Приложение не найдено в заблокированных") async def _handle_allow_command(self, message: types.Message): """Обработка команды /allow""" args = message.text.split()[1:] if not args: await message.answer( "✔️ <b>Разрешение приложения</b>\n\n" "Использование:\n" "<code>/allow название [время]</code>\n\n" "Разрешённые приложения имеют приоритет над блокировками.", parse_mode='HTML' ) return app_name = args[0] duration = int(args[1]) if len(args) > 1 else None if self.blocker.allow_app( app_name=app_name, allowed_by=message.from_user.id, duration_minutes=duration ): text = f"✔️ <b>{app_name}</b> разрешён" if duration: text += f" на {duration} минут" await message.answer(text, parse_mode='HTML') else: await message.answer("❌ Ошибка разрешения") async def _handle_block_callback(self, callback: CallbackQuery): """Обработка callback блокировки""" data = callback.data.replace("block_", "") user_id = callback.from_user.id if data == "scan_games": await callback.message.edit_text("🔍 Сканирование игр...", parse_mode='HTML') games = self.app_scanner.scan_games() for game in games: self.db.add_discovered_app(game['name'], game['exe'], 'game', 'game') await callback.message.edit_text( f"✅ Найдено игр: <b>{len(games)}</b>", parse_mode='HTML' ) elif data == "scan_browsers": await callback.message.edit_text("🔍 Сканирование браузеров...", parse_mode='HTML') browsers = self.app_scanner.scan_browsers() for browser in browsers: self.db.add_discovered_app(browser['name'], browser['exe'], 'browser', 'browser') await callback.message.edit_text( f"✅ Найдено браузеров: <b>{len(browsers)}</b>", parse_mode='HTML' ) elif data == "all_games": count = self.blocker.block_all_games(user_id) await callback.answer(f"✅ Заблокировано: {count} игр") await self._show_games_menu(callback) elif data == "all_browsers": count = self.blocker.block_all_browsers(user_id) await callback.answer(f"✅ Заблокировано: {count} браузеров") await self._show_browsers_menu(callback) elif data == "list_games": await self._show_apps_list(callback, 'game') elif data == "list_browsers": await self._show_apps_list(callback, 'browser') async def _handle_unblock_callback(self, callback: CallbackQuery): """Обработка callback разблокировки""" data = callback.data.replace("unblock_", "") user_id = callback.from_user.id if data == "all_games": count = self.blocker.unblock_all_games(user_id) await callback.answer(f"✅ Разблокировано: {count} игр") await self._show_games_menu(callback) elif data == "all_browsers": count = self.blocker.unblock_all_browsers(user_id) await callback.answer(f"✅ Разблокировано: {count} браузеров") await self._show_browsers_menu(callback) async def _show_apps_list(self, callback: CallbackQuery, app_type: str, page: int = 0): """Показ списка приложений""" apps = self.db.get_discovered_apps(app_type=app_type) blocked = self.db.get_blocked_apps() allowed = self.db.get_allowed_apps() blocked_names = {app['name'] for app in blocked} allowed_names = {app['name'] for app in allowed} per_page = 5 text = f"📋 <b>{'Игры' if app_type == 'game' else 'Браузеры'}</b>\n\n" text += "Статусы:\n" text += "✅ - Доступно | 🚫 - Заблокировано | ✔️ - Разрешено\n\n" if not apps: text += "Список пуст. Выполните сканирование." else: start_idx = page * per_page end_idx = start_idx + per_page page_apps = apps[start_idx:end_idx] for app in page_apps: if app['name'] in allowed_names: status = "✔️" # Разрешено elif app['name'] in blocked_names: status = "🚫" # Заблокировано else: status = "✅" # Доступно text += f"{status} <b>{app['name']}</b>\n" builder = InlineKeyboardBuilder() # Кнопки для каждого приложения на странице if apps: start_idx = page * per_page end_idx = start_idx + per_page page_apps = apps[start_idx:end_idx] for app in page_apps: if app['name'] in allowed_names: # Разрешено - можно отменить разрешение builder.row( InlineKeyboardButton( text=f"❌ Отменить разрешение: {app['name'][:15]}", callback_data=f"allow_disallow_{app_type}_{app['name'][:30]}" ) ) elif app['name'] in blocked_names: # Заблокировано - можно разблокировать или разрешить builder.row( InlineKeyboardButton( text=f"✅ Разблокировать: {app['name'][:15]}", callback_data=f"allow_unblock_{app_type}_{app['name'][:30]}" ), InlineKeyboardButton( text=f"✔️ Разрешить: {app['name'][:15]}", callback_data=f"allow_allow_{app_type}_{app['name'][:30]}" ) ) else: # Доступно - можно заблокировать builder.row( InlineKeyboardButton( text=f"🚫 Заблокировать: {app['name'][:15]}", callback_data=f"allow_block_{app_type}_{app['name'][:30]}" ) ) # Навигация total_pages = (len(apps) + per_page - 1) // per_page if apps else 1 nav_buttons = [] if page > 0: nav_buttons.append(InlineKeyboardButton( text="⬅️ Назад", callback_data=f"allow_page_{app_type}_{page - 1}" )) nav_buttons.append(InlineKeyboardButton( text=f"{page + 1}/{total_pages}", callback_data="allow_noop" )) if page < total_pages - 1: nav_buttons.append(InlineKeyboardButton( text="➡️ Вперёд", callback_data=f"allow_page_{app_type}_{page + 1}" )) if nav_buttons: builder.row(*nav_buttons) builder.row( InlineKeyboardButton(text="🏠 В меню", callback_data=f"menu_{'games' if app_type == 'game' else 'browsers'}") ) await callback.message.edit_text(text, reply_markup=builder.as_markup(), parse_mode='HTML') async def _handle_allow_callback(self, callback: CallbackQuery): """Обработка callback для действий с приложениями""" data = callback.data.replace("allow_", "") user_id = callback.from_user.id if data == "noop": await callback.answer() return if data.startswith("page_"): # Пагинация parts = data.split("_") app_type = parts[1] page = int(parts[2]) await self._show_apps_list(callback, app_type, page) return if data.startswith("block_"): # Блокировка приложения parts = data.split("_", 2) if len(parts) >= 3: app_type = parts[1] app_name = parts[2] self.blocker.block_app(app_name, user_id, app_type=app_type) await callback.answer(f"✅ {app_name} заблокирован") await self._show_apps_list(callback, app_type) return if data.startswith("unblock_"): # Разблокировка приложения parts = data.split("_", 2) if len(parts) >= 3: app_type = parts[1] app_name = parts[2] self.blocker.unblock_app(app_name, user_id) await callback.answer(f"✅ {app_name} разблокирован") await self._show_apps_list(callback, app_type) return if data.startswith("allow_"): # Разрешение приложения parts = data.split("_", 2) if len(parts) >= 3: app_type = parts[1] app_name = parts[2] self.blocker.allow_app(app_name, user_id) await callback.answer(f"✔️ {app_name} разрешён") await self._show_apps_list(callback, app_type) return if data.startswith("disallow_"): # Отмена разрешения parts = data.split("_", 2) if len(parts) >= 3: app_type = parts[1] app_name = parts[2] self.blocker.disallow_app(app_name, user_id) await callback.answer(f"❌ Разрешение для {app_name} отменено") await self._show_apps_list(callback, app_type) return await callback.answer("❌ Неизвестная команда") # ==================== Скачивания ==================== async def _show_downloads_menu(self, callback: CallbackQuery): """Показ меню скачиваний""" blocked_ext = self.download_blocker.get_blocked_extensions() text = "📥 <b>Блокировка скачивания</b>\n\n" text += f"Заблокированные расширения: <b>{len(blocked_ext)}</b>\n\n" if blocked_ext: text += "Список:\n" for ext in blocked_ext[:15]: text += f"🚫 {ext}\n" else: text += "Нет заблокированных расширений.\n" builder = InlineKeyboardBuilder() builder.row( InlineKeyboardButton(text="➕ Добавить расширение", callback_data="download_add") ) # Кнопки для удаления расширений if blocked_ext: for ext in blocked_ext[:8]: builder.row( InlineKeyboardButton( text=f"✅ Разрешить {ext}", callback_data=f"download_remove_{ext}" ) ) builder.row( InlineKeyboardButton(text="◀️ Назад", callback_data="menu_back") ) await callback.message.edit_text(text, reply_markup=builder.as_markup(), parse_mode='HTML') async def _handle_download_callback(self, callback: CallbackQuery, state: FSMContext = None): """Обработка callback скачиваний""" data = callback.data.replace("download_", "") user_id = callback.from_user.id if not await self._is_admin(user_id): await callback.answer("⛔ Нет доступа", show_alert=True) return if data == "add": await callback.message.edit_text( "➕ <b>Добавление расширения</b>\n\n" "Введите расширение файла для блокировки:\n" "Пример: <code>.exe</code> или <code>exe</code>", parse_mode='HTML' ) await state.set_state(AddExtensionStates.waiting_for_extension) elif data.startswith("remove_"): ext = data.replace("remove_", "") self.download_blocker.allow_extension(ext) await callback.answer(f"✅ Расширение {ext} разрешено") await self._show_downloads_menu(callback) async def _process_add_extension(self, message: types.Message, state: FSMContext): """Обработка добавления расширения""" ext = message.text.strip() if not ext.startswith('.'): ext = '.' + ext self.download_blocker.block_extension(ext, message.from_user.id) await state.clear() await message.answer( f"✅ Расширение <b>{ext}</b> заблокировано для скачивания!", parse_mode='HTML' ) # ==================== Обновления ==================== async def _show_updates_menu(self, callback: CallbackQuery): """Показ меню обновлений""" await callback.message.edit_text("🔍 Проверка обновлений...", parse_mode='HTML') update_info = self.updater.check_for_updates() text = "🔄 <b>Обновления</b>\n\n" text += f"Текущая версия: <b>{update_info['current_version']}</b>\n" if update_info['has_update']: text += f"Доступна версия: <b>{update_info['latest_version']}</b>\n\n" text += "📝 Примечания:\n" text += update_info['release_notes'][:500] builder = InlineKeyboardBuilder() builder.row(InlineKeyboardButton(text="⬇️ Обновить", callback_data="update_install")) else: text += "\n✅ Установлена последняя версия." builder = InlineKeyboardBuilder() builder.row(InlineKeyboardButton(text="◀️ Назад", callback_data="menu_back")) await callback.message.edit_text(text, reply_markup=builder.as_markup(), parse_mode='HTML') async def _handle_update(self, message: types.Message): """Обработка команды /update""" if not await self._require_admin(message): return update_info = self.updater.check_for_updates() if update_info['has_update']: await message.answer( f"🔄 Доступно обновление!\n\n" f"Текущая: {update_info['current_version']}\n" f"Новая: {update_info['latest_version']}", parse_mode='HTML' ) else: await message.answer("✅ Установлена последняя версия.", parse_mode='HTML') # ==================== Сканирование ==================== async def _handle_scan(self, message: types.Message): """Обработка команды /scan""" status_msg = await message.answer("🔍 Сканирование системы...") # Сканируем игры games = self.app_scanner.scan_games() for game in games: self.db.add_discovered_app(game['name'], game['exe'], 'game', 'game') # Сканируем браузеры browsers = self.app_scanner.scan_browsers() for browser in browsers: self.db.add_discovered_app(browser['name'], browser['exe'], 'browser', 'browser') await status_msg.edit_text( f"✅ <b>Сканирование завершено</b>\n\n" f"🎮 Найдено игр: <b>{len(games)}</b>\n" f"🌐 Найдено браузеров: <b>{len(browsers)}</b>", parse_mode='HTML' ) # ==================== Настройки ==================== async def _show_settings_menu(self, callback: CallbackQuery): """Показ меню настроек""" from autostart import AutoStartManager autostart = AutoStartManager() text = "⚙️ <b>Настройки</b>\n\n" text += f"🔄 Автозапуск: {autostart.get_status_string()}\n" builder = InlineKeyboardBuilder() if autostart.is_autostart_enabled(): builder.row(InlineKeyboardButton(text="❌ Отключить автозапуск", callback_data="settings_autostart_disable")) else: builder.row(InlineKeyboardButton(text="✅ Включить автозапуск", callback_data="settings_autostart_enable")) builder.row(InlineKeyboardButton(text="◀️ Назад", callback_data="menu_back")) await callback.message.edit_text(text, reply_markup=builder.as_markup(), parse_mode='HTML') async def _handle_settings_callback(self, callback: CallbackQuery): """Обработка callback настроек""" data = callback.data.replace("settings_", "") if not await self._is_admin(callback.from_user.id): await callback.answer("⛔ Нет доступа", show_alert=True) return from autostart import AutoStartManager autostart = AutoStartManager() if data == "autostart_enable": if autostart.enable_autostart(): await callback.answer("✅ Автозапуск включён") else: await callback.answer("❌ Ошибка включения") elif data == "autostart_disable": if autostart.disable_autostart(): await callback.answer("✅ Автозапуск отключён") else: await callback.answer("❌ Ошибка отключения") await self._show_settings_menu(callback) # ==================== Помощь ==================== async def _show_help(self, message: types.Message): """Показ справки""" text = "📚 <b>Справка по командам</b>\n\n" text += "<b>Основные команды:</b>\n" text += "/start - Начало работы\n" text += "/menu - Главное меню\n" text += "/status - Статус системы\n" text += "/history - История действий\n\n" text += "<b>Управление приложениями:</b>\n" text += "/scan - Сканировать игры и браузеры\n" text += "/block название [минуты] - Заблокировать\n" text += "/unblock название - Разблокировать\n" text += "/allow название [минуты] - Разрешить\n\n" text += "<b>Режим сна:</b>\n" text += "/sleep HH:MM HH:MM - Настроить\n" text += "Пример: /sleep 22:00 07:00\n\n" if await self._is_admin(message.from_user.id): text += "<b>Команды администратора:</b>\n" text += "/family - Управление семьёй\n" text += "/shutdown [минуты] - Выключить ПК\n" text += "/update - Проверить обновления\n" await message.answer(text, parse_mode='HTML') # ==================== Дополнительный callback для браузера ==================== async def _handle_addbrowser_callback(self, callback: CallbackQuery, state: FSMContext = None): """Обработка добавления браузера""" user_id = callback.from_user.id if not await self._is_admin(user_id): await callback.answer("⛔ Нет доступа", show_alert=True) return await callback.message.edit_text( "➕ <b>Добавление браузера</b>\n\n" "Введите название браузера:", parse_mode='HTML' ) await state.set_state(AddBrowserStates.waiting_for_name) await callback.answer() # ==================== Запуск ==================== async def start(self): """Запуск бота""" logger.info("Запуск бота...") # Запуск мониторинга процессов self._monitor_task = asyncio.create_task( self.process_monitor.monitor_loop( callback=self._on_blocked_process, interval=5 ) ) # Запуск мониторинга режима сна self._sleep_task = asyncio.create_task( self.sleep_mode.monitor_loop(interval=30) ) # Запуск бота await self.dp.start_polling(self.bot) async def _on_blocked_process(self, process: dict, event_type: str): """Callback при обнаружении заблокированного процесса""" if event_type == 'blocked': # Уведомляем администратора if self.admin_id: try: await self.bot.send_message( self.admin_id, f"🚫 <b>Заблокирован процесс</b>\n\n" f"Приложение: <b>{process['name']}</b>\n" f"Было завершено автоматически.", parse_mode='HTML' ) except TelegramAPIError: pass async def stop(self): """Остановка бота""" logger.info("Остановка бота...") # Остановка мониторинга self.process_monitor.stop_monitoring() self.sleep_mode.stop_monitoring() if self._monitor_task: self._monitor_task.cancel() if self._sleep_task: self._sleep_task.cancel() await self.bot.session.close()