From 57e693fa3762265474339ffbc978149cab5c6c43 Mon Sep 17 00:00:00 2001 From: machka-pasla Date: Wed, 6 Aug 2025 22:26:07 +0300 Subject: [PATCH 1/8] Refactor synchronization messaging for clarity and simplicity - Updated synchronization messages to provide simpler, more concise feedback to admins during the sync process. - Replaced detailed sync status messages with straightforward notifications for success, failure, and errors. - Enhanced localization for new message formats to improve user experience across languages. --- bot/handlers/admin/sync_admin.py | 26 ++++++-------------------- locales/en.json | 10 +++++----- locales/ru.json | 10 +++++----- 3 files changed, 16 insertions(+), 30 deletions(-) diff --git a/bot/handlers/admin/sync_admin.py b/bot/handlers/admin/sync_admin.py index c73139a..0f33482 100644 --- a/bot/handlers/admin/sync_admin.py +++ b/bot/handlers/admin/sync_admin.py @@ -265,7 +265,7 @@ async def sync_command_handler( return if isinstance(message_event, types.Message): - await message_event.answer(_("sync_started")) + await message_event.answer(_("sync_started_simple")) logging.info(f"Admin ({message_event.from_user.id}) triggered panel sync.") @@ -277,27 +277,13 @@ async def sync_command_handler( details = sync_result.get("details", "No details available") errors = sync_result.get("errors", []) + # Simple confirmation message to admin if status == "failed": - await bot.send_message(target_chat_id, _("sync_failed", details=details)) + await bot.send_message(target_chat_id, _("sync_failed_simple")) elif status == "completed_with_errors": - error_preview = "; ".join(errors[:3]) # Show first 3 errors - final_message = _( - "sync_completed_with_errors_details", - total_checked=sync_result.get("users_processed", 0), - users_synced=sync_result.get("users_synced", 0), - subs_synced=sync_result.get("subs_synced", 0), - errors_count=len(errors), - error_details_preview=error_preview - ) - await bot.send_message(target_chat_id, final_message) + await bot.send_message(target_chat_id, _("sync_errors_simple", errors_count=len(errors))) else: - final_message = _( - "sync_completed_details", - total_checked=sync_result.get("users_processed", 0), - users_synced=sync_result.get("users_synced", 0), - subs_synced=sync_result.get("subs_synced", 0) - ) - await bot.send_message(target_chat_id, _("sync_completed", status="Success", details=final_message)) + await bot.send_message(target_chat_id, _("sync_success_simple")) # Send notification to log channel with proper thread handling try: @@ -311,7 +297,7 @@ async def sync_command_handler( except Exception as e_sync_global: logging.error(f"Global error during /sync command: {e_sync_global}", exc_info=True) - await bot.send_message(target_chat_id, _("sync_failed", details=str(e_sync_global))) + await bot.send_message(target_chat_id, _("sync_critical_error")) # Send notification to log channel about failure try: diff --git a/locales/en.json b/locales/en.json index 0811460..43d5ca8 100644 --- a/locales/en.json +++ b/locales/en.json @@ -228,11 +228,11 @@ "admin_log_user_not_found": "User \"{input}\" not found in bot database.", "admin_user_logs_title": "Logs for {user_display} (page {current_page}/{total_pages}):", - "sync_started": "🔄 Starting data sync with panel...", - "sync_failed": "❌ Sync with panel failed. Details: {details}", - "sync_completed": "✅ Sync with panel completed. Status: {status}. Details: {details}", - "sync_completed_details": "Checked: {total_checked} entries.\nUsers synced/updated: {users_synced}.\nSubscriptions synced/updated: {subs_synced}.", - "sync_completed_with_errors_details": "Checked: {total_checked} entries.\nUsers synced/updated: {users_synced}.\nSubscriptions synced/updated: {subs_synced}.\nErrors: {errors_count}.\n\nFirst errors:\n{error_details_preview}", + "sync_started_simple": "🔄 Starting synchronization...", + "sync_success_simple": "✅ Synchronization completed successfully", + "sync_failed_simple": "❌ Synchronization failed", + "sync_errors_simple": "⚠️ Synchronization completed with errors ({errors_count} errors)", + "sync_critical_error": "❌ Critical synchronization error", "no_errors_placeholder": "none", "admin_sync_initiated_from_panel": "Sync initiated...", "admin_panel_user_creation_failed": "❌ Failed to create panel user for TG ID {user_id}. Panel unreachable?", diff --git a/locales/ru.json b/locales/ru.json index 35598cc..1cb6640 100644 --- a/locales/ru.json +++ b/locales/ru.json @@ -238,11 +238,11 @@ "admin_log_user_not_found": "Пользователь по запросу \"{input}\" не найден в базе данных бота.", "admin_user_logs_title": "Логи пользователя {user_display} (стр. {current_page}/{total_pages}):", - "sync_started": "🔄 Начинаю синхронизацию данных с панелью...", - "sync_failed": "❌ Ошибка синхронизации с панелью. Детали: {details}", - "sync_completed": "✅ Синхронизация с панелью завершена. Статус: {status}. Детали: {details}", - "sync_completed_details": "Проверено: {total_checked} записей.\nПользователей синхронизировано/обновлено: {users_synced}.\nПодписок синхронизировано/обновлено: {subs_synced}.", - "sync_completed_with_errors_details": "Проверено: {total_checked} записей.\nПользователей синхронизировано/обновлено: {users_synced}.\nПодписок синхронизировано/обновлено: {subs_synced}.\nОшибок: {errors_count}.\n\nПервые ошибки:\n{error_details_preview}", + "sync_started_simple": "🔄 Начинаю синхронизацию...", + "sync_success_simple": "✅ Синхронизация успешно завершена", + "sync_failed_simple": "❌ Синхронизация завершилась с ошибкой", + "sync_errors_simple": "⚠️ Синхронизация завершена с ошибками ({errors_count} ошибок)", + "sync_critical_error": "❌ Критическая ошибка синхронизации", "no_errors_placeholder": "нет", "admin_sync_initiated_from_panel": "Синхронизация запущена...", "admin_panel_user_creation_failed": "❌ Не удалось создать пользователя на панели для TG ID {user_id}. Панель недоступна?", From df15cfd25e1609050c65ce59cad11e1dc7d1a43f Mon Sep 17 00:00:00 2001 From: machka-pasla Date: Thu, 7 Aug 2025 18:46:19 +0300 Subject: [PATCH 2/8] Refactor promo handler functions to include session management - Updated the promo_delete_handler, promo_edit_select_handler, and promo_edit_field_handler functions to accept an AsyncSession parameter, improving database interaction consistency. - Enhanced the handling of expired subscriptions in the PanelWebhookService by modifying notification logic to only send messages if enabled, ensuring better control over user notifications. - Added an import for the 'and_' function in payment_dal.py to support more complex query conditions. --- bot/handlers/admin/promo/manage.py | 8 ++++---- bot/services/panel_webhook_service.py | 22 ++++++++++++---------- db/dal/payment_dal.py | 2 +- 3 files changed, 17 insertions(+), 15 deletions(-) diff --git a/bot/handlers/admin/promo/manage.py b/bot/handlers/admin/promo/manage.py index f8a5d97..c83b1d7 100644 --- a/bot/handlers/admin/promo/manage.py +++ b/bot/handlers/admin/promo/manage.py @@ -332,7 +332,7 @@ async def promo_export_all_handler(callback: types.CallbackQuery, i18n_data: dic @router.callback_query(F.data.startswith("promo_delete:")) -async def promo_delete_handler(callback: types.CallbackQuery, i18n_data: dict, session: AsyncSession): +async def promo_delete_handler(callback: types.CallbackQuery, i18n_data: dict, settings: Settings, session: AsyncSession): i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") current_lang = i18n_data.get("current_language") if not i18n or not callback.message or not current_lang: @@ -345,7 +345,7 @@ async def promo_delete_handler(callback: types.CallbackQuery, i18n_data: dict, s if promo: await session.commit() await callback.answer(_("admin_promo_deleted_success", code=promo.code), show_alert=True) - await promo_management_handler(callback, i18n_data, get_settings(), session, 0) + await promo_management_handler(callback, i18n_data, settings, session, 0) else: await callback.answer(_("admin_promo_not_found"), show_alert=True) except (ValueError, IndexError): @@ -354,7 +354,7 @@ async def promo_delete_handler(callback: types.CallbackQuery, i18n_data: dict, s # --- Promo Edit Handlers --- @router.callback_query(F.data.startswith("promo_edit_select:")) -async def promo_edit_select_handler(callback: types.CallbackQuery, i18n_data: dict): +async def promo_edit_select_handler(callback: types.CallbackQuery, i18n_data: dict, session: AsyncSession): i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") current_lang = i18n_data.get("current_language") if not i18n or not callback.message or not current_lang: @@ -373,7 +373,7 @@ async def promo_edit_select_handler(callback: types.CallbackQuery, i18n_data: di @router.callback_query(F.data.startswith("promo_edit_field:")) -async def promo_edit_field_handler(callback: types.CallbackQuery, state: FSMContext, i18n_data: dict): +async def promo_edit_field_handler(callback: types.CallbackQuery, state: FSMContext, i18n_data: dict, session: AsyncSession): i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") current_lang = i18n_data.get("current_language") if not i18n or not callback.message or not current_lang: return diff --git a/bot/services/panel_webhook_service.py b/bot/services/panel_webhook_service.py index 818c04e..1d22689 100644 --- a/bot/services/panel_webhook_service.py +++ b/bot/services/panel_webhook_service.py @@ -133,18 +133,20 @@ class PanelWebhookService: user_name=first_name, end_date=user_payload.get("expireAt", "")[:10], ) - elif event_name == "user.expired" and self.settings.SUBSCRIPTION_NOTIFY_ON_EXPIRE: - # Check if this is a tribute user that should be auto-renewed + elif event_name == "user.expired": + # Check if this is a tribute user that should be auto-renewed (regardless of notification settings) await self._handle_expired_subscription(session, user_id, user_payload, lang, markup, first_name) - await self._send_message( - user_id, - lang, - "subscription_expired_notification", - reply_markup=markup, - user_name=first_name, - end_date=user_payload.get("expireAt", "")[:10], - ) + # Send notification only if enabled + if self.settings.SUBSCRIPTION_NOTIFY_ON_EXPIRE: + await self._send_message( + user_id, + lang, + "subscription_expired_notification", + reply_markup=markup, + user_name=first_name, + end_date=user_payload.get("expireAt", "")[:10], + ) elif event_name == "user.expired_24_hours_ago" and self.settings.SUBSCRIPTION_NOTIFY_AFTER_EXPIRE: await self._send_message( user_id, diff --git a/db/dal/payment_dal.py b/db/dal/payment_dal.py index 394e43a..5187e58 100644 --- a/db/dal/payment_dal.py +++ b/db/dal/payment_dal.py @@ -2,7 +2,7 @@ import logging from typing import Optional, List, Dict, Any from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.future import select -from sqlalchemy import update, func +from sqlalchemy import update, func, and_ from sqlalchemy.orm import selectinload from db.models import Payment, User From 194f1b9e49a6956626f032499f8f40a760e7aa21 Mon Sep 17 00:00:00 2001 From: machka-pasla Date: Thu, 7 Aug 2025 19:00:27 +0300 Subject: [PATCH 3/8] Enhance recent payment log retrieval to filter by succeeded status - Updated the `get_recent_payment_logs_with_user` function to include a filter for payments with a 'succeeded' status, improving the relevance of retrieved payment logs. - Adjusted the query structure for better readability and maintainability. --- db/dal/payment_dal.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/db/dal/payment_dal.py b/db/dal/payment_dal.py index 5187e58..5a93e6d 100644 --- a/db/dal/payment_dal.py +++ b/db/dal/payment_dal.py @@ -82,8 +82,10 @@ async def update_payment_status_by_db_id( async def get_recent_payment_logs_with_user(session: AsyncSession, limit: int = 20, offset: int = 0) -> List[Payment]: - stmt = (select(Payment).options(selectinload(Payment.user)).order_by( - Payment.created_at.desc()).limit(limit).offset(offset)) + stmt = (select(Payment).options(selectinload(Payment.user)) + .where(Payment.status == 'succeeded') + .order_by(Payment.created_at.desc()) + .limit(limit).offset(offset)) result = await session.execute(stmt) return result.scalars().all() From 91cfe0baf30f179257576d591f621ed6bfaa6cca Mon Sep 17 00:00:00 2001 From: machka-pasla Date: Thu, 7 Aug 2025 22:48:09 +0300 Subject: [PATCH 4/8] Add payments feature to admin panel - Integrated payments functionality into the admin panel by adding a new payments router and corresponding handlers. - Updated the admin panel actions to include a view payments option, enhancing admin capabilities. - Implemented new database functions to retrieve successful payment counts and details for export. - Enhanced localization with new strings for payments management in both English and Russian. --- bot/handlers/admin/__init__.py | 2 + bot/handlers/admin/common.py | 4 + bot/handlers/admin/payments.py | 245 ++++++++++++++++++++++++ bot/keyboards/inline/admin_keyboards.py | 4 +- db/dal/payment_dal.py | 16 ++ locales/en.json | 22 +++ locales/ru.json | 22 +++ 7 files changed, 314 insertions(+), 1 deletion(-) create mode 100644 bot/handlers/admin/payments.py diff --git a/bot/handlers/admin/__init__.py b/bot/handlers/admin/__init__.py index 2444ab2..ecf8b9f 100644 --- a/bot/handlers/admin/__init__.py +++ b/bot/handlers/admin/__init__.py @@ -7,6 +7,7 @@ from . import user_management from . import statistics from . import sync_admin from . import logs_admin +from . import payments admin_router_aggregate = Router(name="admin_features_router") @@ -17,5 +18,6 @@ admin_router_aggregate.include_router(user_management.router) admin_router_aggregate.include_router(statistics.router) admin_router_aggregate.include_router(sync_admin.router) admin_router_aggregate.include_router(logs_admin.router) +admin_router_aggregate.include_router(payments.router) __all__ = ("admin_router_aggregate", ) diff --git a/bot/handlers/admin/common.py b/bot/handlers/admin/common.py index a43c57d..8204404 100644 --- a/bot/handlers/admin/common.py +++ b/bot/handlers/admin/common.py @@ -123,6 +123,10 @@ async def admin_panel_actions_callback_handler( await callback.answer(_("admin_sync_initiated_from_panel")) elif action == "queue_status": await show_queue_status_handler(callback, i18n_data) + elif action == "view_payments": + from . import payments as admin_payments_handlers + await admin_payments_handlers.view_payments_handler( + callback, i18n_data, settings, session) elif action == "main": try: await callback.message.edit_text( diff --git a/bot/handlers/admin/payments.py b/bot/handlers/admin/payments.py new file mode 100644 index 0000000..2b4a95e --- /dev/null +++ b/bot/handlers/admin/payments.py @@ -0,0 +1,245 @@ +import logging +import csv +import io +from aiogram import Router, F, types +from aiogram.filters import StateFilter +from aiogram.fsm.context import FSMContext +from datetime import datetime, timedelta, timezone +from typing import Optional, List +from sqlalchemy.ext.asyncio import AsyncSession + +from config.settings import Settings +from db.dal import payment_dal +from db.models import Payment +from bot.keyboards.inline.admin_keyboards import get_back_to_admin_panel_keyboard +from aiogram.utils.keyboard import InlineKeyboardBuilder, InlineKeyboardButton +from bot.middlewares.i18n import JsonI18n + +router = Router(name="admin_payments_router") + + +async def get_payments_with_pagination(session: AsyncSession, page: int = 0, + page_size: int = 10) -> tuple[List[Payment], int]: + """Get payments with pagination and total count.""" + offset = page * page_size + + # Get total count + total_count = await payment_dal.get_payments_count(session) + + # Get payments for current page + payments = await payment_dal.get_recent_payment_logs_with_user( + session, limit=page_size, offset=offset + ) + + return payments, total_count + + +def format_payment_text(payment: Payment, i18n: JsonI18n, lang: str) -> str: + """Format single payment info as text.""" + _ = lambda key, **kwargs: i18n.gettext(lang, key, **kwargs) + + status_emoji = "✅" if payment.status == 'succeeded' else ( + "⏳" if payment.status in ['pending', 'pending_yookassa'] 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 = payment.created_at.strftime('%Y-%m-%d %H:%M') if payment.created_at else "N/A" + + provider_text = { + 'yookassa': 'YooKassa', + 'tribute': 'Tribute', + 'telegram_stars': 'Telegram Stars', + 'cryptopay': 'CryptoPay' + }.get(payment.provider, payment.provider or 'Unknown') + + return ( + f"{status_emoji} {payment.amount} {payment.currency}\n" + f"👤 {user_info}\n" + f"💳 {provider_text}\n" + f"📅 {payment_date}\n" + f"📋 {payment.status}\n" + f"📝 {payment.description or 'N/A'}" + ) + + +async def view_payments_handler(callback: types.CallbackQuery, i18n_data: dict, + settings: Settings, session: AsyncSession, page: int = 0): + """Display paginated list of all payments.""" + 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 processing request.", show_alert=True) + return + _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) + + page_size = 5 # Показываем по 5 платежей на странице + payments, total_count = await get_payments_with_pagination(session, page, page_size) + total_pages = (total_count + page_size - 1) // page_size if total_count > 0 else 1 + + if not payments and page == 0: + await callback.message.edit_text( + _("admin_no_payments_found", default="Платежи не найдены."), + reply_markup=get_back_to_admin_panel_keyboard(current_lang, i18n), + parse_mode="HTML" + ) + await callback.answer() + return + + # Format payments text + text_parts = [_("admin_payments_header", default="💰 Все платежи")] + text_parts.append(f"📊 Показано {len(payments)} из {total_count} платежей (стр. {page + 1}/{total_pages})\n") + + for i, payment in enumerate(payments, 1): + text_parts.append(f"{page * page_size + i}. {format_payment_text(payment, i18n, current_lang)}") + text_parts.append("") # Empty line between payments + + # Build keyboard with pagination and export + builder = InlineKeyboardBuilder() + + # Pagination buttons + nav_buttons = [] + if page > 0: + nav_buttons.append(InlineKeyboardButton(text="⬅️", callback_data=f"payments_page:{page-1}")) + + nav_buttons.append(InlineKeyboardButton(text=f"{page + 1}/{total_pages}", callback_data="noop")) + + if page < total_pages - 1: + nav_buttons.append(InlineKeyboardButton(text="➡️", callback_data=f"payments_page:{page+1}")) + + if nav_buttons: + builder.row(*nav_buttons) + + # Export and refresh buttons + builder.row( + InlineKeyboardButton( + text=_("admin_export_payments_csv", default="📊 Экспорт CSV"), + callback_data="payments_export_csv" + ), + InlineKeyboardButton( + text=_("admin_refresh_payments", default="🔄 Обновить"), + callback_data=f"payments_page:{page}" + ) + ) + + # Back button + builder.row(InlineKeyboardButton( + text=_("back_to_admin_panel_button"), + callback_data="admin_section:stats_monitoring" + )) + + await callback.message.edit_text( + "\n".join(text_parts), + reply_markup=builder.as_markup(), + parse_mode="HTML" + ) + await callback.answer() + + +@router.callback_query(F.data.startswith("payments_page:")) +async def payments_pagination_handler(callback: types.CallbackQuery, i18n_data: dict, + settings: Settings, session: AsyncSession): + """Handle pagination for payments list.""" + try: + page = int(callback.data.split(":")[1]) + await view_payments_handler(callback, i18n_data, settings, session, page) + except (ValueError, IndexError): + await callback.answer("Error processing pagination.", show_alert=True) + + +@router.callback_query(F.data == "payments_export_csv") +async def export_payments_csv_handler(callback: types.CallbackQuery, i18n_data: dict, + settings: Settings, session: AsyncSession): + """Export all successful payments to CSV file.""" + current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE) + i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") + if not i18n: + await callback.answer("Language service error.", show_alert=True) + return + _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) + + try: + # Get all successful payments + all_payments = await payment_dal.get_all_succeeded_payments_with_user(session) + + if not all_payments: + await callback.answer( + _("admin_no_payments_to_export", default="Нет платежей для экспорта."), + show_alert=True + ) + return + + # Create CSV in memory + output = io.StringIO() + writer = csv.writer(output) + + # Write header + writer.writerow([ + _("admin_csv_payment_id", default="ID"), + _("admin_csv_user_id", default="User ID"), + _("admin_csv_username", default="Username"), + _("admin_csv_first_name", default="First Name"), + _("admin_csv_amount", default="Amount"), + _("admin_csv_currency", default="Currency"), + _("admin_csv_provider", default="Provider"), + _("admin_csv_status", default="Status"), + _("admin_csv_description", default="Description"), + _("admin_csv_months", default="Months"), + _("admin_csv_created_at", default="Created At"), + _("admin_csv_provider_payment_id", default="Provider Payment ID") + ]) + + # Write payment data + for payment in all_payments: + writer.writerow([ + payment.payment_id, + payment.user_id, + payment.user.username if payment.user and payment.user.username else "", + payment.user.first_name if payment.user and payment.user.first_name else "", + payment.amount, + payment.currency, + payment.provider or "", + payment.status, + payment.description or "", + payment.subscription_duration_months or "", + payment.created_at.strftime('%Y-%m-%d %H:%M:%S') if payment.created_at else "", + payment.provider_payment_id or "" + ]) + + # Prepare file + csv_content = output.getvalue().encode('utf-8-sig') # UTF-8 with BOM for Excel + output.close() + + # Generate filename with current date + current_time = datetime.now().strftime('%Y-%m-%d_%H-%M') + filename = f"payments_export_{current_time}.csv" + + # Send file + from aiogram.types import BufferedInputFile + file = BufferedInputFile(csv_content, filename=filename) + + await callback.message.reply_document( + document=file, + caption=_("admin_payments_export_success", + default="📊 Экспорт платежей завершен!\nВсего записей: {count}", + count=len(all_payments)) + ) + + await callback.answer( + _("admin_export_sent", default="Файл отправлен!"), + show_alert=False + ) + + except Exception as e: + logging.error(f"Failed to export payments CSV: {e}", exc_info=True) + await callback.answer(f"❌ Ошибка экспорта: {str(e)}", show_alert=True) + + +@router.callback_query(F.data == "noop") +async def noop_handler(callback: types.CallbackQuery): + """Handle no-op callback (for pagination display).""" + await callback.answer() \ No newline at end of file diff --git a/bot/keyboards/inline/admin_keyboards.py b/bot/keyboards/inline/admin_keyboards.py index bf0d4e7..293100e 100644 --- a/bot/keyboards/inline/admin_keyboards.py +++ b/bot/keyboards/inline/admin_keyboards.py @@ -39,12 +39,14 @@ def get_stats_monitoring_keyboard(i18n_instance, lang: str) -> InlineKeyboardMar builder.button(text=_(key="admin_stats_button"), callback_data="admin_action:stats") + builder.button(text=_(key="admin_view_payments_button", default="💰 Платежи"), + callback_data="admin_action:view_payments") builder.button(text=_(key="admin_view_logs_menu_button"), callback_data="admin_action:view_logs_menu") builder.button(text=_(key="back_to_admin_panel_button"), callback_data="admin_action:main") - builder.adjust(2, 1) + builder.adjust(2, 1, 1) return builder.as_markup() diff --git a/db/dal/payment_dal.py b/db/dal/payment_dal.py index 5a93e6d..9bb2c12 100644 --- a/db/dal/payment_dal.py +++ b/db/dal/payment_dal.py @@ -90,6 +90,22 @@ async def get_recent_payment_logs_with_user(session: AsyncSession, return result.scalars().all() +async def get_payments_count(session: AsyncSession) -> int: + """Get total count of successful payments.""" + stmt = select(func.count(Payment.payment_id)).where(Payment.status == 'succeeded') + result = await session.execute(stmt) + return result.scalar() or 0 + + +async def get_all_succeeded_payments_with_user(session: AsyncSession) -> List[Payment]: + """Get all successful payments with user data for export.""" + stmt = (select(Payment).options(selectinload(Payment.user)) + .where(Payment.status == 'succeeded') + .order_by(Payment.created_at.desc())) + result = await session.execute(stmt) + return result.scalars().all() + + async def update_provider_payment_and_status( session: AsyncSession, payment_db_id: int, provider_payment_id: str, new_status: str) -> Optional[Payment]: diff --git a/locales/en.json b/locales/en.json index 43d5ca8..5712a5f 100644 --- a/locales/en.json +++ b/locales/en.json @@ -122,6 +122,28 @@ "admin_stats_recent_payments_header": "Recent Payments:", "admin_stats_payment_item": "{status_emoji} {amount} {currency} from {user_info} ({p_status}) [{p_date}]", "admin_stats_no_payments_found": "No payments found yet.", + + "admin_view_payments_button": "💰 Payments", + "admin_payments_header": "💰 All Payments", + "admin_no_payments_found": "No payments found.", + "admin_export_payments_csv": "📊 Export CSV", + "admin_refresh_payments": "🔄 Refresh", + "admin_no_payments_to_export": "No payments to export.", + "admin_payments_export_success": "📊 Payments export completed!\nTotal records: {count}", + "admin_export_sent": "File sent!", + + "admin_csv_payment_id": "ID", + "admin_csv_user_id": "User ID", + "admin_csv_username": "Username", + "admin_csv_first_name": "First Name", + "admin_csv_amount": "Amount", + "admin_csv_currency": "Currency", + "admin_csv_provider": "Provider", + "admin_csv_status": "Status", + "admin_csv_description": "Description", + "admin_csv_months": "Months", + "admin_csv_created_at": "Created At", + "admin_csv_provider_payment_id": "Provider Payment ID", "admin_stats_last_sync_header": "Last Panel Sync:", "admin_stats_sync_time": "Time", "admin_stats_sync_status": "Status", diff --git a/locales/ru.json b/locales/ru.json index 1cb6640..8197e9a 100644 --- a/locales/ru.json +++ b/locales/ru.json @@ -122,6 +122,28 @@ "admin_stats_recent_payments_header": "Последние платежи:", "admin_stats_payment_item": "{status_emoji} {amount} {currency} от {user_info} ({p_status}) [{p_date}]", "admin_stats_no_payments_found": "Платежей пока нет.", + + "admin_view_payments_button": "💰 Платежи", + "admin_payments_header": "💰 Все платежи", + "admin_no_payments_found": "Платежи не найдены.", + "admin_export_payments_csv": "📊 Экспорт CSV", + "admin_refresh_payments": "🔄 Обновить", + "admin_no_payments_to_export": "Нет платежей для экспорта.", + "admin_payments_export_success": "📊 Экспорт платежей завершен!\nВсего записей: {count}", + "admin_export_sent": "Файл отправлен!", + + "admin_csv_payment_id": "ID", + "admin_csv_user_id": "User ID", + "admin_csv_username": "Логин", + "admin_csv_first_name": "Имя", + "admin_csv_amount": "Сумма", + "admin_csv_currency": "Валюта", + "admin_csv_provider": "Платежная система", + "admin_csv_status": "Статус", + "admin_csv_description": "Описание", + "admin_csv_months": "Месяцев", + "admin_csv_created_at": "Дата создания", + "admin_csv_provider_payment_id": "ID платежа в системе", "admin_stats_last_sync_header": "Последняя синхронизация с панелью:", "admin_stats_sync_time": "Время", "admin_stats_sync_status": "Статус", From 5853b9da639cbef999acae36aacaba5f7face1ac Mon Sep 17 00:00:00 2001 From: machka-pasla Date: Thu, 7 Aug 2025 23:02:23 +0300 Subject: [PATCH 5/8] Enhance payment identifier normalization in TributeService - Updated the TributeService to normalize provider payment identifiers, prioritizing true payment identifiers over subscription IDs for better uniqueness. - Implemented fallback logic to append timestamps to subscription IDs, preventing deduplication of renewals. - Expanded the list of successful charge events to include various payment-related events, improving event handling consistency. --- bot/services/tribute_service.py | 40 +++++++++++++++++++++++++++++++-- 1 file changed, 38 insertions(+), 2 deletions(-) diff --git a/bot/services/tribute_service.py b/bot/services/tribute_service.py index 6e2029a..6ea1ae3 100644 --- a/bot/services/tribute_service.py +++ b/bot/services/tribute_service.py @@ -96,8 +96,44 @@ class TributeService: price_rub = price_val / 100 async with async_session_factory() as session: - if event_name == 'new_subscription': - provider_payment_id = str(data.get('subscription_id')) + # Normalize provider payment identifier to be unique per successful charge + # Prefer true payment/transaction identifiers over subscription id + provider_payment_id = ( + data.get('payment_id') + or data.get('invoice_id') + or data.get('order_id') + or data.get('transaction_id') + or data.get('charge_id') + or data.get('subscription_payment_id') + ) + if provider_payment_id is None: + # Fallback to subscription_id which may be stable across renewals + # To avoid deduplicating different renewals under same subscription, + # append a timestamp if available + base_sub_id = data.get('subscription_id') + paid_at = ( + data.get('paid_at') + or data.get('created_at') + or payload.get('timestamp') + or payload.get('id') + ) + if base_sub_id is not None and paid_at is not None: + provider_payment_id = f"{base_sub_id}:{paid_at}" + elif base_sub_id is not None: + provider_payment_id = str(base_sub_id) + else: + provider_payment_id = str(provider_payment_id) + + # Consider multiple Tribute events as successful charge events + success_events = { + 'new_subscription', + 'payment_succeeded', + 'subscription_renewed', + 'subscription_payment_succeeded', + 'invoice_paid', + } + + if event_name in success_events: existing_payment = await payment_dal.get_payment_by_provider_payment_id( session, provider_payment_id) if existing_payment: From 18f65ea493907bb9d0d5f3573f3dc4dbaee5fc87 Mon Sep 17 00:00:00 2001 From: machka-pasla Date: Thu, 7 Aug 2025 23:30:32 +0300 Subject: [PATCH 6/8] Refactor notification handling and streamline router registration - Replaced legacy notification functions with a unified NotificationService for better maintainability and clarity. - Updated the main bot router registration to utilize a root router, simplifying the inclusion of user and admin routes. - Removed unused middleware and helper functions to enhance code cleanliness and focus on essential components. - Improved localization by adding new error messages for user interactions. --- bot/handlers/admin/sync_admin.py | 12 +- bot/handlers/user/trial_handler.py | 20 +--- bot/main_bot.py | 67 +---------- bot/middlewares/db_session.py | 40 +++++++ bot/routers.py | 26 +++++ bot/services/notification_service.py | 49 +------- bot/services/subscription_service.py | 85 +++++++------- bot/services/tribute_service.py | 160 +++++++++------------------ db/dal/payment_dal.py | 32 ++++++ db/dal/subscription_dal.py | 23 ++++ db/dal/user_dal.py | 15 +-- locales/en.json | 1 + locales/ru.json | 1 + 13 files changed, 239 insertions(+), 292 deletions(-) create mode 100644 bot/middlewares/db_session.py create mode 100644 bot/routers.py diff --git a/bot/handlers/admin/sync_admin.py b/bot/handlers/admin/sync_admin.py index 0f33482..b0c306c 100644 --- a/bot/handlers/admin/sync_admin.py +++ b/bot/handlers/admin/sync_admin.py @@ -7,7 +7,7 @@ from datetime import datetime, timezone from config.settings import Settings from bot.services.panel_api_service import PanelApiService -from bot.services.notification_service import notify_admin_panel_sync +from bot.services.notification_service import NotificationService from db.dal import user_dal, subscription_dal, panel_sync_dal @@ -287,8 +287,9 @@ async def sync_command_handler( # Send notification to log channel with proper thread handling try: - await notify_admin_panel_sync( - bot, settings, i18n, status, details, + notification_service = NotificationService(bot, settings, i18n) + await notification_service.notify_panel_sync( + status, details, sync_result.get("users_processed", 0), sync_result.get("subs_synced", 0) ) @@ -301,8 +302,9 @@ async def sync_command_handler( # Send notification to log channel about failure try: - await notify_admin_panel_sync( - bot, settings, i18n, "failed", str(e_sync_global), 0, 0 + notification_service = NotificationService(bot, settings, i18n) + await notification_service.notify_panel_sync( + "failed", str(e_sync_global), 0, 0 ) except Exception as e_notification: logging.error(f"Failed to send sync failure notification: {e_notification}") diff --git a/bot/handlers/user/trial_handler.py b/bot/handlers/user/trial_handler.py index d390086..043fb18 100644 --- a/bot/handlers/user/trial_handler.py +++ b/bot/handlers/user/trial_handler.py @@ -7,7 +7,7 @@ from datetime import datetime from config.settings import Settings from bot.services.subscription_service import SubscriptionService from bot.services.panel_api_service import PanelApiService -from bot.services.notification_service import notify_admin_new_trial +from bot.services.notification_service import NotificationService from bot.keyboards.inline.user_keyboards import ( get_trial_confirmation_keyboard, get_main_menu_inline_keyboard, @@ -97,13 +97,8 @@ async def request_trial_confirmation_handler( ) # Send notification to admin about new trial - await notify_admin_new_trial( - callback.bot, - settings, - i18n, - user_id, - end_date_obj, - ) + notification_service = NotificationService(callback.bot, settings, i18n) + await notification_service.notify_trial_activation(user_id, end_date_obj) else: message_key_from_service = ( activation_result.get("message_key", "trial_activation_failed") @@ -264,13 +259,8 @@ async def confirm_activate_trial_handler( ) if activation_result and activation_result.get("activated") and end_date_obj: - await notify_admin_new_trial( - callback.bot, - settings, - i18n, - user_id, - end_date_obj, - ) + notification_service = NotificationService(callback.bot, settings, i18n) + await notification_service.notify_trial_activation(user_id, end_date_obj) @router.callback_query(F.data == "main_action:cancel_trial") diff --git a/bot/main_bot.py b/bot/main_bot.py index 21d2964..1392e01 100644 --- a/bot/main_bot.py +++ b/bot/main_bot.py @@ -1,17 +1,10 @@ import logging import asyncio -from typing import Callable, Dict, Any, Awaitable, Optional +from typing import Dict, Any, Optional -from aiogram import Bot, Dispatcher, BaseMiddleware, Router, F -from aiogram.types import ( - Update, - MenuButtonDefault, - MenuButtonWebApp, - WebAppInfo, - BotCommand, -) +from aiogram import Bot, Dispatcher +from aiogram.types import (MenuButtonDefault, MenuButtonWebApp, WebAppInfo, BotCommand) from aiogram.enums import ParseMode -from aiogram.filters import CommandStart, Command from aiogram.client.default import DefaultBotProperties from aiogram.webhook.aiohttp_server import SimpleRequestHandler, setup_application from aiogram.fsm.storage.memory import MemoryStorage @@ -24,13 +17,11 @@ from config.settings import Settings from db.database_setup import init_db_connection from bot.middlewares.i18n import I18nMiddleware, get_i18n_instance, JsonI18n +from bot.middlewares.db_session import DBSessionMiddleware from bot.middlewares.ban_check_middleware import BanCheckMiddleware from bot.middlewares.action_logger_middleware import ActionLoggerMiddleware -from bot.handlers.user import user_router_aggregate -from bot.handlers.admin import admin_router_aggregate -from bot.handlers import inline_mode -from bot.filters.admin_filter import AdminFilter +from bot.routers import build_root_router from bot.services.yookassa_service import YooKassaService from bot.services.panel_api_service import PanelApiService @@ -46,54 +37,8 @@ from bot.handlers.admin.sync_admin import perform_sync from bot.utils.message_queue import init_queue_manager -class DBSessionMiddleware(BaseMiddleware): - - def __init__(self, async_session_factory: sessionmaker): - super().__init__() - self.async_session_factory = async_session_factory - - async def __call__( - self, - handler: Callable[[Update, Dict[str, Any]], Awaitable[Any]], - event: Update, - data: Dict[str, Any], - ) -> Any: - if self.async_session_factory is None: - logging.critical("DBSessionMiddleware: async_session_factory is None!") - raise RuntimeError( - "async_session_factory not provided to DBSessionMiddleware" - ) - - async with self.async_session_factory() as session: - data["session"] = session - try: - result = await handler(event, data) - - await session.commit() - return result - except Exception: - await session.rollback() - logging.error( - "DBSessionMiddleware: Exception caused rollback.", exc_info=True - ) - raise - - async def register_all_routers(dp: Dispatcher, settings: Settings): - dp.include_router(user_router_aggregate) - - # Add inline mode router (available for all users) - dp.include_router(inline_mode.router) - - admin_main_router = Router(name="admin_main_filtered_router") - admin_filter_instance = AdminFilter(admin_ids=settings.ADMIN_IDS) - - admin_main_router.message.filter(admin_filter_instance) - admin_main_router.callback_query.filter(admin_filter_instance) - - admin_main_router.include_router(admin_router_aggregate) - - dp.include_router(admin_main_router) + dp.include_router(build_root_router(settings)) logging.info("All application routers registered.") diff --git a/bot/middlewares/db_session.py b/bot/middlewares/db_session.py new file mode 100644 index 0000000..006a2d2 --- /dev/null +++ b/bot/middlewares/db_session.py @@ -0,0 +1,40 @@ +import logging +from typing import Callable, Dict, Any, Awaitable + +from aiogram import BaseMiddleware +from aiogram.types import Update +from sqlalchemy.orm import sessionmaker + + +class DBSessionMiddleware(BaseMiddleware): + + def __init__(self, async_session_factory: sessionmaker): + super().__init__() + self.async_session_factory = async_session_factory + + async def __call__( + self, + handler: Callable[[Update, Dict[str, Any]], Awaitable[Any]], + event: Update, + data: Dict[str, Any], + ) -> Any: + if self.async_session_factory is None: + logging.critical("DBSessionMiddleware: async_session_factory is None!") + raise RuntimeError( + "async_session_factory not provided to DBSessionMiddleware" + ) + + async with self.async_session_factory() as session: + data["session"] = session + try: + result = await handler(event, data) + + await session.commit() + return result + except Exception: + await session.rollback() + logging.error( + "DBSessionMiddleware: Exception caused rollback.", exc_info=True + ) + raise + diff --git a/bot/routers.py b/bot/routers.py new file mode 100644 index 0000000..c79b963 --- /dev/null +++ b/bot/routers.py @@ -0,0 +1,26 @@ +from aiogram import Router + +from bot.handlers.user import user_router_aggregate +from bot.handlers import inline_mode +from bot.handlers.admin import admin_router_aggregate +from bot.filters.admin_filter import AdminFilter +from config.settings import Settings + + +def build_root_router(settings: Settings) -> Router: + root = Router(name="root") + + # Public routers + root.include_router(user_router_aggregate) + root.include_router(inline_mode.router) + + # Admin routers behind filter + admin_main_router = Router(name="admin_main_filtered_router") + admin_filter_instance = AdminFilter(admin_ids=settings.ADMIN_IDS) + admin_main_router.message.filter(admin_filter_instance) + admin_main_router.callback_query.filter(admin_filter_instance) + admin_main_router.include_router(admin_router_aggregate) + root.include_router(admin_main_router) + + return root + diff --git a/bot/services/notification_service.py b/bot/services/notification_service.py index 64f02b9..f14cfb3 100644 --- a/bot/services/notification_service.py +++ b/bot/services/notification_service.py @@ -295,51 +295,4 @@ class NotificationService: if to_admins: await self._send_to_admins(message) - -# Legacy functions for backward compatibility -async def notify_admins(bot: Bot, settings: Settings, i18n: JsonI18n, - message_key: str, parse_mode: str | None = None, - **kwargs) -> None: - if not settings.ADMIN_IDS: - return - admin_lang = settings.DEFAULT_LANGUAGE - msg = i18n.gettext(admin_lang, message_key, **kwargs) - for admin_id in settings.ADMIN_IDS: - try: - await bot.send_message(admin_id, msg, parse_mode=parse_mode) - except Exception as e: - logging.error(f"Failed to send admin notification to {admin_id}: {e}") - - -async def notify_admin_new_trial(bot: Bot, settings: Settings, i18n: JsonI18n, - user_id: int, end_date: datetime) -> None: - """Send notification to admins about new trial activation (legacy)""" - notification_service = NotificationService(bot, settings, i18n) - await notification_service.notify_trial_activation(user_id, end_date) - - - - - -async def notify_admin_promo_activation(bot: Bot, settings: Settings, - i18n: JsonI18n, user_id: int, - code: str, - bonus_days: int) -> None: - await notify_admins( - bot, - settings, - i18n, - "admin_promo_activation_notification", - user_id=user_id, - code=code, - bonus_days=bonus_days, - ) - - -async def notify_admin_panel_sync(bot: Bot, settings: Settings, - i18n: JsonI18n, status: str, - details: str, users_processed: int, - subs_synced: int) -> None: - """Send notification to admins about panel sync (legacy)""" - notification_service = NotificationService(bot, settings, i18n) - await notification_service.notify_panel_sync(status, details, users_processed, subs_synced) \ No newline at end of file +# Removed legacy helper functions that duplicated NotificationService API \ No newline at end of file diff --git a/bot/services/subscription_service.py b/bot/services/subscription_service.py index 61795c3..9546175 100644 --- a/bot/services/subscription_service.py +++ b/bot/services/subscription_service.py @@ -34,10 +34,7 @@ class SubscriptionService: else self.settings.DEFAULT_LANGUAGE ) - async def has_had_any_subscription( - self, session: AsyncSession, user_id: int - ) -> bool: - + async def has_had_any_subscription(self, session: AsyncSession, user_id: int) -> bool: return await subscription_dal.has_any_subscription_for_user(session, user_id) async def _notify_admin_panel_user_creation_failed(self, user_id: int): @@ -348,19 +345,12 @@ class SubscriptionService: "message_key": "trial_activation_failed_db", } - panel_update_payload: Dict[str, Any] = { - "uuid": panel_user_uuid, - "expireAt": end_date.isoformat(timespec="milliseconds").replace( - "+00:00", "Z" - ), - "status": "ACTIVE", - "trafficLimitBytes": self.settings.trial_traffic_limit_bytes, - "trafficLimitStrategy": self.settings.USER_TRAFFIC_STRATEGY, - } - if self.settings.parsed_user_squad_uuids: - panel_update_payload["activeInternalSquads"] = ( - self.settings.parsed_user_squad_uuids - ) + panel_update_payload = self._build_panel_update_payload( + panel_user_uuid=panel_user_uuid, + expire_at=end_date, + status="ACTIVE", + traffic_limit_bytes=self.settings.trial_traffic_limit_bytes, + ) updated_panel_user = await self.panel_service.update_user_details_on_panel( panel_user_uuid, panel_update_payload @@ -495,19 +485,12 @@ class SubscriptionService: ) return None - panel_update_payload = { - "uuid": panel_user_uuid, - "expireAt": final_end_date.isoformat(timespec="milliseconds").replace( - "+00:00", "Z" - ), - "status": "ACTIVE", - "trafficLimitBytes": self.settings.user_traffic_limit_bytes, - "trafficLimitStrategy": self.settings.USER_TRAFFIC_STRATEGY, - } - if self.settings.parsed_user_squad_uuids: - panel_update_payload["activeInternalSquads"] = ( - self.settings.parsed_user_squad_uuids - ) + panel_update_payload = self._build_panel_update_payload( + panel_user_uuid=panel_user_uuid, + expire_at=final_end_date, + status="ACTIVE", + traffic_limit_bytes=self.settings.user_traffic_limit_bytes, + ) updated_panel_user = await self.panel_service.update_user_details_on_panel( panel_user_uuid, panel_update_payload @@ -598,17 +581,13 @@ class SubscriptionService: if updated_sub_model: # Prepare panel update payload - panel_update_payload = { - "expireAt": new_end_date_obj.isoformat( - timespec="milliseconds" - ).replace("+00:00", "Z") - } - - # For promo code activations, remove traffic limit - if "promo code" in reason.lower(): - panel_update_payload["trafficLimitBytes"] = self.settings.user_traffic_limit_bytes - panel_update_payload["trafficLimitStrategy"] = self.settings.USER_TRAFFIC_STRATEGY - logging.info(f"Updating traffic limit for user {user_id} to {self.settings.user_traffic_limit_bytes} bytes due to promo code activation") + panel_update_payload = self._build_panel_update_payload( + expire_at=new_end_date_obj, + traffic_limit_bytes=( + self.settings.user_traffic_limit_bytes if "promo code" in reason.lower() else None + ), + include_uuid=False, + ) panel_update_success = ( await self.panel_service.update_user_details_on_panel( @@ -775,3 +754,27 @@ class SubscriptionService: logging.warning( f"Could not find subscription for user {user_id} ending at {subscription_end_date.isoformat()} to update notification time." ) + + # Helpers + def _build_panel_update_payload( + self, + *, + panel_user_uuid: Optional[str] = None, + expire_at: Optional[datetime] = None, + status: Optional[str] = None, + traffic_limit_bytes: Optional[int] = None, + include_uuid: bool = True, + ) -> Dict[str, Any]: + payload: Dict[str, Any] = {} + if include_uuid and panel_user_uuid: + payload["uuid"] = panel_user_uuid + if expire_at is not None: + payload["expireAt"] = expire_at.isoformat(timespec="milliseconds").replace("+00:00", "Z") + if status is not None: + payload["status"] = status + if traffic_limit_bytes is not None: + payload["trafficLimitBytes"] = traffic_limit_bytes + payload["trafficLimitStrategy"] = self.settings.USER_TRAFFIC_STRATEGY + if self.settings.parsed_user_squad_uuids: + payload["activeInternalSquads"] = self.settings.parsed_user_squad_uuids + return payload diff --git a/bot/services/tribute_service.py b/bot/services/tribute_service.py index 6ea1ae3..5401922 100644 --- a/bot/services/tribute_service.py +++ b/bot/services/tribute_service.py @@ -39,11 +39,16 @@ def convert_period_to_months(period: Optional[str]) -> int: class TributeService: - def __init__(self, bot: Bot, settings: Settings, i18n: JsonI18n, - async_session_factory: sessionmaker, - panel_service: PanelApiService, - subscription_service: SubscriptionService, - referral_service: ReferralService): + def __init__( + self, + bot: Bot, + settings: Settings, + i18n: JsonI18n, + async_session_factory: sessionmaker, + panel_service: PanelApiService, + subscription_service: SubscriptionService, + referral_service: ReferralService, + ): self.bot = bot self.settings = settings self.i18n = i18n @@ -52,8 +57,7 @@ class TributeService: self.subscription_service = subscription_service self.referral_service = referral_service - async def handle_webhook(self, raw_body: bytes, - signature_header: Optional[str]) -> web.Response: + async def handle_webhook(self, raw_body: bytes, signature_header: Optional[str]) -> web.Response: settings = self.settings bot = self.bot i18n = self.i18n @@ -79,96 +83,53 @@ class TributeService: json.dumps(payload, ensure_ascii=False), ) - event_name = payload.get('name') - data = payload.get('payload', {}) - user_id = data.get('telegram_user_id') - price_val = ( - data.get('amount') - or data.get('amount_paid') - or data.get('price') - ) + # Tribute webhook spec: only two events are sent + # name: new_subscription | cancelled_subscription + event_name = payload.get("name") + data = payload.get("payload", {}) - if not user_id or price_val is None: - return web.Response(status=200, text="ok_missing_fields") + # Mandatory routing fields + user_id = data.get("telegram_user_id") + if not user_id: + return web.Response(status=400, text="missing_telegram_user_id") - period_val = data.get('period') + period_val = data.get("period") months = convert_period_to_months(period_val) - price_rub = price_val / 100 + + # Price/amount from spec is integer cents in currency; we store float in Payment + amount_value = data.get("amount") or data.get("price") + currency = (data.get("currency") or settings.DEFAULT_CURRENCY_SYMBOL or "RUB").upper() + amount_float = float(amount_value) if amount_value is not None else 0.0 async with async_session_factory() as session: - # Normalize provider payment identifier to be unique per successful charge - # Prefer true payment/transaction identifiers over subscription id - provider_payment_id = ( - data.get('payment_id') - or data.get('invoice_id') - or data.get('order_id') - or data.get('transaction_id') - or data.get('charge_id') - or data.get('subscription_payment_id') - ) - if provider_payment_id is None: - # Fallback to subscription_id which may be stable across renewals - # To avoid deduplicating different renewals under same subscription, - # append a timestamp if available - base_sub_id = data.get('subscription_id') - paid_at = ( - data.get('paid_at') - or data.get('created_at') - or payload.get('timestamp') - or payload.get('id') + if event_name == "new_subscription": + # Build a stable provider payment id from subscription and timestamps + provider_payment_id = str(data.get("subscription_id")) + # Idempotent ensure payment + payment_record = await payment_dal.ensure_payment_with_provider_id( + session, + user_id=int(user_id), + amount=amount_float, + currency=currency, + months=months, + description="Tribute subscription", + provider="tribute", + provider_payment_id=provider_payment_id, ) - if base_sub_id is not None and paid_at is not None: - provider_payment_id = f"{base_sub_id}:{paid_at}" - elif base_sub_id is not None: - provider_payment_id = str(base_sub_id) - else: - provider_payment_id = str(provider_payment_id) - # Consider multiple Tribute events as successful charge events - success_events = { - 'new_subscription', - 'payment_succeeded', - 'subscription_renewed', - 'subscription_payment_succeeded', - 'invoice_paid', - } - - if event_name in success_events: - existing_payment = await payment_dal.get_payment_by_provider_payment_id( - session, provider_payment_id) - if existing_payment: - logging.info( - "Duplicate Tribute payment webhook ignored for provider_payment_id %s", - provider_payment_id, - ) - payment_record = existing_payment - else: - payment_record = await payment_dal.create_payment_record( - session, - { - 'user_id': user_id, - 'amount': float(price_rub), - 'currency': 'RUB', - 'status': 'succeeded', - 'description': 'Tribute subscription', - 'subscription_duration_months': months, - 'provider_payment_id': provider_payment_id, - 'provider': 'tribute', - }, - ) activation_details = await subscription_service.activate_subscription( session, - user_id, + int(user_id), months, - float(price_rub), + float(amount_float), payment_record.payment_id, - provider='tribute', + provider="tribute", ) referral_bonus = await referral_service.apply_referral_bonuses_for_payment( - session, user_id, months) + session, int(user_id), months) await session.commit() - db_user = await user_dal.get_user_by_id(session, user_id) + db_user = await user_dal.get_user_by_id(session, int(user_id)) lang = db_user.language_code if db_user and db_user.language_code else settings.DEFAULT_LANGUAGE _ = lambda k, **kw: i18n.gettext(lang, k, **kw) @@ -213,7 +174,7 @@ class TributeService: try: await bot.send_message( - user_id, + int(user_id), success_msg, reply_markup=markup, parse_mode="HTML", @@ -226,21 +187,19 @@ class TributeService: # Send notification about payment try: notification_service = NotificationService(bot, settings, i18n) - user = await user_dal.get_user_by_id(session, user_id) + user = await user_dal.get_user_by_id(session, int(user_id)) await notification_service.notify_payment_received( - user_id=user_id, - amount=float(price_rub), - currency="RUB", + user_id=int(user_id), + amount=float(amount_float), + currency=currency, months=months, payment_provider="tribute", username=user.username if user else None ) except Exception as e: logging.error(f"Failed to send tribute payment notification: {e}") - - elif event_name == 'subscription_cancelled': - # Handle tribute subscription cancellation - await self._handle_tribute_cancellation(session, user_id, bot, i18n) + elif event_name == "cancelled_subscription": + await self._handle_tribute_cancellation(session, int(user_id), bot, i18n) else: await session.commit() @@ -254,22 +213,7 @@ class TributeService: try: # Set all user's subscriptions to expire in 1 day (grace period) - grace_end_date = datetime.now(timezone.utc) + timedelta(days=1) - - # Get all active subscriptions for the user - user_subs = await subscription_dal.get_active_subscriptions_for_user(session, user_id) - - for sub in user_subs: - await subscription_dal.update_subscription( - session, - sub.subscription_id, - { - 'end_date': grace_end_date, - 'status_from_panel': 'CANCELLED', - 'skip_notifications': True # Skip future notifications for cancelled subs - } - ) - + await subscription_dal.set_user_subscriptions_cancelled_with_grace(session, user_id, grace_days=1) await session.commit() # Send notification about cancellation if enabled @@ -292,7 +236,7 @@ class TributeService: try: await bot.send_message( - user_id, + int(user_id), cancellation_msg, reply_markup=markup, parse_mode="HTML" diff --git a/db/dal/payment_dal.py b/db/dal/payment_dal.py index 9bb2c12..a52b825 100644 --- a/db/dal/payment_dal.py +++ b/db/dal/payment_dal.py @@ -47,6 +47,38 @@ async def get_payment_by_provider_payment_id( return result.scalar_one_or_none() +async def ensure_payment_with_provider_id( + session: AsyncSession, + *, + user_id: int, + amount: float, + currency: str, + months: int, + description: str, + provider: str, + provider_payment_id: str) -> Payment: + """Idempotently create a payment record for a provider event. + + If a payment with the same provider_payment_id already exists, returns it. + Otherwise creates a new succeeded payment with provided data. + """ + existing = await get_payment_by_provider_payment_id(session, provider_payment_id) + if existing: + return existing + + payment_payload: Dict[str, Any] = { + "user_id": user_id, + "amount": float(amount), + "currency": currency, + "status": "succeeded", + "description": description, + "subscription_duration_months": months, + "provider_payment_id": provider_payment_id, + "provider": provider, + } + return await create_payment_record(session, payment_payload) + + async def get_payment_by_db_id(session: AsyncSession, payment_db_id: int) -> Optional[Payment]: diff --git a/db/dal/subscription_dal.py b/db/dal/subscription_dal.py index 91ede4a..6a0b0e6 100644 --- a/db/dal/subscription_dal.py +++ b/db/dal/subscription_dal.py @@ -53,6 +53,29 @@ async def update_subscription( return sub +async def set_user_subscriptions_cancelled_with_grace( + session: AsyncSession, user_id: int, grace_days: int = 1) -> int: + """Mark all active user subscriptions as cancelled with a short grace period. + + Sets end_date to now + grace_days, status_from_panel to 'CANCELLED', and + skip future notifications to reduce noise after cancellation. + Returns number of updated rows. + """ + from datetime import datetime, timezone, timedelta + grace_end = datetime.now(timezone.utc) + timedelta(days=grace_days) + stmt = ( + update(Subscription) + .where(Subscription.user_id == user_id, Subscription.is_active == True) + .values( + end_date=grace_end, + status_from_panel="CANCELLED", + skip_notifications=True, + ) + ) + result = await session.execute(stmt) + return result.rowcount or 0 + + async def upsert_subscription(session: AsyncSession, sub_payload: Dict[str, Any]) -> Subscription: panel_sub_uuid = sub_payload.get("panel_subscription_uuid") diff --git a/db/dal/user_dal.py b/db/dal/user_dal.py index af8f58c..40854eb 100644 --- a/db/dal/user_dal.py +++ b/db/dal/user_dal.py @@ -30,20 +30,7 @@ async def get_user_by_panel_uuid( return result.scalar_one_or_none() -async def get_user( - session: AsyncSession, - *, - user_id: Optional[int] = None, - username: Optional[str] = None, - panel_uuid: Optional[str] = None, -) -> Optional[User]: - if user_id is not None: - return await get_user_by_id(session, user_id) - if username is not None: - return await get_user_by_username(session, username) - if panel_uuid is not None: - return await get_user_by_panel_uuid(session, panel_uuid) - return None +## Removed unused generic get_user helper to keep DAL explicit and simple async def create_user(session: AsyncSession, user_data: Dict[str, Any]) -> User: diff --git a/locales/en.json b/locales/en.json index 5712a5f..a7e77f4 100644 --- a/locales/en.json +++ b/locales/en.json @@ -16,6 +16,7 @@ "choose_language": "Choose language / Выберите язык:", "language_set_alert": "Language changed!", "error_occurred_try_again": "An error occurred, please try again.", + "error_try_again": "Please try again.", "error_displaying_menu": "Error displaying menu.", "main_menu_unknown_action": "Unknown action.", diff --git a/locales/ru.json b/locales/ru.json index 8197e9a..34063e2 100644 --- a/locales/ru.json +++ b/locales/ru.json @@ -16,6 +16,7 @@ "choose_language": "Выберите язык / Select language:", "language_set_alert": "Язык изменен!", "error_occurred_try_again": "Произошла ошибка, попробуйте снова.", + "error_try_again": "Попробуйте еще раз.", "error_displaying_menu": "Ошибка отображения меню.", "main_menu_unknown_action": "Неизвестное действие.", From bb7641bb744732537bd54ddc4e659b7036db50df Mon Sep 17 00:00:00 2001 From: machka-pasla Date: Thu, 7 Aug 2025 23:39:48 +0300 Subject: [PATCH 7/8] Improve amount conversion in TributeService to handle minor units - Updated the amount handling logic to convert minor currency units (kopecks/cents) to major units before persisting. - Added error handling for invalid amount inputs to ensure robustness in processing payment data. - Ensured that the amount is rounded to two decimal places for accurate representation in the system. --- bot/services/tribute_service.py | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/bot/services/tribute_service.py b/bot/services/tribute_service.py index 5401922..b65b7a6 100644 --- a/bot/services/tribute_service.py +++ b/bot/services/tribute_service.py @@ -96,10 +96,17 @@ class TributeService: period_val = data.get("period") months = convert_period_to_months(period_val) - # Price/amount from spec is integer cents in currency; we store float in Payment + # Tribute sends amount in minor units (kopecks/cents). Convert to major units before persisting. amount_value = data.get("amount") or data.get("price") currency = (data.get("currency") or settings.DEFAULT_CURRENCY_SYMBOL or "RUB").upper() - amount_float = float(amount_value) if amount_value is not None else 0.0 + if amount_value is not None: + try: + amount_minor_units = float(amount_value) + except (TypeError, ValueError): + amount_minor_units = 0.0 + amount_float = round(amount_minor_units / 100.0, 2) + else: + amount_float = 0.0 async with async_session_factory() as session: if event_name == "new_subscription": From 149ff057a92a40d8b4a63563a723e517d64b8aaa Mon Sep 17 00:00:00 2001 From: machka-pasla Date: Thu, 7 Aug 2025 23:47:19 +0300 Subject: [PATCH 8/8] Refactor subscription synchronization logic for improved handling - Enhanced the subscription syncing process to prioritize concrete subscription UUIDs for updates and creations, ensuring idempotency. - Implemented atomic updates for existing subscriptions and streamlined the creation of new subscriptions when a UUID is available. - Improved logging for subscription updates and creations to provide clearer feedback on sync actions. - Added handling for cases where no subscription UUID is present, avoiding unnecessary record creation. --- bot/handlers/admin/sync_admin.py | 112 +++++++++++++++++++++---------- 1 file changed, 76 insertions(+), 36 deletions(-) diff --git a/bot/handlers/admin/sync_admin.py b/bot/handlers/admin/sync_admin.py index b0c306c..81c4e34 100644 --- a/bot/handlers/admin/sync_admin.py +++ b/bot/handlers/admin/sync_admin.py @@ -122,48 +122,88 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession, panel_expire_at_iso.replace("Z", "+00:00") ) - # Update or create subscription - active_sub = await subscription_dal.get_active_subscription_by_user_id( - session, actual_user_id, panel_uuid + # Prefer syncing by concrete subscription UUID (shortUuid/subscriptionUuid) + subscription_uuid_from_panel = ( + panel_user_dict.get("subscriptionUuid") + or panel_user_dict.get("shortUuid") ) - - if active_sub: - # Check if subscription needs update - if (active_sub.end_date != panel_expire_at or - active_sub.status_from_panel != panel_status or - active_sub.is_active != (panel_status == "ACTIVE")): - - await subscription_dal.update_subscription_end_date( - session, active_sub.subscription_id, panel_expire_at + + if subscription_uuid_from_panel: + # Try to find subscription by its panel_subscription_uuid first (idempotent) + existing_sub_by_uuid = ( + await subscription_dal.get_subscription_by_panel_subscription_uuid( + session, subscription_uuid_from_panel + ) + ) + + if existing_sub_by_uuid: + # Atomic update of all relevant fields + await subscription_dal.update_subscription( + session, + existing_sub_by_uuid.subscription_id, + { + "user_id": actual_user_id, + "panel_user_uuid": panel_uuid, + "end_date": panel_expire_at, + "is_active": panel_status == "ACTIVE", + "status_from_panel": panel_status, + }, ) - # Update status fields - active_sub.status_from_panel = panel_status - active_sub.is_active = (panel_status == "ACTIVE") subscriptions_synced_count += 1 subscriptions_updated += 1 user_was_updated = True - logging.info(f"Updated subscription for user {actual_user_id}: expires {panel_expire_at}, status {panel_status}") + logging.info( + f"Synced existing subscription {existing_sub_by_uuid.subscription_id} for user {actual_user_id}: expires {panel_expire_at}, status {panel_status}" + ) + else: + # Create a new subscription only when we have a concrete subscription UUID + sub_payload = { + "user_id": actual_user_id, + "panel_user_uuid": panel_uuid, + "panel_subscription_uuid": subscription_uuid_from_panel, + # Do not guess precise start_date from panel; keep nullable + "start_date": None, + "end_date": panel_expire_at, + "duration_months": None, + "is_active": panel_status == "ACTIVE", + "status_from_panel": panel_status, + "traffic_limit_bytes": settings.user_traffic_limit_bytes, + } + created_sub = await subscription_dal.upsert_subscription( + session, sub_payload + ) + subscriptions_synced_count += 1 + subscriptions_created += 1 + user_was_updated = True + logging.info( + f"Created subscription {created_sub.subscription_id} for user {actual_user_id} by panel_sub_uuid {subscription_uuid_from_panel}" + ) else: - # Create new subscription record - subscription_uuid_to_use = panel_subscription_uuid or panel_uuid - - logging.info(f"Creating new subscription for user {actual_user_id} with UUID {subscription_uuid_to_use}") - - sub_payload = { - "user_id": actual_user_id, - "panel_user_uuid": panel_uuid, - "panel_subscription_uuid": subscription_uuid_to_use, - "start_date": datetime.now(timezone.utc), - "end_date": panel_expire_at, - "duration_months": 1, # Default - "is_active": panel_status == "ACTIVE", - "status_from_panel": panel_status, - "traffic_limit_bytes": settings.user_traffic_limit_bytes, - } - await subscription_dal.upsert_subscription(session, sub_payload) - subscriptions_synced_count += 1 - subscriptions_created += 1 - user_was_updated = True + # No subscription UUID from panel: only update an already active subscription for this user/panel UUID + active_sub = await subscription_dal.get_active_subscription_by_user_id( + session, actual_user_id, panel_uuid + ) + if active_sub: + await subscription_dal.update_subscription( + session, + active_sub.subscription_id, + { + "end_date": panel_expire_at, + "is_active": panel_status == "ACTIVE", + "status_from_panel": panel_status, + }, + ) + subscriptions_synced_count += 1 + subscriptions_updated += 1 + user_was_updated = True + logging.info( + f"Updated active subscription {active_sub.subscription_id} for user {actual_user_id}: expires {panel_expire_at}, status {panel_status}" + ) + else: + # Without a concrete subscription UUID we avoid creating new records to keep sync idempotent + logging.debug( + f"No subscriptionUuid for panel user {panel_uuid}; skipped creation for user {actual_user_id}" + ) except Exception as e: sync_errors.append(f"Error syncing subscription for user {actual_user_id}: {str(e)}")