• XSS.stack #1 – первый литературный журнал от юзеров форума

Как сделать админ панель для бота?

Отслеживать
asyncTraffic сказал(а):
брать sqlite3 для тг бота любого уровня уже не совсем акутально, рано или поздно все кто sqlite3 юзают при масштабировании проекта или написании какого то сложно, как говорится, "в*ебываются рогом" в ошибку database locked) лучше какой нибудь postgresql использовать надежнее, либа asyncpg подходит замечательно
Поддержка многопоточности в SQLite3 была добавлена в версии 3.3.1, выпущенной в начале 2006 года
 
Пожалуйста, обратите внимание, что пользователь заблокирован
Насчёт бд: советую MariaDB.
Насчёт либы: aiogram - самое простое, быстрое и универсальное что я видел.

Насчёт того как сделать веб интерфейс:
Варианты работы:
1. Бот + интерфейс работают с 1 бд
2. Бот имеет хуки или простое апи для выполнения действий/выдачи нужной информации, которые будут вызываться в бэкенде.

В случае выбора 1 варианта: веб делать либо отдельно, либо сразу в проекте бота (я выбираю всегда этот вариант, если это не полностью отдельное большое веб приложение с фронтом, бэком и огромным функционалом, а что-то базовое до 2 тысяч строк кода).
В случае выбора 2 варианта: тут легко, просто запросы в бд выполняет сам бот и выполняет прописанную тобой логику (допустим оповещение о действии), смотря что тебе нужно.

И пожалуйста, не забудь сделать валидацию данных дабы твоего бота не взломали. Буквально 90% ботов работают без какой либо валидации данных, а тем более без защит в их веб приложениях, ведь создатели даже не подозревают насколько легко взломать и выкачать фулл код бота через него, если проект не имеет должной логики для создания защиты, либо не написан на Фреймворке в котором эти уязвимости уже закрыты.
 
Пожалуйста, обратите внимание, что пользователь заблокирован
cassary сказал(а):
Пишу на aiogram 3
Вот написанный бот с помощью моей модели, обученной на документации aiogram 3.18.0:

Python: 
import asyncio
import logging
import datetime
import hashlib
import html
import secrets
import aiomysql
from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command
from aiogram.utils.keyboard import InlineKeyboardBuilder
from aiohttp import web

# -------------------- Конфигурация --------------------
TOKEN = "ВАШ_ТОКЕН"  # Укажите ваш токен бота
DB_CONFIG = {
    "host": "localhost",
    "user": "root",
    "password": "password",
    "db": "bot_db"
}
ADMINS = [123456789]  # Список Telegram ID администраторов
ADMIN_PASSWORD_HASH = hashlib.sha256("securepass".encode()).hexdigest()  # Хешированный пароль
LOG_FILE = "bot_logs.txt"

bot = Bot(token=TOKEN, parse_mode="HTML")
dp = Dispatcher()

# Хранилище для "сессий" админов в веб-интерфейсе
# Можно заменить на более надёжную БД или Redis, если нужно
ACTIVE_SESSIONS = set()

# -------------------- Инициализация БД --------------------
async def init_db():
    """
    Подключение к БД и создание таблицы users, если её нет
    """
    conn = await aiomysql.connect(**DB_CONFIG)
    async with conn.cursor() as cur:
        await cur.execute("""
        CREATE TABLE IF NOT EXISTS users (
            user_id BIGINT PRIMARY KEY,
            banned BOOLEAN DEFAULT FALSE
        )
        """)
    await conn.commit()
    conn.close()

# -------------------- Логирование --------------------
def write_log(action: str, user_id: int, admin_id: str):
    """
    Запись действия (бан/разбан и т.д.) в текстовый файл LOG_FILE.
    action: строка ("Бан"/"Разбан" и т.д.)
    user_id: ID пользователя, над которым совершается действие
    admin_id: "WEB-Admin" или Telegram ID админа
    """
    with open(LOG_FILE, "a", encoding="utf-8") as log:
        log.write(f"{datetime.datetime.now()} - {action} пользователя {user_id} админом {admin_id}\n")

# -------------------- Обработчики Telegram --------------------
@dp.message(Command("start"))
async def start_cmd(message: types.Message):
    """
    Обработчик команды /start
    Вставляет пользователя в БД, если его там нет
    """
    user_id = message.from_user.id
    # Проверяем валидность ID (например, если от API пришёл user_id <= 0)
    if user_id <= 0:
        return

    conn = await aiomysql.connect(**DB_CONFIG)
    async with conn.cursor() as cur:
        await cur.execute("INSERT IGNORE INTO users (user_id) VALUES (%s)", (user_id,))
    await conn.commit()
    conn.close()

    await message.answer("Добро пожаловать! Используйте /admin для управления, если у вас есть права админа.")

@dp.message(Command("admin"))
async def admin_panel(message: types.Message):
    """
    Встроенная админ-панель в самом боте
    Показывает кнопки «Список пользователей» и «Логи»
    """
    if message.from_user.id not in ADMINS:
        return await message.answer("У вас нет прав доступа.")

    keyboard = InlineKeyboardBuilder()
    keyboard.button(text="📋 Список пользователей", callback_data="list_users")
    keyboard.button(text="📜 Логи", callback_data="view_logs")
    keyboard.adjust(1)

    await message.answer("Админ-панель:", reply_markup=keyboard.as_markup())

@dp.callback_query(lambda call: call.data == "list_users")
async def show_users(call: types.CallbackQuery):
    """
    Кнопка «Список пользователей» в Telegram
    """
    conn = await aiomysql.connect(**DB_CONFIG)
    async with conn.cursor() as cur:
        await cur.execute("SELECT user_id, banned FROM users")
        users = await cur.fetchall()
    conn.close()

    # Формируем текст со статусом
    text = "\n".join(f"{row[0]} - {'Забанен' if row[1] else 'Активен'}" for row in users)

    # Ограничение Telegram на 4096 символов. Если мало ли слишком много пользователей,
    # придётся разбивать, но для примера оставим так.
    await call.message.edit_text(f"Список пользователей:\n{text}")

@dp.callback_query(lambda call: call.data == "view_logs")
async def show_logs(call: types.CallbackQuery):
    """
    Кнопка «Логи» в Telegram
    Показываем последние 10 строк логов
    """
    try:
        with open(LOG_FILE, "r", encoding="utf-8") as log:
            logs = log.readlines()
    except FileNotFoundError:
        logs = ["Лог-файл пока пуст или не создан."]

    log_text = "".join(logs[-10:])  # последние 10 записей
    await call.message.edit_text(f"Последние логи:\n{log_text}")

# -------------------- Вспомогательные функции веб-интерфейса --------------------
def is_session_valid(request: web.Request) -> bool:
    """
    Проверяет, есть ли у пользователя валидная куки 'admin_session'
    """
    session_token = request.cookies.get("admin_session")
    return session_token in ACTIVE_SESSIONS

def create_session_cookie() -> str:
    """
    Создаёт новую случайную строку для сессии
    """
    return secrets.token_urlsafe(32)

# -------------------- Обработчики веб-сервера --------------------
async def handle_login(request):
    """
    Отображает форму логина (GET) или обрабатывает ввод пароля (POST).
    При успешном вводе пароля устанавливает куки, иначе 403.
    """
    if request.method == "GET":
        # Просто отображаем HTML-страницу
        return web.Response(text="""
        <html>
        <head>
            <title>Админ-панель</title>
            <link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css">
        </head>
        <body class="container mt-5">
            <h2>Вход в админ-панель</h2>
            <form method='post' class="form-group">
                <input type='password' name='password' class="form-control w-25 d-inline" placeholder="Введите пароль" required/>
                <input type='submit' class="btn btn-primary" value='Войти'/>
            </form>
        </body>
        </html>
        """, content_type='text/html')
    else:
        # POST-запрос: пользователь ввёл пароль
        data = await request.post()
        password = data.get("password", "")
        if hashlib.sha256(password.encode()).hexdigest() == ADMIN_PASSWORD_HASH:
            # Успешный вход
            session_token = create_session_cookie()
            ACTIVE_SESSIONS.add(session_token)
            resp = web.HTTPFound("/admin")  # перенаправляем на /admin
            resp.set_cookie("admin_session", session_token, httponly=True, secure=False)  # secure=True для HTTPS
            return resp
        else:
            return web.Response(text="Неверный пароль!", status=403)

async def handle_admin_panel(request):
    """
    Отображает список пользователей после проверки валидной сессии.
    """
    if not is_session_valid(request):
        return web.Response(text="Доступ запрещён! Войдите в систему.", status=403)

    conn = await aiomysql.connect(**DB_CONFIG)
    async with conn.cursor() as cur:
        await cur.execute("SELECT user_id, banned FROM users")
        users = await cur.fetchall()
    conn.close()

    user_rows = ""
    for user in users:
        uid = user[0]
        banned_status = "Забанен" if user[1] else "Активен"
        # HTML-эскейп на всякий случай
        uid_html = html.escape(str(uid))
        banned_html = html.escape(banned_status)
        user_rows += f"""
            <tr>
                <td>{uid_html}</td>
                <td>{banned_html}</td>
                <td>
                    <a href="/ban/{uid_html}" class='btn btn-danger'>🔨 Бан</a>
                    <a href="/unban/{uid_html}" class='btn btn-success'>✅ Разбан</a>
                </td>
            </tr>
        """

    return web.Response(text=f"""
    <html>
    <head>
        <title>Админ-панель</title>
        <link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css">
    </head>
    <body class="container mt-5">
        <h2>Список пользователей</h2>
        <table class="table table-bordered">
            <thead><tr><th>ID</th><th>Статус</th><th>Действия</th></tr></thead>
            <tbody>
                {user_rows}
            </tbody>
        </table>
        <a href="/logs" class="btn btn-secondary">Посмотреть логи</a>
    </body>
    </html>
    """, content_type='text/html')

async def handle_ban(request):
    """
    Бан пользователя (обновление в БД) только при валидной сессии.
    """
    if not is_session_valid(request):
        return web.Response(text="Доступ запрещён! Войдите в систему.", status=403)

    user_id = request.match_info.get('user_id', "")
    if not user_id.isdigit() or int(user_id) <= 0:
        return web.Response(text="Некорректный ID пользователя", status=400)

    uid = int(user_id)
    conn = await aiomysql.connect(**DB_CONFIG)
    async with conn.cursor() as cur:
        await cur.execute("UPDATE users SET banned = TRUE WHERE user_id = %s", (uid,))
    await conn.commit()
    conn.close()

    write_log("Бан", uid, "WEB-Admin")
    return web.Response(text="Пользователь забанен!")

async def handle_unban(request):
    """
    Разбан пользователя (обновление в БД) только при валидной сессии.
    """
    if not is_session_valid(request):
        return web.Response(text="Доступ запрещён! Войдите в систему.", status=403)

    user_id = request.match_info.get('user_id', "")
    if not user_id.isdigit() or int(user_id) <= 0:
        return web.Response(text="Некорректный ID пользователя", status=400)

    uid = int(user_id)
    conn = await aiomysql.connect(**DB_CONFIG)
    async with conn.cursor() as cur:
        await cur.execute("UPDATE users SET banned = FALSE WHERE user_id = %s", (uid,))
    await conn.commit()
    conn.close()

    write_log("Разбан", uid, "WEB-Admin")
    return web.Response(text="Пользователь разбанен!")

async def handle_logs(request):
    """
    Показывает логи из LOG_FILE только при валидной сессии.
    """
    if not is_session_valid(request):
        return web.Response(text="Доступ запрещён! Войдите в систему.", status=403)

    try:
        with open(LOG_FILE, "r", encoding="utf-8") as log:
            logs = log.read()
    except FileNotFoundError:
        logs = "Лог-файл пока пуст или не создан."

    # HTML-эскейп содержимого логов на случай, если там есть подозрительные данные
    logs_html = html.escape(logs).replace("\n", "<br>")

    return web.Response(text=f"""
    <html>
    <head>
        <title>Логи действий</title>
        <link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css">
    </head>
    <body class="container mt-5">
        <h2>Логи действий</h2>
        <pre>{logs_html}</pre>
        <a href="/admin" class="btn btn-primary">Назад</a>
    </body>
    </html>
    """, content_type='text/html')

# -------------------- Запуск веб-сервера --------------------
async def start_web_server():
    """
    Создаёт Aiohttp-приложение, регистрирует роуты и запускает на 8080 порту
    """
    app = web.Application()

    # Маршруты
    app.router.add_route("GET", "/", handle_login)        # Страница логина (GET)
    app.router.add_route("POST", "/", handle_login)       # Обработка пароля (POST)
    app.router.add_get("/admin", handle_admin_panel)
    app.router.add_get("/ban/{user_id}", handle_ban)
    app.router.add_get("/unban/{user_id}", handle_unban)
    app.router.add_get("/logs", handle_logs)

    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, "localhost", 8080)
    await site.start()

# -------------------- Основная точка входа --------------------
async def main():
    await init_db()
    logging.basicConfig(level=logging.INFO)
    asyncio.create_task(start_web_server())
    await dp.start_polling(bot)

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

Но веб интерфейс все же советую писать с помощью tailwind css, так будет как минимум легче, а как максимум удобнее, добавится поддержка самых разных форматов экранов и в принципе будет более красивый интерфейс.
 
serenityweb сказал(а):
Занятное, конечно, дело ты себе выбрал — писать ботов на aiogram. Не мучай себя, Telebot все еще актуален)
Не согласен с таким мнением хоть и сам пишу на telebot просто из-за того что привык =}
 
admin


Напишите ответ...
  • Вставить:
Прикрепить файлы
Верх