Andrey
2d40f4496e
- Added voice message handling capabilities, including saving and deleting audio messages via callback queries. - Refactored audio record management in the database to remove unnecessary fields and streamline operations. - Introduced new keyboard options for voice interactions in the bot. - Updated `.gitignore` to include voice user files for better project organization. - Removed obsolete voice bot handler files to simplify the codebase.
242 lines
11 KiB
Python
242 lines
11 KiB
Python
"""Main private handlers module for Telegram bot"""
|
|
|
|
# Standard library imports
|
|
import asyncio
|
|
from datetime import datetime
|
|
|
|
# Third-party imports
|
|
from aiogram import types, Router, F
|
|
from aiogram.filters import Command, StateFilter
|
|
from aiogram.fsm.context import FSMContext
|
|
|
|
# Local imports - filters and middlewares
|
|
from helper_bot.filters.main import ChatTypeFilter
|
|
from helper_bot.middlewares.album_middleware import AlbumMiddleware
|
|
from helper_bot.middlewares.blacklist_middleware import BlacklistMiddleware
|
|
|
|
# Local imports - utilities
|
|
from helper_bot.keyboards import get_reply_keyboard, get_reply_keyboard_for_post
|
|
from helper_bot.keyboards.keyboards import get_reply_keyboard_leave_chat
|
|
from helper_bot.utils import messages
|
|
from helper_bot.utils.helper_func import (
|
|
get_first_name,
|
|
update_user_info,
|
|
check_user_emoji
|
|
)
|
|
|
|
# Local imports - metrics
|
|
from helper_bot.utils.metrics import (
|
|
metrics,
|
|
track_time,
|
|
track_errors
|
|
)
|
|
|
|
# Local imports - modular components
|
|
from .constants import FSM_STATES, BUTTON_TEXTS, ERROR_MESSAGES
|
|
from .services import BotSettings, UserService, PostService, StickerService
|
|
from .decorators import error_handler
|
|
|
|
# Expose sleep for tests (tests patch helper_bot.handlers.private.private_handlers.sleep)
|
|
sleep = asyncio.sleep
|
|
|
|
|
|
class PrivateHandlers:
|
|
"""Main handler class for private messages"""
|
|
|
|
def __init__(self, db, settings: BotSettings):
|
|
self.db = db
|
|
self.settings = settings
|
|
self.user_service = UserService(db, settings)
|
|
self.post_service = PostService(db, settings)
|
|
self.sticker_service = StickerService(settings)
|
|
|
|
# Create router
|
|
self.router = Router()
|
|
self.router.message.middleware(AlbumMiddleware())
|
|
self.router.message.middleware(BlacklistMiddleware())
|
|
|
|
# Register handlers
|
|
self._register_handlers()
|
|
|
|
def _register_handlers(self):
|
|
"""Register all message handlers"""
|
|
# Command handlers
|
|
self.router.message.register(self.handle_emoji_message, ChatTypeFilter(chat_type=["private"]), Command("emoji"))
|
|
self.router.message.register(self.handle_restart_message, ChatTypeFilter(chat_type=["private"]), Command("restart"))
|
|
self.router.message.register(self.handle_start_message, ChatTypeFilter(chat_type=["private"]), Command("start"))
|
|
self.router.message.register(self.handle_start_message, ChatTypeFilter(chat_type=["private"]), F.text == BUTTON_TEXTS["RETURN_TO_BOT"])
|
|
|
|
# Button handlers
|
|
self.router.message.register(self.suggest_post, StateFilter(FSM_STATES["START"]), ChatTypeFilter(chat_type=["private"]), F.text == BUTTON_TEXTS["SUGGEST_POST"])
|
|
self.router.message.register(self.end_message, ChatTypeFilter(chat_type=["private"]), F.text == BUTTON_TEXTS["SAY_GOODBYE"])
|
|
self.router.message.register(self.end_message, ChatTypeFilter(chat_type=["private"]), F.text == BUTTON_TEXTS["LEAVE_CHAT"])
|
|
self.router.message.register(self.stickers, ChatTypeFilter(chat_type=["private"]), F.text == BUTTON_TEXTS["WANT_STICKERS"])
|
|
self.router.message.register(self.connect_with_admin, StateFilter(FSM_STATES["START"]), ChatTypeFilter(chat_type=["private"]), F.text == BUTTON_TEXTS["CONNECT_ADMIN"])
|
|
|
|
|
|
# State handlers
|
|
self.router.message.register(self.suggest_router, StateFilter(FSM_STATES["SUGGEST"]), ChatTypeFilter(chat_type=["private"]))
|
|
self.router.message.register(self.resend_message_in_group_for_message, StateFilter(FSM_STATES["PRE_CHAT"]), ChatTypeFilter(chat_type=["private"]))
|
|
self.router.message.register(self.resend_message_in_group_for_message, StateFilter(FSM_STATES["CHAT"]), ChatTypeFilter(chat_type=["private"]))
|
|
|
|
@error_handler
|
|
async def handle_emoji_message(self, message: types.Message, state: FSMContext, **kwargs):
|
|
"""Handle emoji command"""
|
|
await self.user_service.log_user_message(message)
|
|
user_emoji = check_user_emoji(message)
|
|
await state.set_state(FSM_STATES["START"])
|
|
if user_emoji is not None:
|
|
await message.answer(f'Твоя эмодзя - {user_emoji}', parse_mode='HTML')
|
|
|
|
@error_handler
|
|
async def handle_restart_message(self, message: types.Message, state: FSMContext, **kwargs):
|
|
"""Handle restart command"""
|
|
markup = get_reply_keyboard(self.db, message.from_user.id)
|
|
await self.user_service.log_user_message(message)
|
|
await state.set_state(FSM_STATES["START"])
|
|
await update_user_info('love', message)
|
|
check_user_emoji(message)
|
|
await message.answer('Я перезапущен!', reply_markup=markup, parse_mode='HTML')
|
|
|
|
@error_handler
|
|
async def handle_start_message(self, message: types.Message, state: FSMContext, **kwargs):
|
|
"""Handle start command and return to bot button with metrics tracking"""
|
|
# User service operations with metrics
|
|
await self.user_service.log_user_message(message)
|
|
await self.user_service.ensure_user_exists(message)
|
|
await state.set_state(FSM_STATES["START"])
|
|
|
|
# Send sticker with metrics
|
|
await self.sticker_service.send_random_hello_sticker(message)
|
|
|
|
# Send welcome message with metrics
|
|
markup = get_reply_keyboard(self.db, message.from_user.id)
|
|
hello_message = messages.get_message(get_first_name(message), 'HELLO_MESSAGE')
|
|
await message.answer(hello_message, reply_markup=markup, parse_mode='HTML')
|
|
|
|
@error_handler
|
|
async def suggest_post(self, message: types.Message, state: FSMContext, **kwargs):
|
|
"""Handle suggest post button"""
|
|
# User service operations with metrics
|
|
await self.user_service.update_user_activity(message.from_user.id)
|
|
await self.user_service.log_user_message(message)
|
|
await state.set_state(FSM_STATES["SUGGEST"])
|
|
|
|
markup = types.ReplyKeyboardRemove()
|
|
suggest_news = messages.get_message(get_first_name(message), 'SUGGEST_NEWS')
|
|
await message.answer(suggest_news, reply_markup=markup)
|
|
|
|
@error_handler
|
|
async def end_message(self, message: types.Message, state: FSMContext, **kwargs):
|
|
"""Handle goodbye button"""
|
|
# User service operations with metrics
|
|
await self.user_service.update_user_activity(message.from_user.id)
|
|
await self.user_service.log_user_message(message)
|
|
|
|
# Send sticker
|
|
await self.sticker_service.send_random_goodbye_sticker(message)
|
|
|
|
# Send goodbye message
|
|
markup = types.ReplyKeyboardRemove()
|
|
bye_message = messages.get_message(get_first_name(message), 'BYE_MESSAGE')
|
|
await message.answer(bye_message, reply_markup=markup)
|
|
await state.set_state(FSM_STATES["START"])
|
|
|
|
@error_handler
|
|
async def suggest_router(self, message: types.Message, state: FSMContext, album: list = None, **kwargs):
|
|
"""Handle post submission in suggest state"""
|
|
# Post service operations with metrics
|
|
await self.user_service.update_user_activity(message.from_user.id)
|
|
await self.user_service.log_user_message(message)
|
|
await self.post_service.process_post(message, album)
|
|
|
|
# Send success message and return to start state
|
|
markup_for_user = get_reply_keyboard(self.db, message.from_user.id)
|
|
success_send_message = messages.get_message(get_first_name(message), 'SUCCESS_SEND_MESSAGE')
|
|
await message.answer(success_send_message, reply_markup=markup_for_user)
|
|
await state.set_state(FSM_STATES["START"])
|
|
|
|
@error_handler
|
|
async def stickers(self, message: types.Message, state: FSMContext, **kwargs):
|
|
"""Handle stickers request"""
|
|
# User service operations with metrics
|
|
markup = get_reply_keyboard(self.db, message.from_user.id)
|
|
self.db.update_info_about_stickers(user_id=message.from_user.id)
|
|
await self.user_service.log_user_message(message)
|
|
await message.answer(
|
|
text=ERROR_MESSAGES["STICKERS_LINK"],
|
|
reply_markup=markup
|
|
)
|
|
await state.set_state(FSM_STATES["START"])
|
|
|
|
@error_handler
|
|
async def connect_with_admin(self, message: types.Message, state: FSMContext, **kwargs):
|
|
"""Handle connect with admin button"""
|
|
# User service operations with metrics
|
|
await self.user_service.update_user_activity(message.from_user.id)
|
|
admin_message = messages.get_message(get_first_name(message), 'CONNECT_WITH_ADMIN')
|
|
await message.answer(admin_message, parse_mode="html")
|
|
await self.user_service.log_user_message(message)
|
|
await state.set_state(FSM_STATES["PRE_CHAT"])
|
|
|
|
@error_handler
|
|
async def resend_message_in_group_for_message(self, message: types.Message, state: FSMContext, **kwargs):
|
|
"""Handle messages in admin chat states"""
|
|
# User service operations with metrics
|
|
await self.user_service.update_user_activity(message.from_user.id)
|
|
await message.forward(chat_id=self.settings.group_for_message)
|
|
|
|
current_date = datetime.now()
|
|
date = current_date.strftime("%Y-%m-%d %H:%M:%S")
|
|
self.db.add_new_message_in_db(message.text, message.from_user.id, message.message_id + 1, date)
|
|
|
|
question = messages.get_message(get_first_name(message), 'QUESTION')
|
|
user_state = await state.get_state()
|
|
|
|
if user_state == FSM_STATES["PRE_CHAT"]:
|
|
markup = get_reply_keyboard(self.db, message.from_user.id)
|
|
await message.answer(question, reply_markup=markup)
|
|
await state.set_state(FSM_STATES["START"])
|
|
elif user_state == FSM_STATES["CHAT"]:
|
|
markup = get_reply_keyboard_leave_chat()
|
|
await message.answer(question, reply_markup=markup)
|
|
|
|
|
|
# Factory function to create handlers with dependencies
|
|
def create_private_handlers(db, settings: BotSettings) -> PrivateHandlers:
|
|
"""Create private handlers instance with dependencies"""
|
|
return PrivateHandlers(db, settings)
|
|
|
|
|
|
# Legacy router for backward compatibility
|
|
private_router = Router()
|
|
|
|
# Initialize with global dependencies (for backward compatibility)
|
|
def init_legacy_router():
|
|
"""Initialize legacy router with global dependencies"""
|
|
global private_router
|
|
|
|
from helper_bot.utils.base_dependency_factory import get_global_instance
|
|
|
|
bdf = get_global_instance()
|
|
settings = BotSettings(
|
|
group_for_posts=bdf.settings['Telegram']['group_for_posts'],
|
|
group_for_message=bdf.settings['Telegram']['group_for_message'],
|
|
main_public=bdf.settings['Telegram']['main_public'],
|
|
group_for_logs=bdf.settings['Telegram']['group_for_logs'],
|
|
important_logs=bdf.settings['Telegram']['important_logs'],
|
|
preview_link=bdf.settings['Telegram']['preview_link'],
|
|
logs=bdf.settings['Settings']['logs'],
|
|
test=bdf.settings['Settings']['test']
|
|
)
|
|
|
|
db = bdf.get_db()
|
|
handlers = create_private_handlers(db, settings)
|
|
|
|
# Instead of trying to copy handlers, we'll use the new router directly
|
|
# This maintains backward compatibility while using the new architecture
|
|
private_router = handlers.router
|
|
|
|
# Initialize legacy router
|
|
init_legacy_router()
|