import logging from aiogram import Router, F, types from typing import Optional, Dict, List from datetime import datetime import html from sqlalchemy.ext.asyncio import AsyncSession from config.settings import Settings from db.dal import user_dal, payment_dal, panel_sync_dal from db.models import Payment, PanelSyncStatus from bot.services.panel_api_service import PanelApiService from bot.keyboards.inline.admin_keyboards import ( get_back_to_admin_panel_keyboard, get_back_to_user_management_keyboard, ) from bot.middlewares.i18n import JsonI18n router = Router(name="admin_statistics_router") def _format_rating_user_label(user_row: Dict[str, object], bot_username: Optional[str] = None) -> str: user_id = int(user_row.get("user_id", 0) or 0) username = user_row.get("username") first_name = user_row.get("first_name") user_id_text = str(user_id) user_id_html = html.escape(user_id_text) if bot_username: safe_bot_username = html.escape(bot_username) user_id_html = ( f'' f"{user_id_html}" ) parts: List[str] = [] if username: parts.append(f"@{html.escape(str(username))}") elif first_name: parts.append(html.escape(str(first_name))) if not parts: parts.append(f"ID {user_id_html}") else: parts.append(f"(ID {user_id_html})") return " ".join(parts) async def show_statistics_handler(callback: types.CallbackQuery, i18n_data: dict, settings: Settings, session: AsyncSession): current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE) i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") if not i18n or not callback.message: await callback.answer("Error displaying statistics.", show_alert=True) return _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) await callback.answer() stats_text_parts = [f"{_('admin_stats_header')}"] # Enhanced user statistics user_stats = await user_dal.get_enhanced_user_statistics(session) stats_text_parts.append( f"\n👥 {_('admin_enhanced_users_stats_header')}" ) stats_text_parts.append( f"📊 {_('admin_user_stats_total_label')}: {user_stats['total_users']}" ) # Removed: Active today moved to panel stats stats_text_parts.append( f"💳 {_('admin_user_stats_paid_subs_label')}: {user_stats['paid_subscriptions']}" ) stats_text_parts.append( f"🆓 {_('admin_user_stats_trial_label')}: {user_stats['trial_users']}" ) stats_text_parts.append( f"😴 {_('admin_user_stats_inactive_label')}: {user_stats['inactive_users']}" ) stats_text_parts.append( f"🚫 {_('admin_user_stats_banned_label')}: {user_stats['banned_users']}" ) stats_text_parts.append( f"🎁 {_('admin_user_stats_referral_label')}: {user_stats['referral_users']}" ) # Panel Statistics - moved above financial stats_text_parts.append(f"\n🖥 {_('admin_panel_stats_header')}") try: async with PanelApiService(settings) as panel_service: # Get system stats system_stats = await panel_service.get_system_stats() bandwidth_stats = await panel_service.get_bandwidth_stats() nodes_stats = await panel_service.get_nodes_statistics() logging.info(f"Panel stats response: system={system_stats}, bandwidth={bandwidth_stats}, nodes={nodes_stats}") if system_stats: users = system_stats.get('users', {}) status_counts = users.get('statusCounts', {}) online_stats = system_stats.get('onlineStats', {}) active_users = status_counts.get('ACTIVE', 0) disabled_users = status_counts.get('DISABLED', 0) expired_users = status_counts.get('EXPIRED', 0) limited_users = status_counts.get('LIMITED', 0) total_users = users.get('totalUsers', 0) online_now = online_stats.get('onlineNow', 0) stats_text_parts.append(f"🟢 {_('admin_panel_online_label')}: {online_now}") stats_text_parts.append(f"📊 {_('admin_panel_active_label')}: {active_users}") stats_text_parts.append(f"🔴 {_('admin_panel_disabled_label')}: {disabled_users}") stats_text_parts.append(f"⏰ {_('admin_panel_expired_label')}: {expired_users}") stats_text_parts.append(f"⚠️ {_('admin_panel_limited_label')}: {limited_users}") stats_text_parts.append(f"👥 {_('admin_panel_total_users_label')}: {total_users}") # System resources memory = system_stats.get('memory', {}) if memory: memory_total = memory.get('total', 1) memory_used = memory.get('used', 0) memory_usage = (memory_used / memory_total) * 100 if memory_total > 0 else 0 stats_text_parts.append(f"💾 {_('admin_panel_memory_usage_label')}: {memory_usage:.1f}%") else: stats_text_parts.append(f"⚠️ {_('admin_panel_system_stats_error')}") # Bandwidth stats if bandwidth_stats: week_traffic = bandwidth_stats.get('bandwidthLastSevenDays', {}) month_traffic = bandwidth_stats.get('bandwidthLast30Days', {}) # Fallback to the actual key name from API if the above doesn't exist if not month_traffic: month_traffic = bandwidth_stats.get('bandwidthLastThirtyDays', {}) if week_traffic: week_total = week_traffic.get('current', '0 B') stats_text_parts.append(f"📊 {_('admin_panel_traffic_week_label')}: {week_total}") if month_traffic: month_total = month_traffic.get('current', '0 B') stats_text_parts.append(f"📊 {_('admin_panel_traffic_month_label')}: {month_total}") else: stats_text_parts.append(f"⚠️ {_('admin_panel_bandwidth_stats_error')}") # Nodes stats if nodes_stats and 'lastSevenDays' in nodes_stats: last_seven_days = nodes_stats.get('lastSevenDays', []) # Get unique node names from the data unique_nodes = set() for node_data in last_seven_days: unique_nodes.add(node_data.get('nodeName', '')) total_nodes_count = len(unique_nodes) # Assume all nodes are active since we don't have status info stats_text_parts.append(f"🔗 {_('admin_panel_nodes_label')}: {total_nodes_count}/{total_nodes_count}") else: # Use nodes total from system stats as fallback nodes_info = system_stats.get('nodes', {}) if system_stats else {} total_online = nodes_info.get('totalOnline', 0) stats_text_parts.append(f"🔗 {_('admin_panel_nodes_label')}: {total_online}") except Exception as e: logging.error(f"Failed to fetch panel statistics: {e}", exc_info=True) stats_text_parts.append(f"❌ {_('admin_panel_stats_fetch_error')}") stats_text_parts.append(f"⚠️ {_('admin_panel_stats_error_details')}: {str(e)}") # Financial statistics financial_stats = await payment_dal.get_financial_statistics(session) stats_text_parts.append( f"\n💰 {_('admin_financial_stats_header')}" ) stats_text_parts.append( f"📅 {_('admin_financial_today_label')}: {financial_stats['today_revenue']:.2f} RUB ({financial_stats['today_payments_count']} {_('admin_financial_payments_label')})" ) stats_text_parts.append( f"📅 {_('admin_financial_week_label')}: {financial_stats['week_revenue']:.2f} RUB" ) stats_text_parts.append( f"📅 {_('admin_financial_month_label')}: {financial_stats['month_revenue']:.2f} RUB" ) stats_text_parts.append( f"🏆 {_('admin_financial_all_time_label')}: {financial_stats['all_time_revenue']:.2f} RUB" ) last_payments_models: List[ Payment] = await payment_dal.get_recent_payment_logs_with_user(session, limit=5) if last_payments_models: stats_text_parts.append( f"\n{_('admin_stats_recent_payments_header')}") for payment in last_payments_models: pending_statuses = [ "pending", "pending_yookassa", "pending_freekassa", "pending_platega", "pending_severpay", "pending_cryptopay", ] status_emoji = ( "✅" if payment.status == "succeeded" else "⏳" if payment.status in pending_statuses else "❌" ) user_info = f"User {payment.user_id}" if payment.user and payment.user.username: user_info += f" (@{payment.user.username})" elif payment.user and payment.user.first_name: user_info += f" ({payment.user.first_name})" payment_date_str = payment.created_at.strftime( '%Y-%m-%d') if payment.created_at else "N/A" stats_text_parts.append( _("admin_stats_payment_item", status_emoji=status_emoji, amount=payment.amount, currency=payment.currency, user_info=user_info, p_status=payment.status, p_date=payment_date_str)) else: stats_text_parts.append(f"\n{_('admin_stats_no_payments_found')}") sync_status_model: Optional[ PanelSyncStatus] = await panel_sync_dal.get_panel_sync_status(session) if sync_status_model and sync_status_model.status != "never_run": stats_text_parts.append( f"\n{_('admin_stats_last_sync_header')}") sync_time_val = sync_status_model.last_sync_time sync_time_str = sync_time_val.strftime( '%Y-%m-%d %H:%M:%S UTC') if sync_time_val else "N/A" details_val = sync_status_model.details details_str = details_val or "N/A" stats_text_parts.append( f" {_('admin_stats_sync_time')}: {sync_time_str}") stats_text_parts.append( f" {_('admin_stats_sync_status')}: {sync_status_model.status}") stats_text_parts.append( f" {_('admin_stats_sync_users_processed')}: {sync_status_model.users_processed_from_panel}" ) stats_text_parts.append( f" {_('admin_stats_sync_subs_synced')}: {sync_status_model.subscriptions_synced}" ) stats_text_parts.append( f" {_('admin_stats_sync_details_label')}: {details_str}") else: stats_text_parts.append(f"\n{_('admin_sync_status_never_run')}") final_text = "\n".join(stats_text_parts) try: await callback.message.edit_text( final_text, reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n), parse_mode="HTML") except Exception as e_edit: logging.error(f"Error editing message for statistics: {e_edit}", exc_info=True) max_chunk_size = 4000 for i in range(0, len(final_text), max_chunk_size): chunk = final_text[i:i + max_chunk_size] is_last_chunk = (i + max_chunk_size) >= len(final_text) try: await callback.message.answer( chunk, reply_markup=get_back_to_admin_panel_keyboard( current_lang, i18n) if is_last_chunk else None, parse_mode="HTML") except Exception as e_chunk: logging.error(f"Failed to send statistics chunk: {e_chunk}") if i == 0: await callback.message.answer( _("error_displaying_statistics"), reply_markup=get_back_to_admin_panel_keyboard( current_lang, i18n)) break async def show_user_ratings_handler( callback: types.CallbackQuery, i18n_data: dict, settings: Settings, session: AsyncSession, ): current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE) i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") if not i18n or not callback.message: await callback.answer("Error displaying ratings.", show_alert=True) return _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) await callback.answer() top_limit = 10 bot_username: Optional[str] = None try: me = await callback.bot.get_me() bot_username = me.username except Exception as e_get_me: logging.warning("Failed to resolve bot username for ratings links: %s", e_get_me) traffic_top = await user_dal.get_top_users_by_traffic_used(session, limit=top_limit) lifetime_traffic_top = await user_dal.get_top_users_by_lifetime_traffic_used( session, limit=top_limit ) invited_top = await user_dal.get_top_users_by_referrals_count(session, limit=top_limit) revenue_top = await user_dal.get_top_users_by_referral_revenue(session, limit=top_limit) text_parts: List[str] = [ _("admin_user_ratings_header", top_limit=top_limit), "", f"{_('admin_user_ratings_traffic_month_title')}", ] if traffic_top: for idx, row in enumerate(traffic_top, start=1): traffic_gb = float(row.get("traffic_used_bytes") or 0) / (1024**3) text_parts.append( _( "admin_user_ratings_traffic_item", rank=idx, user=_format_rating_user_label(row, bot_username), traffic_gb=f"{traffic_gb:.2f}", ) ) else: text_parts.append(_("admin_user_ratings_empty")) text_parts.extend(["", f"{_('admin_user_ratings_traffic_lifetime_title')}"]) if lifetime_traffic_top: for idx, row in enumerate(lifetime_traffic_top, start=1): traffic_gb = float(row.get("lifetime_used_traffic_bytes") or 0) / (1024**3) text_parts.append( _( "admin_user_ratings_traffic_item", rank=idx, user=_format_rating_user_label(row, bot_username), traffic_gb=f"{traffic_gb:.2f}", ) ) else: text_parts.append(_("admin_user_ratings_empty")) text_parts.extend(["", f"{_('admin_user_ratings_invited_title')}"]) if invited_top: for idx, row in enumerate(invited_top, start=1): text_parts.append( _( "admin_user_ratings_invited_item", rank=idx, user=_format_rating_user_label(row, bot_username), invited_count=int(row.get("invited_count") or 0), ) ) else: text_parts.append(_("admin_user_ratings_empty")) text_parts.extend(["", f"{_('admin_user_ratings_revenue_title')}"]) if revenue_top: for idx, row in enumerate(revenue_top, start=1): text_parts.append( _( "admin_user_ratings_revenue_item", rank=idx, user=_format_rating_user_label(row, bot_username), revenue=f"{float(row.get('referral_revenue') or 0):.2f}", ) ) else: text_parts.append(_("admin_user_ratings_empty")) await callback.message.edit_text( "\n".join(text_parts), reply_markup=get_back_to_user_management_keyboard(current_lang, i18n), parse_mode="HTML", )