"""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 # 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""" def user_exists(self, user_id: int) -> bool: ... def add_new_user_in_db(self, user_id: int, first_name: str, full_name: str, username: str, is_bot: bool, language_code: str, emoji: str, created_date: str, updated_date: str) -> None: ... def update_username_and_full_name(self, user_id: int, username: str, full_name: str) -> None: ... def update_date_for_user(self, date: str, user_id: int) -> None: ... def add_post_in_db(self, message_id: int, text: str, user_id: int) -> None: ... def update_info_about_stickers(self, user_id: int) -> None: ... def add_new_message_in_db(self, text: str, user_id: int, message_id: int, date: str) -> None: ... def update_helper_message_in_db(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", "users", "update") async def update_user_activity(self, user_id: int) -> None: """Update user's last activity timestamp with metrics tracking""" current_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S") self.db.update_date_for_user(current_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 current_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S") if not self.db.user_exists(user_id): # Record database operation self.db.add_new_user_in_db( user_id, first_name, full_name, username, is_bot, language_code, "", current_date, current_date ) metrics.record_db_query("add_new_user", 0.0, "users", "insert") else: is_need_update = check_username_and_full_name(user_id, username, full_name, self.db) if is_need_update: self.db.update_username_and_full_name(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}') self.db.update_date_for_user(current_date, user_id) metrics.record_db_query("update_date_for_user", 0.0, "users", "update") @track_time("log_user_message", "user_service") @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 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) self.db.add_post_in_db(sent_message_id, message.text, message.from_user.id) 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 ) self.db.add_post_in_db(sent_message.message_id, sent_message.caption, message.from_user.id) await add_in_db_media(sent_message, self.db) 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 ) self.db.add_post_in_db(sent_message.message_id, sent_message.caption, message.from_user.id) await add_in_db_media(sent_message, self.db) 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 ) self.db.add_post_in_db(sent_message.message_id, sent_message.caption, message.from_user.id) await add_in_db_media(sent_message, self.db) 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 ) self.db.add_post_in_db(sent_message.message_id, sent_message.caption, message.from_user.id) await add_in_db_media(sent_message, self.db) 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 ) self.db.add_post_in_db(sent_message.message_id, sent_message.caption, message.from_user.id) await add_in_db_media(sent_message, self.db) 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) 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) markup = get_reply_keyboard_for_post() help_message_id = await send_text_message(self.settings.group_for_posts, message, "^", markup) self.db.update_helper_message_in_db( message_id=media_group_message_id, helper_message_id=help_message_id ) 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) 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)