max_bot.py - master - Befun/FamalyControlle

max_bot.py
695 строк30 KB
BeFun
Init project
07 май 2026, 10:43
07 май 2026, 10:437e96900
"""
Бот мессенджера Макс.
Long-polling клиент для API Макс, обрабатывает команды чата.

Команды:
  /start, /help, /block, /unblock, /shutdown, /restart,
  /sleep, /rest, /allow, /limit, /status, /history,
  /schedules, /users, /addfamily, /remove
"""

import asyncio
import logging
import re
import time
from datetime import datetime, timedelta
from typing import Any, Optional

import httpx

from core.config import Config
from core.permissions import Role, PermissionChecker
from utils.helpers import parse_day_range, now_iso, parse_duration

logger = logging.getLogger("bot")


class MaxBot:
    """Бот мессенджера Макс — long polling."""

    def __init__(self, db=None):
        self.config = Config
        self.token = self.config.max_bot_token
        self.api_base = self.config.max_bot_api_base
        self.poll_interval = self.config.max_bot_poll_interval
        self._db = db
        self._running = False
        self._last_update_id: int = 0

        # HTTP-клиент для запросов к API Макс
        self._client = httpx.Client(timeout=30)

    # ── Связь с сервером ──────────────────────────────────────

    def set_db(self, db) -> None:
        """Установить ссылку на Database (вызывается из main.py)."""
        self._db = db

    def _server_request(self, method: str, path: str, json_data: dict = None) -> Optional[dict]:
        """HTTP-запрос к локальному API-серверу."""
        try:
            server_url = self.config.agent_server_url
            resp = httpx.request(
                method,
                f"{server_url}{path}",
                json=json_data,
                timeout=5,
            )
            if resp.status_code in (200, 201):
                return resp.json()
            logger.warning("Сервер вернул %d: %s", resp.status_code, resp.text)
            return None
        except Exception as e:
            logger.error("Ошибка запроса к серверу: %s", e)
            return None

    # ── API Макс ──────────────────────────────────────────────

    def _max_get_updates(self) -> list[dict]:
        """Получить новые сообщения через long polling."""
        try:
            resp = self._client.get(
                f"{self.api_base}/bot{self.token}/getUpdates",
                params={
                    "offset": self._last_update_id + 1,
                    "timeout": self.poll_interval,
                },
            )
            if resp.status_code == 200:
                data = resp.json()
                return data.get("result", [])
            return []
        except Exception as e:
            logger.error("Ошибка long-poll Макс: %s", e)
            return []

    def _max_send_message(self, chat_id: str, text: str) -> None:
        """Отправить сообщение в чат Макс."""
        try:
            self._client.post(
                f"{self.api_base}/bot{self.token}/sendMessage",
                json={
                    "chat_id": chat_id,
                    "text": text,
                },
            )
        except Exception as e:
            logger.error("Ошибка отправки сообщения: %s", e)

    # ── Идентификация ─────────────────────────────────────────

    def _get_or_register_user(self, max_user_id: str) -> Optional[dict]:
        """Найти или зарегистрировать пользователя. Возвращает dict с user-данными."""
        # Ищем существующего
        result = self._server_request("POST", "/api/auth/identify", {
            "max_user_id": max_user_id,
        })
        if result and result.get("ok"):
            return result.get("user")

        # Регистрируем нового
        result = self._server_request("POST", "/api/auth/register", {
            "max_user_id": max_user_id,
        })
        if result and result.get("ok"):
            return result.get("user")
        return None

    def _get_user_role(self, max_user_id: str) -> tuple[Optional[dict], Role]:
        """Получить пользователя и его роль."""
        user = self._get_or_register_user(max_user_id)
        if not user:
            return None, Role.CHILD
        role = PermissionChecker.get_role(user.get("role", "child"))
        return user, role

    # ── Основной цикл ─────────────────────────────────────────

    def start(self) -> None:
        """Запуск бота (блокирующий)."""
        if not self.token or self.token == "ТОКЕН_ВАШЕГО_БОТА":
            logger.error("ОШИБКА: Не задан токен бота в config.yaml")
            return

        logger.info("Бот Макс запущен")
        self._running = True

        while self._running:
            try:
                updates = self._max_get_updates()
                for update in updates:
                    self._process_update(update)
                    self._last_update_id = update.get("update_id", self._last_update_id)
            except Exception as e:
                logger.error("Ошибка в цикле бота: %s", e, exc_info=True)

            time.sleep(1)

    def stop(self) -> None:
        self._running = False

    def _process_update(self, update: dict) -> None:
        """Обработать одно обновление (сообщение)."""
        message = update.get("message", {})
        if not message:
            return

        chat_id = str(message.get("chat", {}).get("id", ""))
        text = message.get("text", "").strip()
        from_user = message.get("from", {})
        max_user_id = str(from_user.get("id", ""))

        if not text or not chat_id:
            return

        logger.info("Сообщение от %s: %s", max_user_id, text)
        self._handle_command(chat_id, max_user_id, text)

    # ── Обработчик команд ─────────────────────────────────────

    def _handle_command(self, chat_id: str, max_user_id: str, text: str) -> None:
        """Маршрутизация команд."""
        parts = text.split()
        cmd = parts[0].lower()
        args = parts[1:] if len(parts) > 1 else []

        handlers = {
            "/start": self._cmd_start,
            "/help": self._cmd_help,
            "/status": self._cmd_status,
            "/history": self._cmd_history,
            "/schedules": self._cmd_schedules,
            "/users": self._cmd_users,
            "/block": self._cmd_block,
            "/unblock": self._cmd_unblock,
            "/shutdown": self._cmd_shutdown,
            "/restart": self._cmd_restart,
            "/sleep": self._cmd_sleep,
            "/rest": self._cmd_rest,
            "/allow": self._cmd_allow,
            "/limit": self._cmd_limit,
            "/addfamily": self._cmd_add_family,
            "/remove": self._cmd_remove,
        }

        handler = handlers.get(cmd)
        if handler:
            handler(chat_id, max_user_id, args)
        else:
            self._max_send_message(chat_id, "Неизвестная команда. Отправьте /help для списка команд.")

    # ── Реализации команд ─────────────────────────────────────

    def _cmd_start(self, chat_id: str, max_user_id: str, args: list) -> None:
        user = self._get_or_register_user(max_user_id)
        if user:
            role = user.get("role", "child")
            role_names = {"admin": "Администратор", "family": "Член семьи", "child": "Ребёнок"}
            self._max_send_message(chat_id,
                f"Привет, {user.get('display_name', 'Пользователь')}!\n\n"
                f"Ваша роль: {role_names.get(role, role)}\n"
                f"Ваш ID: {user.get('id')}\n\n"
                f"Отправьте /help для списка команд."
            )
        else:
            self._max_send_message(chat_id, "Ошибка регистрации. Обратитесь к администратору.")

    def _cmd_help(self, chat_id: str, max_user_id: str, args: list) -> None:
        _, role = self._get_user_role(max_user_id)

        lines = ["Список команд:\n"]

        if role == Role.ADMIN:
            lines.extend([
                "/block games — Блокировать игры",
                "/block browsers — Блокировать браузеры",
                "/block downloads — Блокировать загрузки",
                "/block app <имя> — Блокировать приложение",
                "/unblock <id> — Снять запрет",
                "/shutdown — Выключить ПК",
                "/restart — Перезагрузить ПК",
                "/sleep 22:00 07:00 пн-пт — Расписание сна",
                "/rest 12:00 13:00 пн-пт — Расписание отдыха",
                "/allow 30 — Разрешить на 30 мин",
                "/limit 15 — Заблокировать через 15 мин",
                "/addfamily <id> <имя> — Добавить члена семьи",
                "/remove <id> — Удалить расписание/ограничение",
            ])
        elif role == Role.FAMILY:
            lines.extend([
                "/block games — Блокировать игры",
                "/block browsers — Блокировать браузеры",
                "/block downloads — Блокировать загрузки",
                "/block app <имя> — Блокировать приложение",
                "/unblock <id> — Снять свой запрет",
                "/restart — Перезагрузить ПК",
                "/sleep 22:00 07:00 пн-пт — Расписание сна",
                "/rest 12:00 13:00 пн-пт — Расписание отдыха",
                "/allow 30 — Разрешить на 30 мин",
                "/limit 15 — Заблокировать через 15 мин",
                "/remove <id> — Удалить расписание",
            ])
        else:
            lines.append("У вас нет команд управления.")

        lines.extend([
            "",
            "/status — Текущие ограничения",
            "/history — История действий",
            "/schedules — Список расписаний",
            "/users — Список семьи",
        ])

        self._max_send_message(chat_id, "\n".join(lines))

    def _cmd_block(self, chat_id: str, max_user_id: str, args: list) -> None:
        user, role = self._get_user_role(max_user_id)
        if not user:
            self._max_send_message(chat_id, "Ошибка: пользователь не найден.")
            return

        if not PermissionChecker.can(role, "block_games"):
            self._max_send_message(chat_id, "У вас нет прав на создание ограничений.")
            return

        if not args:
            self._max_send_message(chat_id, "Укажите категорию: /block games|browsers|downloads|app <имя>")
            return

        category = args[0].lower()

        if category == "app" and len(args) > 1:
            target = " ".join(args[1:])
            category = "app"
        elif category in ("games", "browsers", "downloads"):
            target = "*"
        else:
            # Может быть, указано как: /block app chrome.exe
            if len(args) >= 2 and args[0].lower() == "app":
                category = "app"
                target = " ".join(args[1:])
            else:
                self._max_send_message(chat_id, "Категория: games, browsers, downloads или app <имя>")
                return

        result = self._server_request("POST", "/api/restrictions", {
            "category": category,
            "target": target,
        }, )
        # Нужно передать заголовок X-User-ID
        # Переделаем через direct HTTP call
        result = self._server_request_with_auth("POST", "/api/restrictions", user["id"], {
            "category": category,
            "target": target,
        })

        if result and result.get("ok"):
            r = result.get("restriction", {})
            self._max_send_message(chat_id,
                f"Ограничение создано!\n"
                f"ID: {r.get('id')}\n"
                f"Категория: {r.get('category')}\n"
                f"Цель: {r.get('target')}\n"
                f"Срок: {r.get('expires_at') or 'Бессрочно'}"
            )
        else:
            self._max_send_message(chat_id, "Ошибка создания ограничения.")

    def _cmd_unblock(self, chat_id: str, max_user_id: str, args: list) -> None:
        user, role = self._get_user_role(max_user_id)
        if not user:
            self._max_send_message(chat_id, "Ошибка: пользователь не найден.")
            return

        if not args:
            self._max_send_message(chat_id, "Укажите ID ограничения: /unblock <id>")
            return

        try:
            restriction_id = int(args[0])
        except ValueError:
            self._max_send_message(chat_id, "ID должен быть числом.")
            return

        result = self._server_request_with_auth(
            "DELETE", f"/api/restrictions/{restriction_id}", user["id"]
        )

        if result and result.get("ok"):
            self._max_send_message(chat_id, f"Ограничение #{restriction_id} снято.")
        else:
            self._max_send_message(chat_id, "Не удалось снять ограничение. Проверьте ID или права.")

    def _cmd_shutdown(self, chat_id: str, max_user_id: str, args: list) -> None:
        user, role = self._get_user_role(max_user_id)
        if not user:
            return
        if not PermissionChecker.can(role, "shutdown"):
            self._max_send_message(chat_id, "У вас нет прав на выключение ПК.")
            return

        delay = int(args[0]) if args and args[0].isdigit() else 60
        result = self._server_request_with_auth("POST", "/api/system/shutdown", user["id"], {
            "delay_seconds": delay,
        })
        if result and result.get("ok"):
            self._max_send_message(chat_id, f"ПК будет выключен через {delay} секунд.")
        else:
            self._max_send_message(chat_id, "Ошибка выключения.")

    def _cmd_restart(self, chat_id: str, max_user_id: str, args: list) -> None:
        user, role = self._get_user_role(max_user_id)
        if not user:
            return
        if not PermissionChecker.can(role, "restart"):
            self._max_send_message(chat_id, "У вас нет прав на перезагрузку ПК.")
            return

        delay = int(args[0]) if args and args[0].isdigit() else 60
        result = self._server_request_with_auth("POST", "/api/system/restart", user["id"], {
            "delay_seconds": delay,
        })
        if result and result.get("ok"):
            self._max_send_message(chat_id, f"ПК будет перезагружен через {delay} секунд.")
        else:
            self._max_send_message(chat_id, "Ошибка перезагрузки.")

    def _cmd_sleep(self, chat_id: str, max_user_id: str, args: list) -> None:
        user, role = self._get_user_role(max_user_id)
        if not user:
            return
        if not PermissionChecker.can(role, "set_schedule"):
            self._max_send_message(chat_id, "У вас нет прав на создание расписаний.")
            return

        if len(args) < 2:
            self._max_send_message(chat_id, "Формат: /sleep 22:00 07:00 пн-пт")
            return

        start_time = args[0]
        end_time = args[1]
        days = parse_day_range(" ".join(args[2:])) if len(args) > 2 else list(range(7))

        result = self._server_request_with_auth("POST", "/api/schedules", user["id"], {
            "name": "Сон",
            "schedule_type": "sleep",
            "start_time": start_time,
            "end_time": end_time,
            "days": days,
        })

        if result and result.get("ok"):
            self._max_send_message(chat_id,
                f"Расписание сна создано:\n"
                f"{start_time} — {end_time}\n"
                f"Дни: {', '.join(str(d) for d in days)}\n"
                f"ID: {result.get('schedule', {}).get('id')}"
            )
        else:
            self._max_send_message(chat_id, "Ошибка создания расписания.")

    def _cmd_rest(self, chat_id: str, max_user_id: str, args: list) -> None:
        user, role = self._get_user_role(max_user_id)
        if not user:
            return
        if not PermissionChecker.can(role, "set_schedule"):
            self._max_send_message(chat_id, "У вас нет прав на создание расписаний.")
            return

        if len(args) < 2:
            self._max_send_message(chat_id, "Формат: /rest 12:00 13:00 пн-пт")
            return

        start_time = args[0]
        end_time = args[1]
        days = parse_day_range(" ".join(args[2:])) if len(args) > 2 else list(range(7))

        result = self._server_request_with_auth("POST", "/api/schedules", user["id"], {
            "name": "Отдых",
            "schedule_type": "rest",
            "start_time": start_time,
            "end_time": end_time,
            "days": days,
        })

        if result and result.get("ok"):
            self._max_send_message(chat_id,
                f"Расписание отдыха создано:\n"
                f"{start_time} — {end_time}\n"
                f"Дни: {', '.join(str(d) for d in days)}\n"
                f"ID: {result.get('schedule', {}).get('id')}"
            )
        else:
            self._max_send_message(chat_id, "Ошибка создания расписания.")

    def _cmd_allow(self, chat_id: str, max_user_id: str, args: list) -> None:
        """Временное разрешение — снимает все ограничения на N минут."""
        user, role = self._get_user_role(max_user_id)
        if not user:
            return
        if not PermissionChecker.can(role, "allow_temporary"):
            self._max_send_message(chat_id, "У вас нет прав на временное разрешение.")
            return

        if not args or not args[0].isdigit():
            self._max_send_message(chat_id, "Формат: /allow <минуты>")
            return

        minutes = int(args[0])
        # Устанавливаем ограничение с обратным эффектом через N минут
        # Сначала снимаем все текущие
        active = self._server_request("GET", "/api/restrictions/active")
        if active:
            for r in active.get("restrictions", []):
                self._server_request_with_auth(
                    "DELETE", f"/api/restrictions/{r['id']}", user["id"]
                )

        # Планируем обратную блокировку через N минут
        # (в простейшем случае — уведомляем)
        self._max_send_message(chat_id,
            f"Все ограничения сняты на {parse_duration(minutes)}.\n"
            f"Через {parse_duration(minutes)} ограничения будут возвращены."
        )

    def _cmd_limit(self, chat_id: str, max_user_id: str, args: list) -> None:
        """Полная блокировка через N минут."""
        user, role = self._get_user_role(max_user_id)
        if not user:
            return
        if not PermissionChecker.can(role, "set_limit"):
            self._max_send_message(chat_id, "У вас нет прав на установку лимитов.")
            return

        if not args or not args[0].isdigit():
            self._max_send_message(chat_id, "Формат: /limit <минуты>")
            return

        minutes = int(args[0])
        # Создаём ограничения games + browsers с expires_at = now + minutes
        for cat in ("games", "browsers", "downloads"):
            self._server_request_with_auth("POST", "/api/restrictions", user["id"], {
                "category": cat,
                "target": "*",
                "expires_minutes": minutes,
            })

        self._max_send_message(chat_id,
            f"Полная блокировка через {parse_duration(minutes)}.\n"
            f"Блокируются: игры, браузеры, загрузки.\n"
            f"Срок ограничения: {parse_duration(minutes)} с момента создания."
        )

    def _cmd_status(self, chat_id: str, max_user_id: str, args: list) -> None:
        """Текущие ограничения."""
        result = self._server_request("GET", "/api/restrictions/active")
        if not result:
            self._max_send_message(chat_id, "Не удалось получить статус.")
            return

        restrictions = result.get("restrictions", [])
        if not restrictions:
            self._max_send_message(chat_id, "Нет активных ограничений.")
            return

        lines = ["Активные ограничения:\n"]
        for r in restrictions:
            expires = r.get("expires_at", "Бессрочно")
            if expires and expires != "Бессрочно":
                # Укорачиваем для читаемости
                expires = expires[:19].replace("T", " ")
            lines.append(
                f"#{r['id']} | {r['category']} | {r['target']} | до: {expires}"
            )
        lines.append(f"\nВсего: {len(restrictions)}")
        self._max_send_message(chat_id, "\n".join(lines))

    def _cmd_history(self, chat_id: str, max_user_id: str, args: list) -> None:
        """История действий."""
        limit = 20
        result = self._server_request("GET", f"/api/restrictions/history?limit={limit}")
        if not result:
            self._max_send_message(chat_id, "Не удалось получить историю.")
            return

        entries = result.get("history", [])
        if not entries:
            self._max_send_message(chat_id, "История пуста.")
            return

        lines = ["Последние действия:\n"]
        for e in entries[-15:]:  # Показываем последние 15
            ts = e.get("created_at", "")[:19].replace("T", " ") if e.get("created_at") else "?"
            lines.append(
                f"[{ts}] {e.get('action', '?')} → {e.get('target', '')} "
                f"(от пользователя #{e.get('user_id', '?')})"
            )
        self._max_send_message(chat_id, "\n".join(lines))

    def _cmd_schedules(self, chat_id: str, max_user_id: str, args: list) -> None:
        """Список расписаний."""
        result = self._server_request("GET", "/api/schedules")
        if not result:
            self._max_send_message(chat_id, "Не удалось получить расписания.")
            return

        schedules = result.get("schedules", [])
        if not schedules:
            self._max_send_message(chat_id, "Нет активных расписаний.")
            return

        type_names = {"sleep": "Сон", "rest": "Отдых"}
        lines = ["Расписания:\n"]
        for s in schedules:
            lines.append(
                f"#{s['id']} | {type_names.get(s['schedule_type'], s['schedule_type'])} "
                f"| {s['start_time']}—{s['end_time']} | дни: {s.get('days', [])}"
            )
        self._max_send_message(chat_id, "\n".join(lines))

    def _cmd_users(self, chat_id: str, max_user_id: str, args: list) -> None:
        """Список пользователей."""
        result = self._server_request("GET", "/api/users")
        if not result:
            self._max_send_message(chat_id, "Не удалось получить список.")
            return

        users = result.get("users", [])
        role_names = {"admin": "Админ", "family": "Семья", "child": "Ребёнок"}
        lines = ["Семья:\n"]
        for u in users:
            lines.append(
                f"#{u['id']} | {u.get('display_name', '?')} | "
                f"{role_names.get(u.get('role', '?'), u.get('role', '?'))} | "
                f"МаксID: {u.get('max_user_id', '?')}"
            )
        self._max_send_message(chat_id, "\n".join(lines))

    def _cmd_add_family(self, chat_id: str, max_user_id: str, args: list) -> None:
        """Добавить члена семьи."""
        user, role = self._get_user_role(max_user_id)
        if not user:
            return
        if not PermissionChecker.can(role, "add_family"):
            self._max_send_message(chat_id, "Только администратор может добавлять пользователей.")
            return

        if len(args) < 2:
            self._max_send_message(chat_id, "Формат: /addfamily <max_id> <имя> [роль]")
            return

        target_max_id = args[0]
        display_name = args[1]
        target_role = args[2] if len(args) > 2 else "family"

        if target_role not in PermissionChecker.get_all_roles():
            self._max_send_message(chat_id, f"Роль: {', '.join(PermissionChecker.get_all_roles())}")
            return

        result = self._server_request("POST", "/api/auth/register", {
            "max_user_id": target_max_id,
            "display_name": display_name,
            "role": target_role,
        })

        if result and result.get("ok"):
            added = result.get("user", {})
            self._max_send_message(chat_id,
                f"Пользователь добавлен:\n"
                f"ID: {added.get('id')}\n"
                f"Имя: {added.get('display_name')}\n"
                f"Роль: {added.get('role')}"
            )
        else:
            self._max_send_message(chat_id, "Не удалось добавить пользователя.")

    def _cmd_remove(self, chat_id: str, max_user_id: str, args: list) -> None:
        """Удалить расписание или ограничение."""
        user, role = self._get_user_role(max_user_id)
        if not user:
            return

        if not args:
            self._max_send_message(chat_id, "Формат: /remove <id>")
            return

        try:
            target_id = int(args[0])
        except ValueError:
            self._max_send_message(chat_id, "ID должен быть числом.")
            return

        # Пробуем удалить как ограничение
        result = self._server_request_with_auth(
            "DELETE", f"/api/restrictions/{target_id}", user["id"]
        )
        if result and result.get("ok"):
            self._max_send_message(chat_id, f"Ограничение #{target_id} удалено.")
            return

        # Пробуем удалить как расписание
        result = self._server_request_with_auth(
            "DELETE", f"/api/schedules/{target_id}", user["id"]
        )
        if result and result.get("ok"):
            self._max_send_message(chat_id, f"Расписание #{target_id} удалено.")
            return

        self._max_send_message(chat_id, f"Не удалось удалить #{target_id}. Проверьте ID и права.")

    # ── Вспомогательный метод с авторизацией ──────────────────

    def _server_request_with_auth(self, method: str, path: str,
                                   user_id: int, json_data: dict = None) -> Optional[dict]:
        """HTTP-запрос к серверу с заголовком X-User-ID."""
        try:
            server_url = self.config.agent_server_url
            resp = httpx.request(
                method,
                f"{server_url}{path}",
                json=json_data,
                headers={"X-User-ID": str(user_id)},
                timeout=5,
            )
            if resp.status_code in (200, 201):
                return resp.json()
            logger.warning("Сервер вернул %d: %s", resp.status_code, resp.text)
            return None
        except Exception as e:
            logger.error("Ошибка запроса к серверу: %s", e)
            return None


# ── Экспорт ───────────────────────────────────────────────────

def run_bot(db=None):
    """Точка входа для бота Макс."""
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(name)s] %(levelname)s: %(message)s",
    )
    bot = MaxBot(db=db)
    try:
        bot.start()
    except KeyboardInterrupt:
        bot.stop()