Добавлены основные файлы проекта: конфигурация, обработчики команд, мидлвары и утилиты для бота на Python.

.env.template

@@ -0,0 +1,2 @@
+TOKEN=
+USER_ID=

.gitignore

@@ -174,3 +174,4 @@ cython_debug/
 # PyPI configuration file
 .pypirc
 
+uv.lock

.python-version

@@ -0,0 +1 @@
+3.12

pyproject.toml

@@ -0,0 +1,46 @@
+[project]
+name = "server-bot-python"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+    "aiogram>=3.27.0",
+    "psutil>=7.2.2",
+    "pydantic-settings>=2.14.0",
+    "python-dotenv>=1.2.2",
+    "types-psutil>=7.2.2.20260408",
+]
+
+[dependency-groups]
+dev = [
+    "deptry>=0.25.1",
+    "mypy>=1.20.1",
+    "ruff>=0.15.11",
+]
+
+
+[tool.ruff]
+line-length = 160
+target-version = "py313"
+
+[tool.ruff.format]
+quote-style = "single"
+indent-style = "space"
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "UP", "B", "Q", "ERA", "D", "SIM", "ARG", "RUF", "C4", "RET", "ASYNC", "PERF", "PL"]
+ignore = ["D100", "RUF002", "D107", "RUF001", "D104", "D203", "D213", "ARG001"]
+
+[tool.ruff.lint.flake8-quotes]
+inline-quotes = "single"
+multiline-quotes = "double"
+docstring-quotes = "double"
+avoid-escape = true
+
+[tool.mypy]
+python_version = "3.13"
+strict = true
+warn_unused_ignores = true
+warn_redundant_casts = true
+warn_unreachable = true

requirements.txt

@@ -0,0 +1,3 @@
+python-dotenv
+psutil
+aiogram

src/config.py

@@ -0,0 +1,21 @@
+from pydantic import Field, ValidationError
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    """App settings loaded from environment."""
+
+    TOKEN: str = Field(default='default_value', description='Токен Telegram-бота')
+    USER_ID: int = Field(default=0, description='ID пользователя для доступа')
+
+    model_config = SettingsConfigDict(env_file='.env', env_file_encoding='utf-8')
+
+
+try:
+    settings = Settings()
+except ValidationError as exc:
+    msg = f'Ошибка валидации настроек: {exc}'
+    raise RuntimeError(msg) from exc
+
+TOKEN: str = settings.TOKEN
+USER_ID: str = str(settings.USER_ID)

src/handlers/commands.py

@@ -0,0 +1,96 @@
+import logging
+import os
+
+import psutil
+from aiogram import Router
+from aiogram.enums import ParseMode
+from aiogram.filters import Command
+from aiogram.types import Message
+
+from src.config import USER_ID
+from src.utils.system import execute_command, get_cpu_temperature
+
+router = Router()
+
+current_directory = os.getcwd()
+
+
+@router.message(Command('start'))
+async def cmd_start(message: Message) -> None:
+    """Команда /start - приветствие и краткая инструкция."""
+    await message.answer('Привет! Я бот для мониторинга сервера. Используй /help для списка команд.')
+
+
+@router.message(Command('help'))
+async def cmd_help(message: Message) -> None:
+    """Команда /help - показать список доступных команд и их описание."""
+    help_text = (
+        '📋 <b>Список команд:</b>\n'
+        '/start - Начать работу с ботом\n'
+        '/help - Показать это сообщение\n'
+        '/status - Показать текущее состояние сервера\n'
+        '/c &lt;команда&gt; - Выполнить команду на сервере\n'
+        '/sysinfo - Показать информацию о системе'
+    )
+    await message.answer(help_text, parse_mode=ParseMode.HTML)
+
+
+@router.message(Command('status'))
+async def cmd_status(message: Message) -> None:
+    """Выводит текущую загрузку CPU, RAM, диска и температуру CPU."""
+    cpu_usage = psutil.cpu_percent(interval=1)
+    ram_usage = psutil.virtual_memory().percent
+    disk_usage = psutil.disk_usage('/').percent
+    cpu_temp = get_cpu_temperature()
+
+    status_message = f'📊 <b>Состояние сервера:</b>\n• <b>CPU:</b> {cpu_usage}%\n• <b>RAM:</b> {ram_usage}%\n• <b>Диск:</b> {disk_usage}%\n'
+    if cpu_temp is not None:
+        status_message += f'• <b>Температура CPU:</b> {cpu_temp}°C\n'
+    else:
+        status_message += '• <b>Температура CPU:</b> N/A\n'
+    await message.answer(status_message, parse_mode=ParseMode.HTML)
+
+
+@router.message(Command('c'))
+async def cmd_execute(message: Message) -> None:
+    """Команда /c &lt;команда&gt; - выполняет указанную команду на сервере и возвращает результат."""
+    global current_directory
+    user_id = message.from_user.id if message.from_user else 'Unknown'
+    if str(user_id) != str(USER_ID):
+        await message.answer('❌ У вас нет прав для использования этого бота.')
+        return
+
+    if not message.text:
+        await message.answer('❌ Укажите команду для выполнения.')
+        return
+    command = message.text.split(maxsplit=1)[1] if len(message.text.split()) > 1 else None
+    if not command:
+        await message.answer('❌ Укажите команду для выполнения.')
+        return
+
+    logging.info(f'Пользователь {user_id} выполнил команду: {command}')
+    result, current_directory = execute_command(command, current_directory)
+    await message.answer(f'✅ Результат выполнения команды:\n<pre>{result}</pre>', parse_mode=ParseMode.HTML)
+
+
+@router.message(Command('sysinfo'))
+async def cmd_sysinfo(message: Message) -> None:
+    """Sysinfo - Показать информацию о системе, включая имя хоста, ОС и количество ядер CPU."""
+    """Команда /sysinfo - выводит информацию о системе, включая имя хоста, ОС и количество ядер CPU."""
+    user_id = message.from_user.id if message.from_user else 'Unknown'
+    if str(user_id) != str(USER_ID):
+        await message.answer('❌ У вас нет прав для использования этого бота.')
+        return
+    logging.info(f'Пользователь {user_id} выполнил команду /sysinfo')
+    try:
+        hostname = os.uname().nodename if hasattr(os, 'uname') else os.getenv('COMPUTERNAME', 'N/A')
+        os_info = os.uname().sysname + ' ' + os.uname().release if hasattr(os, 'uname') else os.name
+        cpu_count = os.cpu_count()
+
+        sysinfo_message = (
+            f'📊 <b>Информация о системе:</b>\n• <b>Имя хоста:</b> {hostname}\n• <b>ОС:</b> {os_info}\n• <b>Количество ядер CPU:</b> {cpu_count}\n'
+        )
+
+        await message.answer(sysinfo_message, parse_mode=ParseMode.HTML)
+    except Exception as e:
+        await message.answer(f'❌ Ошибка:\n<pre>{e!s}</pre>', parse_mode=ParseMode.HTML)

src/main.py

@@ -0,0 +1,35 @@
+import logging
+
+from aiogram import Bot, Dispatcher
+from aiogram.enums import ParseMode
+
+from src.config import settings
+from src.handlers import commands
+from src.middlewares.access import AccessMiddleware
+
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(levelname)s - %(message)s',
+    handlers=[logging.FileHandler('bot.log'), logging.StreamHandler()],
+)
+
+
+def setup_routers(dp: Dispatcher) -> None:
+    """Регистрирует роутеры и мидлвары в диспетчере."""
+    dp.message.middleware(AccessMiddleware())
+    dp.include_router(commands.router)
+
+
+async def main() -> None:
+    """Создает экземпляр бота, регистрирует роутеры и запускает поллинг."""
+    bot = Bot(token=settings.TOKEN, parse_mode=ParseMode.HTML)
+    dp = Dispatcher()
+    setup_routers(dp)
+    logging.info('Бот запущен.')
+    await dp.start_polling(bot)
+
+
+if __name__ == '__main__':
+    import asyncio
+
+    asyncio.run(main())

src/middlewares/access.py

@@ -0,0 +1,28 @@
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+from aiogram.dispatcher.middlewares.base import BaseMiddleware
+from aiogram.types import Message, TelegramObject
+
+from src.config import USER_ID
+
+
+class AccessMiddleware(BaseMiddleware):
+    """Мидлвар для проверки доступа по USER_ID. Если ID пользователя не совпадает, отправляет сообщение об ошибке и не пропускает дальше."""
+
+    async def __call__(
+        self,
+        handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]],
+        event: TelegramObject,
+        data: dict[str, Any],
+    ) -> Any:
+        """Проверяет ID пользователя только для Message."""
+        if isinstance(event, Message):
+            if event.from_user is None:
+                return None
+
+            if str(event.from_user.id) != USER_ID:
+                await event.answer('❌ У вас нет прав для использования этого бота.')
+                return None
+
+        return await handler(event, data)

src/middlewares/logging.py

@@ -0,0 +1,20 @@
+import logging
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+from aiogram.dispatcher.middlewares.base import BaseMiddleware
+from aiogram.types import TelegramObject
+
+
+class LoggingMiddleware(BaseMiddleware):
+    """Мидлвар для логирования всех входящих обновлений. Логирует тип и содержимое обновления. Логи сохраняются в файл bot.log и выводятся в консоль."""
+
+    async def __call__(
+        self,
+        handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]],
+        event: TelegramObject,
+        data: dict[str, Any],
+    ) -> Any:
+        """Выполняет логирование входящего обновления и передает его дальше по цепочке обработки."""
+        logging.info('Update: %s', event)
+        return await handler(event, data)

src/utils/system.py

@@ -0,0 +1,38 @@
+import logging
+import os
+import subprocess
+from typing import Any
+
+import psutil
+
+
+def get_cpu_temperature() -> Any:
+    """Выполняет попытку получения температуры CPU. Сначала через psutil, затем через чтение из системного файла."""
+    try:
+        if hasattr(psutil, 'sensors_temperatures'):
+            temps = psutil.sensors_temperatures()
+            if 'coretemp' in temps:
+                return temps['coretemp'][0].current
+        with open('/sys/class/thermal/thermal_zone0/temp') as f:
+            return int(f.read()) / 1000
+    except Exception as e:
+        logging.error(f'Не удалось получить температуру CPU: {e}')
+    return None
+
+
+def execute_command(command: str, current_directory: str) -> tuple[str, str]:
+    """Выполняет команду в текущей директории. Если команда начинается с "cd", меняет директорию. Возвращает результат выполнения и текущую директорию."""
+    try:
+        if command.startswith('cd '):
+            new_dir = command.split(' ', 1)[1].strip()
+            try:
+                os.chdir(new_dir)
+                return f'Текущая директория изменена на: {os.getcwd()}', os.getcwd()
+            except Exception as e:
+                return f'Ошибка при смене директории: {e}', current_directory
+        else:
+            result = subprocess.run(command, shell=True, capture_output=True, text=True, cwd=current_directory, check=True)
+            return result.stdout or result.stderr, current_directory
+    except Exception as e:
+        logging.error(f'Ошибка при выполнении команды: {e}')
+        return f'Ошибка: {e}', current_directory