diff --git a/handlers/admin/users/__init__.py b/handlers/admin/users/__init__.py index c38e8ea5..63dcc1e2 100644 --- a/handlers/admin/users/__init__.py +++ b/handlers/admin/users/__init__.py @@ -1,6 +1,6 @@ from aiogram import Router -from . import users_balance, users_bans, users_hwid, users_keys, users_manage, users_tariffs +from . import users_balance, users_bans, users_gifts, users_hwid, users_keys, users_manage, users_tariffs router = Router() @@ -10,3 +10,4 @@ router.include_router(users_hwid.router) router.include_router(users_keys.router) router.include_router(users_bans.router) router.include_router(users_tariffs.router) +router.include_router(users_gifts.router) diff --git a/handlers/admin/users/keyboard.py b/handlers/admin/users/keyboard.py index 792dea99..feda6537 100644 --- a/handlers/admin/users/keyboard.py +++ b/handlers/admin/users/keyboard.py @@ -69,7 +69,11 @@ async def build_user_edit_kb(tg_id: int, key_records: list, is_banned: bool = Fa InlineKeyboardButton( text="🤝 Выгрузить рефералов", callback_data=AdminUserEditorCallback(action="users_export_referrals", tg_id=tg_id).pack(), - ) + ), + InlineKeyboardButton( + text="🎁 Подарки", + callback_data=AdminUserEditorCallback(action="users_gifts", tg_id=tg_id).pack(), + ), ) builder.row( @@ -263,13 +267,9 @@ def build_key_edit_kb(key_details: dict, email: str, is_configurable: bool = Fal ).pack(), ) builder.button( - text="🔄 Перевыпустить", - callback_data=AdminUserEditorCallback(action="users_update_key", data=email, tg_id=key_details["tg_id"]).pack(), - ) - builder.button( - text="🔁 Пересоздать", + text="🔄 Перевыпуск подписки", callback_data=AdminUserEditorCallback( - action="users_recreate_key", data=email, tg_id=key_details["tg_id"] + action="users_reissue_menu", data=email, tg_id=key_details["tg_id"] ).pack(), ) builder.button( @@ -324,6 +324,24 @@ def build_key_edit_kb(key_details: dict, email: str, is_configurable: bool = Fal return builder.as_markup() +def build_reissue_menu_kb(email: str, tg_id: int) -> InlineKeyboardMarkup: + builder = InlineKeyboardBuilder() + builder.button( + text="📦 Полный перевыпуск", + callback_data=AdminUserEditorCallback(action="users_update_key", data=email, tg_id=tg_id).pack(), + ) + builder.button( + text="🔗 Сменить ссылку", + callback_data=AdminUserEditorCallback(action="users_recreate_key", data=email, tg_id=tg_id).pack(), + ) + builder.button( + text=BACK, + callback_data=AdminUserEditorCallback(action="users_key_edit", data=email, tg_id=tg_id).pack(), + ) + builder.adjust(1) + return builder.as_markup() + + def build_hwid_menu_kb(email: str, tg_id: int) -> InlineKeyboardMarkup: builder = InlineKeyboardBuilder() builder.button( @@ -415,3 +433,81 @@ def build_user_ban_type_kb(tg_id: int) -> InlineKeyboardMarkup: ) return builder.as_markup() + + +class AdminUserGiftCallback(CallbackData, prefix="admin_gift"): + action: str + tg_id: int + gift_id: str | None = None + page: int = 0 + + +GIFTS_PER_PAGE = 10 + + +def build_user_gifts_kb(tg_id: int, gifts: list, page: int = 0) -> InlineKeyboardMarkup: + builder = InlineKeyboardBuilder() + + total_pages = (len(gifts) + GIFTS_PER_PAGE - 1) // GIFTS_PER_PAGE if gifts else 1 + start_idx = page * GIFTS_PER_PAGE + end_idx = start_idx + GIFTS_PER_PAGE + page_gifts = gifts[start_idx:end_idx] + + row_buttons = [] + for gift in page_gifts: + created_str = gift.created_at.strftime("%d.%m.%Y") if gift.created_at else "—" + row_buttons.append( + InlineKeyboardButton( + text=f"Удалить {created_str}", + callback_data=f"user_gift_del|{tg_id}|{gift.gift_id}|{page}", + ) + ) + if len(row_buttons) == 1: + builder.row(*row_buttons) + row_buttons = [] + if row_buttons: + builder.row(*row_buttons) + + if total_pages > 1: + nav_buttons = [] + if page > 0: + nav_buttons.append( + InlineKeyboardButton( + text="◀️", + callback_data=f"user_gift_page|{tg_id}|{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"user_gift_page|{tg_id}|{page + 1}", + ) + ) + builder.row(*nav_buttons) + + builder.row(build_editor_back_btn(tg_id, True)) + return builder.as_markup() + + +def build_gift_delete_confirm_kb(tg_id: int, gift_id: str, page: int = 0) -> InlineKeyboardMarkup: + builder = InlineKeyboardBuilder() + + builder.row( + InlineKeyboardButton( + text="✅ Да, удалить", + callback_data=f"user_gift_del_c|{tg_id}|{gift_id}", + ) + ) + + builder.row( + InlineKeyboardButton( + text=BACK, + callback_data=f"user_gift_page|{tg_id}|{page}", + ) + ) + + return builder.as_markup() diff --git a/handlers/admin/users/users_gifts.py b/handlers/admin/users/users_gifts.py new file mode 100644 index 00000000..b91f9040 --- /dev/null +++ b/handlers/admin/users/users_gifts.py @@ -0,0 +1,152 @@ +import pytz + +from aiogram import F, Router, types +from sqlalchemy import delete, select +from sqlalchemy.ext.asyncio import AsyncSession + +from database.models import Gift, GiftUsage +from filters.admin import IsAdminFilter + +from .keyboard import ( + AdminUserEditorCallback, + build_gift_delete_confirm_kb, + build_user_gifts_kb, +) + + +MOSCOW_TZ = pytz.timezone("Europe/Moscow") + +router = Router() + + +async def get_user_gifts(session: AsyncSession, tg_id: int) -> list: + stmt = select(Gift).where(Gift.sender_tg_id == tg_id).order_by(Gift.created_at.desc()) + result = await session.execute(stmt) + return result.scalars().all() + + +async def show_gifts_list(message: types.Message, session: AsyncSession, tg_id: int, page: int = 0): + gifts = await get_user_gifts(session, tg_id) + + if not gifts: + text = ( + f"🎁 Подарки пользователя {tg_id}\n\n" + f"У пользователя нет созданных подарков." + ) + await message.edit_text( + text=text, + reply_markup=build_user_gifts_kb(tg_id, [], page), + ) + return + + from .keyboard import GIFTS_PER_PAGE + start_idx = page * GIFTS_PER_PAGE + end_idx = start_idx + GIFTS_PER_PAGE + page_gifts = gifts[start_idx:end_idx] + + gift_ids = [g.gift_id for g in page_gifts] + usages_stmt = select(GiftUsage).where(GiftUsage.gift_id.in_(gift_ids)) + usages_result = await session.execute(usages_stmt) + usages = usages_result.scalars().all() + usage_map = {u.gift_id: u.tg_id for u in usages} + + lines = [f"🎁 Подарки пользователя {tg_id}\n"] + + for i, gift in enumerate(page_gifts, start=start_idx + 1): + if gift.is_used: + used_by = usage_map.get(gift.gift_id) + status = f"✅ Использован: {used_by}" if used_by else "✅ Использован" + else: + status = "⏳ Не использован" + + created_str = gift.created_at.replace(tzinfo=pytz.UTC).astimezone(MOSCOW_TZ).strftime("%d.%m.%Y %H:%M") + + lines.append( + f"\n{i}.🎁 {gift.selected_months} мес.\n" + f" 📅 Создан: {created_str}\n" + f" {status}" + ) + + lines.append("\n\nНажмите кнопку для удаления:") + + await message.edit_text( + text="".join(lines), + reply_markup=build_user_gifts_kb(tg_id, gifts, page), + ) + + +@router.callback_query( + AdminUserEditorCallback.filter(F.action == "users_gifts"), + IsAdminFilter(), +) +async def handle_users_gifts( + callback: types.CallbackQuery, + callback_data: AdminUserEditorCallback, + session: AsyncSession, +): + await show_gifts_list(callback.message, session, callback_data.tg_id, page=0) + + +@router.callback_query( + F.data.startswith("user_gift_page|"), + IsAdminFilter(), +) +async def handle_gifts_page( + callback: types.CallbackQuery, + session: AsyncSession, +): + _, tg_id, page = callback.data.split("|") + await show_gifts_list(callback.message, session, int(tg_id), page=int(page)) + + +@router.callback_query( + F.data.startswith("user_gift_del|"), + IsAdminFilter(), +) +async def handle_gift_delete( + callback: types.CallbackQuery, + session: AsyncSession, +): + _, tg_id, gift_id, page = callback.data.split("|") + tg_id, page = int(tg_id), int(page) + + stmt = select(Gift).where(Gift.gift_id == gift_id) + result = await session.execute(stmt) + gift = result.scalar_one_or_none() + + if not gift: + await callback.answer("❌ Подарок не найден", show_alert=True) + return + + created_str = gift.created_at.replace(tzinfo=pytz.UTC).astimezone(MOSCOW_TZ).strftime("%d.%m.%Y %H:%M") + status = "✅ Использован" if gift.is_used else "⏳ Не использован" + + await callback.message.edit_text( + text=( + f"❓ Удалить подарок?\n\n" + f"📆 Длительность: {gift.selected_months} дн.\n" + f"📅 Создан: {created_str}\n" + f"📊 Статус: {status}\n\n" + f"⚠️ Это действие необратимо!" + ), + reply_markup=build_gift_delete_confirm_kb(tg_id, gift_id, page), + ) + + +@router.callback_query( + F.data.startswith("user_gift_del_c|"), + IsAdminFilter(), +) +async def handle_gift_delete_confirm( + callback: types.CallbackQuery, + session: AsyncSession, +): + _, tg_id, gift_id = callback.data.split("|") + tg_id = int(tg_id) + + await session.execute(delete(GiftUsage).where(GiftUsage.gift_id == gift_id)) + await session.execute(delete(Gift).where(Gift.gift_id == gift_id)) + await session.commit() + + await callback.answer("✅ Подарок удалён", show_alert=True) + await show_gifts_list(callback.message, session, tg_id, page=0) diff --git a/handlers/admin/users/users_keys.py b/handlers/admin/users/users_keys.py index 018807ab..f1a18dea 100644 --- a/handlers/admin/users/users_keys.py +++ b/handlers/admin/users/users_keys.py @@ -52,6 +52,7 @@ from .keyboard import ( build_editor_kb, build_key_delete_kb, build_key_edit_kb, + build_reissue_menu_kb, build_user_delete_kb, build_users_key_expiry_kb, build_users_key_show_kb, @@ -343,6 +344,33 @@ async def handle_expiry_time_input(message: Message, state: FSMContext, session: await message.answer(text=text, reply_markup=build_users_key_show_kb(tg_id, email)) +@router.callback_query( + AdminUserEditorCallback.filter(F.action == "users_reissue_menu"), + IsAdminFilter(), +) +async def handle_reissue_menu( + callback_query: CallbackQuery, + callback_data: AdminUserEditorCallback, +): + tg_id = callback_data.tg_id + email = callback_data.data + + text = ( + "🔄 Перевыпуск подписки\n\n" + "📦 Полный перевыпуск\n" + "Пересоздаёт подписку на сервере с возможностью выбора кластера. " + "Используйте для переноса на другой сервер или обновления данных.\n\n" + "🔗 Сменить ссылку\n" + "Генерирует новую ссылку подписки. Старая ссылка перестанет работать. " + "Все данные подписки сохранятся." + ) + + await callback_query.message.edit_text( + text=text, + reply_markup=build_reissue_menu_kb(email, tg_id), + ) + + @router.callback_query( AdminUserEditorCallback.filter(F.action == "users_update_key"), IsAdminFilter(), diff --git a/handlers/admin/users/users_manage.py b/handlers/admin/users/users_manage.py index 14153681..bc327a49 100644 --- a/handlers/admin/users/users_manage.py +++ b/handlers/admin/users/users_manage.py @@ -339,7 +339,7 @@ async def process_user_search( ) -> None: await state.clear() - stmt_user = select(User.username, User.balance, User.created_at, User.updated_at).where(User.tg_id == tg_id) + stmt_user = select(User.username, User.balance, User.created_at, User.updated_at, User.trial).where(User.tg_id == tg_id) result_user = await session.execute(stmt_user) user_data = result_user.first() @@ -350,11 +350,13 @@ async def process_user_search( ) return - username, balance, created_at, updated_at = user_data + username, balance, created_at, updated_at, trial = user_data balance = int(balance or 0) created_at_str = created_at.replace(tzinfo=pytz.UTC).astimezone(MOSCOW_TZ).strftime("%H:%M:%S %d.%m.%Y") updated_at_str = updated_at.replace(tzinfo=pytz.UTC).astimezone(MOSCOW_TZ).strftime("%H:%M:%S %d.%m.%Y") + trial_status = "использован" if trial == 1 else "доступен" + stmt_ref_count = select(func.count()).select_from(Referral).where(Referral.referrer_tg_id == tg_id) result_ref = await session.execute(stmt_ref_count) referral_count = result_ref.scalar_one() @@ -384,28 +386,49 @@ async def process_user_search( result_keys = await session.execute(stmt_keys) key_records = result_keys.scalars().all() + stmt_ban = select(ManualBan).where(ManualBan.tg_id == tg_id).limit(1) + result_ban = await session.execute(stmt_ban) + ban_record = result_ban.scalar_one_or_none() + + ban_info = None + ban_reason = None + is_banned = ban_record is not None + if ban_record: + if ban_record.reason == "shadow": + ban_info = "🚫 Блокировка: 👻 Теневой бан" + elif ban_record.until: + until_str = ban_record.until.replace(tzinfo=pytz.UTC).astimezone(MOSCOW_TZ).strftime("%d.%m.%Y %H:%M") + ban_info = f"🚫 Блокировка: до {until_str}" + if ban_record.reason: + ban_reason = ban_record.reason + else: + ban_info = "🚫 Блокировка: навсегда" + if ban_record.reason: + ban_reason = ban_record.reason + body = Text( f"🆔 ID: {tg_id}\n", - f"📄 Логин: @{username}" if username else "📄 Логин: —", - "\n", + f"📄 Логин: @{username}\n" if username else "📄 Логин: —\n", f"📅 Дата регистрации: {created_at_str}\n", f"🏃 Дата активности: {updated_at_str}\n", f"💰 Баланс: {balance} Р.\n", f"💳 Пополнения: {topups_sum} Р. ({topups_amount} шт.)\n", f"👥 Количество рефералов: {referral_count}\n", + f"🎁 Триал: {trial_status}\n", ) if referrer_text: body += Text(referrer_text, "\n") + if ban_info: + body += Text(ban_info, "\n") + if ban_reason: + body += Text(f"📝 Причина: {ban_reason}\n") + text_builder = Text(Bold("📊 Информация о пользователе"), "\n\n", BlockQuote(body)) text = text_builder.as_html() - stmt_ban = select(1).where(ManualBan.tg_id == tg_id).limit(1) - result_ban = await session.execute(stmt_ban) - is_banned = result_ban.scalar_one_or_none() is not None - kb = await build_user_edit_kb(tg_id, key_records, is_banned=is_banned) if edit: diff --git a/handlers/buttons.py b/handlers/buttons.py index 2b6f5500..be137aca 100644 --- a/handlers/buttons.py +++ b/handlers/buttons.py @@ -18,6 +18,7 @@ GIFTS = "🎁 Подарить" INSTRUCTIONS = "📘 Инструкции" TOP_FIVE = "🏆 Топ-5" TRIAL_SUB = "🎁 Пробная подписка" +TRIAL_BONUS = "🚀 Активировать пробный период" MY_SUB = "🔐 Моя подписка" RENEW_SUB = "🔄 Обновить подписку" diff --git a/handlers/notifications/general_notifications.py b/handlers/notifications/general_notifications.py index c73c7ae7..74aaf96c 100644 --- a/handlers/notifications/general_notifications.py +++ b/handlers/notifications/general_notifications.py @@ -1,10 +1,11 @@ import asyncio - +from dataclasses import dataclass from datetime import datetime, timedelta +from typing import Any, Optional import pytz - from aiogram import Bot, Router +from sqlalchemy import select, text, update from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from config import ( @@ -35,6 +36,7 @@ from database import ( update_key_expiry, update_key_tariff, ) +from database.models import Key, Tariff, User from database.tariffs import ( check_tariff_exists, get_tariff_by_id, @@ -69,607 +71,634 @@ moscow_tz = pytz.timezone("Europe/Moscow") notification_lock = asyncio.Lock() -async def periodic_notifications(bot: Bot, *, sessionmaker: async_sessionmaker): - while True: - notification_interval = int(NOTIFICATIONS_CONFIG.get("BASE_NOTIFICATION_MINUTE", NOTIFICATION_TIME)) - - if notification_lock.locked(): - logger.warning("Уведомления уже выполняются. Пропуск...") - await asyncio.sleep(notification_interval) - continue - - async with notification_lock: - try: - async with sessionmaker() as session: - logger.info("Запуск обработки уведомлений") - - current_time = int(datetime.now(moscow_tz).timestamp() * 1000) - - try: - keys = await get_all_keys(session=session) - keys = [k for k in keys if not k.is_frozen] - except Exception as error: - logger.error(f"Ошибка при получении ключей: {error}") - keys = [] - - trial_time_disable = bool(MODES_CONFIG.get("TRIAL_TIME_DISABLED", TRIAL_TIME_DISABLE)) - - if not trial_time_disable: - try: - await notify_inactive_trial_users(bot, session) - except Exception as error: - logger.error(f"Ошибка в notify_inactive_trial_users: {error}") - - notify_24_enabled = bool(NOTIFICATIONS_CONFIG.get("EXPIRY_24H_ENABLED", NOTIFY_24H_ENABLED)) - notify_24_hours = int(NOTIFICATIONS_CONFIG.get("EXPIRY_24H_BEFORE_HOURS", NOTIFY_24H_HOURS)) - notify_10_enabled = bool(NOTIFICATIONS_CONFIG.get("EXPIRY_10H_ENABLED", NOTIFY_10H_ENABLED)) - notify_10_hours = int(NOTIFICATIONS_CONFIG.get("EXPIRY_10H_BEFORE_HOURS", NOTIFY_10H_HOURS)) - notify_renew_enabled = bool(NOTIFICATIONS_CONFIG.get("RENEW_ENABLED", NOTIFY_RENEW)) - inactive_traffic_enabled = bool( - NOTIFICATIONS_CONFIG.get("INACTIVE_TRAFFIC_ENABLED", NOTIFY_INACTIVE_TRAFFIC) - ) - notify_hot_leads_enabled = bool(NOTIFICATIONS_CONFIG.get("HOT_LEADS_ENABLED", NOTIFY_HOT_LEADS)) - - if notify_24_enabled: - try: - threshold_24h = int( - (datetime.now(moscow_tz) + timedelta(hours=notify_24_hours)).timestamp() * 1000 - ) - await notify_24h_keys( - bot, - session, - current_time, - threshold_24h, - keys, - notify_24_hours, - notify_renew_enabled, - ) - except Exception as error: - logger.error(f"Ошибка в notify_24h_keys: {error}") - - if notify_10_enabled: - try: - threshold_10h = int( - (datetime.now(moscow_tz) + timedelta(hours=notify_10_hours)).timestamp() * 1000 - ) - await notify_10h_keys( - bot, - session, - current_time, - threshold_10h, - keys, - notify_10_hours, - notify_renew_enabled, - ) - except Exception as error: - logger.error(f"Ошибка в notify_10h_keys: {error}") - - try: - await handle_expired_keys(bot, session, current_time, keys) - except Exception as error: - logger.error(f"Ошибка в handle_expired_keys: {error}") - - if inactive_traffic_enabled: - try: - await notify_users_no_traffic(bot, session, current_time, keys) - except Exception as error: - logger.error(f"Ошибка в notify_users_no_traffic: {error}") - - try: - await run_hooks("periodic_notifications", bot=bot, session=session, keys=keys) - except Exception as error: - logger.error(f"Ошибка в хуках periodic_notifications: {error}") - - if notify_hot_leads_enabled: - try: - await notify_hot_leads(bot, session) - except Exception as error: - logger.error(f"Ошибка в notify_hot_leads: {error}") - - logger.info("Уведомления завершены") - except Exception as error: - logger.error(f"Ошибка в periodic_notifications: {error}") - - await asyncio.sleep(notification_interval) +@dataclass +class NotificationContext: + bot: Bot + session: AsyncSession + current_time: int + preload_data: Optional[dict] = None + bulk_updates: Optional[dict] = None + + def get_balance(self, tg_id: int) -> float: + if self.preload_data and tg_id in self.preload_data.get("balances_cache", {}): + return self.preload_data["balances_cache"][tg_id] + return 0.0 + + def get_tariff(self, tariff_id: int) -> Optional[dict]: + if self.preload_data and tariff_id in self.preload_data.get("tariffs_cache", {}): + return self.preload_data["tariffs_cache"][tariff_id] + return None -async def notify_24h_keys( - bot: Bot, - session: AsyncSession, - current_time: int, - threshold_time_24h: int, - keys: list, - notification_hours: int, - notify_renew_enabled: bool, -): - logger.info(f"Начало проверки подписок, истекающих через {notification_hours} часов.") - expiring_keys = [key for key in keys if key.expiry_time and current_time < key.expiry_time <= threshold_time_24h] - logger.info(f"Найдено {len(expiring_keys)} подписок, истекающих через {notification_hours} часов.") - - tg_ids = [getattr(key, "tg_id", key["tg_id"]) for key in expiring_keys] - emails = [key.email or "" for key in expiring_keys] - allowed = await check_notifications_bulk(session, "key_24h", notification_hours, tg_ids=tg_ids, emails=emails) - - allowed_set = {(user["tg_id"], user["email"]) for user in allowed} - messages = [] - - for key in expiring_keys: - tg_id = getattr(key, "tg_id", key["tg_id"]) - email = key.email or "" - if (tg_id, email) not in allowed_set: - continue - - notification_id = f"{email}_key_24h" - - can_notify = await check_notification_time(session, tg_id, notification_id, hours=notification_hours) - if not can_notify: - continue - - expiry_data = await prepare_key_expiry_data(key, session, current_time) - - notification_text = KEY_EXPIRY.format( - email=email, - hours_left_formatted=expiry_data["hours_left_formatted"], - formatted_expiry_date=expiry_data["formatted_expiry_date"], - tariff_name=expiry_data["tariff_name"], - tariff_details=expiry_data["tariff_details"], +async def preload_notification_data(session: AsyncSession) -> dict[str, Any]: + stmt = ( + select( + Key, + Tariff, + User.balance.label("user_balance"), ) + .outerjoin(Tariff, Key.tariff_id == Tariff.id) + .outerjoin(User, Key.tg_id == User.tg_id) + .where(Key.is_frozen.is_(False)) + ) - if notify_renew_enabled: - try: - await process_auto_renew_or_notify( - bot, - session, - key, - notification_id, - 1, - "notify_24h.jpg", - notification_text, + result = await session.execute(stmt) + rows = result.all() + + keys_data = {} + tariffs_cache = {} + balances_cache = {} + + for row in rows: + key = row[0] + tariff = row[1] + balance = row[2] or 0.0 + + key_dict = { + "key": key, + "tariff": dict(tariff.__dict__) if tariff else None, + "balance": float(balance), + } + keys_data[key.client_id] = key_dict + + if tariff and tariff.id not in tariffs_cache: + tariffs_cache[tariff.id] = dict(tariff.__dict__) + + balances_cache[key.tg_id] = float(balance) + + return { + "keys_data": keys_data, + "tariffs_cache": tariffs_cache, + "balances_cache": balances_cache, + } + + +async def execute_bulk_updates(session: AsyncSession, bulk_updates: dict[str, Any]) -> None: + try: + if bulk_updates["balance_changes"]: + for tg_id, balance_change in bulk_updates["balance_changes"].items(): + await session.execute( + text("UPDATE users SET balance = balance + :change WHERE tg_id = :tg_id"), + {"change": balance_change, "tg_id": tg_id} ) - except Exception as error: - logger.error(f"Ошибка авто-продления/уведомления для пользователя {tg_id}: {error}") - continue + logger.info(f"Bulk: обновлено {len(bulk_updates['balance_changes'])} балансов") + + if bulk_updates["key_expiry_updates"]: + for client_id, new_expiry in bulk_updates["key_expiry_updates"]: + await session.execute( + update(Key) + .where(Key.client_id == client_id) + .values(expiry_time=new_expiry) + ) + logger.info(f"Bulk: обновлено {len(bulk_updates['key_expiry_updates'])} сроков действия ключей") + + if bulk_updates["key_tariff_updates"]: + for client_id, new_tariff_id in bulk_updates["key_tariff_updates"]: + await session.execute( + update(Key) + .where(Key.client_id == client_id) + .values(tariff_id=new_tariff_id) + ) + logger.info(f"Bulk: обновлено {len(bulk_updates['key_tariff_updates'])} тарифов ключей") + + for tg_id, notification_type in bulk_updates["notifications_to_add"]: + await add_notification(session, tg_id, notification_type) + + for tg_id, notification_type in bulk_updates["notifications_to_delete"]: + await delete_notification(session, tg_id, notification_type) + + if bulk_updates["notifications_to_add"] or bulk_updates["notifications_to_delete"]: + logger.info( + f"Bulk: обработано {len(bulk_updates['notifications_to_add'])} добавлений " + f"и {len(bulk_updates['notifications_to_delete'])} удалений уведомлений" + ) + + await session.commit() + + except Exception as error: + logger.error(f"Ошибка в bulk-обновлениях: {error}") + await session.rollback() + raise + + +async def send_expiry_warning(ctx: NotificationContext, key, hours_left: int, photo: str) -> bool: + tg_id = key.tg_id + email = key.email or "" + + expiry_data = await prepare_key_expiry_data(key, ctx.session, ctx.current_time) + + message_text = KEY_EXPIRY.format( + email=email, + hours_left_formatted=expiry_data["hours_left_formatted"], + formatted_expiry_date=expiry_data["formatted_expiry_date"], + tariff_name=expiry_data["tariff_name"], + tariff_details=expiry_data["tariff_details"], + ) + + keyboard = build_notification_kb(email) + return await send_notification(ctx.bot, tg_id, photo, message_text, keyboard) + + +async def send_cannot_renew(ctx: NotificationContext, key, photo: str) -> bool: + tg_id = key.tg_id + email = key.email or "" + + expiry_data = await prepare_key_expiry_data(key, ctx.session, ctx.current_time) + + message_text = KEY_CANNOT_RENEW_CURRENT.format( + email=email, + hours_left_formatted=expiry_data["hours_left_formatted"], + formatted_expiry_date=expiry_data["formatted_expiry_date"], + tariff_name=expiry_data["tariff_name"], + tariff_details=expiry_data["tariff_details"], + ) + + keyboard = build_change_tariff_kb(email) + return await send_notification(ctx.bot, tg_id, photo, message_text, keyboard) + + +async def send_expired_notification(ctx: NotificationContext, key, delay_minutes: int) -> bool: + tg_id = key.tg_id + email = key.email or "" + + if delay_minutes > 0: + hours = delay_minutes // 60 + minutes = delay_minutes % 60 + if hours > 0 and minutes > 0: + time_formatted = f"{format_hours(hours)} и {format_minutes(minutes)}" + elif hours > 0: + time_formatted = format_hours(hours) else: - keyboard = build_notification_kb(email) - messages.append({ - "tg_id": tg_id, - "text": notification_text, - "photo": "notify_24h.jpg", - "keyboard": keyboard, - "notification_id": notification_id, - "email": email, - }) - - if messages: - results = await send_messages_with_limit(bot, messages, session=session) - sent_count = 0 - for msg, result in zip(messages, results, strict=False): - tg_id = msg["tg_id"] - - await add_notification(session, tg_id, msg["notification_id"]) - if result: - sent_count += 1 - logger.info(f"Отправлено уведомление об истекающей подписке {msg['email']} пользователю {tg_id}.") - else: - logger.warning( - f"Не удалось отправить уведомление об истекающей подписке {msg['email']} пользователю {tg_id}." - ) - logger.info(f"Отправлено {sent_count} уведомлений об истечении подписки через {notification_hours} часов.") - - logger.info(f"Обработка всех уведомлений за {notification_hours} часов завершена.") - await asyncio.sleep(1) + time_formatted = format_minutes(minutes) + message_text = KEY_EXPIRED_DELAY_MSG.format(email=email, time_formatted=time_formatted) + else: + message_text = KEY_EXPIRED_NO_DELAY_MSG.format(email=email) + + keyboard = build_notification_kb(email) + return await send_notification(ctx.bot, tg_id, "notify_expired.jpg", message_text, keyboard) -async def notify_10h_keys( - bot: Bot, - session: AsyncSession, - current_time: int, - threshold_time_10h: int, - keys: list, - notification_hours: int, - notify_renew_enabled: bool, -): - logger.info(f"Начало проверки подписок, истекающих через {notification_hours} часов.") - expiring_keys = [key for key in keys if key.expiry_time and current_time < key.expiry_time <= threshold_time_10h] - logger.info(f"Найдено {len(expiring_keys)} подписок, истекающих через {notification_hours} часов.") - - tg_ids = [key.tg_id for key in expiring_keys] - emails = [key.email or "" for key in expiring_keys] - allowed = await check_notifications_bulk(session, "key_10h", notification_hours, tg_ids=tg_ids, emails=emails) - - allowed_set = {(user["tg_id"], user["email"]) for user in allowed} - messages = [] - - for key in expiring_keys: - tg_id = key.tg_id - email = key.email or "" - if (tg_id, email) not in allowed_set: - continue - - notification_id = f"{email}_key_10h" - - can_notify = await check_notification_time(session, tg_id, notification_id, hours=notification_hours) - if not can_notify: - continue - - expiry_data = await prepare_key_expiry_data(key, session, current_time) - - notification_text = KEY_EXPIRY.format( - email=email, - hours_left_formatted=expiry_data["hours_left_formatted"], - formatted_expiry_date=expiry_data["formatted_expiry_date"], - tariff_name=expiry_data["tariff_name"], - tariff_details=expiry_data["tariff_details"], - ) - - if notify_renew_enabled: - try: - await process_auto_renew_or_notify( - bot, - session, - key, - notification_id, - 1, - "notify_10h.jpg", - notification_text, - ) - except Exception as error: - logger.error(f"Ошибка авто-продления/уведомления для пользователя {tg_id}: {error}") - continue - else: - keyboard = build_notification_kb(email) - messages.append({ - "tg_id": tg_id, - "text": notification_text, - "photo": "notify_10h.jpg", - "keyboard": keyboard, - "notification_id": notification_id, - "email": email, - }) - - if messages: - results = await send_messages_with_limit(bot, messages, session=session) - sent_count = 0 - for msg, result in zip(messages, results, strict=False): - tg_id = msg["tg_id"] - - await add_notification(session, tg_id, msg["notification_id"]) - if result: - sent_count += 1 - logger.info(f"Отправлено уведомление об истекающей подписке {msg['email']} пользователю {tg_id}.") - else: - logger.warning( - f"Не удалось отправить уведомление об истекающей подписке {msg['email']} пользователю {tg_id}." - ) - logger.info(f"Отправлено {sent_count} уведомлений об истечении подписки через {notification_hours} часов.") - - logger.info(f"Обработка всех уведомлений за {notification_hours} часов завершена.") - await asyncio.sleep(1) +async def send_deleted_notification(ctx: NotificationContext, key) -> bool: + tg_id = key.tg_id + email = key.email or "" + + message_text = KEY_DELETED_MSG.format(email=email) + keyboard = build_notification_expired_kb() + return await send_notification(ctx.bot, tg_id, "notify_expired.jpg", message_text, keyboard) -async def handle_expired_keys( - bot: Bot, - session: AsyncSession, - current_time: int, - keys: list, -): - logger.info("Начало обработки истекших ключей.") - - expired_keys = [key for key in keys if key.expiry_time and key.expiry_time < current_time] - logger.info(f"Найдено {len(expired_keys)} истекших ключей.") - - tg_ids = [key.tg_id for key in expired_keys] - emails = [key.email or "" for key in expired_keys] - users = await check_notifications_bulk(session, "key_expired", 0, tg_ids=tg_ids, emails=emails) - - notify_renew_expired_enabled = bool(NOTIFICATIONS_CONFIG.get("RENEW_EXPIRED_ENABLED", NOTIFY_RENEW_EXPIRED)) - notify_delete_key_enabled = bool(NOTIFICATIONS_CONFIG.get("DELETE_KEY_ENABLED", NOTIFY_DELETE_KEY)) - delete_key_delay_minutes = int(NOTIFICATIONS_CONFIG.get("DELETE_KEY_DELAY_MINUTES", NOTIFY_DELETE_DELAY)) - - messages = [] - - for key in expired_keys: - tg_id = key.tg_id - email = key.email or "" - client_id = key.client_id - server_id = key.server_id - notification_id = f"{email}_key_expired" - - last_notification_time = await get_last_notification_time(session, tg_id, notification_id) - - if notify_renew_expired_enabled: - try: - standard_caption = "" - tariff_id = getattr(key, "tariff_id", None) - if tariff_id and await check_tariff_exists(session, int(tariff_id)): - tariff = await get_tariff_by_id(session, int(tariff_id)) - if tariff: - selected_device_limit = getattr(key, "selected_device_limit", None) - selected_traffic_limit = getattr(key, "selected_traffic_limit", None) - selected_traffic_gb = int(selected_traffic_limit) if selected_traffic_limit is not None else None - - device_limit_effective, traffic_limit_bytes_effective = await get_effective_limits_for_key( - session=session, - tariff_id=int(tariff["id"]), - selected_device_limit=int(selected_device_limit) if selected_device_limit is not None else None, - selected_traffic_gb=selected_traffic_gb, - ) - traffic_limit_gb_effective = ( - int(traffic_limit_bytes_effective / GB) if traffic_limit_bytes_effective else 0 - ) - - standard_caption = get_renewal_message( - tariff_name=tariff.get("name", ""), - traffic_limit=traffic_limit_gb_effective, - device_limit=device_limit_effective, - subgroup_title=tariff.get("subgroup_title", ""), - ) - - renewed = await process_auto_renew_or_notify( - bot, - session, - key, - notification_id, - 1, - "notify_expired.jpg", - standard_caption, - ) - - if renewed: - continue - - except Exception as error: - logger.error(f"Ошибка авто-продления для пользователя {tg_id}: {error}") - continue - - if notify_delete_key_enabled: - delete_immediately = delete_key_delay_minutes == 0 - delete_after_delay = False - - if last_notification_time is not None: - delete_after_delay = (current_time - last_notification_time) / (1000 * 60) >= delete_key_delay_minutes - logger.info( - f"Прошло минут={(current_time - last_notification_time) / (1000 * 60):.2f} " - f"DELETE_KEY_DELAY_MINUTES={delete_key_delay_minutes}" - ) - - if delete_immediately or delete_after_delay: - try: - await delete_key_from_cluster(server_id, email, client_id, session) - await delete_key(session, client_id) - logger.info(f"🗑 Ключ {client_id} для пользователя {tg_id} успешно удалён.") - - keyboard = build_notification_expired_kb() - messages.append({ - "tg_id": tg_id, - "text": KEY_DELETED_MSG.format(email=email), - "photo": "notify_expired.jpg", - "keyboard": keyboard, - "notification_id": notification_id, - "email": email, - }) - except Exception as error: - logger.error(f"Ошибка удаления ключа {client_id} для пользователя {tg_id}: {error}") - continue - - if last_notification_time is None and any(user["tg_id"] == tg_id and user["email"] == email for user in users): - keyboard = build_notification_kb(email) - - if delete_key_delay_minutes > 0: - hours = delete_key_delay_minutes // 60 - minutes = delete_key_delay_minutes % 60 - if hours > 0 and minutes > 0: - time_formatted = f"{format_hours(hours)} и {format_minutes(minutes)}" - elif hours > 0: - time_formatted = format_hours(hours) - else: - time_formatted = format_minutes(minutes) - - delay_message = KEY_EXPIRED_DELAY_MSG.format(email=email, time_formatted=time_formatted) - else: - delay_message = KEY_EXPIRED_NO_DELAY_MSG.format(email=email) - - messages.append({ - "tg_id": tg_id, - "text": delay_message, - "photo": "notify_expired.jpg", - "keyboard": keyboard, - "notification_id": notification_id, - "email": email, - }) - - if messages: - results = await send_messages_with_limit(bot, messages, session=session) - sent_count = 0 - for msg, result in zip(messages, results, strict=False): - await add_notification(session, msg["tg_id"], msg["notification_id"]) - if result: - sent_count += 1 - logger.info(f"📢 Уведомление об истекшем ключе {msg['email']} отправлено пользователю {msg['tg_id']}.") - else: - logger.warning( - f"📢 Не удалось отправить уведомление об истекшем ключе {msg['email']} пользователю {msg['tg_id']}." - ) - - logger.info(f"Отправлено {sent_count} уведомлений об истекших ключах.") - - logger.info("Обработка истекших ключей завершена.") - await asyncio.sleep(1) +async def send_renewed_notification(ctx: NotificationContext, key, tariff: dict, new_expiry_time: int) -> bool: + tg_id = key.tg_id + email = key.email or "" + + selected_device_limit = getattr(key, "selected_device_limit", None) + selected_traffic_limit = getattr(key, "selected_traffic_limit", None) + selected_traffic_gb = int(selected_traffic_limit) if selected_traffic_limit is not None else None + + device_limit_effective, traffic_limit_bytes_effective = await get_effective_limits_for_key( + session=ctx.session, + tariff_id=int(tariff["id"]), + selected_device_limit=int(selected_device_limit) if selected_device_limit is not None else None, + selected_traffic_gb=selected_traffic_gb, + ) + traffic_limit_gb = int(traffic_limit_bytes_effective / GB) if traffic_limit_bytes_effective else 0 + + formatted_expiry_date = datetime.fromtimestamp(new_expiry_time / 1000, tz=moscow_tz).strftime("%d %B %Y, %H:%M") + formatted_expiry_date = formatted_expiry_date.replace( + datetime.fromtimestamp(new_expiry_time / 1000, tz=moscow_tz).strftime("%B"), + get_russian_month(datetime.fromtimestamp(new_expiry_time / 1000, tz=moscow_tz)), + ) + + message_text = get_renewal_message( + tariff_name=tariff["name"], + traffic_limit=traffic_limit_gb, + device_limit=device_limit_effective, + expiry_date=formatted_expiry_date, + subgroup_title=tariff.get("subgroup_title", ""), + ) + + keyboard = build_notification_expired_kb() + result = await send_notification(ctx.bot, tg_id, "notify_expired.jpg", message_text, keyboard) + + if result: + logger.info(f"✅ Уведомление о продлении подписки {email} отправлено пользователю {tg_id}.") + else: + logger.warning(f"📢 Не удалось отправить уведомление о продлении подписки {email} пользователю {tg_id}.") + + return result -async def process_auto_renew_or_notify( - bot, - conn, - key, - notification_id: str, - renewal_period_months: int, - standard_photo: str, - standard_caption: str, -): - """Пытается автопродлить ключ или отправить уведомление.""" +async def try_auto_renew(ctx: NotificationContext, key) -> tuple[bool, Optional[dict], Optional[int]]: tg_id = key.tg_id email = key.email or "" renew_notification_id = f"{email}_renew" - + + can_renew = await check_notification_time(ctx.session, tg_id, renew_notification_id, hours=24) + if not can_renew: + logger.debug(f"⏳ Подписка {email} уже продлевалась в течение последних 24 часов.") + return False, None, None + + if ctx.preload_data and tg_id in ctx.preload_data.get("balances_cache", {}): + balance = ctx.preload_data["balances_cache"][tg_id] + else: + balance = await get_balance(ctx.session, tg_id) + + server_id = key.server_id + tariff_id = key.tariff_id + + tariffs = await get_tariffs_for_cluster(ctx.session, server_id) + if not tariffs: + logger.warning(f"⛔ Нет доступных тарифов для продления подписки {email}") + return False, None, None + + current_tariff = None + if tariff_id: + current_tariff = ctx.get_tariff(tariff_id) + if not current_tariff and await check_tariff_exists(ctx.session, tariff_id): + current_tariff = await get_tariff_by_id(ctx.session, tariff_id) + + if not current_tariff: + return False, None, None + + forbidden_groups = ["discounts", "discounts_max", "gifts", "trial"] try: - can_renew = await check_notification_time(conn, tg_id, renew_notification_id, hours=24) - if not can_renew: - logger.debug( - f"⏳ Подписка {email} уже продлевалась в течение последних 24 часов, повторное продление отменено." - ) - return False - - balance = await get_balance(conn, tg_id) - server_id = key.server_id - tariff_id = key.tariff_id - - tariffs = await get_tariffs_for_cluster(conn, server_id) - if not tariffs: - logger.warning(f"⛔ Нет доступных тарифов для продления подписки {email} (сервер: {server_id})") - return False - - selected_tariff = None - - if tariff_id and await check_tariff_exists(conn, tariff_id): - current_tariff = await get_tariff_by_id(conn, tariff_id) - - forbidden_groups = ["discounts", "discounts_max", "gifts", "trial"] - try: - hook_results = await run_hooks("renewal_forbidden_groups", chat_id=tg_id, admin=False, session=conn) - for hook_result in hook_results: - additional_groups = hook_result.get("additional_groups", []) - forbidden_groups.extend(additional_groups) - except Exception as error: - logger.warning(f"[AUTO_RENEW] Ошибка при получении дополнительных групп: {error}") - - if current_tariff and current_tariff["group_code"] not in forbidden_groups: - renewal_cost = await resolve_price_to_charge( - conn, - { - "tariff_id": current_tariff.get("id"), - "selected_device_limit": getattr(key, "selected_device_limit", None), - "selected_traffic_limit": getattr(key, "selected_traffic_limit", None), - "selected_price_rub": getattr(key, "selected_price_rub", None), - }, - ) - - if renewal_cost is not None and balance >= renewal_cost: - selected_tariff = current_tariff - else: - selected_tariff = None - else: - selected_tariff = None + hook_results = await run_hooks("renewal_forbidden_groups", chat_id=tg_id, admin=False, session=ctx.session) + for hook_result in hook_results: + additional_groups = hook_result.get("additional_groups", []) + forbidden_groups.extend(additional_groups) + except Exception as error: + logger.warning(f"[AUTO_RENEW] Ошибка при получении дополнительных групп: {error}") + + if current_tariff["group_code"] in forbidden_groups: + return False, None, None + + renewal_cost = await resolve_price_to_charge( + ctx.session, + { + "tariff_id": current_tariff.get("id"), + "selected_device_limit": getattr(key, "selected_device_limit", None), + "selected_traffic_limit": getattr(key, "selected_traffic_limit", None), + "selected_price_rub": getattr(key, "selected_price_rub", None), + }, + ) + + if renewal_cost is None or balance < renewal_cost: + return False, None, None + + client_id = key.client_id + current_expiry = key.expiry_time + duration_days = current_tariff["duration_days"] + + selected_device_limit = getattr(key, "selected_device_limit", None) + selected_traffic_limit = getattr(key, "selected_traffic_limit", None) + selected_traffic_gb = int(selected_traffic_limit) if selected_traffic_limit is not None else None + + device_limit_effective, traffic_limit_bytes_effective = await get_effective_limits_for_key( + session=ctx.session, + tariff_id=int(current_tariff["id"]), + selected_device_limit=int(selected_device_limit) if selected_device_limit is not None else None, + selected_traffic_gb=selected_traffic_gb, + ) + traffic_limit_gb = int(traffic_limit_bytes_effective / GB) if traffic_limit_bytes_effective else 0 + + new_expiry_time = ( + current_expiry + if current_expiry > datetime.utcnow().timestamp() * 1000 + else datetime.utcnow().timestamp() * 1000 + ) + duration_days * 24 * 60 * 60 * 1000 + + logger.info( + f"Продление подписки {email} на {duration_days} дней для пользователя {tg_id}. " + f"Баланс: {balance}, списываем: {renewal_cost}" + ) + + key_subgroup = current_tariff.get("subgroup_title") + + await renew_key_in_cluster( + cluster_id=server_id, + email=email, + client_id=client_id, + new_expiry_time=int(new_expiry_time), + total_gb=traffic_limit_gb, + hwid_device_limit=device_limit_effective, + session=ctx.session, + target_subgroup=key_subgroup, + old_subgroup=key_subgroup, + plan=current_tariff["id"], + ) + + if ctx.bulk_updates is not None: + if tg_id in ctx.bulk_updates["balance_changes"]: + ctx.bulk_updates["balance_changes"][tg_id] -= renewal_cost else: - selected_tariff = None + ctx.bulk_updates["balance_changes"][tg_id] = -renewal_cost + + ctx.bulk_updates["key_expiry_updates"].append((client_id, int(new_expiry_time))) + ctx.bulk_updates["key_tariff_updates"].append((client_id, current_tariff["id"])) + ctx.bulk_updates["notifications_to_add"].append((tg_id, renew_notification_id)) + else: + await update_balance(ctx.session, tg_id, -renewal_cost) + await update_key_expiry(ctx.session, client_id, int(new_expiry_time)) + await update_key_tariff(ctx.session, client_id, current_tariff["id"]) + await add_notification(ctx.session, tg_id, renew_notification_id) + + return True, current_tariff, int(new_expiry_time) - if not selected_tariff: - expiry_data = await prepare_key_expiry_data( - key, - conn, - int(datetime.now(moscow_tz).timestamp() * 1000), - ) - last_notification_time = await get_last_notification_time(conn, tg_id, notification_id) - if last_notification_time is not None: - return False - - message_text = KEY_CANNOT_RENEW_CURRENT.format( +async def notify_expiring_keys( + ctx: NotificationContext, + keys: list, + min_hours: int, + max_hours: int, + notify_type: str, + photo: str, + notify_renew_enabled: bool, +): + if min_hours > 0: + logger.info(f"Начало проверки подписок, истекающих через {min_hours}-{max_hours} часов.") + else: + logger.info(f"Начало проверки подписок, истекающих через {max_hours} часов.") + + min_threshold = int((datetime.now(moscow_tz) + timedelta(hours=min_hours)).timestamp() * 1000) + max_threshold = int((datetime.now(moscow_tz) + timedelta(hours=max_hours)).timestamp() * 1000) + expiring_keys = [key for key in keys if key.expiry_time and min_threshold < key.expiry_time <= max_threshold] + + if min_hours > 0: + logger.info(f"Найдено {len(expiring_keys)} подписок, истекающих через {min_hours}-{max_hours} часов.") + else: + logger.info(f"Найдено {len(expiring_keys)} подписок, истекающих через {max_hours} часов.") + + tg_ids = [key.tg_id for key in expiring_keys] + emails = [key.email or "" for key in expiring_keys] + allowed = await check_notifications_bulk(ctx.session, notify_type, max_hours, tg_ids=tg_ids, emails=emails) + allowed_set = {(user["tg_id"], user["email"]) for user in allowed} + + messages = [] + + for key in expiring_keys: + tg_id = key.tg_id + email = key.email or "" + + if (tg_id, email) not in allowed_set: + continue + + notification_id = f"{email}_{notify_type}" + + can_notify = await check_notification_time(ctx.session, tg_id, notification_id, hours=max_hours) + if not can_notify: + continue + + if notify_renew_enabled: + try: + renewed, tariff, new_expiry = await try_auto_renew(ctx, key) + + if renewed and tariff and new_expiry: + await send_renewed_notification(ctx, key, tariff, new_expiry) + await add_notification(ctx.session, tg_id, notification_id) + else: + await send_cannot_renew(ctx, key, photo) + await add_notification(ctx.session, tg_id, notification_id) + + except Exception as error: + logger.error(f"Ошибка авто-продления/уведомления для пользователя {tg_id}: {error}") + else: + expiry_data = await prepare_key_expiry_data(key, ctx.session, ctx.current_time) + notification_text = KEY_EXPIRY.format( email=email, hours_left_formatted=expiry_data["hours_left_formatted"], formatted_expiry_date=expiry_data["formatted_expiry_date"], tariff_name=expiry_data["tariff_name"], tariff_details=expiry_data["tariff_details"], ) + keyboard = build_notification_kb(email) + messages.append({ + "tg_id": tg_id, + "text": notification_text, + "photo": photo, + "keyboard": keyboard, + "notification_id": notification_id, + "email": email, + }) + + if messages: + results = await send_messages_with_limit(ctx.bot, messages, session=ctx.session) + sent_count = 0 + for msg, result in zip(messages, results, strict=False): + await add_notification(ctx.session, msg["tg_id"], msg["notification_id"]) + if result: + sent_count += 1 + logger.info(f"Отправлено уведомление об истекающей подписке {msg['email']} пользователю {msg['tg_id']}.") + logger.info(f"Отправлено {sent_count} уведомлений типа {notify_type}.") + + logger.info(f"Обработка уведомлений {notify_type} завершена.") + await asyncio.sleep(1) - keyboard = build_change_tariff_kb(email) - await add_notification(conn, tg_id, notification_id) - await send_notification(bot, tg_id, standard_photo, message_text, keyboard) - return False +async def handle_expired_keys(ctx: NotificationContext, keys: list): + logger.info("Начало обработки истекших ключей.") + + expired_keys = [key for key in keys if key.expiry_time and key.expiry_time < ctx.current_time] + logger.info(f"Найдено {len(expired_keys)} истекших ключей.") + + tg_ids = [key.tg_id for key in expired_keys] + emails = [key.email or "" for key in expired_keys] + users = await check_notifications_bulk(ctx.session, "key_expired", 0, tg_ids=tg_ids, emails=emails) + users_set = {(user["tg_id"], user["email"]) for user in users} + + notify_renew_expired_enabled = bool(NOTIFICATIONS_CONFIG.get("RENEW_EXPIRED_ENABLED", NOTIFY_RENEW_EXPIRED)) + notify_delete_key_enabled = bool(NOTIFICATIONS_CONFIG.get("DELETE_KEY_ENABLED", NOTIFY_DELETE_KEY)) + delete_key_delay_minutes = int(NOTIFICATIONS_CONFIG.get("DELETE_KEY_DELAY_MINUTES", NOTIFY_DELETE_DELAY)) + + for key in expired_keys: + tg_id = key.tg_id + email = key.email or "" client_id = key.client_id - current_expiry = key.expiry_time - duration_days = selected_tariff["duration_days"] - - renewal_cost = await resolve_price_to_charge( - conn, - { - "tariff_id": selected_tariff.get("id"), - "selected_device_limit": getattr(key, "selected_device_limit", None), - "selected_traffic_limit": getattr(key, "selected_traffic_limit", None), - "selected_price_rub": getattr(key, "selected_price_rub", None), - }, - ) - if renewal_cost is None: - logger.warning(f"[AUTO_RENEW] Не удалось определить стоимость продления для {email}. Продление отменено.") - return False - - selected_device_limit = getattr(key, "selected_device_limit", None) - selected_traffic_limit = getattr(key, "selected_traffic_limit", None) - selected_traffic_gb = int(selected_traffic_limit) if selected_traffic_limit is not None else None - - device_limit_effective, traffic_limit_bytes_effective = await get_effective_limits_for_key( - session=conn, - tariff_id=int(selected_tariff["id"]), - selected_device_limit=int(selected_device_limit) if selected_device_limit is not None else None, - selected_traffic_gb=selected_traffic_gb, - ) - traffic_limit_gb_effective = int(traffic_limit_bytes_effective / GB) if traffic_limit_bytes_effective else 0 - total_gb = traffic_limit_gb_effective - - new_expiry_time = ( - current_expiry - if current_expiry > datetime.utcnow().timestamp() * 1000 - else datetime.utcnow().timestamp() * 1000 - ) + duration_days * 24 * 60 * 60 * 1000 - - formatted_expiry_date = datetime.fromtimestamp(new_expiry_time / 1000, tz=moscow_tz).strftime("%d %B %Y, %H:%M") - formatted_expiry_date = formatted_expiry_date.replace( - datetime.fromtimestamp(new_expiry_time / 1000, tz=moscow_tz).strftime("%B"), - get_russian_month(datetime.fromtimestamp(new_expiry_time / 1000, tz=moscow_tz)), - ) - - logger.info( - f"Продление подписки {email} на {duration_days} дней для пользователя {tg_id}. " - f"Баланс: {balance}, списываем: {renewal_cost}" - ) - - key_subgroup = selected_tariff.get("subgroup_title") - - await renew_key_in_cluster( - cluster_id=server_id, - email=email, - client_id=client_id, - new_expiry_time=int(new_expiry_time), - total_gb=total_gb, - hwid_device_limit=device_limit_effective, - session=conn, - target_subgroup=key_subgroup, - old_subgroup=key_subgroup, - plan=selected_tariff["id"], - ) - await update_balance(conn, tg_id, -renewal_cost) - await update_key_expiry(conn, client_id, int(new_expiry_time)) - await update_key_tariff(conn, client_id, selected_tariff["id"]) - await add_notification(conn, tg_id, renew_notification_id) - await delete_notification(conn, tg_id, notification_id) - - renewed_message = get_renewal_message( - tariff_name=selected_tariff["name"], - traffic_limit=traffic_limit_gb_effective, - device_limit=device_limit_effective, - expiry_date=formatted_expiry_date, - subgroup_title=selected_tariff.get("subgroup_title", ""), - ) - - keyboard = build_notification_expired_kb() - result = await send_notification(bot, tg_id, "notify_expired.jpg", renewed_message, keyboard) - if result: - logger.info(f"✅ Уведомление о продлении подписки {email} отправлено пользователю {tg_id}.") - else: - logger.warning(f"📢 Не удалось отправить уведомление о продлении подписки {email} пользователю {tg_id}.") + server_id = key.server_id + notification_id = f"{email}_key_expired" - return True + last_notification_time = await get_last_notification_time(ctx.session, tg_id, notification_id) + + if notify_renew_expired_enabled: + try: + renewed, tariff, new_expiry = await try_auto_renew(ctx, key) + + if renewed and tariff and new_expiry: + await send_renewed_notification(ctx, key, tariff, new_expiry) + if ctx.bulk_updates: + ctx.bulk_updates["notifications_to_delete"].append((tg_id, notification_id)) + else: + await delete_notification(ctx.session, tg_id, notification_id) + continue + + except Exception as error: + logger.error(f"Ошибка авто-продления для пользователя {tg_id}: {error}") + continue + + if notify_delete_key_enabled: + should_delete = False + + if delete_key_delay_minutes == 0: + should_delete = True + elif last_notification_time is not None: + minutes_passed = (ctx.current_time - last_notification_time) / (1000 * 60) + should_delete = minutes_passed >= delete_key_delay_minutes + logger.info(f"Прошло минут={minutes_passed:.2f} DELETE_KEY_DELAY_MINUTES={delete_key_delay_minutes}") + + if should_delete: + try: + await delete_key_from_cluster(server_id, email, client_id, ctx.session) + await delete_key(ctx.session, client_id) + logger.info(f"🗑 Ключ {client_id} для пользователя {tg_id} успешно удалён.") + + await send_deleted_notification(ctx, key) + except Exception as error: + logger.error(f"Ошибка удаления ключа {client_id} для пользователя {tg_id}: {error}") + continue + + if last_notification_time is None and (tg_id, email) in users_set: + await send_expired_notification(ctx, key, delete_key_delay_minutes) + await add_notification(ctx.session, tg_id, notification_id) + + logger.info("Обработка истекших ключей завершена.") + await asyncio.sleep(1) - except Exception as error: - logger.error(f"❌ Ошибка в process_auto_renew_or_notify: {error}") - return False + +async def periodic_notifications(bot: Bot, *, sessionmaker: async_sessionmaker): + while True: + notification_interval = int(NOTIFICATIONS_CONFIG.get("BASE_NOTIFICATION_MINUTE", NOTIFICATION_TIME)) + + if notification_lock.locked(): + logger.warning("Уведомления уже выполняются. Пропуск...") + await asyncio.sleep(notification_interval) + continue + + async with notification_lock: + try: + async with sessionmaker() as session: + logger.info("Запуск обработки уведомлений") + + current_time = int(datetime.now(moscow_tz).timestamp() * 1000) + start_time = datetime.now() + + try: + preload_data = await preload_notification_data(session) + keys_data = preload_data["keys_data"] + keys = [data["key"] for data in keys_data.values()] + preload_time = (datetime.now() - start_time).total_seconds() + logger.info( + f"Предзагружено данных: {len(keys)} ключей, " + f"{len(preload_data['tariffs_cache'])} тарифов за {preload_time:.2f}s" + ) + + bulk_updates = { + "balance_changes": {}, + "key_expiry_updates": [], + "key_tariff_updates": [], + "notifications_to_add": [], + "notifications_to_delete": [], + } + + except Exception as error: + logger.error(f"Ошибка при предварительной загрузке данных: {error}") + try: + keys = await get_all_keys(session=session) + keys = [k for k in keys if not k.is_frozen] + preload_data = None + bulk_updates = None + preload_time = (datetime.now() - start_time).total_seconds() + logger.info(f"Fallback: получено {len(keys)} ключей за {preload_time:.2f}s") + except Exception as fallback_error: + logger.error(f"Ошибка fallback получения ключей: {fallback_error}") + keys = [] + preload_data = None + bulk_updates = None + + ctx = NotificationContext( + bot=bot, + session=session, + current_time=current_time, + preload_data=preload_data, + bulk_updates=bulk_updates, + ) + + trial_time_disable = bool(MODES_CONFIG.get("TRIAL_TIME_DISABLED", TRIAL_TIME_DISABLE)) + if not trial_time_disable: + try: + await notify_inactive_trial_users(bot, session) + except Exception as error: + logger.error(f"Ошибка в notify_inactive_trial_users: {error}") + + notify_24_enabled = bool(NOTIFICATIONS_CONFIG.get("EXPIRY_24H_ENABLED", NOTIFY_24H_ENABLED)) + notify_24_hours = int(NOTIFICATIONS_CONFIG.get("EXPIRY_24H_BEFORE_HOURS", NOTIFY_24H_HOURS)) + notify_10_enabled = bool(NOTIFICATIONS_CONFIG.get("EXPIRY_10H_ENABLED", NOTIFY_10H_ENABLED)) + notify_10_hours = int(NOTIFICATIONS_CONFIG.get("EXPIRY_10H_BEFORE_HOURS", NOTIFY_10H_HOURS)) + notify_renew_enabled = bool(NOTIFICATIONS_CONFIG.get("RENEW_ENABLED", NOTIFY_RENEW)) + inactive_traffic_enabled = bool(NOTIFICATIONS_CONFIG.get("INACTIVE_TRAFFIC_ENABLED", NOTIFY_INACTIVE_TRAFFIC)) + notify_hot_leads_enabled = bool(NOTIFICATIONS_CONFIG.get("HOT_LEADS_ENABLED", NOTIFY_HOT_LEADS)) + + if notify_24_enabled: + try: + await notify_expiring_keys( + ctx, keys, + min_hours=notify_10_hours if notify_10_enabled else 0, + max_hours=notify_24_hours, + notify_type="key_24h", + photo="notify_24h.jpg", + notify_renew_enabled=notify_renew_enabled + ) + except Exception as error: + logger.error(f"Ошибка в notify_expiring_keys (24h): {error}") + + if notify_10_enabled: + try: + await notify_expiring_keys( + ctx, keys, + min_hours=0, + max_hours=notify_10_hours, + notify_type="key_10h", + photo="notify_10h.jpg", + notify_renew_enabled=notify_renew_enabled + ) + except Exception as error: + logger.error(f"Ошибка в notify_expiring_keys (10h): {error}") + + try: + await handle_expired_keys(ctx, keys) + except Exception as error: + logger.error(f"Ошибка в handle_expired_keys: {error}") + + if inactive_traffic_enabled: + try: + await notify_users_no_traffic(bot, session, current_time, keys) + except Exception as error: + logger.error(f"Ошибка в notify_users_no_traffic: {error}") + + try: + await run_hooks("periodic_notifications", bot=bot, session=session, keys=keys) + except Exception as error: + logger.error(f"Ошибка в хуках periodic_notifications: {error}") + + if notify_hot_leads_enabled: + try: + await notify_hot_leads(bot, session) + except Exception as error: + logger.error(f"Ошибка в notify_hot_leads: {error}") + + if bulk_updates: + bulk_start = datetime.now() + await execute_bulk_updates(session, bulk_updates) + bulk_time = (datetime.now() - bulk_start).total_seconds() + total_renewals = len(bulk_updates["balance_changes"]) + total_key_updates = len(bulk_updates["key_expiry_updates"]) + len(bulk_updates["key_tariff_updates"]) + total_notification_updates = len(bulk_updates["notifications_to_add"]) + len(bulk_updates["notifications_to_delete"]) + logger.info( + f"Bulk-операции выполнены за {bulk_time:.2f}s. Обработано: {total_renewals} продлений, {total_key_updates} ключей, {total_notification_updates} уведомлений" + ) + + total_time = (datetime.now() - start_time).total_seconds() + logger.info(f"Уведомления завершены за {total_time:.2f}s") + + except Exception as error: + logger.error(f"Ошибка в periodic_notifications: {error}") + + await asyncio.sleep(notification_interval) diff --git a/handlers/notifications/notify_utils.py b/handlers/notifications/notify_utils.py index d966612d..4e59ae71 100644 --- a/handlers/notifications/notify_utils.py +++ b/handlers/notifications/notify_utils.py @@ -1,6 +1,8 @@ import asyncio import os +import time +from collections import deque from datetime import datetime import aiofiles @@ -24,44 +26,194 @@ from logger import logger moscow_tz = pytz.timezone("Europe/Moscow") +class NotificationRateLimiter: + def __init__(self, max_rate: int = 35, window: float = 1.0) -> None: + self.max_rate = max_rate + self.window = window + self.send_times = deque() + self.lock = asyncio.Lock() + + def _clean_old_timestamps(self, current_time: float): + cutoff_time = current_time - self.window + while self.send_times and self.send_times[0] <= cutoff_time: + self.send_times.popleft() + + async def acquire(self): + async with self.lock: + while True: + now = time.time() + self._clean_old_timestamps(now) + if len(self.send_times) < self.max_rate: + self.send_times.append(now) + return + oldest_timestamp = self.send_times[0] + time_to_wait = (oldest_timestamp + self.window) - now + if time_to_wait > 0: + await asyncio.sleep(time_to_wait + 0.001) + + +class NotificationMessage: + def __init__(self, tg_id: int, text: str, photo: str | None = None, keyboard=None) -> None: + self.tg_id = tg_id + self.text = text + self.photo = photo + self.keyboard = keyboard + self.retry_after = None + self.attempts = 0 + + +class FastNotificationSender: + def __init__(self, bot: Bot, session: AsyncSession | None, messages_per_second: int = 35) -> None: + self.bot = bot + self.session = session + self.rate_limiter = NotificationRateLimiter(max_rate=messages_per_second) + self.blocked_users = set() + self.queue = asyncio.Queue() + self.delayed_queue = asyncio.Queue() + self.results = [] + self.total_sent = 0 + self.is_running = False + + async def _send_single_message(self, msg: NotificationMessage) -> bool: + try: + await self.rate_limiter.acquire() + + if msg.photo: + photo_path = os.path.join("img", msg.photo) + if os.path.isfile(photo_path): + async with aiofiles.open(photo_path, "rb") as f: + image_data = await f.read() + buffered_photo = BufferedInputFile(image_data, filename=msg.photo) + await self.bot.send_photo( + chat_id=msg.tg_id, photo=buffered_photo, caption=msg.text, reply_markup=msg.keyboard + ) + else: + await self.bot.send_message(chat_id=msg.tg_id, text=msg.text, reply_markup=msg.keyboard) + else: + await self.bot.send_message(chat_id=msg.tg_id, text=msg.text, reply_markup=msg.keyboard) + return True + + except TelegramRetryAfter as e: + msg.retry_after = e.retry_after + msg.attempts += 1 + await self.delayed_queue.put(msg) + return False + + except TelegramForbiddenError: + self.blocked_users.add(msg.tg_id) + return False + + except TelegramBadRequest as e: + if "chat not found" in str(e).lower(): + self.blocked_users.add(msg.tg_id) + return False + + except Exception: + return False + + async def _process_delayed_messages(self): + while self.is_running: + try: + if not self.delayed_queue.empty(): + msg = await asyncio.wait_for(self.delayed_queue.get(), timeout=0.1) + if msg.retry_after: + await asyncio.sleep(msg.retry_after) + msg.retry_after = None + if msg.attempts < 3: + await self.queue.put(msg) + else: + self.results.append(False) + else: + await asyncio.sleep(0.1) + except TimeoutError: + continue + except Exception: + await asyncio.sleep(0.1) + + async def _worker(self): + while self.is_running: + try: + msg = await asyncio.wait_for(self.queue.get(), timeout=0.1) + success = await self._send_single_message(msg) + if success: + self.total_sent += 1 + self.results.append(True) + elif msg.attempts == 0: + self.results.append(False) + self.queue.task_done() + except TimeoutError: + continue + except Exception: + await asyncio.sleep(0.1) + + async def _save_blocked_users(self): + if not self.blocked_users or not self.session: + return + try: + from sqlalchemy.dialects.postgresql import insert + from database.models import BlockedUser + values = [{"tg_id": tg_id} for tg_id in self.blocked_users] + stmt = insert(BlockedUser).values(values).on_conflict_do_nothing(index_elements=[BlockedUser.tg_id]) + await self.session.execute(stmt) + await self.session.commit() + logger.info(f"📝 Добавлено {len(self.blocked_users)} пользователей в blocked_users") + except Exception as e: + logger.error(f"❌ Ошибка при сохранении заблокированных пользователей: {e}") + await self.session.rollback() + + async def send_all(self, messages: list[dict], workers: int = 15) -> list[bool]: + if not messages: + return [] + + self.is_running = True + self.results = [] + self.total_sent = 0 + self.blocked_users = set() + start_time = time.time() + + for msg_data in messages: + msg = NotificationMessage( + tg_id=msg_data["tg_id"], + text=msg_data["text"], + photo=msg_data.get("photo"), + keyboard=msg_data.get("keyboard"), + ) + await self.queue.put(msg) + + worker_tasks = [asyncio.create_task(self._worker()) for _ in range(workers)] + delayed_task = asyncio.create_task(self._process_delayed_messages()) + + await self.queue.join() + + await asyncio.sleep(0.5) + while not self.delayed_queue.empty(): + await asyncio.sleep(0.5) + + self.is_running = False + + for task in worker_tasks: + task.cancel() + delayed_task.cancel() + + await asyncio.gather(*worker_tasks, delayed_task, return_exceptions=True) + await self._save_blocked_users() + + duration = time.time() - start_time + speed = self.total_sent / duration if duration > 0 else 0 + logger.info(f"📨 Уведомления: {self.total_sent}/{len(messages)} за {duration:.1f}s ({speed:.1f} msg/s)") + + return self.results + + async def send_messages_with_limit( bot: Bot, messages: list[dict], session: AsyncSession = None, source_file: str = None, - messages_per_second: int = 25, + messages_per_second: int = 35, ): - batch_size = messages_per_second - results = [] - - for i in range(0, len(messages), batch_size): - batch = messages[i : i + batch_size] - tasks = [ - send_notification(bot, msg["tg_id"], msg.get("photo"), msg["text"], msg.get("keyboard")) for msg in batch - ] - batch_results = await asyncio.gather(*tasks, return_exceptions=True) - - for msg, result in zip(batch, batch_results, strict=False): - tg_id = msg["tg_id"] - - if isinstance(result, bool) and result: - results.append(True) - elif isinstance(result, TelegramForbiddenError): - logger.warning(f"🚫 Бот заблокирован пользователем {tg_id}.") - await try_add_blocked_user(tg_id, session, source_file) - results.append(False) - elif isinstance(result, TelegramBadRequest) and "chat not found" in str(result).lower(): - logger.warning(f"🚫 Чат не найден для пользователя {tg_id}.") - await try_add_blocked_user(tg_id, session, source_file) - results.append(False) - else: - logger.warning(f"📩 Не удалось отправить уведомление пользователю {tg_id}.") - await try_add_blocked_user(tg_id, session, source_file) - results.append(False) - - await asyncio.sleep(1.0) - - return results + sender = FastNotificationSender(bot, session, messages_per_second) + return await sender.send_all(messages) async def try_add_blocked_user(tg_id: int, session: AsyncSession, source_file: str | None): diff --git a/handlers/notifications/special_notifications.py b/handlers/notifications/special_notifications.py index 146b6d37..37e25efb 100644 --- a/handlers/notifications/special_notifications.py +++ b/handlers/notifications/special_notifications.py @@ -22,7 +22,7 @@ from database import ( update_key_notified, ) from database.tariffs import get_tariffs -from handlers.buttons import CONNECT_DEVICE, MAIN_MENU +from handlers.buttons import CONNECT_DEVICE, MAIN_MENU, SUPPORT, TRIAL_BONUS from handlers.keys.operations import get_user_traffic from handlers.notifications.notify_utils import send_messages_with_limit from handlers.texts import ( @@ -69,7 +69,7 @@ async def notify_inactive_trial_users(bot: Bot, session: AsyncSession): display_name = username or first_name or last_name or "Пользователь" builder = InlineKeyboardBuilder() - builder.row(types.InlineKeyboardButton(text="🚀 Активировать пробный период", callback_data="create_key")) + builder.row(types.InlineKeyboardButton(text=TRIAL_BONUS, callback_data="create_key")) builder.row(types.InlineKeyboardButton(text=MAIN_MENU, callback_data="profile")) keyboard = builder.as_markup() @@ -176,7 +176,7 @@ async def notify_users_no_traffic(bot: Bot, session: AsyncSession, current_time: logger.error(f"Ошибка при определении типа панели для {email}: {error}") builder.row(InlineKeyboardButton(text=CONNECT_DEVICE, callback_data=f"connect_device|{email}")) - builder.row(InlineKeyboardButton(text="🔧 Написать в поддержку", url=SUPPORT_CHAT_URL)) + builder.row(InlineKeyboardButton(text=SUPPORT, url=SUPPORT_CHAT_URL)) builder.row(InlineKeyboardButton(text=MAIN_MENU, callback_data="profile")) try: diff --git a/panels/remnawave.cpython-312-x86_64-linux-gnu.so b/panels/remnawave.cpython-312-x86_64-linux-gnu.so index b7dd4a1c..3d1f07e9 100644 Binary files a/panels/remnawave.cpython-312-x86_64-linux-gnu.so and b/panels/remnawave.cpython-312-x86_64-linux-gnu.so differ