Создание админ-панели для Telegram бота
В этой статье мы создадим полноценную веб-админ-панель для управления Telegram ботом: статистика, управление пользователями, настройки и мониторинг.Содержание
- Архитектура админ-панели
- Настройка веб-сервера
- Аутентификация
- Дашборд и статистика
- Управление пользователями
- Настройки бота
- Мониторинг и логи
- API для админ-панели
Архитектура админ-панели
Технологический стек
# 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 комментариев
Комментарии (0)
Пока нет комментариев. Будьте первым!