import logging import math import re from aiogram import Router, F, types, Bot from aiogram.fsm.context import FSMContext from typing import Optional, List, Dict, Any import aiosqlite from config.settings import Settings from db.database import (get_all_message_logs_paginated, count_all_message_logs, get_user_message_logs_paginated, count_user_message_logs, get_user, get_user_by_telegram_username) from bot.states.admin_states import AdminStates from bot.keyboards.inline.admin_keyboards import ( get_logs_menu_keyboard, get_logs_pagination_keyboard, get_back_to_admin_panel_keyboard, get_admin_panel_keyboard) from bot.middlewares.i18n import JsonI18n router = Router(name="admin_logs_router") USERNAME_REGEX = re.compile(r"^[a-zA-Z0-9_]{5,32}$") async def display_logs_menu(callback: types.CallbackQuery, i18n_data: dict, settings: Settings): current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE) i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") if not i18n: logging.error("i18n_instance missing in display_logs_menu") await callback.answer("Language service error.", show_alert=True) return if not callback.message: logging.error("CallbackQuery has no message in display_logs_menu") await callback.answer("Error processing request.", show_alert=True) return _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) try: await callback.message.edit_text(text=_(key="admin_logs_menu_title"), reply_markup=get_logs_menu_keyboard( i18n, current_lang)) except Exception as e: logging.warning(f"Failed to edit message for logs menu: {e}") await callback.message.answer(text=_(key="admin_logs_menu_title"), reply_markup=get_logs_menu_keyboard( i18n, current_lang)) await callback.answer() async def _display_formatted_logs(target_message: types.Message, logs: List[aiosqlite.Row], total_logs: int, current_page: int, settings: Settings, title_key: str, base_pagination_callback_data: str, i18n: JsonI18n, current_lang: str, title_kwargs: Optional[Dict[str, Any]] = None): _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) page_size = settings.LOGS_PAGE_SIZE actual_title_kwargs = title_kwargs or {} if not logs and total_logs == 0: text = _( title_key, current_page=1, total_pages=1, ** actual_title_kwargs) + "\n\n" + _("admin_no_logs_found") reply_markup = get_logs_pagination_keyboard( current_page, 1, base_pagination_callback_data, i18n, current_lang, back_to_logs_menu=True) else: total_pages = math.ceil(total_logs / page_size) if page_size > 0 else 1 text = _(title_key, current_page=current_page + 1, total_pages=max(1, total_pages), **actual_title_kwargs) + "\n" log_entries_text = [] for log_entry in logs: user_display_parts = [] telegram_first_name = log_entry[ 'telegram_first_name'] if 'telegram_first_name' in log_entry.keys( ) and log_entry['telegram_first_name'] else None telegram_username = log_entry[ 'telegram_username'] if 'telegram_username' in log_entry.keys( ) and log_entry['telegram_username'] else None user_id_from_log = log_entry[ 'user_id'] if 'user_id' in log_entry.keys( ) and log_entry['user_id'] else None if telegram_first_name: user_display_parts.append(telegram_first_name) if telegram_username: user_display_parts.append(f"(@{telegram_username})") user_display = " ".join(user_display_parts).strip() if not user_display: user_display = _( "system_or_unknown_user" ) if not user_id_from_log else f"ID: {user_id_from_log}" user_id_display = str( user_id_from_log) if user_id_from_log is not None else "N/A" content_raw = log_entry['content'] if 'content' in log_entry.keys( ) and log_entry['content'] else "" content_preview = (content_raw[:100] + "...") if len(content_raw) > 100 else ( content_raw or "N/A") log_entries_text.append( _("admin_log_entry_format", timestamp_str=log_entry['timestamp_str'] if 'timestamp_str' in log_entry.keys() else 'N/A', user_display=user_display, user_id=user_id_display, event_type=log_entry['event_type'] if 'event_type' in log_entry.keys() else 'N/A', content_preview=content_preview).replace("\n", "\n ")) text += "\n\n".join(log_entries_text) reply_markup = get_logs_pagination_keyboard( current_page, total_pages, base_pagination_callback_data, i18n, current_lang, back_to_logs_menu=True) try: await target_message.edit_text(text, reply_markup=reply_markup, parse_mode="HTML", disable_web_page_preview=True) except Exception as e: logging.warning( f"Failed to edit message for logs display: {e}. Content length: {len(text)}" ) chunk_size = 4000 for i in range(0, len(text), chunk_size): chunk = text[i:i + chunk_size] is_last_chunk = (i + chunk_size) >= len(text) await target_message.answer( chunk, reply_markup=reply_markup if is_last_chunk else None, parse_mode="HTML", disable_web_page_preview=True) @router.callback_query(F.data.startswith("admin_logs:view_all")) async def view_all_logs_handler(callback: types.CallbackQuery, settings: Settings, i18n_data: dict): page = 0 parts = callback.data.split(":") if len(parts) == 3: try: page = int(parts[2]) except ValueError: page = 0 i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE) if not i18n or not callback.message: await callback.answer("Error processing request.", show_alert=True) return logs, total_logs = await get_all_message_logs_paginated( settings.LOGS_PAGE_SIZE, page * settings.LOGS_PAGE_SIZE), await count_all_message_logs() await _display_formatted_logs( target_message=callback.message, logs=logs, total_logs=total_logs, current_page=page, settings=settings, title_key="admin_all_logs_title", base_pagination_callback_data="admin_logs:view_all", i18n=i18n, current_lang=current_lang) await callback.answer() @router.callback_query(F.data == "admin_logs:prompt_user") async def prompt_user_for_logs_handler(callback: types.CallbackQuery, state: FSMContext, i18n_data: dict, settings: Settings): i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE) if not i18n or not callback.message: await callback.answer("Error") return _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) await callback.message.edit_text( text=_("admin_prompt_for_user_id_or_username_logs"), reply_markup=get_logs_menu_keyboard(i18n, current_lang)) await state.set_state(AdminStates.waiting_for_user_id_for_logs) await callback.answer() @router.message(AdminStates.waiting_for_user_id_for_logs, F.text) async def process_user_id_for_logs_handler(message: types.Message, state: FSMContext, settings: Settings, i18n_data: dict): current_state_fsm = await state.get_state() logging.info( f"Processing user input for logs in state {current_state_fsm}: '{message.text}'" ) await state.clear() i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE) if not i18n: await message.reply("Language error.") return _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) input_text = message.text.strip() user_data_for_logs: Optional[aiosqlite.Row] = None if input_text.isdigit(): try: user_data_for_logs = await get_user(int(input_text)) except ValueError: pass elif input_text.startswith("@") and USERNAME_REGEX.match(input_text[1:]): user_data_for_logs = await get_user_by_telegram_username(input_text[1:] ) elif USERNAME_REGEX.match(input_text): user_data_for_logs = await get_user_by_telegram_username(input_text) if not user_data_for_logs: await message.answer(_("admin_log_user_not_found", input=input_text)) return target_user_id = user_data_for_logs['user_id'] user_display = user_data_for_logs['first_name'] or ( f"@{user_data_for_logs['username']}" if user_data_for_logs.get('username') else f"ID {target_user_id}") logs, total_logs = await get_user_message_logs_paginated( target_user_id, settings.LOGS_PAGE_SIZE, 0), await count_user_message_logs(target_user_id) await _display_formatted_logs( target_message=message, logs=logs, total_logs=total_logs, current_page=0, settings=settings, title_key="admin_user_logs_title", base_pagination_callback_data=f"admin_logs:view_user:{target_user_id}", i18n=i18n, current_lang=current_lang, title_kwargs={"user_display": user_display}) @router.callback_query(F.data.startswith("admin_logs:view_user:")) async def view_user_logs_paginated_handler(callback: types.CallbackQuery, settings: Settings, i18n_data: dict): try: parts = callback.data.split(":") target_user_id = int(parts[2]) page = int(parts[3]) except (IndexError, ValueError): await callback.answer("Invalid log request.", show_alert=True) return i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE) if not i18n or not callback.message: await callback.answer("Error") return user_data_for_logs = await get_user(target_user_id) if not user_data_for_logs: await callback.message.edit_text("User not found for logs.") await callback.answer() return user_display = user_data_for_logs['first_name'] or ( f"@{user_data_for_logs['username']}" if user_data_for_logs.get('username') else f"ID {target_user_id}") logs, total_logs = await get_user_message_logs_paginated( target_user_id, settings.LOGS_PAGE_SIZE, page * settings.LOGS_PAGE_SIZE), await count_user_message_logs(target_user_id) await _display_formatted_logs( target_message=callback.message, logs=logs, total_logs=total_logs, current_page=page, settings=settings, title_key="admin_user_logs_title", base_pagination_callback_data=f"admin_logs:view_user:{target_user_id}", i18n=i18n, current_lang=current_lang, title_kwargs={"user_display": user_display}) await callback.answer() @router.callback_query(F.data == "admin_action:view_logs_menu", AdminStates.waiting_for_user_id_for_logs) async def cancel_log_user_input_state_to_menu(callback: types.CallbackQuery, state: FSMContext, settings: Settings, i18n_data: dict): await state.clear() await display_logs_menu(callback, i18n_data, settings)