fixes, chat handlers
Some checks reported errors
continuous-integration/drone/push Build encountered an error

This commit is contained in:
2025-11-16 14:53:23 +09:00
parent a0e6a385b6
commit 4e06e6296c
6 changed files with 292 additions and 71 deletions

51
main.py
View File

@@ -17,6 +17,7 @@ from src.core.config import BOT_TOKEN, ADMIN_IDS
from src.core.database import async_session_maker, init_db
from src.core.services import UserService, LotteryService, ParticipationService
from src.core.models import User
from src.core.permissions import is_admin, format_commands_help
from src.handlers.admin_panel import admin_router
from src.handlers.account_handlers import account_router
from src.handlers.registration_handlers import router as registration_router
@@ -63,11 +64,6 @@ dp.message.middleware(TaskManagerMiddleware())
dp.callback_query.middleware(TaskManagerMiddleware())
def is_admin(user_id: int) -> bool:
"""Проверка, является ли пользователь администратором"""
return user_id in ADMIN_IDS
def get_main_keyboard(is_admin_user: bool = False) -> InlineKeyboardMarkup:
"""Главная клавиатура"""
buttons = [
@@ -139,6 +135,13 @@ async def cmd_start(message: Message):
)
@router.message(Command("help"))
async def cmd_help(message: Message):
"""Показать список доступных команд с учетом прав пользователя"""
help_text = format_commands_help(message.from_user.id)
await message.answer(help_text, parse_mode="HTML")
@router.callback_query(F.data == "list_lotteries")
async def show_active_lotteries(callback: CallbackQuery):
"""Показать активные розыгрыши"""
@@ -992,10 +995,42 @@ async def back_to_main(callback: CallbackQuery, state: FSMContext):
async def set_commands():
"""Установка команд бота"""
commands = [
BotCommand(command="start", description="🚀 Запустить бота"),
# Команды для обычных пользователей
user_commands = [
BotCommand(command="start", description="🚀 Начать работу с ботом"),
BotCommand(command="help", description="📋 Показать список команд"),
BotCommand(command="my_code", description="🔑 Мой реферальный код"),
BotCommand(command="my_accounts", description="💳 Мои счета"),
]
await bot.set_my_commands(commands)
# Команды для администраторов (добавляются к пользовательским)
admin_commands = user_commands + [
BotCommand(command="add_account", description=" Добавить счет"),
BotCommand(command="remove_account", description=" Удалить счет"),
BotCommand(command="verify_winner", description="✅ Верифицировать победителя"),
BotCommand(command="check_unclaimed", description="🔍 Проверить невостребованные"),
BotCommand(command="redraw", description="🎲 Повторный розыгрыш"),
BotCommand(command="chat_mode", description="💬 Режим чата"),
BotCommand(command="ban", description="🚫 Забанить пользователя"),
BotCommand(command="unban", description="✅ Разбанить"),
BotCommand(command="banlist", description="📋 Список банов"),
BotCommand(command="chat_stats", description="📊 Статистика чата"),
]
# Устанавливаем команды для обычных пользователей
await bot.set_my_commands(user_commands)
# Для админов устанавливаем расширенный набор команд
from aiogram.types import BotCommandScopeChat
for admin_id in ADMIN_IDS:
try:
await bot.set_my_commands(
admin_commands,
scope=BotCommandScopeChat(chat_id=admin_id)
)
except Exception as e:
logging.warning(f"Не удалось установить команды для админа {admin_id}: {e}")
async def main():

202
src/core/permissions.py Normal file
View File

@@ -0,0 +1,202 @@
"""
Система управления правами доступа к командам бота
"""
from functools import wraps
from aiogram.types import Message
from src.core.config import ADMIN_IDS
def is_admin(user_id: int) -> bool:
"""Проверка является ли пользователь администратором"""
return user_id in ADMIN_IDS
def admin_only(func):
"""
Декоратор для команд, доступных только администраторам.
Если пользователь не админ - отправляется сообщение об отказе в доступе.
"""
@wraps(func)
async def wrapper(message: Message, *args, **kwargs):
if not is_admin(message.from_user.id):
await message.answer("У вас нет прав для выполнения этой команды")
return
return await func(message, *args, **kwargs)
return wrapper
def user_command(func):
"""
Декоратор для пользовательских команд.
Доступны всем зарегистрированным пользователям.
"""
@wraps(func)
async def wrapper(message: Message, *args, **kwargs):
# Здесь можно добавить дополнительные проверки для пользователей
# Например, проверку регистрации
return await func(message, *args, **kwargs)
return wrapper
# Реестр команд с описанием и уровнем доступа
COMMAND_REGISTRY = {
# Пользовательские команды
'start': {
'description': 'Начать работу с ботом',
'access': 'user',
'handler': 'main.py'
},
'my_code': {
'description': 'Показать мой реферальный код',
'access': 'user',
'handler': 'registration_handlers.py'
},
'my_accounts': {
'description': 'Показать мои счета',
'access': 'user',
'handler': 'registration_handlers.py'
},
# Административные команды - Управление счетами
'add_account': {
'description': 'Добавить новый счет в систему',
'access': 'admin',
'category': 'Управление счетами',
'handler': 'admin_account_handlers.py'
},
'remove_account': {
'description': 'Удалить счет из системы',
'access': 'admin',
'category': 'Управление счетами',
'handler': 'admin_account_handlers.py'
},
'verify_winner': {
'description': 'Верифицировать победителя',
'access': 'admin',
'category': 'Управление счетами',
'handler': 'admin_account_handlers.py'
},
'winner_status': {
'description': 'Проверить статус победителя',
'access': 'admin',
'category': 'Управление счетами',
'handler': 'admin_account_handlers.py'
},
'user_info': {
'description': 'Получить информацию о пользователе',
'access': 'admin',
'category': 'Управление счетами',
'handler': 'admin_account_handlers.py'
},
# Административные команды - Розыгрыши
'check_unclaimed': {
'description': 'Проверить невостребованные выигрыши',
'access': 'admin',
'category': 'Розыгрыши',
'handler': 'redraw_handlers.py'
},
'redraw': {
'description': 'Провести повторный розыгрыш',
'access': 'admin',
'category': 'Розыгрыши',
'handler': 'redraw_handlers.py'
},
# Административные команды - Управление чатом
'chat_mode': {
'description': 'Управление режимом чата (рассылка/пересылка)',
'access': 'admin',
'category': 'Управление чатом',
'handler': 'admin_chat_handlers.py'
},
'set_forward': {
'description': 'Установить канал для пересылки',
'access': 'admin',
'category': 'Управление чатом',
'handler': 'admin_chat_handlers.py'
},
'global_ban': {
'description': 'Глобальная блокировка пользователя',
'access': 'admin',
'category': 'Управление чатом',
'handler': 'admin_chat_handlers.py'
},
'ban': {
'description': 'Забанить пользователя по ID или ответом',
'access': 'admin',
'category': 'Управление чатом',
'handler': 'admin_chat_handlers.py'
},
'unban': {
'description': 'Разбанить пользователя',
'access': 'admin',
'category': 'Управление чатом',
'handler': 'admin_chat_handlers.py'
},
'banlist': {
'description': 'Показать список забаненных',
'access': 'admin',
'category': 'Управление чатом',
'handler': 'admin_chat_handlers.py'
},
'delete_msg': {
'description': 'Удалить сообщение у всех пользователей',
'access': 'admin',
'category': 'Управление чатом',
'handler': 'admin_chat_handlers.py'
},
'chat_stats': {
'description': 'Статистика чата',
'access': 'admin',
'category': 'Управление чатом',
'handler': 'admin_chat_handlers.py'
},
}
def get_user_commands():
"""Получить список пользовательских команд"""
return {cmd: info for cmd, info in COMMAND_REGISTRY.items() if info['access'] == 'user'}
def get_admin_commands():
"""Получить список административных команд"""
return {cmd: info for cmd, info in COMMAND_REGISTRY.items() if info['access'] == 'admin'}
def get_admin_commands_by_category():
"""Получить административные команды, сгруппированные по категориям"""
commands_by_category = {}
for cmd, info in COMMAND_REGISTRY.items():
if info['access'] == 'admin':
category = info.get('category', 'Прочее')
if category not in commands_by_category:
commands_by_category[category] = {}
commands_by_category[category][cmd] = info
return commands_by_category
def format_commands_help(user_id: int) -> str:
"""
Форматировать справку по командам в зависимости от прав пользователя
"""
help_text = "📋 <b>Доступные команды:</b>\n\n"
# Пользовательские команды
help_text += "👤 <b>Пользовательские команды:</b>\n"
for cmd, info in get_user_commands().items():
help_text += f"/{cmd} - {info['description']}\n"
# Если админ - показываем административные команды
if is_admin(user_id):
help_text += "\n" + "=" * 30 + "\n\n"
help_text += "🔐 <b>Административные команды:</b>\n\n"
for category, commands in get_admin_commands_by_category().items():
help_text += f"<b>{category}:</b>\n"
for cmd, info in commands.items():
help_text += f"/{cmd} - {info['description']}\n"
help_text += "\n"
return help_text

View File

@@ -11,6 +11,7 @@ from src.core.registration_services import AccountService, WinnerNotificationSer
from src.core.services import UserService, LotteryService, ParticipationService
from src.core.models import User, Winner, Account, Participation
from src.core.config import ADMIN_IDS
from src.core.permissions import admin_only
router = Router()
@@ -21,21 +22,14 @@ class AddAccountStates(StatesGroup):
choosing_lottery = State()
def is_admin(user_id: int) -> bool:
"""Проверка прав администратора"""
return user_id in ADMIN_IDS
@router.message(Command("add_account"))
@admin_only
async def add_account_command(message: Message, state: FSMContext):
"""
Добавить счет пользователю по клубной карте
Формат: /add_account <club_card> <account_number>
Или: /add_account (затем вводить данные построчно)
"""
if not is_admin(message.from_user.id):
await message.answer("❌ Недостаточно прав")
return
parts = message.text.split(maxsplit=2)
@@ -308,14 +302,12 @@ async def skip_lottery_add(callback: CallbackQuery, state: FSMContext):
@router.message(Command("remove_account"))
@admin_only
async def remove_account_command(message: Message):
"""
Деактивировать счет
Формат: /remove_account <account_number>
"""
if not is_admin(message.from_user.id):
await message.answer("❌ Недостаточно прав")
return
parts = message.text.split()
if len(parts) != 2:
@@ -341,15 +333,13 @@ async def remove_account_command(message: Message):
@router.message(Command("verify_winner"))
@admin_only
async def verify_winner_command(message: Message):
"""
Подтвердить выигрыш по коду верификации
Формат: /verify_winner <verification_code> <lottery_id>
Пример: /verify_winner AB12CD34 1
"""
if not is_admin(message.from_user.id):
await message.answer("❌ Недостаточно прав")
return
parts = message.text.split()
if len(parts) != 3:
@@ -434,14 +424,12 @@ async def verify_winner_command(message: Message):
@router.message(Command("winner_status"))
@admin_only
async def winner_status_command(message: Message):
"""
Показать статус всех победителей розыгрыша
Формат: /winner_status <lottery_id>
"""
if not is_admin(message.from_user.id):
await message.answer("❌ Недостаточно прав")
return
parts = message.text.split()
if len(parts) != 2:
@@ -509,14 +497,12 @@ async def winner_status_command(message: Message):
@router.message(Command("user_info"))
@admin_only
async def user_info_command(message: Message):
"""
Показать информацию о пользователе
Формат: /user_info <club_card>
"""
if not is_admin(message.from_user.id):
await message.answer("❌ Недостаточно прав")
return
parts = message.text.split()
if len(parts) != 2:

View File

@@ -12,16 +12,12 @@ from src.core.chat_services import (
from src.core.services import UserService
from src.core.database import async_session_maker
from src.core.config import ADMIN_IDS
from src.core.permissions import admin_only
router = Router(name='admin_chat_router')
def is_admin(user_id: int) -> bool:
"""Проверка является ли пользователь админом"""
return user_id in ADMIN_IDS
def get_chat_mode_keyboard() -> InlineKeyboardMarkup:
"""Клавиатура выбора режима чата"""
return InlineKeyboardMarkup(inline_keyboard=[
@@ -34,11 +30,9 @@ def get_chat_mode_keyboard() -> InlineKeyboardMarkup:
@router.message(Command("chat_mode"))
@admin_only
async def cmd_chat_mode(message: Message):
"""Команда управления режимом чата"""
if not is_admin(message.from_user.id):
await message.answer("У вас нет прав для выполнения этой команды")
return
async with async_session_maker() as session:
settings = await ChatSettingsService.get_or_create_settings(session)
@@ -57,9 +51,6 @@ async def cmd_chat_mode(message: Message):
@router.callback_query(F.data.startswith("chat_mode:"))
async def process_chat_mode(callback: CallbackQuery):
"""Обработка выбора режима чата"""
if not is_admin(callback.from_user.id):
await callback.answer("У вас нет прав", show_alert=True)
return
mode = callback.data.split(":")[1]
@@ -78,11 +69,9 @@ async def process_chat_mode(callback: CallbackQuery):
@router.message(Command("set_forward"))
@admin_only
async def cmd_set_forward(message: Message):
"""Установить ID канала для пересылки"""
if not is_admin(message.from_user.id):
await message.answer("У вас нет прав для выполнения этой команды")
return
args = message.text.split(maxsplit=1)
if len(args) < 2:
@@ -112,11 +101,9 @@ async def cmd_set_forward(message: Message):
@router.message(Command("global_ban"))
@admin_only
async def cmd_global_ban(message: Message):
"""Включить/выключить глобальный бан чата"""
if not is_admin(message.from_user.id):
await message.answer("У вас нет прав для выполнения этой команды")
return
async with async_session_maker() as session:
settings = await ChatSettingsService.get_or_create_settings(session)
@@ -140,11 +127,9 @@ async def cmd_global_ban(message: Message):
@router.message(Command("ban"))
@admin_only
async def cmd_ban(message: Message):
"""Забанить пользователя"""
if not is_admin(message.from_user.id):
await message.answer("У вас нет прав для выполнения этой команды")
return
# Проверяем является ли это ответом на сообщение
if message.reply_to_message:
@@ -201,11 +186,9 @@ async def cmd_ban(message: Message):
@router.message(Command("unban"))
@admin_only
async def cmd_unban(message: Message):
"""Разбанить пользователя"""
if not is_admin(message.from_user.id):
await message.answer("У вас нет прав для выполнения этой команды")
return
# Проверяем является ли это ответом на сообщение
if message.reply_to_message:
@@ -244,11 +227,9 @@ async def cmd_unban(message: Message):
@router.message(Command("banlist"))
@admin_only
async def cmd_banlist(message: Message):
"""Показать список забаненных пользователей"""
if not is_admin(message.from_user.id):
await message.answer("У вас нет прав для выполнения этой команды")
return
async with async_session_maker() as session:
banned_users = await BanService.get_banned_users(session, active_only=True)
@@ -276,11 +257,9 @@ async def cmd_banlist(message: Message):
@router.message(Command("delete_msg"))
@admin_only
async def cmd_delete_message(message: Message):
"""Удалить сообщение из чата (пометить как удаленное)"""
if not is_admin(message.from_user.id):
await message.answer("У вас нет прав для выполнения этой команды")
return
if not message.reply_to_message:
await message.answer(
@@ -339,11 +318,9 @@ async def cmd_delete_message(message: Message):
@router.message(Command("chat_stats"))
@admin_only
async def cmd_chat_stats(message: Message):
"""Статистика чата"""
if not is_admin(message.from_user.id):
await message.answer("У вас нет прав для выполнения этой команды")
return
async with async_session_maker() as session:
settings = await ChatSettingsService.get_or_create_settings(session)

View File

@@ -105,6 +105,35 @@ async def forward_to_channel(message: Message, channel_id: str) -> tuple[bool, O
@router.message(F.text)
async def handle_text_message(message: Message):
"""Обработчик текстовых сообщений"""
# Проверяем является ли это командой
if message.text and message.text.startswith('/'):
# Список пользовательских команд, которые НЕ нужно пересылать
user_commands = ['/start', '/help', '/my_code', '/my_accounts']
admin_commands = ['/start',
'/add_account', '/remove_account', '/verify_winner', '/winner_status', '/user_info',
'/check_unclaimed', '/redraw',
'/chat_mode', '/set_forward', '/global_ban', '/ban', '/unban', '/banlist', '/delete_msg', '/chat_stats'
]
# Извлекаем команду (первое слово)
command = message.text.split()[0] if message.text else ''
# Если это пользовательская команда - пропускаем, она будет обработана другими обработчиками
if command in user_commands:
return
# Если это админская команда
if command in admin_commands:
# Проверяем права админа
if not is_admin(message.from_user.id):
await message.answer("У вас нет прав для выполнения этой команды")
return
# Если админ - команда будет обработана другими обработчиками, пропускаем пересылку
return
# Если неизвестная команда - тоже не пересылаем
return
async with async_session_maker() as session:
# Проверяем права на отправку
can_send, reason = await ChatPermissionService.can_send_message(

View File

@@ -11,25 +11,19 @@ from src.core.registration_services import AccountService, WinnerNotificationSer
from src.core.services import LotteryService
from src.core.models import User, Winner
from src.core.config import ADMIN_IDS
from src.core.permissions import admin_only
router = Router()
def is_admin(user_id: int) -> bool:
"""Проверка прав администратора"""
return user_id in ADMIN_IDS
@router.message(Command("check_unclaimed"))
@admin_only
async def check_unclaimed_winners(message: Message):
"""
Проверить неподтвержденные выигрыши (более 24 часов)
Формат: /check_unclaimed <lottery_id>
"""
if not is_admin(message.from_user.id):
await message.answer("❌ Недостаточно прав")
return
parts = message.text.split()
if len(parts) != 2:
@@ -125,14 +119,12 @@ async def check_unclaimed_winners(message: Message):
@router.message(Command("redraw"))
@admin_only
async def redraw_lottery(message: Message):
"""
Переиграть розыгрыш для неподтвержденных выигрышей
Формат: /redraw <lottery_id>
"""
if not is_admin(message.from_user.id):
await message.answer("❌ Недостаточно прав")
return
parts = message.text.split()
if len(parts) != 2: