Создание админ-панели для Telegram бота

В этой статье мы создадим полноценную веб-админ-панель для управления Telegram ботом: статистика, управление пользователями, настройки и мониторинг.

Содержание

Архитектура админ-панели

Технологический стек

# requirements.txt для админ-панели
Flask==2.3.3
Flask-SQLAlchemy==3.0.5
Flask-Login==0.6.3
Flask-WTF==1.1.1
WTForms==3.0.1
Werkzeug==2.3.7
python-telegram-bot==20.5
psycopg2-binary==2.9.7
redis==4.6.0
celery==5.3.1
gunicorn==21.2.0

Структура проекта

telegram-bot-admin/
├── app/
│   ├── __init__.py
│   ├── models/
│   │   ├── __init__.py
│   │   ├── user.py
│   │   ├── bot_stats.py
│   │   └── admin_log.py
│   ├── routes/
│   │   ├── __init__.py
│   │   ├── auth.py
│   │   ├── dashboard.py
│   │   ├── users.py
│   │   └── settings.py
│   ├── templates/
│   │   ├── base.html
│   │   ├── login.html
│   │   ├── dashboard.html
│   │   ├── users.html
│   │   └── settings.html
│   ├── static/
│   │   ├── css/
│   │   ├── js/
│   │   └── images/
│   └── utils/
│       ├── __init__.py
│       ├── decorators.py
│       └── helpers.py
├── bot/
│   ├── bot.py
│   └── handlers/
├── config.py
├── run.py
└── requirements.txt

Настройка веб-сервера

Основное приложение Flask

# app/__init__.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager
from flask_wtf.csrf import CSRFProtect
import os

db = SQLAlchemy()
login_manager = LoginManager()
csrf = CSRFProtect()

def create_app():
    app = Flask(__name__)
    
    # Конфигурация
    app.config['SECRET_KEY'] = os.getenv('SECRET_KEY', 'your-secret-key-here')
    app.config['SQLALCHEMY_DATABASE_URI'] = os.getenv('DATABASE_URL', 'sqlite:///admin.db')
    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
    app.config['WTF_CSRF_ENABLED'] = True
    
    # Инициализация расширений
    db.init_app(app)
    login_manager.init_app(app)
    csrf.init_app(app)
    
    # Настройка Flask-Login
    login_manager.login_view = 'auth.login'
    login_manager.login_message = 'Пожалуйста, войдите в систему для доступа к этой странице.'
    login_manager.login_message_category = 'info'
    
    # Регистрация Blueprint'ов
    from app.routes.auth import auth_bp
    from app.routes.dashboard import dashboard_bp
    from app.routes.users import users_bp
    from app.routes.settings import settings_bp
    
    app.register_blueprint(auth_bp, url_prefix='/auth')
    app.register_blueprint(dashboard_bp, url_prefix='/dashboard')
    app.register_blueprint(users_bp, url_prefix='/users')
    app.register_blueprint(settings_bp, url_prefix='/settings')
    
    # Главная страница
    @app.route('/')
    def index():
        return redirect(url_for('dashboard.index'))
    
    return app

Конфигурация

# config.py
import os
from datetime import timedelta

class Config:
    SECRET_KEY = os.getenv('SECRET_KEY', 'dev-secret-key-change-in-production')
    SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL', 'sqlite:///admin.db')
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    
    # Telegram Bot
    BOT_TOKEN = os.getenv('BOT_TOKEN')
    ADMIN_IDS = [int(x) for x in os.getenv('ADMIN_IDS', '').split(',') if x]
    
    # Redis
    REDIS_URL = os.getenv('REDIS_URL', 'redis://localhost:6379/0')
    
    # Сессии
    PERMANENT_SESSION_LIFETIME = timedelta(hours=24)
    
    # Загрузка файлов
    UPLOAD_FOLDER = 'uploads'
    MAX_CONTENT_LENGTH = 16  1024  1024  # 16MB
    
    # Логирование
    LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')
    LOG_FILE = os.getenv('LOG_FILE', 'logs/admin.log')

class DevelopmentConfig(Config):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///admin_dev.db'

class ProductionConfig(Config):
    DEBUG = False
    SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL')

config = {
    'development': DevelopmentConfig,
    'production': ProductionConfig,
    'default': DevelopmentConfig
}

Аутентификация

Модель администратора

# app/models/user.py
from flask_sqlalchemy import SQLAlchemy
from flask_login import UserMixin
from werkzeug.security import generate_password_hash, check_password_hash
from datetime import datetime

db = SQLAlchemy()

class AdminUser(UserMixin, db.Model):
    """Модель администратора"""
    __tablename__ = 'admin_users'
    
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    password_hash = db.Column(db.String(255), nullable=False)
    telegram_id = db.Column(db.Integer, unique=True, nullable=True)
    is_active = db.Column(db.Boolean, default=True)
    is_superuser = db.Column(db.Boolean, default=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    last_login = db.Column(db.DateTime, nullable=True)
    
    def set_password(self, password):
        """Установка пароля"""
        self.password_hash = generate_password_hash(password)
    
    def check_password(self, password):
        """Проверка пароля"""
        return check_password_hash(self.password_hash, password)
    
    def __repr__(self):
        return f'<AdminUser {self.username}>'

class AdminLog(db.Model):
    """Лог действий администратора"""
    __tablename__ = 'admin_logs'
    
    id = db.Column(db.Integer, primary_key=True)
    admin_id = db.Column(db.Integer, db.ForeignKey('admin_users.id'), nullable=False)
    action = db.Column(db.String(100), nullable=False)
    details = db.Column(db.Text, nullable=True)
    ip_address = db.Column(db.String(45), nullable=True)
    user_agent = db.Column(db.String(255), nullable=True)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    
    admin = db.relationship('AdminUser', backref=db.backref('logs', lazy=True))

Маршруты аутентификации

# app/routes/auth.py
from flask import Blueprint, render_template, request, redirect, url_for, flash, session
from flask_login import login_user, logout_user, login_required, current_user
from app.models.user import AdminUser, AdminLog, db
from app.utils.decorators import log_admin_action
from datetime import datetime

auth_bp = Blueprint('auth', __name__)

@auth_bp.route('/login', methods=['GET', 'POST'])
def login():
    """Страница входа"""
    if current_user.is_authenticated:
        return redirect(url_for('dashboard.index'))
    
    if request.method == 'POST':
        username = request.form.get('username')
        password = request.form.get('password')
        remember = bool(request.form.get('remember'))
        
        user = AdminUser.query.filter_by(username=username).first()
        
        if user and user.check_password(password) and user.is_active:
            login_user(user, remember=remember)
            
            # Обновление времени последнего входа
            user.last_login = datetime.utcnow()
            db.session.commit()
            
            # Логирование входа
            log_admin_action('LOGIN', f'Вход в систему: {username}')
            
            flash('Успешный вход в систему!', 'success')
            return redirect(url_for('dashboard.index'))
        else:
            flash('Неверное имя пользователя или пароль!', 'error')
    
    return render_template('login.html')

@auth_bp.route('/logout')
@login_required
def logout():
    """Выход из системы"""
    log_admin_action('LOGOUT', 'Выход из системы')
    logout_user()
    flash('Вы вышли из системы.', 'info')
    return redirect(url_for('auth.login'))

@auth_bp.route('/profile')
@login_required
def profile():
    """Профиль администратора"""
    return render_template('profile.html', user=current_user)

Декораторы безопасности

# app/utils/decorators.py
from functools import wraps
from flask import request, redirect, url_for, flash, abort
from flask_login import current_user
from app.models.user import AdminLog, db
from datetime import datetime

def log_admin_action(action, details=None):
    """Логирование действий администратора"""
    if current_user.is_authenticated:
        log_entry = AdminLog(
            admin_id=current_user.id,
            action=action,
            details=details,
            ip_address=request.remote_addr,
            user_agent=request.headers.get('User-Agent')
        )
        db.session.add(log_entry)
        db.session.commit()

def admin_required(f):
    """Декоратор для проверки прав администратора"""
    @wraps(f)
    def decorated_function(args, *kwargs):
        if not current_user.is_authenticated:
            return redirect(url_for('auth.login'))
        if not current_user.is_active:
            flash('Ваш аккаунт деактивирован!', 'error')
            return redirect(url_for('auth.login'))
        return f(args, *kwargs)
    return decorated_function

def superuser_required(f):
    """Декоратор для проверки прав суперпользователя"""
    @wraps(f)
    def decorated_function(args, *kwargs):
        if not current_user.is_authenticated:
            return redirect(url_for('auth.login'))
        if not current_user.is_superuser:
            abort(403)
        return f(args, *kwargs)
    return decorated_function

Дашборд и статистика

Модель статистики

# app/models/bot_stats.py
from app.models.user import db
from datetime import datetime, timedelta

class BotStats(db.Model):
    """Статистика бота"""
    __tablename__ = 'bot_stats'
    
    id = db.Column(db.Integer, primary_key=True)
    date = db.Column(db.Date, nullable=False)
    total_users = db.Column(db.Integer, default=0)
    new_users = db.Column(db.Integer, default=0)
    active_users = db.Column(db.Integer, default=0)
    messages_sent = db.Column(db.Integer, default=0)
    commands_used = db.Column(db.Integer, default=0)
    errors_count = db.Column(db.Integer, default=0)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

class UserActivity(db.Model):
    """Активность пользователей"""
    __tablename__ = 'user_activity'
    
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, nullable=False)
    action = db.Column(db.String(50), nullable=False)
    details = db.Column(db.Text, nullable=True)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

Маршруты дашборда

# app/routes/dashboard.py
from flask import Blueprint, render_template, jsonify
from flask_login import login_required
from app.models.user import db
from app.models.bot_stats import BotStats, UserActivity
from app.utils.decorators import admin_required
from datetime import datetime, timedelta
import json

dashboard_bp = Blueprint('dashboard', __name__)

@dashboard_bp.route('/')
@admin_required
def index():
    """Главная страница дашборда"""
    # Получение статистики за последние 30 дней
    thirty_days_ago = datetime.now() - timedelta(days=30)
    
    # Общая статистика
    total_users = db.session.query(db.func.count()).select_from(db.text('users')).scalar() or 0
    active_users = db.session.query(db.func.count()).select_from(db.text('users')).where(
        db.text('last_activity > :cutoff')
    ).params(cutoff=thirty_days_ago).scalar() or 0
    
    # Статистика за сегодня
    today_stats = BotStats.query.filter_by(date=datetime.now().date()).first()
    
    # Статистика за последние 7 дней
    week_stats = BotStats.query.filter(
        BotStats.date >= datetime.now().date() - timedelta(days=7)
    ).order_by(BotStats.date).all()
    
    # Топ команд
    top_commands = db.session.execute(db.text("""
        SELECT command, COUNT(*) as count 
        FROM user_activity 
        WHERE action = 'command' 
        AND created_at >= :cutoff
        GROUP BY command 
        ORDER BY count DESC 
        LIMIT 10
    """), {'cutoff': thirty_days_ago}).fetchall()
    
    return render_template('dashboard.html',
                         total_users=total_users,
                         active_users=active_users,
                         today_stats=today_stats,
                         week_stats=week_stats,
                         top_commands=top_commands)

@dashboard_bp.route('/api/stats')
@admin_required
def api_stats():
    """API для получения статистики"""
    # Статистика за последние 30 дней
    thirty_days_ago = datetime.now() - timedelta(days=30)
    stats = BotStats.query.filter(
        BotStats.date >= thirty_days_ago
    ).order_by(BotStats.date).all()
    
    data = {
        'labels': [stat.date.strftime('%Y-%m-%d') for stat in stats],
        'datasets': [
            {
                'label': 'Новые пользователи',
                'data': [stat.new_users for stat in stats],
                'borderColor': 'rgb(75, 192, 192)',
                'backgroundColor': 'rgba(75, 192, 192, 0.2)'
            },
            {
                'label': 'Активные пользователи',
                'data': [stat.active_users for stat in stats],
                'borderColor': 'rgb(255, 99, 132)',
                'backgroundColor': 'rgba(255, 99, 132, 0.2)'
            }
        ]
    }
    
    return jsonify(data)

@dashboard_bp.route('/api/realtime')
@admin_required
def api_realtime():
    """API для получения данных в реальном времени"""
    # Количество пользователей онлайн (активных за последние 5 минут)
    five_minutes_ago = datetime.now() - timedelta(minutes=5)
    online_users = db.session.query(db.func.count()).select_from(db.text('users')).where(
        db.text('last_activity > :cutoff')
    ).params(cutoff=five_minutes_ago).scalar() or 0
    
    # Количество сообщений за последний час
    one_hour_ago = datetime.now() - timedelta(hours=1)
    messages_last_hour = db.session.query(db.func.count()).select_from(db.text('messages')).where(
        db.text('created_at > :cutoff')
    ).params(cutoff=one_hour_ago).scalar() or 0
    
    return jsonify({
        'online_users': online_users,
        'messages_last_hour': messages_last_hour,
        'timestamp': datetime.now().isoformat()
    })

HTML шаблон дашборда

<!-- app/templates/dashboard.html -->
{% extends "base.html" %}

{% block title %}Дашборд - Админ-панель{% endblock %}

{% block content %}
<div class="container-fluid">
    <div class="row">
        <div class="col-12">
            <h1 class="h3 mb-4">Дашборд</h1>
        </div>
    </div>
    
    <!-- Статистические карточки -->
    <div class="row mb-4">
        <div class="col-xl-3 col-md-6 mb-4">
            <div class="card border-left-primary shadow h-100 py-2">
                <div class="card-body">
                    <div class="row no-gutters align-items-center">
                        <div class="col mr-2">
                            <div class="text-xs font-weight-bold text-primary text-uppercase mb-1">
                                Всего пользователей</div>
                            <div class="h5 mb-0 font-weight-bold text-gray-800">{{ total_users }}</div>
                        </div>
                        <div class="col-auto">
                            <i class="fas fa-users fa-2x text-gray-300"></i>
                        </div>
                    </div>
                </div>
            </div>
        </div>
        
        <div class="col-xl-3 col-md-6 mb-4">
            <div class="card border-left-success shadow h-100 py-2">
                <div class="card-body">
                    <div class="row no-gutters align-items-center">
                        <div class="col mr-2">
                            <div class="text-xs font-weight-bold text-success text-uppercase mb-1">
                                Активные пользователи</div>
                            <div class="h5 mb-0 font-weight-bold text-gray-800">{{ active_users }}</div>
                        </div>
                        <div class="col-auto">
                            <i class="fas fa-user-check fa-2x text-gray-300"></i>
                        </div>
                    </div>
                </div>
            </div>
        </div>
        
        <div class="col-xl-3 col-md-6 mb-4">
            <div class="card border-left-info shadow h-100 py-2">
                <div class="card-body">
                    <div class="row no-gutters align-items-center">
                        <div class="col mr-2">
                            <div class="text-xs font-weight-bold text-info text-uppercase mb-1">
                                Сообщений сегодня</div>
                            <div class="h5 mb-0 font-weight-bold text-gray-800">
                                {{ today_stats.messages_sent if today_stats else 0 }}
                            </div>
                        </div>
                        <div class="col-auto">
                            <i class="fas fa-comments fa-2x text-gray-300"></i>
                        </div>
                    </div>
                </div>
            </div>
        </div>
        
        <div class="col-xl-3 col-md-6 mb-4">
            <div class="card border-left-warning shadow h-100 py-2">
                <div class="card-body">
                    <div class="row no-gutters align-items-center">
                        <div class="col mr-2">
                            <div class="text-xs font-weight-bold text-warning text-uppercase mb-1">
                                Команд сегодня</div>
                            <div class="h5 mb-0 font-weight-bold text-gray-800">
                                {{ today_stats.commands_used if today_stats else 0 }}
                            </div>
                        </div>
                        <div class="col-auto">
                            <i class="fas fa-terminal fa-2x text-gray-300"></i>
                        </div>
                    </div>
                </div>
            </div>
        </div>
    </div>
    
    <!-- Графики -->
    <div class="row">
        <div class="col-xl-8 col-lg-7">
            <div class="card shadow mb-4">
                <div class="card-header py-3 d-flex flex-row align-items-center justify-content-between">
                    <h6 class="m-0 font-weight-bold text-primary">Статистика пользователей</h6>
                </div>
                <div class="card-body">
                    <div class="chart-area">
                        <canvas id="userStatsChart"></canvas>
                    </div>
                </div>
            </div>
        </div>
        
        <div class="col-xl-4 col-lg-5">
            <div class="card shadow mb-4">
                <div class="card-header py-3 d-flex flex-row align-items-center justify-content-between">
                    <h6 class="m-0 font-weight-bold text-primary">Топ команд</h6>
                </div>
                <div class="card-body">
                    <div class="chart-pie pt-4 pb-2">
                        <canvas id="commandsChart"></canvas>
                    </div>
                </div>
            </div>
        </div>
    </div>
</div>

<script>
// Инициализация графиков
document.addEventListener('DOMContentLoaded', function() {
    // График статистики пользователей
    fetch('/dashboard/api/stats')
        .then(response => response.json())
        .then(data => {
            const ctx = document.getElementById('userStatsChart').getContext('2d');
            new Chart(ctx, {
                type: 'line',
                data: data,
                options: {
                    responsive: true,
                    scales: {
                        y: {
                            beginAtZero: true
                        }
                    }
                }
            });
        });
    
    // Обновление данных в реальном времени
    setInterval(function() {
        fetch('/dashboard/api/realtime')
            .then(response => response.json())
            .then(data => {
                // Обновление счетчиков
                console.log('Real-time data:', data);
            });
    }, 30000); // Обновление каждые 30 секунд
});
</script>
{% endblock %}

Управление пользователями

Маршруты управления пользователями

# app/routes/users.py
from flask import Blueprint, render_template, request, jsonify, flash, redirect, url_for
from flask_login import login_required
from app.models.user import db
from app.utils.decorators import admin_required, log_admin_action
from sqlalchemy import or_, and_
import json

users_bp = Blueprint('users', __name__)

@users_bp.route('/')
@admin_required
def index():
    """Список пользователей"""
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 20, type=int)
    search = request.args.get('search', '')
    status = request.args.get('status', 'all')
    
    # Базовый запрос
    query = db.session.execute(db.text("SELECT * FROM users WHERE 1=1"))
    
    # Фильтрация по поиску
    if search:
        query = db.session.execute(db.text("""
            SELECT * FROM users 
            WHERE first_name ILIKE :search 
            OR username ILIKE :search 
            OR telegram_id::text ILIKE :search
        """), {'search': f'%{search}%'})
    
    # Фильтрация по статусу
    if status == 'active':
        query = db.session.execute(db.text("""
            SELECT * FROM users 
            WHERE last_activity > NOW() - INTERVAL '7 days'
        """))
    elif status == 'inactive':
        query = db.session.execute(db.text("""
            SELECT * FROM users 
            WHERE last_activity <= NOW() - INTERVAL '7 days'
        """))
    
    users = query.fetchall()
    
    return render_template('users.html', 
                         users=users, 
                         page=page, 
                         search=search, 
                         status=status)

@users_bp.route('/<int:user_id>')
@admin_required
def view_user(user_id):
    """Просмотр пользователя"""
    user = db.session.execute(db.text("SELECT * FROM users WHERE id = :user_id"), 
                            {'user_id': user_id}).fetchone()
    
    if not user:
        flash('Пользователь не найден!', 'error')
        return redirect(url_for('users.index'))
    
    # Статистика пользователя
    user_stats = db.session.execute(db.text("""
        SELECT 
            COUNT(*) as messages_count,
            COUNT(DISTINCT DATE(created_at)) as active_days,
            MAX(created_at) as last_message
        FROM messages 
        WHERE user_id = :user_id
    """), {'user_id': user_id}).fetchone()
    
    # Последние сообщения
    recent_messages = db.session.execute(db.text("""
        SELECT * FROM messages 
        WHERE user_id = :user_id 
        ORDER BY created_at DESC 
        LIMIT 10
    """), {'user_id': user_id}).fetchall()
    
    return render_template('user_detail.html', 
                         user=user, 
                         user_stats=user_stats,
                         recent_messages=recent_messages)

@users_bp.route('/<int:user_id>/ban', methods=['POST'])
@admin_required
def ban_user(user_id):
    """Блокировка пользователя"""
    user = db.session.execute(db.text("SELECT * FROM users WHERE id = :user_id"), 
                            {'user_id': user_id}).fetchone()
    
    if not user:
        flash('Пользователь не найден!', 'error')
        return redirect(url_for('users.index'))
    
    # Обновление статуса пользователя
    db.session.execute(db.text("""
        UPDATE users 
        SET is_banned = true, banned_at = NOW() 
        WHERE id = :user_id
    """), {'user_id': user_id})
    
    db.session.commit()
    
    log_admin_action('BAN_USER', f'Заблокирован пользователь {user_id}')
    flash('Пользователь заблокирован!', 'success')
    
    return redirect(url_for('users.view_user', user_id=user_id))

@users_bp.route('/<int:user_id>/unban', methods=['POST'])
@admin_required
def unban_user(user_id):
    """Разблокировка пользователя"""
    user = db.session.execute(db.text("SELECT * FROM users WHERE id = :user_id"), 
                            {'user_id': user_id}).fetchone()
    
    if not user:
        flash('Пользователь не найден!', 'error')
        return redirect(url_for('users.index'))
    
    # Обновление статуса пользователя
    db.session.execute(db.text("""
        UPDATE users 
        SET is_banned = false, banned_at = NULL 
        WHERE id = :user_id
    """), {'user_id': user_id})
    
    db.session.commit()
    
    log_admin_action('UNBAN_USER', f'Разблокирован пользователь {user_id}')
    flash('Пользователь разблокирован!', 'success')
    
    return redirect(url_for('users.view_user', user_id=user_id))

@users_bp.route('/export')
@admin_required
def export_users():
    """Экспорт пользователей"""
    users = db.session.execute(db.text("""
        SELECT 
            telegram_id,
            username,
            first_name,
            last_name,
            created_at,
            last_activity,
            is_premium
        FROM users 
        ORDER BY created_at DESC
    """)).fetchall()
    
    # Создание CSV
    import csv
    import io
    
    output = io.StringIO()
    writer = csv.writer(output)
    
    # Заголовки
    writer.writerow(['Telegram ID', 'Username', 'First Name', 'Last Name', 'Created At', 'Last Activity', 'Premium'])
    
    # Данные
    for user in users:
        writer.writerow([
            user.telegram_id,
            user.username or '',
            user.first_name or '',
            user.last_name or '',
            user.created_at.strftime('%Y-%m-%d %H:%M:%S') if user.created_at else '',
            user.last_activity.strftime('%Y-%m-%d %H:%M:%S') if user.last_activity else '',
            'Да' if user.is_premium else 'Нет'
        ])
    
    output.seek(0)
    
    log_admin_action('EXPORT_USERS', f'Экспортировано {len(users)} пользователей')
    
    return output.getvalue(), 200, {
        'Content-Type': 'text/csv',
        'Content-Disposition': 'attachment; filename=users_export.csv'
    }

Настройки бота

Маршруты настроек

# app/routes/settings.py
from flask import Blueprint, render_template, request, flash, redirect, url_for, jsonify
from flask_login import login_required
from app.models.user import db
from app.utils.decorators import admin_required, superuser_required, log_admin_action
import json
import os

settings_bp = Blueprint('settings', __name__)

@settings_bp.route('/')
@admin_required
def index():
    """Настройки бота"""
    # Получение текущих настроек
    settings = get_bot_settings()
    
    return render_template('settings.html', settings=settings)

@settings_bp.route('/update', methods=['POST'])
@admin_required
def update_settings():
    """Обновление настроек"""
    try:
        # Получение данных из формы
        settings_data = {
            'bot_name': request.form.get('bot_name'),
            'welcome_message': request.form.get('welcome_message'),
            'help_message': request.form.get('help_message'),
            'maintenance_mode': bool(request.form.get('maintenance_mode')),
            'max_file_size': int(request.form.get('max_file_size', 10)),
            'allowed_file_types': request.form.get('allowed_file_types', '').split(','),
            'auto_delete_messages': bool(request.form.get('auto_delete_messages')),
            'message_delete_delay': int(request.form.get('message_delete_delay', 60)),
            'enable_logging': bool(request.form.get('enable_logging')),
            'log_level': request.form.get('log_level', 'INFO')
        }
        
        # Сохранение настроек
        save_bot_settings(settings_data)
        
        log_admin_action('UPDATE_SETTINGS', 'Обновлены настройки бота')
        flash('Настройки успешно обновлены!', 'success')
        
    except Exception as e:
        flash(f'Ошибка при обновлении настроек: {str(e)}', 'error')
    
    return redirect(url_for('settings.index'))

@settings_bp.route('/bot-status')
@admin_required
def bot_status():
    """Статус бота"""
    try:
        # Проверка статуса бота через API
        import requests
        
        bot_token = os.getenv('BOT_TOKEN')
        if not bot_token:
            return jsonify({'status': 'error', 'message': 'BOT_TOKEN не настроен'})
        
        # Получение информации о боте
        response = requests.get(f'https://api.telegram.org/bot{bot_token}/getMe')
        
        if response.status_code == 200:
            bot_info = response.json()
            if bot_info['ok']:
                return jsonify({
                    'status': 'online',
                    'bot_info': bot_info['result']
                })
            else:
                return jsonify({'status': 'error', 'message': bot_info['description']})
        else:
            return jsonify({'status': 'error', 'message': 'Ошибка подключения к API'})
            
    except Exception as e:
        return jsonify({'status': 'error', 'message': str(e)})

@settings_bp.route('/restart-bot', methods=['POST'])
@superuser_required
def restart_bot():
    """Перезапуск бота"""
    try:
        # Здесь должна быть логика перезапуска бота
        # Например, отправка сигнала процессу или перезапуск через Docker
        
        log_admin_action('RESTART_BOT', 'Перезапуск бота')
        flash('Команда перезапуска бота отправлена!', 'success')
        
    except Exception as e:
        flash(f'Ошибка при перезапуске бота: {str(e)}', 'error')
    
    return redirect(url_for('settings.index'))

def get_bot_settings():
    """Получение настроек бота"""
    # Здесь можно получать настройки из базы данных или файла конфигурации
    default_settings = {
        'bot_name': 'Мой Telegram Бот',
        'welcome_message': 'Добро пожаловать!',
        'help_message': 'Используйте /help для получения помощи',
        'maintenance_mode': False,
        'max_file_size': 10,
        'allowed_file_types': ['jpg', 'png', 'pdf', 'txt'],
        'auto_delete_messages': False,
        'message_delete_delay': 60,
        'enable_logging': True,
        'log_level': 'INFO'
    }
    
    # Попытка загрузить из базы данных
    try:
        settings_row = db.session.execute(db.text("SELECT settings FROM bot_config WHERE id = 1")).fetchone()
        if settings_row:
            return json.loads(settings_row.settings)
    except:
        pass
    
    return default_settings

def save_bot_settings(settings_data):
    """Сохранение настроек бота"""
    # Создание таблицы настроек если не существует
    db.session.execute(db.text("""
        CREATE TABLE IF NOT EXISTS bot_config (
            id INTEGER PRIMARY KEY,
            settings TEXT,
            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """))
    
    # Сохранение настроек
    db.session.execute(db.text("""
        INSERT INTO bot_config (id, settings) 
        VALUES (1, :settings)
        ON CONFLICT(id) DO UPDATE SET 
            settings = :settings,
            updated_at = CURRENT_TIMESTAMP
    """), {'settings': json.dumps(settings_data)})
    
    db.session.commit()

Мониторинг и логи

Система логирования

# app/utils/logging.py
import logging
import os
from datetime import datetime
from logging.handlers import RotatingFileHandler

def setup_logging(app):
    """Настройка логирования"""
    if not app.debug:
        # Создание директории для логов
        log_dir = 'logs'
        os.makedirs(log_dir, exist_ok=True)
        
        # Настройка файлового логгера
        file_handler = RotatingFileHandler(
            os.path.join(log_dir, 'admin.log'),
            maxBytes=10240000,  # 10MB
            backupCount=10
        )
        
        file_handler.setFormatter(logging.Formatter(
            '%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]'
        ))
        
        file_handler.setLevel(logging.INFO)
        app.logger.addHandler(file_handler)
        
        app.logger.setLevel(logging.INFO)
        app.logger.info('Admin panel startup')

# Маршрут для просмотра логов
@settings_bp.route('/logs')
@admin_required
def view_logs():
    """Просмотр логов"""
    log_file = 'logs/admin.log'
    
    if not os.path.exists(log_file):
        flash('Файл логов не найден!', 'warning')
        return render_template('logs.html', logs=[])
    
    try:
        with open(log_file, 'r', encoding='utf-8') as f:
            logs = f.readlines()
        
        # Последние 100 строк
        logs = logs[-100:]
        
        return render_template('logs.html', logs=logs)
        
    except Exception as e:
        flash(f'Ошибка при чтении логов: {str(e)}', 'error')
        return render_template('logs.html', logs=[])

API для админ-панели

REST API

# app/routes/api.py
from flask import Blueprint, jsonify, request
from flask_login import login_required
from app.utils.decorators import admin_required
from app.models.user import db
import json

api_bp = Blueprint('api', __name__)

@api_bp.route('/users')
@admin_required
def api_users():
    """API для получения списка пользователей"""
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 20, type=int)
    
    users = db.session.execute(db.text("""
        SELECT 
            id, telegram_id, username, first_name, last_name,
            created_at, last_activity, is_premium
        FROM users 
        ORDER BY created_at DESC
        LIMIT :limit OFFSET :offset
    """), {
        'limit': per_page,
        'offset': (page - 1) * per_page
    }).fetchall()
    
    return jsonify({
        'users': [dict(user._mapping) for user in users],
        'page': page,
        'per_page': per_page
    })

@api_bp.route('/stats')
@admin_required
def api_stats():
    """API для получения статистики"""
    stats = db.session.execute(db.text("""
        SELECT 
            COUNT(*) as total_users,
            COUNT(CASE WHEN last_activity > NOW() - INTERVAL '7 days' THEN 1 END) as active_users,
            COUNT(CASE WHEN created_at > CURRENT_DATE THEN 1 END) as new_users_today
        FROM users
    """)).fetchone()
    
    return jsonify(dict(stats._mapping))

@api_bp.route('/bot/status')
@admin_required
def api_bot_status():
    """API для проверки статуса бота"""
    # Проверка статуса бота
    try:
        import requests
        bot_token = os.getenv('BOT_TOKEN')
        
        response = requests.get(f'https://api.telegram.org/bot{bot_token}/getMe', timeout=5)
        
        if response.status_code == 200:
            return jsonify({'status': 'online', 'data': response.json()})
        else:
            return jsonify({'status': 'offline', 'error': 'API недоступен'})
            
    except Exception as e:
        return jsonify({'status': 'error', 'error': str(e)})

Заключение

В этой статье мы создали полноценную админ-панель для Telegram бота, включающую:
  • ✅ Систему аутентификации и авторизации
  • ✅ Дашборд с статистикой и графиками
  • ✅ Управление пользователями
  • ✅ Настройки бота
  • ✅ Мониторинг и логирование
  • ✅ REST API для интеграции
Админ-панель предоставляет полный контроль над ботом и позволяет эффективно управлять пользователями и настройками.

Полезные ссылки

1881 просмотров
13 лайков
0 комментариев