Третий стрим

app/callbacks/callback_docker.py

@@ -0,0 +1,10 @@
+from aiogram.filters.callback_data import CallbackData
+
+
+class ContainerCallback(CallbackData, prefix="container"):
+    name: str
+
+
+class ActionCallback(CallbackData, prefix="action"):
+    name: str
+    action: str

app/callbacks/callback_favorites.py

@@ -0,0 +1,13 @@
+from aiogram.filters.callback_data import CallbackData
+
+
+class FavoriteCallback(CallbackData, prefix="favorite"):
+    user_id: int
+
+
+class AddFavoriteCallback(FavoriteCallback, prefix="add"):
+    pass
+
+
+class DelFavoriteCallback(FavoriteCallback, prefix="del"):
+    pass

app/handlers/docker_commands.py

@@ -2,47 +2,42 @@ import subprocess
 
 from aiogram.types import Message, CallbackQuery
 
+from app import views
+from app.callbacks.callback_docker import ContainerCallback, ActionCallback
 from app.keyboards.docker_keyboards import container_names_keyboard, container_actions_keyboard
+from app.utils.text_splitter import split_text
 
 
 async def get_containers(message: Message):
-    sub = subprocess.check_output("docker ps -a").decode()
+    sub = subprocess.check_output("docker ps -a", shell=True).decode()
+    messages = split_text(sub)
+
+    for m in messages:
+        await message.answer(
+            text=f"<pre><code>{m}</code></pre>",
+            parse_mode="HTML",
+        )
+
     keyboard = container_names_keyboard(sub)
-    await message.answer(
-        text=f"<pre><code>{sub}</code></pre>",
-        parse_mode="HTML",
-        reply_markup=keyboard.as_markup()
-    )
+    await message.answer(text="Выберите контейнер:", reply_markup=keyboard.as_markup())
 
 
-async def container_actions(call: CallbackQuery):
-    name = call.data.split("_")[-1]
+async def container_actions(callback: CallbackQuery, callback_data: ContainerCallback):
+    name = callback_data.name
     keyboard = container_actions_keyboard(name)
-    await call.message.answer(
+    await callback.message.answer(
         text=f"Выберите действие для контейнера {name}",
         parse_mode="HTML",
         reply_markup=keyboard.as_markup()
     )
 
 
-async def do_container_action(call: CallbackQuery):
-    _, action, name = call.data.split("_")
-    match action:
-        case "start":
-            subprocess.run(f"docker start {name}")
-            message = f"Контейнер {name} успешно запущен"
-        case "stop":
-            subprocess.run(f"docker stop {name}")
-            message = f"Контейнер {name} успешно остановлен"
-        case "restart":
-            subprocess.run(f"docker restart {name}")
-            message = f"Контейнер {name} успешно перезапущен"
-        case "delete":
-            subprocess.run(f"docker rm -f {name}")
-            message = f"Контейнер {name} успешно удалён"
-        case _:
-            message = f"Произошла необъяснимая ошибка"
+async def do_container_action(callback: CallbackQuery, callback_data: ActionCallback):
+    action, name = callback_data.name, callback_data.action
 
-    await call.message.answer(
+    subprocess.run(f"docker {action} {name}", shell=True)
+    message = views.actions.get(action).format(name)
+
+    await callback.message.answer(
         text=message,
     )

app/handlers/events.py

@@ -1,12 +1,12 @@
-from app.settings import bot, Secrets
+from app.settings import bot, secrets
 from app.utils.commands import set_commands
 from app import views
 
 
 async def start_bot():
     await set_commands(bot)
-    await bot.send_message(Secrets.admin_id, views.start_bot_message())
+    await bot.send_message(secrets.admin_id, views.start_bot_message())
 
 
 async def stop_bot():
-    await bot.send_message(Secrets.admin_id, views.stop_bot_message())
+    await bot.send_message(secrets.admin_id, views.stop_bot_message())

app/handlers/favorites.py

@@ -0,0 +1,60 @@
+from aiogram.fsm.context import FSMContext
+from aiogram.types import Message, CallbackQuery
+
+from app.callbacks.callback_favorites import AddFavoriteCallback, DelFavoriteCallback
+from app.keyboards.favorites_keyboards import favorite_list, add_favorite_inline, add_del_favorite_inline
+from app.utils.db_actions import get_user, create_user, get_favorites, add_favorite, del_favorite
+from app.utils.statesform import FavoritesCommandsSteps
+
+
+async def favorites_command(message: Message):
+    user = await get_user(message.from_user.id)
+    if not user:
+        user = await create_user(message.from_user.id)
+
+    favorites = await get_favorites(user.id)
+
+    if not favorites:
+        inline_keyboard = await add_favorite_inline(user)
+        await message.answer("Нет команд в избранном.", reply_markup=inline_keyboard)
+    else:
+        inline_keyboard = await add_del_favorite_inline(user)
+        await message.answer("Выберите действие:", reply_markup=inline_keyboard)
+        keyboard = await favorite_list(favorites)
+        await message.answer("Выберите команду из списка ниже:", reply_markup=keyboard)
+
+
+async def add_favorite_callback(callback: CallbackQuery, callback_data: AddFavoriteCallback, state: FSMContext):
+    await callback.message.answer("Введите команду.")
+    await state.set_data({"user_id": callback_data.user_id})
+    await state.set_state(FavoritesCommandsSteps.ADD)
+
+
+async def add_favorite_state(message: Message, state: FSMContext):
+    data = await state.get_data()
+    new_command = await add_favorite(user_id=data.get("user_id"), command=message.text)
+    await message.answer(f"Добавлена команда: {new_command.command}")
+    await state.clear()
+
+
+async def del_favorite_callback(callback: CallbackQuery, callback_data: DelFavoriteCallback, state: FSMContext):
+    favorites = await get_favorites(callback_data.user_id)
+    favorites_text = "\n".join([f"{i}: {favorite.command}" for i, favorite in enumerate(favorites)])
+    await callback.message.answer(f"Введите номер команды для удаления:\n{favorites_text}")
+    await state.set_data({"favorites": favorites})
+    await state.set_state(FavoritesCommandsSteps.DEL)
+
+
+async def del_favorite_state(message: Message, state: FSMContext):
+    data = await state.get_data()
+
+    try:
+        choice = int(message.text)
+    except ValueError:
+        await message.answer("Введите число!")
+        await state.set_state(FavoritesCommandsSteps.DEL)
+    else:
+        favorite = data.get("favorites")[choice]
+        await del_favorite(favorite.id)
+        await message.answer("Команда успешно удалена.")
+        await state.clear()

app/handlers/multiply_commands.py

@@ -6,6 +6,7 @@ from aiogram.types import Message
 
 from app import views
 from app.utils.statesform import ExecuteCommandsSteps
+from app.utils.text_splitter import split_text
 
 
 async def multiply_commands(message: Message, state: FSMContext):
@@ -34,6 +35,9 @@ async def execute_commands(message: Message, state: FSMContext):
             except subprocess.CalledProcessError as e:
                 res = views.subprocess_error(e)
 
-        await message.answer(views.user_command(res), parse_mode="HTML")
+        messages = split_text(res)
+
+        for m in messages:
+            await message.answer(views.user_command(m), parse_mode="HTML")
 
         await state.set_state(ExecuteCommandsSteps.EXECUTE)

app/handlers/single_command.py

@@ -3,10 +3,15 @@ import subprocess
 from aiogram.types import Message
 
 from app import views
+from app.utils.text_splitter import split_text
 
 
 async def execute_command(message: Message):
     user_command = message.text.split()[1:]
     sub = subprocess.check_output(user_command, shell=True)
     res = sub.decode().replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
-    await message.answer(views.user_command(res), parse_mode="HTML")
+
+    messages = split_text(res)
+
+    for m in messages:
+        await message.answer(views.user_command(m), parse_mode="HTML")

app/keyboards/docker_keyboards.py

@@ -1,16 +1,17 @@
 from aiogram.types import InlineKeyboardButton
 from aiogram.utils.keyboard import InlineKeyboardBuilder
 
+from app.callbacks.callback_docker import ContainerCallback, ActionCallback
+
 
 def container_names_keyboard(stdout: str):
     container_names = [line.split("   ")[-1].strip() for line in stdout.splitlines()[1:]]
     builder = InlineKeyboardBuilder()
     for name in container_names:
-        data = f"container_{name}"
         builder.add(
             InlineKeyboardButton(
                 text=name,
-                callback_data=data,
+                callback_data=ContainerCallback(name=name).pack(),
             )
         )
     builder.adjust(1)
@@ -23,19 +24,19 @@ def container_actions_keyboard(name: str):
 
     builder.add(InlineKeyboardButton(
         text="Запустить контейнер",
-        callback_data=f"action_start_{name}",
+        callback_data=ActionCallback(name=name, action="start").pack(),
     ))
     builder.add(InlineKeyboardButton(
         text="Остановить контейнер",
-        callback_data=f"action_stop_{name}",
+        callback_data=ActionCallback(name=name, action="stop").pack(),
     ))
     builder.add(InlineKeyboardButton(
         text="Перезапустить контейнер",
-        callback_data=f"action_restart_{name}",
+        callback_data=ActionCallback(name=name, action="restart").pack(),
     ))
     builder.add(InlineKeyboardButton(
         text="Удалить контейнер",
-        callback_data=f"action_delete_{name}",
+        callback_data=ActionCallback(name=name, action="delete").pack(),
     ))
     builder.adjust(1)
 

app/keyboards/favorites_keyboards.py

@@ -0,0 +1,52 @@
+from aiogram.types import KeyboardButton, InlineKeyboardButton
+from aiogram.utils.keyboard import ReplyKeyboardBuilder, InlineKeyboardBuilder
+
+from app.callbacks.callback_favorites import AddFavoriteCallback, DelFavoriteCallback
+from app.schemas.favorites_schema import FavoritesSchemaOutput
+from app.schemas.user_schema import UserSchemaOutput
+
+
+async def favorite_list(favorites: list[FavoritesSchemaOutput]):
+    builder = ReplyKeyboardBuilder()
+
+    for favorite in favorites:
+        builder.add(
+            KeyboardButton(text=f"/command {favorite.command}")
+        )
+
+    builder.adjust(2)
+    return builder.as_markup(resize_keyboard=True, one_time_keyboard=True)
+
+
+async def add_favorite_inline(user: UserSchemaOutput):
+    builder = InlineKeyboardBuilder()
+
+    builder.add(
+        InlineKeyboardButton(
+            text="Добавить команду",
+            callback_data=AddFavoriteCallback(user_id=user.id).pack()
+        )
+    )
+
+    builder.adjust(1)
+    return builder.as_markup()
+
+
+async def add_del_favorite_inline(user: UserSchemaOutput):
+    builder = InlineKeyboardBuilder()
+
+    builder.add(
+        InlineKeyboardButton(
+            text="Добавить команду",
+            callback_data=AddFavoriteCallback(user_id=user.id).pack()
+        )
+    )
+    builder.add(
+        InlineKeyboardButton(
+            text="Удалить команду",
+            callback_data=DelFavoriteCallback(user_id=user.id).pack()
+        )
+    )
+
+    builder.adjust(1)
+    return builder.as_markup()

app/middlewares/admin_middleware.py

@@ -0,0 +1,18 @@
+from typing import Callable, Dict, Any, Awaitable
+
+from aiogram import BaseMiddleware
+from aiogram.types import TelegramObject, User
+
+from app.settings import secrets
+
+
+class AdminMiddleware(BaseMiddleware):
+    async def __call__(
+            self,
+            handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
+            event: TelegramObject,
+            data: Dict[str, Any]
+    ) -> Any:
+        user: User = data.get("event_from_user")
+        if user.id == secrets.admin_id:
+            return await handler(event, data)

app/models/__init__.py

@@ -0,0 +1,9 @@
+from .base import Base
+from .user import User
+from .favorites import Favorites
+
+__all__ = [
+    "Base",
+    "User",
+    "Favorites",
+]

app/models/base.py

@@ -0,0 +1,7 @@
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+
+
+class Base(DeclarativeBase):
+    __abstract__ = True
+
+    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)

app/models/favorites.py

@@ -0,0 +1,17 @@
+from typing import TYPE_CHECKING
+
+from sqlalchemy import String, ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from app.models.base import Base
+
+if TYPE_CHECKING:
+    from .user import User
+
+
+class Favorites(Base):
+    __tablename__ = "favorites"
+
+    command: Mapped[str] = mapped_column(String(200), nullable=False)
+    user_id: Mapped[int] = mapped_column(ForeignKey("user.id"))
+    user: Mapped["User"] = relationship("User", back_populates="favorites")

app/models/user.py

@@ -0,0 +1,18 @@
+from datetime import datetime
+from typing import TYPE_CHECKING
+
+from sqlalchemy import BigInteger, func
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from app.models.base import Base
+
+if TYPE_CHECKING:
+    from .favorites import Favorites
+
+
+class User(Base):
+    __tablename__ = "user"
+
+    telegram_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
+    added_at: Mapped[datetime] = mapped_column(server_default=func.now(), default=datetime.now)
+    favorites: Mapped["Favorites"] = relationship("Favorites", back_populates="user")

app/schemas/favorites_schema.py

@@ -0,0 +1,10 @@
+from pydantic import BaseModel
+
+
+class FavoritesSchemaInput(BaseModel):
+    command: str
+    user_id: int
+
+
+class FavoritesSchemaOutput(FavoritesSchemaInput):
+    id: int

app/schemas/user_schema.py

@@ -0,0 +1,12 @@
+from datetime import datetime
+
+from pydantic import BaseModel
+
+
+class UserSchemaInput(BaseModel):
+    telegram_id: int
+
+
+class UserSchemaOutput(UserSchemaInput):
+    id: int
+    added_at: datetime

app/settings.py

@@ -1,16 +1,25 @@
 import os
-from dataclasses import dataclass
+from typing import Union
 
 from aiogram import Bot
-from dotenv import load_dotenv
-
-load_dotenv()
+from pydantic import SecretStr
+from pydantic_settings import BaseSettings
+from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
 
 
-@dataclass
-class Secrets:
-    token: str = os.environ.get("token")
-    admin_id: int = os.environ.get("admin_id")
+class Secrets(BaseSettings):
+    token: SecretStr
+    admin_id: Union[SecretStr.get_secret_value, int]
+    db_url: str = "sqlite+aiosqlite:///db.sqlite3"
+
+    class Config:
+        env_file = ".env"
+        env_file_encoding = "utf-8"
 
 
-bot = Bot(token=Secrets.token)
+secrets = Secrets()
+
+engine = create_async_engine(url=secrets.db_url)
+sessionmaker = async_sessionmaker(engine, expire_on_commit=False, autocommit=False)
+
+bot = Bot(token=secrets.token.get_secret_value())

app/utils/db_actions.py

@@ -0,0 +1,50 @@
+from sqlalchemy import select, insert, delete
+
+from app.models import User, Favorites
+from app.schemas.favorites_schema import FavoritesSchemaOutput
+from app.schemas.user_schema import UserSchemaOutput
+from app.settings import sessionmaker
+
+
+async def get_user(telegram_id: int):
+    async with sessionmaker() as session:
+        query = select(User).where(User.telegram_id == telegram_id)
+        result = await session.execute(query)
+        user = result.scalar_one_or_none()
+        if user:
+            return UserSchemaOutput(**user.__dict__)
+        return user
+
+
+async def create_user(telegram_id: int):
+    async with sessionmaker() as session:
+        query = insert(User).values(telegram_id=telegram_id).returning(User.id, User.telegram_id, User.added_at)
+        result = await session.execute(query)
+        await session.commit()
+        user = result.mappings().first()
+        return UserSchemaOutput(**user)
+
+
+async def get_favorites(user_id: int):
+    async with sessionmaker() as session:
+        query = select(Favorites).where(Favorites.user_id == user_id)
+        result = await session.execute(query)
+        favorites = result.scalars().all()
+        return [FavoritesSchemaOutput(**favorite.__dict__) for favorite in favorites]
+
+
+async def add_favorite(user_id: int, command: str):
+    async with sessionmaker() as session:
+        query = insert(Favorites).values(user_id=user_id, command=command).returning(Favorites.id, Favorites.user_id,
+                                                                                     Favorites.command)
+        result = await session.execute(query)
+        await session.commit()
+        favorite = result.mappings().first()
+        return FavoritesSchemaOutput(**favorite)
+
+
+async def del_favorite(favorite_id: int):
+    async with sessionmaker() as session:
+        query = delete(Favorites).where(Favorites.id == favorite_id)
+        await session.execute(query)
+        await session.commit()

app/utils/statesform.py

@@ -3,3 +3,8 @@ from aiogram.fsm.state import StatesGroup, State
 
 class ExecuteCommandsSteps(StatesGroup):
     EXECUTE = State()
+
+
+class FavoritesCommandsSteps(StatesGroup):
+    ADD = State()
+    DEL = State()

app/utils/text_splitter.py

@@ -0,0 +1,20 @@
+import re
+
+
+def split_text(message: str):
+    messages = []
+    lines = re.split(r"\n", message)
+
+    temp = ""
+
+    for line in lines:
+        if len(temp) + len(line) < 4096:
+            temp += line + "\n"
+        else:
+            messages.append(temp)
+            temp = line + "\n"
+
+    if temp:
+        messages.append(temp)
+
+    return messages

app/views.py

@@ -5,6 +5,12 @@ menu = {
     "docker_list": "Список Docker-контейнеров",
 }
 
+actions = {
+    "start": "Контейнер {} успешно запущен",
+    "stop": "Контейнер {} успешно остановлен",
+    "restart": "Контейнер {} успешно перезапущен",
+    "delete": "Контейнер {} успешно удалён",
+}
 
 def start_bot_message():
     return "Бот запущен"