"""Service classes for private handlers""" # Standard library imports import random import asyncio import html from datetime import datetime from pathlib import Path from typing import Dict, Callable, Any, Protocol, Union from dataclasses import dataclass # Third-party imports from aiogram import types from aiogram.types import FSInputFile from database.models import TelegramPost, User # Local imports - utilities from helper_bot.utils.helper_func import ( get_first_name, get_text_message, send_text_message, send_photo_message, send_media_group_message_to_private_chat, prepare_media_group_from_middlewares, send_video_message, send_video_note_message, send_audio_message, send_voice_message, add_in_db_media, check_username_and_full_name ) from helper_bot.keyboards import get_reply_keyboard_for_post # Local imports - metrics from helper_bot.utils.metrics import ( metrics, track_time, track_errors, db_query_time ) class DatabaseProtocol(Protocol): """Protocol for database operations""" async def user_exists(self, user_id: int) -> bool: ... async def add_user(self, user: User) -> None: ... async def update_user_info(self, user_id: int, username: str = None, full_name: str = None) -> None: ... async def update_user_date(self, user_id: int) -> None: ... async def add_post(self, post: TelegramPost) -> None: ... async def update_stickers_info(self, user_id: int) -> None: ... async def add_message(self, message_text: str, user_id: int, message_id: int, date: int = None) -> None: ... async def update_helper_message(self, message_id: int, helper_message_id: int) -> None: ... @dataclass class BotSettings: """Bot configuration settings""" group_for_posts: str group_for_message: str main_public: str group_for_logs: str important_logs: str preview_link: str logs: str test: str class UserService: """Service for user-related operations""" def __init__(self, db: DatabaseProtocol, settings: BotSettings) -> None: self.db = db self.settings = settings @track_time("update_user_activity", "user_service") @track_errors("user_service", "update_user_activity") @db_query_time("update_user_activity", "user_service") async def update_user_activity(self, user_id: int) -> None: """Update user's last activity timestamp with metrics tracking""" await self.db.update_user_date(user_id) @track_time("ensure_user_exists", "user_service") @track_errors("user_service", "ensure_user_exists") async def ensure_user_exists(self, message: types.Message) -> None: """Ensure user exists in database, create if needed with metrics tracking""" user_id = message.from_user.id full_name = message.from_user.full_name username = message.from_user.username or "private_username" first_name = get_first_name(message) is_bot = message.from_user.is_bot language_code = message.from_user.language_code if not await self.db.user_exists(user_id): # Create User object with current timestamp current_timestamp = int(datetime.now().timestamp()) user = User( user_id=user_id, first_name=first_name, full_name=full_name, username=username, is_bot=is_bot, language_code=language_code, emoji="", has_stickers=False, date_added=current_timestamp, date_changed=current_timestamp, voice_bot_welcome_received=False ) await self.db.add_user(user) metrics.record_db_query("add_user", 0.0, "users", "insert") else: is_need_update = await check_username_and_full_name(user_id, username, full_name, self.db) if is_need_update: await self.db.update_user_info(user_id, username, full_name) metrics.record_db_query("update_username_fullname", 0.0, "users", "update") safe_full_name = html.escape(full_name) if full_name else "Неизвестный пользователь" safe_username = html.escape(username) if username else "Без никнейма" await message.answer( f"Давно не виделись! Вижу что ты изменился;) Теперь буду звать тебя: {safe_full_name} и ник @{safe_username}") await message.bot.send_message( chat_id=self.settings.group_for_logs, text=f'Для пользователя: {user_id} обновлены данные в БД.\nНовое имя: {safe_full_name}\nНовый ник:{safe_username}') await self.db.update_user_date(user_id) metrics.record_db_query("update_user_date", 0.0, "users", "update") @track_errors("user_service", "log_user_message") async def log_user_message(self, message: types.Message) -> None: """Forward user message to logs group with metrics tracking""" await message.forward(chat_id=self.settings.group_for_logs) def get_safe_user_info(self, message: types.Message) -> tuple[str, str]: """Get safely escaped user information for logging""" full_name = message.from_user.full_name or "Неизвестный пользователь" username = message.from_user.username or "Без никнейма" return html.escape(full_name), html.escape(username) class PostService: """Service for post-related operations""" def __init__(self, db: DatabaseProtocol, settings: BotSettings) -> None: self.db = db self.settings = settings @track_time("handle_text_post", "post_service") @track_errors("post_service", "handle_text_post") async def handle_text_post(self, message: types.Message, first_name: str) -> None: """Handle text post submission""" post_text = get_text_message(message.text.lower(), first_name, message.from_user.username) markup = get_reply_keyboard_for_post() sent_message_id = await send_text_message(self.settings.group_for_posts, message, post_text, markup) post = TelegramPost( message_id=sent_message_id, text=message.text, author_id=message.from_user.id, created_at=int(datetime.now().timestamp()) ) await self.db.add_post(post) @track_time("handle_photo_post", "post_service") @track_errors("post_service", "handle_photo_post") async def handle_photo_post(self, message: types.Message, first_name: str) -> None: """Handle photo post submission""" post_caption = "" if message.caption: post_caption = get_text_message(message.caption.lower(), first_name, message.from_user.username) markup = get_reply_keyboard_for_post() sent_message = await send_photo_message( self.settings.group_for_posts, message, message.photo[-1].file_id, post_caption, markup ) post = TelegramPost( message_id=sent_message.message_id, text=sent_message.caption or "", author_id=message.from_user.id, created_at=int(datetime.now().timestamp()) ) await self.db.add_post(post) await add_in_db_media(sent_message, self.db) @track_time("handle_video_post", "post_service") @track_errors("post_service", "handle_video_post") async def handle_video_post(self, message: types.Message, first_name: str) -> None: """Handle video post submission""" post_caption = "" if message.caption: post_caption = get_text_message(message.caption.lower(), first_name, message.from_user.username) markup = get_reply_keyboard_for_post() sent_message = await send_video_message( self.settings.group_for_posts, message, message.video.file_id, post_caption, markup ) post = TelegramPost( message_id=sent_message.message_id, text=sent_message.caption or "", author_id=message.from_user.id, created_at=int(datetime.now().timestamp()) ) await self.db.add_post(post) await add_in_db_media(sent_message, self.db) @track_time("handle_video_note_post", "post_service") @track_errors("post_service", "handle_video_note_post") async def handle_video_note_post(self, message: types.Message) -> None: """Handle video note post submission""" markup = get_reply_keyboard_for_post() sent_message = await send_video_note_message( self.settings.group_for_posts, message, message.video_note.file_id, markup ) post = TelegramPost( message_id=sent_message.message_id, text=sent_message.caption or "", author_id=message.from_user.id, created_at=int(datetime.now().timestamp()) ) await self.db.add_post(post) await add_in_db_media(sent_message, self.db) @track_time("handle_audio_post", "post_service") @track_errors("post_service", "handle_audio_post") async def handle_audio_post(self, message: types.Message, first_name: str) -> None: """Handle audio post submission""" post_caption = "" if message.caption: post_caption = get_text_message(message.caption.lower(), first_name, message.from_user.username) markup = get_reply_keyboard_for_post() sent_message = await send_audio_message( self.settings.group_for_posts, message, message.audio.file_id, post_caption, markup ) post = TelegramPost( message_id=sent_message.message_id, text=sent_message.caption or "", author_id=message.from_user.id, created_at=int(datetime.now().timestamp()) ) await self.db.add_post(post) await add_in_db_media(sent_message, self.db) @track_time("handle_voice_post", "post_service") @track_errors("post_service", "handle_voice_post") async def handle_voice_post(self, message: types.Message) -> None: """Handle voice post submission""" markup = get_reply_keyboard_for_post() sent_message = await send_voice_message( self.settings.group_for_posts, message, message.voice.file_id, markup ) post = TelegramPost( message_id=sent_message.message_id, text=sent_message.caption or "", author_id=message.from_user.id, created_at=int(datetime.now().timestamp()) ) await self.db.add_post(post) await add_in_db_media(sent_message, self.db) @track_time("handle_media_group_post", "post_service") @track_errors("post_service", "handle_media_group_post") async def handle_media_group_post(self, message: types.Message, album: list, first_name: str) -> None: """Handle media group post submission""" post_caption = " " if album and album[0].caption: post_caption = get_text_message(album[0].caption.lower(), first_name, message.from_user.username) # Создаем основной пост для медиагруппы main_post = TelegramPost( message_id=message.message_id, # ID основного сообщения медиагруппы text=post_caption, author_id=message.from_user.id, created_at=int(datetime.now().timestamp()) ) await self.db.add_post(main_post) # Отправляем медиагруппу в группу для модерации media_group = await prepare_media_group_from_middlewares(album, post_caption) media_group_message_id = await send_media_group_message_to_private_chat( self.settings.group_for_posts, message, media_group, self.db ) await asyncio.sleep(0.2) # Создаем helper сообщение с кнопками markup = get_reply_keyboard_for_post() help_message_id = await send_text_message(self.settings.group_for_posts, message, "ВРУЧНУЮ ВЫКЛАДЫВАТЬ, ПОСЛЕ ВЫКЛАДКИ УДАЛИТЬ ОБА ПОСТА") # Создаем helper пост и связываем его с основным helper_post = TelegramPost( message_id=help_message_id, # ID helper сообщения text="^", # Специальный маркер для медиагруппы author_id=message.from_user.id, helper_text_message_id=main_post.message_id, # Ссылка на основной пост created_at=int(datetime.now().timestamp()) ) await self.db.add_post(helper_post) # Обновляем основной пост, чтобы он ссылался на helper await self.db.update_helper_message( message_id=main_post.message_id, helper_message_id=help_message_id ) @track_time("process_post", "post_service") @track_errors("post_service", "process_post") async def process_post(self, message: types.Message, album: Union[list, None] = None) -> None: """Process post based on content type""" first_name = get_first_name(message) # TODO: Бесит меня этот функционал if message.media_group_id is not None: safe_username = html.escape(message.from_user.username) if message.from_user.username else "Без никнейма" await send_text_message( self.settings.group_for_logs, message, f'Закинул медиагруппу, пользователь: имя - {first_name}, ник - {safe_username}' ) await self.handle_media_group_post(message, album, first_name) return content_handlers: Dict[str, Callable] = { 'text': lambda: self.handle_text_post(message, first_name), 'photo': lambda: self.handle_photo_post(message, first_name), 'video': lambda: self.handle_video_post(message, first_name), 'video_note': lambda: self.handle_video_note_post(message), 'audio': lambda: self.handle_audio_post(message, first_name), 'voice': lambda: self.handle_voice_post(message) } handler = content_handlers.get(message.content_type) if handler: await handler() else: from .constants import ERROR_MESSAGES await message.bot.send_message( message.chat.id, ERROR_MESSAGES["UNSUPPORTED_CONTENT"] ) class StickerService: """Service for sticker-related operations""" def __init__(self, settings: BotSettings) -> None: self.settings = settings @track_time("send_random_hello_sticker", "sticker_service") @track_errors("sticker_service", "send_random_hello_sticker") async def send_random_hello_sticker(self, message: types.Message) -> None: """Send random hello sticker with metrics tracking""" name_stick_hello = list(Path('Stick').rglob('Hello_*')) if not name_stick_hello: return random_stick_hello = random.choice(name_stick_hello) random_stick_hello = FSInputFile(path=random_stick_hello) await message.answer_sticker(random_stick_hello) await asyncio.sleep(0.3) @track_time("send_random_goodbye_sticker", "sticker_service") @track_errors("sticker_service", "send_random_goodbye_sticker") async def send_random_goodbye_sticker(self, message: types.Message) -> None: """Send random goodbye sticker with metrics tracking""" name_stick_bye = list(Path('Stick').rglob('Universal_*')) if not name_stick_bye: return random_stick_bye = random.choice(name_stick_bye) random_stick_bye = FSInputFile(path=random_stick_bye) await message.answer_sticker(random_stick_bye)