diff --git a/core/tasks/loop_tasks.py b/core/tasks/loop_tasks.py index 87d0bf04..aa138ac9 100644 --- a/core/tasks/loop_tasks.py +++ b/core/tasks/loop_tasks.py @@ -4,7 +4,7 @@ from logger import logger async def notifications_loop(bot, sessionmaker) -> None: - from handlers.notifications.general_notifications import periodic_notifications + from handlers.notifications.engine import periodic_notifications await periodic_notifications(bot, sessionmaker=sessionmaker) diff --git a/handlers/keys/key_mode/key_discount_mode.py b/handlers/keys/key_mode/key_discount_mode.py index 1eb74805..df56b462 100644 --- a/handlers/keys/key_mode/key_discount_mode.py +++ b/handlers/keys/key_mode/key_discount_mode.py @@ -14,7 +14,7 @@ from database.access.resolution import resolve_user_optional from database.models import Notification from handlers.buttons import MAIN_MENU, RENEW_KEY_NOTIFICATION from handlers.keys.utils import build_key_callback -from handlers.notifications.notify_kb import build_tariffs_keyboard +from handlers.notifications.keyboards import build_tariffs_keyboard from handlers.tariffs.buy.key_tariffs import select_tariff_plan from handlers.texts import DISCOUNT_TARIFF, DISCOUNT_TARIFF_MAX from handlers.utils import format_discount_time_left, get_least_loaded_cluster diff --git a/handlers/notifications/__init__.py b/handlers/notifications/__init__.py index 5ec8845a..852039de 100644 --- a/handlers/notifications/__init__.py +++ b/handlers/notifications/__init__.py @@ -4,10 +4,5 @@ from aiogram import Router from core.tasks import lifecycle as _task_lifecycle -from .general_notifications import router as general_notifications_router -from .special_notifications import router as special_notifications_router - router = Router(name="notifications_main_router") - -router.include_routers(general_notifications_router, special_notifications_router) diff --git a/handlers/notifications/bulk.py b/handlers/notifications/bulk.py new file mode 100644 index 00000000..ac6a2b2f --- /dev/null +++ b/handlers/notifications/bulk.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +from typing import Any + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession + +from database import bulk_add_notifications, bulk_delete_notifications +from database.models import Key +from logger import logger + + +def create_bulk_updates() -> dict[str, Any]: + return { + "balance_changes": {}, + "key_expiry_updates": [], + "key_tariff_updates": [], + "notifications_to_add": [], + "notifications_to_delete": [], + } + + +async def execute_bulk_updates(session: AsyncSession, bulk_updates: dict[str, Any]) -> None: + try: + balance_changes = bulk_updates.get("balance_changes") or {} + if balance_changes: + tg_ids = list(balance_changes.keys()) + changes = [balance_changes[tg_id] for tg_id in tg_ids] + await session.execute( + text( + "UPDATE users SET balance = balance + v.change FROM " + "(SELECT unnest(CAST(:tg_ids AS bigint[])) AS tg_id, unnest(CAST(:changes AS double precision[])) AS change) AS v " + "WHERE users.tg_id = v.tg_id" + ), + {"tg_ids": tg_ids, "changes": changes}, + ) + logger.info(f"Bulk: обновлено {len(balance_changes)} балансов") + + key_expiry = bulk_updates.get("key_expiry_updates") or [] + if key_expiry: + await session.run_sync( + lambda sync_sess: sync_sess.bulk_update_mappings( + Key, + [{"client_id": cid, "expiry_time": exp} for cid, exp in key_expiry], + ) + ) + logger.info(f"Bulk: обновлено {len(key_expiry)} сроков действия ключей") + + key_tariff = bulk_updates.get("key_tariff_updates") or [] + if key_tariff: + await session.run_sync( + lambda sync_sess: sync_sess.bulk_update_mappings( + Key, + [{"client_id": cid, "tariff_id": tid} for cid, tid in key_tariff], + ) + ) + logger.info(f"Bulk: обновлено {len(key_tariff)} тарифов ключей") + + to_add = bulk_updates.get("notifications_to_add") or [] + if to_add: + await bulk_add_notifications(session, to_add) + + to_delete = bulk_updates.get("notifications_to_delete") or [] + if to_delete: + await bulk_delete_notifications(session, to_delete) + + if to_add or to_delete: + logger.info(f"Bulk: {len(to_add)} добавлений, {len(to_delete)} удалений уведомлений") + + await session.commit() + + except Exception as error: + logger.error(f"Ошибка в bulk-обновлениях: {error}") + await session.rollback() + raise diff --git a/handlers/notifications/context.py b/handlers/notifications/context.py new file mode 100644 index 00000000..f5836710 --- /dev/null +++ b/handlers/notifications/context.py @@ -0,0 +1,26 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +from aiogram import Bot +from sqlalchemy.ext.asyncio import AsyncSession + + +@dataclass +class NotificationContext: + bot: Bot + session: AsyncSession + current_time: int + preload_data: dict | None = None + bulk_updates: dict | None = 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) -> dict | None: + if self.preload_data and tariff_id in self.preload_data.get("tariffs_cache", {}): + return self.preload_data["tariffs_cache"][tariff_id] + return None diff --git a/handlers/notifications/engine.py b/handlers/notifications/engine.py new file mode 100644 index 00000000..a2dd10e6 --- /dev/null +++ b/handlers/notifications/engine.py @@ -0,0 +1,165 @@ +from __future__ import annotations + +import asyncio + +from datetime import datetime + +import pytz + +from aiogram import Bot +from sqlalchemy.ext.asyncio import async_sessionmaker + +from config import ( + NOTIFICATION_TIME, + NOTIFY_10H_ENABLED, + NOTIFY_10H_HOURS, + NOTIFY_24H_ENABLED, + NOTIFY_24H_HOURS, + NOTIFY_DELETE_DELAY, + NOTIFY_DELETE_KEY, + NOTIFY_HOT_LEADS, + NOTIFY_INACTIVE_TRAFFIC, + NOTIFY_RENEW, + NOTIFY_RENEW_EXPIRED, + TRIAL_TIME_DISABLE, +) +from core.bootstrap import MODES_CONFIG, NOTIFICATIONS_CONFIG +from handlers.notifications.bulk import create_bulk_updates, execute_bulk_updates +from handlers.notifications.context import NotificationContext +from handlers.notifications.preload import preload_with_fallback +from handlers.notifications.processors.expired import process_expired_keys +from handlers.notifications.processors.expiring import process_expiring_keys +from handlers.notifications.processors.hot_leads import process_hot_leads +from handlers.notifications.processors.inactive_trial import process_inactive_trial +from handlers.notifications.processors.zero_traffic import process_zero_traffic +from hooks.hooks import run_hooks +from logger import logger +from middlewares.session import wrap_session + +moscow_tz = pytz.timezone("Europe/Moscow") +notification_lock = asyncio.Lock() + + +async def periodic_notifications(bot: Bot, *, sessionmaker: async_sessionmaker): + while True: + interval = int(NOTIFICATIONS_CONFIG.get("BASE_NOTIFICATION_MINUTE", NOTIFICATION_TIME)) + + if notification_lock.locked(): + logger.warning("Уведомления уже выполняются. Пропуск...") + await asyncio.sleep(interval) + continue + + async with notification_lock: + try: + await _run_cycle(bot, sessionmaker) + except Exception as error: + logger.error(f"Ошибка в periodic_notifications: {error}") + + await asyncio.sleep(interval) + + +async def _run_cycle(bot: Bot, sessionmaker: async_sessionmaker): + start_time = datetime.now() + current_time = int(datetime.now(moscow_tz).timestamp() * 1000) + + async with sessionmaker() as preload_session: + keys, preload_data = await preload_with_fallback(preload_session) + + if not keys and not preload_data: + logger.info("Нет данных для обработки") + return + + bulk_updates = create_bulk_updates() + + async with sessionmaker() as session: + session = wrap_session(session, sessionmaker) + ctx = NotificationContext( + bot=bot, + session=session, + current_time=current_time, + preload_data=preload_data, + bulk_updates=bulk_updates, + ) + + trial_disabled = bool(MODES_CONFIG.get("TRIAL_TIME_DISABLED", TRIAL_TIME_DISABLE)) + 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)) + renew_enabled = bool(NOTIFICATIONS_CONFIG.get("RENEW_ENABLED", NOTIFY_RENEW)) + renew_expired_enabled = bool(NOTIFICATIONS_CONFIG.get("RENEW_EXPIRED_ENABLED", NOTIFY_RENEW_EXPIRED)) + delete_key_enabled = bool(NOTIFICATIONS_CONFIG.get("DELETE_KEY_ENABLED", NOTIFY_DELETE_KEY)) + delete_delay = int(NOTIFICATIONS_CONFIG.get("DELETE_KEY_DELAY_MINUTES", NOTIFY_DELETE_DELAY)) + inactive_traffic = bool(NOTIFICATIONS_CONFIG.get("INACTIVE_TRAFFIC_ENABLED", NOTIFY_INACTIVE_TRAFFIC)) + hot_leads_enabled = bool(NOTIFICATIONS_CONFIG.get("HOT_LEADS_ENABLED", NOTIFY_HOT_LEADS)) + + if not trial_disabled: + try: + await process_inactive_trial(bot, session, sessionmaker=sessionmaker) + except Exception as e: + logger.error(f"Ошибка inactive_trial: {e}") + + if notify_24_enabled: + try: + await process_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=renew_enabled, + sessionmaker=sessionmaker, + ) + except Exception as e: + logger.error(f"Ошибка expiring 24h: {e}") + + if notify_10_enabled: + try: + await process_expiring_keys( + ctx, keys, + min_hours=0, + max_hours=notify_10_hours, + notify_type="key_10h", + photo="notify_10h.jpg", + notify_renew_enabled=renew_enabled, + sessionmaker=sessionmaker, + ) + except Exception as e: + logger.error(f"Ошибка expiring 10h: {e}") + + try: + await process_expired_keys( + ctx, keys, + notify_renew_expired=renew_expired_enabled, + notify_delete_key=delete_key_enabled, + delete_delay_minutes=delete_delay, + ) + except Exception as e: + logger.error(f"Ошибка expired: {e}") + + if inactive_traffic: + try: + await process_zero_traffic(bot, session, current_time, keys) + except Exception as e: + logger.error(f"Ошибка zero_traffic: {e}") + + try: + await run_hooks("periodic_notifications", bot=bot, session=session, keys=keys) + except Exception as e: + logger.error(f"Ошибка хуков: {e}") + + if hot_leads_enabled: + try: + await process_hot_leads(bot, session) + except Exception as e: + logger.error(f"Ошибка hot_leads: {e}") + + if bulk_updates: + bulk_start = datetime.now() + await execute_bulk_updates(session, bulk_updates) + bulk_time = (datetime.now() - bulk_start).total_seconds() + logger.info(f"Bulk за {bulk_time:.2f}s") + + total_time = (datetime.now() - start_time).total_seconds() + logger.info(f"Уведомления завершены за {total_time:.2f}s") + await session.commit() diff --git a/handlers/notifications/general_notifications.py b/handlers/notifications/general_notifications.py deleted file mode 100644 index a443c37c..00000000 --- a/handlers/notifications/general_notifications.py +++ /dev/null @@ -1,855 +0,0 @@ -import asyncio - -from dataclasses import dataclass -from datetime import datetime, timedelta, timezone -from typing import Any, Optional - -import pytz - -from aiogram import Bot, Router -from sqlalchemy import exists, or_, select, text, update -from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker - -from config import ( - EXECUTOR_POOL_SIZE, - NOTIFICATION_TIME, - NOTIFY_10H_ENABLED, - NOTIFY_10H_HOURS, - NOTIFY_24H_ENABLED, - NOTIFY_24H_HOURS, - NOTIFY_DELETE_DELAY, - NOTIFY_DELETE_KEY, - NOTIFY_HOT_LEADS, - NOTIFY_INACTIVE_TRAFFIC, - NOTIFY_RENEW, - NOTIFY_RENEW_EXPIRED, - TRIAL_TIME_DISABLE, -) -from core.bootstrap import MODES_CONFIG, NOTIFICATIONS_CONFIG -from database import ( - add_notification, - bulk_add_notifications, - bulk_delete_notifications, - check_notification_time, - check_notification_time_bulk, - check_notifications_bulk, - delete_key, - delete_notification, - get_all_keys, - get_balance, - get_last_notification_time, - get_last_notification_times_bulk, - update_balance, - update_key_expiry, - update_key_tariff, -) -from database.models import Key, Tariff, User -from database.models.users import BlockedUser, ManualBan -from database.tariffs import ( - check_tariff_exists, - get_tariff_by_id, - get_tariffs_for_cluster, -) -from handlers.notifications.notify_kb import ( - build_change_tariff_kb, - build_notification_expired_kb, - build_notification_kb, -) -from handlers.texts import ( - KEY_CANNOT_RENEW_CURRENT, - KEY_DELETED_MSG, - KEY_EXPIRED_DELAY_MSG, - KEY_EXPIRED_NO_DELAY_MSG, - KEY_EXPIRY, - get_renewal_message, -) -from handlers.utils import format_hours, format_minutes, get_russian_month -from hooks.hooks import run_hooks -from logger import logger -from middlewares.session import release_session_early, wrap_session -from services.operations import delete_key_from_cluster, renew_key_in_cluster -from services.tariffs.tariff_display import GB, get_effective_limits_for_key, resolve_price_to_charge - -from .hot_leads_notifications import notify_hot_leads -from .notify_utils import ( - NotificationRateLimiter, - prepare_key_expiry_data, - send_messages_with_limit, - send_notification, -) -from .special_notifications import notify_inactive_trial_users, notify_users_no_traffic - - -router = Router() -moscow_tz = pytz.timezone("Europe/Moscow") -notification_lock = asyncio.Lock() - - -@dataclass -class NotificationContext: - bot: Bot - session: AsyncSession - current_time: int - preload_data: dict | None = None - bulk_updates: dict | None = 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) -> dict | None: - 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 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.user_id == User.id) - .where( - Key.is_frozen.is_(False), - ~exists().where(BlockedUser.tg_id == Key.tg_id), - ~exists().where( - ManualBan.tg_id == Key.tg_id, - or_(ManualBan.until.is_(None), ManualBan.until > datetime.now(timezone.utc)), - ), - ) - ) - - 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: - balance_changes = bulk_updates.get("balance_changes") or {} - if balance_changes: - tg_ids = list(balance_changes.keys()) - changes = [balance_changes[tg_id] for tg_id in tg_ids] - await session.execute( - text( - "UPDATE users SET balance = balance + v.change FROM " - "(SELECT unnest(CAST(:tg_ids AS bigint[])) AS tg_id, unnest(CAST(:changes AS double precision[])) AS change) AS v " - "WHERE users.tg_id = v.tg_id" - ), - {"tg_ids": tg_ids, "changes": changes}, - ) - logger.info(f"Bulk: обновлено {len(balance_changes)} балансов") - - key_expiry = bulk_updates.get("key_expiry_updates") or [] - if key_expiry: - await session.run_sync( - lambda sync_sess: sync_sess.bulk_update_mappings( - Key, - [{"client_id": cid, "expiry_time": exp} for cid, exp in key_expiry], - ) - ) - logger.info(f"Bulk: обновлено {len(key_expiry)} сроков действия ключей") - - key_tariff = bulk_updates.get("key_tariff_updates") or [] - if key_tariff: - await session.run_sync( - lambda sync_sess: sync_sess.bulk_update_mappings( - Key, - [{"client_id": cid, "tariff_id": tid} for cid, tid in key_tariff], - ) - ) - logger.info(f"Bulk: обновлено {len(key_tariff)} тарифов ключей") - - to_add = bulk_updates.get("notifications_to_add") or [] - if to_add: - await bulk_add_notifications(session, to_add) - - to_delete = bulk_updates.get("notifications_to_delete") or [] - if to_delete: - await bulk_delete_notifications(session, to_delete) - - if to_add or to_delete: - logger.info(f"Bulk: обработано {len(to_add)} добавлений и {len(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, getattr(key, "client_id", None)) - 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, getattr(key, "client_id", None)) - 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: - 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, getattr(key, "client_id", None)) - return await send_notification(ctx.bot, tg_id, "notify_expired.jpg", message_text, keyboard) - - -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 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 try_auto_renew(ctx: NotificationContext, key) -> tuple[bool, dict | None, int | None]: - 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: - 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.now(timezone.utc).timestamp() * 1000 - else datetime.now(timezone.utc).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 release_session_early(ctx.session) - 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"], - ) - - new_tariff_device_limit = current_tariff.get("device_limit") - new_tariff_traffic_limit = current_tariff.get("traffic_limit") - reset_values = { - "selected_device_limit": new_tariff_device_limit, - "current_device_limit": new_tariff_device_limit, - "selected_traffic_limit": new_tariff_traffic_limit, - "current_traffic_limit": new_tariff_traffic_limit, - "selected_price_rub": int(current_tariff["price_rub"]) if current_tariff.get("price_rub") is not None else None, - } - await ctx.session.execute(update(Key).where(Key.client_id == client_id).values(**reset_values)) - - 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: - 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) - - -async def notify_expiring_keys( - ctx: NotificationContext, - keys: list, - min_hours: int, - max_hours: int, - notify_type: str, - photo: str, - notify_renew_enabled: bool, - sessionmaker: async_sessionmaker | None = None, -): - 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} - - notify_pairs = [ - (key.tg_id, f"{(key.email or '')}_{notify_type}") - for key in expiring_keys - if (key.tg_id, key.email or "") in allowed_set - ] - can_notify_set = await check_notification_time_bulk(ctx.session, notify_pairs, max_hours) - - messages = [] - renew_candidates: list[tuple[Any, str]] = [] - - 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}" - if (tg_id, notification_id) not in can_notify_set: - continue - - if notify_renew_enabled: - renew_candidates.append((key, notification_id)) - 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, getattr(key, "client_id", None)) - messages.append({ - "tg_id": tg_id, - "text": notification_text, - "photo": photo, - "keyboard": keyboard, - "notification_id": notification_id, - "email": email, - }) - try: - from database.web_notifications import notify_web - - await notify_web( - ctx.session, tg_id=tg_id, type="key_expiry", template_vars={"email": email}, data={"email": email} - ) - except Exception as e: - logger.warning("[Notifications] Ошибка web-уведомления key_expiry tg_id={}: {}", tg_id, e) - - renew_results: list[tuple[Any, str, bool, dict | None, int | None]] = [] - use_parallel = notify_renew_enabled and renew_candidates and sessionmaker is not None and EXECUTOR_POOL_SIZE > 1 - if use_parallel: - semaphore = asyncio.Semaphore(EXECUTOR_POOL_SIZE) - - async def do_one_renew(key: Any, notification_id: str) -> tuple[Any, str, bool, dict | None, int | None]: - async with semaphore: - async with sessionmaker() as session: - session = wrap_session(session, sessionmaker) - ctx_key = NotificationContext( - bot=ctx.bot, - session=session, - current_time=ctx.current_time, - preload_data=ctx.preload_data, - bulk_updates=None, - ) - try: - renewed, tariff, new_expiry = await try_auto_renew(ctx_key, key) - await session.commit() - return (key, notification_id, bool(renewed), tariff, new_expiry) - except Exception as error: - logger.error( - "Ошибка авто-продления для пользователя %s (%s): %s", - key.tg_id, - getattr(key, "email", ""), - error, - ) - return (key, notification_id, False, None, None) - - tasks = [do_one_renew(key, nid) for key, nid in renew_candidates] - results = await asyncio.gather(*tasks, return_exceptions=True) - for r in results: - if isinstance(r, Exception): - logger.error("Ошибка в задаче продления: {}", r) - continue - renew_results.append(r) - else: - for key, notification_id in renew_candidates: - tg_id = key.tg_id - try: - renewed, tariff, new_expiry = await try_auto_renew(ctx, key) - renew_results.append((key, notification_id, bool(renewed), tariff, new_expiry)) - except Exception as error: - logger.error(f"Ошибка авто-продления/уведомления для пользователя {tg_id}: {error}") - renew_results.append((key, notification_id, False, None, None)) - - renew_rate_limiter = NotificationRateLimiter(max_rate=30, window=1.0) - for key, notification_id, renewed, tariff, new_expiry in renew_results: - tg_id = key.tg_id - await renew_rate_limiter.acquire() - 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) - - 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) - - -async def _get_blocked_expired_keys(session: AsyncSession, current_time: int) -> list: - """Получает истекшие ключи заблокированных/забаненных пользователей.""" - stmt = ( - select(Key) - .where( - Key.is_frozen.is_(False), - Key.expiry_time.isnot(None), - Key.expiry_time < current_time, - or_( - exists().where(BlockedUser.tg_id == Key.tg_id), - exists().where( - ManualBan.tg_id == Key.tg_id, - or_(ManualBan.until.is_(None), ManualBan.until > datetime.now(timezone.utc)), - ), - ), - ) - ) - result = await session.execute(stmt) - return list(result.scalars().all()) - - -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] - - try: - blocked_expired = await _get_blocked_expired_keys(ctx.session, ctx.current_time) - if blocked_expired: - logger.info(f"Дополнительно найдено {len(blocked_expired)} истекших ключей заблокированных пользователей.") - existing_ids = {key.client_id for key in expired_keys} - for bk in blocked_expired: - if bk.client_id not in existing_ids: - expired_keys.append(bk) - except Exception as error: - logger.error(f"Ошибка получения ключей заблокированных пользователей: {error}") - - 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)) - - notification_pairs = [(key.tg_id, f"{key.email or ''}_key_expired") for key in expired_keys] - last_times = await get_last_notification_times_bulk(ctx.session, notification_pairs) - - 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 = last_times.get((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) - - -async def periodic_notifications(bot: Bot, *, sessionmaker: async_sessionmaker): - from middlewares.session import wrap_session - - 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: - current_time = int(datetime.now(moscow_tz).timestamp() * 1000) - start_time = datetime.now() - preload_data = None - keys = [] - bulk_updates = { - "balance_changes": {}, - "key_expiry_updates": [], - "key_tariff_updates": [], - "notifications_to_add": [], - "notifications_to_delete": [], - } - - async with sessionmaker() as preload_session: - logger.info("Запуск обработки уведомлений") - try: - preload_data = await preload_notification_data(preload_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" - ) - except Exception as error: - logger.error(f"Ошибка при предварительной загрузке данных: {error}") - try: - keys = await get_all_keys(session=preload_session) - keys = [k for k in keys if not k.is_frozen] - preload_data = 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 - - async with sessionmaker() as session: - session = wrap_session(session, sessionmaker) - 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, sessionmaker=sessionmaker) - 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, - sessionmaker=sessionmaker, - ) - 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, - sessionmaker=sessionmaker, - ) - 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") - await session.commit() - - except Exception as error: - logger.error(f"Ошибка в periodic_notifications: {error}") - - await asyncio.sleep(notification_interval) diff --git a/handlers/notifications/notify_kb.py b/handlers/notifications/keyboards.py similarity index 69% rename from handlers/notifications/notify_kb.py rename to handlers/notifications/keyboards.py index 7113a39e..f2edebbf 100644 --- a/handlers/notifications/notify_kb.py +++ b/handlers/notifications/keyboards.py @@ -1,16 +1,11 @@ from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup +from aiogram.utils.keyboard import InlineKeyboardBuilder from handlers.buttons import CHANGE_TARIFF, DISCOUNT_TARIFF, MAIN_MENU, MAX_DISCOUNT_TARIFF, RENEW_KEY_NOTIFICATION from handlers.keys.utils import build_key_callback def build_notification_kb(email: str, client_id: str | None = None) -> InlineKeyboardMarkup: - """ - Формирует inline-клавиатуру для уведомлений. - Кнопки: "🔄 Продлить VPN" (callback_data содержит email) и "👤 Личный кабинет". - """ - from aiogram.utils.keyboard import InlineKeyboardBuilder - builder = InlineKeyboardBuilder() builder.button(text=RENEW_KEY_NOTIFICATION, callback_data=build_key_callback("renew_key", client_id, email)) builder.button(text=MAIN_MENU, callback_data="profile") @@ -19,12 +14,6 @@ def build_notification_kb(email: str, client_id: str | None = None) -> InlineKey def build_change_tariff_kb(email: str, client_id: str | None = None) -> InlineKeyboardMarkup: - """ - Формирует inline-клавиатуру для смены тарифа. - Кнопки: "🔄 Сменить тариф" (callback_data содержит email) и "👤 Личный кабинет". - """ - from aiogram.utils.keyboard import InlineKeyboardBuilder - builder = InlineKeyboardBuilder() builder.button(text=CHANGE_TARIFF, callback_data=build_key_callback("renew_key", client_id, email)) builder.button(text=MAIN_MENU, callback_data="profile") @@ -33,12 +22,6 @@ def build_change_tariff_kb(email: str, client_id: str | None = None) -> InlineKe def build_notification_expired_kb() -> InlineKeyboardMarkup: - """ - Формирует inline-клавиатуру для уведомлений после удаления или продления. - Кнопка: "👤 Личный кабинет" - """ - from aiogram.utils.keyboard import InlineKeyboardBuilder - builder = InlineKeyboardBuilder() builder.button(text=MAIN_MENU, callback_data="profile") return builder.as_markup() @@ -67,5 +50,4 @@ def build_tariffs_keyboard(tariffs: list[dict], prefix: str = "tariff") -> Inlin ] for t in tariffs ] - return InlineKeyboardMarkup(inline_keyboard=buttons) diff --git a/handlers/notifications/preload.py b/handlers/notifications/preload.py new file mode 100644 index 00000000..1ce8f5a6 --- /dev/null +++ b/handlers/notifications/preload.py @@ -0,0 +1,80 @@ +from __future__ import annotations + +from datetime import datetime, timezone + +from sqlalchemy import exists, or_, select +from sqlalchemy.ext.asyncio import AsyncSession + +from database import get_all_keys +from database.models import Key, Tariff, User +from database.models.users import BlockedUser, ManualBan +from logger import logger + + +async def preload_notification_data(session: AsyncSession) -> dict: + stmt = ( + select( + Key, + Tariff, + User.balance.label("user_balance"), + ) + .outerjoin(Tariff, Key.tariff_id == Tariff.id) + .outerjoin(User, Key.user_id == User.id) + .where( + Key.is_frozen.is_(False), + ~exists().where(BlockedUser.tg_id == Key.tg_id), + ~exists().where( + ManualBan.tg_id == Key.tg_id, + or_(ManualBan.until.is_(None), ManualBan.until > datetime.now(timezone.utc)), + ), + ) + ) + + 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 + + keys_data[key.client_id] = { + "key": key, + "tariff": dict(tariff.__dict__) if tariff else None, + "balance": float(balance), + } + + 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 preload_with_fallback(session: AsyncSession) -> tuple[list, dict | None]: + try: + preload_data = await preload_notification_data(session) + keys = [data["key"] for data in preload_data["keys_data"].values()] + logger.info( + f"Предзагружено: {len(keys)} ключей, {len(preload_data['tariffs_cache'])} тарифов" + ) + return keys, preload_data + 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] + logger.info(f"Fallback: {len(keys)} ключей") + return keys, None + except Exception as fallback_error: + logger.error(f"Ошибка fallback: {fallback_error}") + return [], None diff --git a/handlers/notifications/processors/__init__.py b/handlers/notifications/processors/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/handlers/notifications/processors/expired.py b/handlers/notifications/processors/expired.py new file mode 100644 index 00000000..c0e050ca --- /dev/null +++ b/handlers/notifications/processors/expired.py @@ -0,0 +1,176 @@ +from __future__ import annotations + +from datetime import datetime, timezone + +from sqlalchemy import exists, or_, select + +from database import add_notification, check_notifications_bulk, delete_key, delete_notification, get_last_notification_times_bulk +from database.models import Key +from database.models.users import BlockedUser, ManualBan +from handlers.notifications.context import NotificationContext +from handlers.notifications.keyboards import build_notification_expired_kb, build_notification_kb +from handlers.notifications.renewal import RenewalStatus, try_auto_renew +from handlers.notifications.sender import send_notification +from handlers.texts import KEY_DELETED_MSG, KEY_EXPIRED_DELAY_MSG, KEY_EXPIRED_NO_DELAY_MSG +from handlers.utils import format_hours, format_minutes +from logger import logger +from services.operations import delete_key_from_cluster + +from .expiring import _send_renewed + + +async def process_expired_keys( + ctx: NotificationContext, + keys: list, + notify_renew_expired: bool, + notify_delete_key: bool, + delete_delay_minutes: int, +): + expired_keys = [k for k in keys if k.expiry_time and k.expiry_time < ctx.current_time] + + try: + blocked_expired = await _get_blocked_expired_keys(ctx.session, ctx.current_time) + if blocked_expired: + existing_ids = {k.client_id for k in expired_keys} + for bk in blocked_expired: + if bk.client_id not in existing_ids: + expired_keys.append(bk) + logger.info(f"[Expired] +{len(blocked_expired)} ключей заблокированных") + except Exception as e: + logger.error(f"Ошибка получения ключей заблокированных: {e}") + + if not expired_keys: + return + + logger.info(f"[Expired] Найдено {len(expired_keys)} истекших ключей") + + tg_ids = [k.tg_id for k in expired_keys] + emails = [k.email or "" for k in expired_keys] + users = await check_notifications_bulk(ctx.session, "key_expired", 0, tg_ids=tg_ids, emails=emails) + users_set = {(u["tg_id"], u["email"]) for u in users} + + notification_pairs = [(k.tg_id, f"{k.email or ''}_key_expired") for k in expired_keys] + last_times = await get_last_notification_times_bulk(ctx.session, notification_pairs) + + 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 = last_times.get((tg_id, notification_id)) + + expired_ms = ctx.current_time - key.expiry_time + delay_ms = delete_delay_minutes * 60 * 1000 + is_grace = notify_delete_key and delete_delay_minutes > 0 and expired_ms < delay_ms + is_delete = not is_grace + + if notify_renew_expired: + try: + result = await try_auto_renew(ctx, key) + + if result.status == RenewalStatus.SUCCESS: + await _send_renewed(ctx, key, result.tariff, result.new_expiry_time) + 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 e: + logger.error(f"Ошибка продления для {tg_id}: {e}") + continue + + if is_grace: + if last_notification_time is None and (tg_id, email) in users_set: + remaining_ms = delay_ms - expired_ms + remaining_minutes = max(1, int(remaining_ms / (60 * 1000))) + await _send_expired_grace(ctx, key, remaining_minutes) + await add_notification(ctx.session, tg_id, notification_id) + continue + + if is_delete and notify_delete_key: + should_delete = False + if delete_delay_minutes == 0: + should_delete = True + elif last_notification_time is not None: + minutes_passed = expired_ms / (60 * 1000) + should_delete = minutes_passed >= delete_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(ctx, key) + except Exception as e: + logger.error(f"Ошибка удаления ключа {client_id}: {e}") + continue + + if last_notification_time is None and (tg_id, email) in users_set: + await _send_expired(ctx, key, delete_delay_minutes) + await add_notification(ctx.session, tg_id, notification_id) + + logger.info("[Expired] Обработка завершена") + + +async def _get_blocked_expired_keys(session, current_time: int) -> list: + stmt = ( + select(Key) + .where( + Key.is_frozen.is_(False), + Key.expiry_time.isnot(None), + Key.expiry_time < current_time, + or_( + exists().where(BlockedUser.tg_id == Key.tg_id), + exists().where( + ManualBan.tg_id == Key.tg_id, + or_(ManualBan.until.is_(None), ManualBan.until > datetime.now(timezone.utc)), + ), + ), + ) + ) + result = await session.execute(stmt) + return list(result.scalars().all()) + + +async def _send_expired_grace(ctx: NotificationContext, key, remaining_minutes: int) -> bool: + email = key.email or "" + hours = remaining_minutes // 60 + minutes = remaining_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) + + text = KEY_EXPIRED_DELAY_MSG.format(email=email, time_formatted=time_formatted) + keyboard = build_notification_kb(email, getattr(key, "client_id", None)) + return await send_notification(ctx.bot, key.tg_id, "notify_expired.jpg", text, keyboard) + + +async def _send_expired(ctx: NotificationContext, key, delay_minutes: int) -> bool: + 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: + time_formatted = format_minutes(minutes) + text = KEY_EXPIRED_DELAY_MSG.format(email=email, time_formatted=time_formatted) + else: + text = KEY_EXPIRED_NO_DELAY_MSG.format(email=email) + + keyboard = build_notification_kb(email, getattr(key, "client_id", None)) + return await send_notification(ctx.bot, key.tg_id, "notify_expired.jpg", text, keyboard) + + +async def _send_deleted(ctx: NotificationContext, key) -> bool: + email = key.email or "" + text = KEY_DELETED_MSG.format(email=email) + keyboard = build_notification_expired_kb() + return await send_notification(ctx.bot, key.tg_id, "notify_expired.jpg", text, keyboard) diff --git a/handlers/notifications/processors/expiring.py b/handlers/notifications/processors/expiring.py new file mode 100644 index 00000000..57e54923 --- /dev/null +++ b/handlers/notifications/processors/expiring.py @@ -0,0 +1,242 @@ +from __future__ import annotations + +import asyncio + +from datetime import datetime, timedelta + +import pytz + +from sqlalchemy.ext.asyncio import async_sessionmaker + +from config import EXECUTOR_POOL_SIZE +from database import add_notification, check_notification_time_bulk, check_notifications_bulk +from database.web_notifications import notify_web +from handlers.notifications.context import NotificationContext +from handlers.notifications.keyboards import build_change_tariff_kb, build_notification_expired_kb, build_notification_kb +from handlers.notifications.renewal import RenewalResult, RenewalStatus, try_auto_renew +from handlers.notifications.sender import ( + NotificationRateLimiter, + prepare_key_expiry_data, + send_messages_with_limit, + send_notification, +) +from handlers.texts import KEY_CANNOT_RENEW_CURRENT, KEY_EXPIRY, get_renewal_message +from handlers.utils import get_russian_month +from logger import logger +from middlewares.session import wrap_session +from services.tariffs.tariff_display import GB, get_effective_limits_for_key + +moscow_tz = pytz.timezone("Europe/Moscow") + + +async def process_expiring_keys( + ctx: NotificationContext, + keys: list, + min_hours: int, + max_hours: int, + notify_type: str, + photo: str, + notify_renew_enabled: bool, + sessionmaker: async_sessionmaker | None = None, +): + 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 = [k for k in keys if k.expiry_time and min_threshold < k.expiry_time <= max_threshold] + + if not expiring_keys: + return + + logger.info(f"[{notify_type}] Найдено {len(expiring_keys)} истекающих ключей") + + tg_ids = [k.tg_id for k in expiring_keys] + emails = [k.email or "" for k in expiring_keys] + allowed = await check_notifications_bulk(ctx.session, notify_type, max_hours, tg_ids=tg_ids, emails=emails) + allowed_set = {(u["tg_id"], u["email"]) for u in allowed} + + notify_pairs = [ + (k.tg_id, f"{(k.email or '')}_{notify_type}") + for k in expiring_keys + if (k.tg_id, k.email or "") in allowed_set + ] + can_notify_set = await check_notification_time_bulk(ctx.session, notify_pairs, max_hours) + + renew_candidates = [] + simple_notify = [] + + for key in expiring_keys: + tg_id = key.tg_id + email = key.email or "" + notification_id = f"{email}_{notify_type}" + + if (tg_id, email) not in allowed_set: + continue + if (tg_id, notification_id) not in can_notify_set: + continue + + if notify_renew_enabled: + renew_candidates.append((key, notification_id)) + else: + simple_notify.append((key, notification_id)) + + if renew_candidates: + await _process_renew_candidates(ctx, renew_candidates, photo, sessionmaker) + + if simple_notify: + await _send_simple_warnings(ctx, simple_notify, photo, notify_type) + + +async def _process_renew_candidates( + ctx: NotificationContext, + candidates: list[tuple], + photo: str, + sessionmaker: async_sessionmaker | None, +): + use_parallel = sessionmaker is not None and EXECUTOR_POOL_SIZE > 1 + rate_limiter = NotificationRateLimiter(max_rate=30, window=1.0) + + if use_parallel: + semaphore = asyncio.Semaphore(EXECUTOR_POOL_SIZE) + + async def do_one(key, notification_id): + async with semaphore: + async with sessionmaker() as session: + session = wrap_session(session, sessionmaker) + one_ctx = NotificationContext( + bot=ctx.bot, + session=session, + current_time=ctx.current_time, + preload_data=ctx.preload_data, + bulk_updates=None, + ) + try: + result = await try_auto_renew(one_ctx, key) + await session.commit() + return (key, notification_id, result) + except Exception as e: + logger.error(f"Ошибка продления {key.tg_id} ({key.email}): {e}") + return (key, notification_id, RenewalResult(RenewalStatus.NO_TARIFF)) + + tasks = [do_one(k, nid) for k, nid in candidates] + results = await asyncio.gather(*tasks, return_exceptions=True) + else: + results = [] + for key, notification_id in candidates: + try: + result = await try_auto_renew(ctx, key) + results.append((key, notification_id, result)) + except Exception as e: + logger.error(f"Ошибка продления {key.tg_id}: {e}") + results.append((key, notification_id, RenewalResult(RenewalStatus.NO_TARIFF))) + + for item in results: + if isinstance(item, Exception): + logger.error(f"Ошибка задачи продления: {item}") + continue + + key, notification_id, renewal_result = item + tg_id = key.tg_id + email = key.email or "" + + await rate_limiter.acquire() + + if renewal_result.status == RenewalStatus.SUCCESS: + await _send_renewed(ctx, key, renewal_result.tariff, renewal_result.new_expiry_time) + elif renewal_result.status in (RenewalStatus.FORBIDDEN_TARIFF, RenewalStatus.NO_TARIFF): + await _send_change_tariff(ctx, key, photo) + else: + await _send_expiry_warning(ctx, key, photo) + + await add_notification(ctx.session, tg_id, notification_id) + + +async def _send_simple_warnings(ctx: NotificationContext, items: list[tuple], photo: str, notify_type: str): + messages = [] + for key, notification_id in items: + tg_id = key.tg_id + email = key.email or "" + + expiry_data = await prepare_key_expiry_data(key, ctx.session, ctx.current_time) + 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, getattr(key, "client_id", None)) + messages.append({ + "tg_id": tg_id, + "text": text, + "photo": photo, + "keyboard": keyboard, + "notification_id": notification_id, + }) + + try: + await notify_web(ctx.session, tg_id=tg_id, type="key_expiry", template_vars={"email": email}, data={"email": email}) + except Exception as e: + logger.warning(f"[Notifications] web-уведомление key_expiry tg_id={tg_id}: {e}") + + if messages: + results = await send_messages_with_limit(ctx.bot, messages) + for msg, result in zip(messages, results, strict=False): + await add_notification(ctx.session, msg["tg_id"], msg["notification_id"]) + if result: + logger.info(f"Уведомление {notify_type} отправлено {msg['tg_id']}") + + +async def _send_expiry_warning(ctx: NotificationContext, key, photo: str) -> bool: + expiry_data = await prepare_key_expiry_data(key, ctx.session, ctx.current_time) + text = KEY_EXPIRY.format( + email=key.email or "", + 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(key.email or "", getattr(key, "client_id", None)) + return await send_notification(ctx.bot, key.tg_id, photo, text, keyboard) + + +async def _send_change_tariff(ctx: NotificationContext, key, photo: str) -> bool: + expiry_data = await prepare_key_expiry_data(key, ctx.session, ctx.current_time) + text = KEY_CANNOT_RENEW_CURRENT.format( + email=key.email or "", + 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(key.email or "", getattr(key, "client_id", None)) + return await send_notification(ctx.bot, key.tg_id, photo, text, keyboard) + + +async def _send_renewed(ctx: NotificationContext, key, tariff: dict, new_expiry_time: int) -> bool: + 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)), + ) + + 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() + return await send_notification(ctx.bot, key.tg_id, "notify_expired.jpg", text, keyboard) diff --git a/handlers/notifications/hot_leads_notifications.py b/handlers/notifications/processors/hot_leads.py similarity index 51% rename from handlers/notifications/hot_leads_notifications.py rename to handlers/notifications/processors/hot_leads.py index f94903da..fe8e49d7 100644 --- a/handlers/notifications/hot_leads_notifications.py +++ b/handlers/notifications/processors/hot_leads.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from aiogram import Bot from aiogram.types import InlineKeyboardButton from aiogram.utils.keyboard import InlineKeyboardBuilder @@ -5,51 +7,41 @@ from sqlalchemy.ext.asyncio import AsyncSession from config import DISCOUNT_ACTIVE_HOURS, HOT_LEAD_INTERVAL_HOURS from core.bootstrap import NOTIFICATIONS_CONFIG -from database import ( - add_notification, - check_notification_time_bulk, - get_hot_lead_notification_flags, - get_hot_leads, -) +from database import add_notification, check_notification_time_bulk, get_hot_lead_notification_flags, get_hot_leads from database.tariffs import get_tariffs from handlers.buttons import MAIN_MENU -from handlers.notifications.notify_kb import build_hot_lead_kb -from handlers.notifications.notify_utils import send_notification -from handlers.texts import ( - HOT_LEAD_FINAL_MESSAGE, - HOT_LEAD_LOST_OPPORTUNITY, - HOT_LEAD_MESSAGE, -) +from handlers.notifications.keyboards import build_hot_lead_kb +from handlers.notifications.sender import send_notification +from handlers.texts import HOT_LEAD_FINAL_MESSAGE, HOT_LEAD_LOST_OPPORTUNITY, HOT_LEAD_MESSAGE from logger import logger -async def notify_hot_leads(bot: Bot, session: AsyncSession): - logger.info("Запуск уведомлений для горячих лидов.") +async def process_hot_leads(bot: Bot, session: AsyncSession): + logger.info("[HotLeads] Запуск") - hot_lead_interval_hours = int(NOTIFICATIONS_CONFIG.get("HOT_LEADS_INTERVAL_HOURS", HOT_LEAD_INTERVAL_HOURS)) - discount_active_hours = int(NOTIFICATIONS_CONFIG.get("DISCOUNT_ACTIVE_HOURS", DISCOUNT_ACTIVE_HOURS)) + hot_lead_interval = int(NOTIFICATIONS_CONFIG.get("HOT_LEADS_INTERVAL_HOURS", HOT_LEAD_INTERVAL_HOURS)) + discount_active = int(NOTIFICATIONS_CONFIG.get("DISCOUNT_ACTIVE_HOURS", DISCOUNT_ACTIVE_HOURS)) try: leads = await get_hot_leads(session) if not leads: - logger.info("Нет горячих лидов для уведомлений.") return flags = await get_hot_lead_notification_flags(session, leads) can_send_after_step1 = await check_notification_time_bulk( - session, [(tid, "hot_lead_step_1") for tid in leads], hot_lead_interval_hours + session, [(tid, "hot_lead_step_1") for tid in leads], hot_lead_interval, ) step2_expired_can_send = await check_notification_time_bulk( - session, [(tid, "hot_lead_step_2") for tid in leads], discount_active_hours + session, [(tid, "hot_lead_step_2") for tid in leads], discount_active, ) can_send_after_step2 = await check_notification_time_bulk( - session, [(tid, "hot_lead_step_2") for tid in leads], hot_lead_interval_hours + session, [(tid, "hot_lead_step_2") for tid in leads], hot_lead_interval, ) discount_tariffs = await get_tariffs(session, group_code="discounts") - active_discount_tariffs = [t for t in discount_tariffs if t.get("is_active")] + active_discounts = [t for t in discount_tariffs if t.get("is_active")] discount_max_tariffs = await get_tariffs(session, group_code="discounts_max") - active_discount_max_tariffs = [t for t in discount_max_tariffs if t.get("is_active")] + active_max_discounts = [t for t in discount_max_tariffs if t.get("is_active")] notified = 0 @@ -58,61 +50,46 @@ async def notify_hot_leads(bot: Bot, session: AsyncSession): has_step_1 = "hot_lead_step_1" in step_flags has_step_2 = "hot_lead_step_2" in step_flags has_step_3 = "hot_lead_step_3" in step_flags - has_expired_notification = "hot_lead_step_2_expired" in step_flags + has_expired = "hot_lead_step_2_expired" in step_flags if not has_step_1: await add_notification(session, tg_id, "hot_lead_step_1") - logger.info(f"[HOT LEAD] Шаг 1 — зафиксировано без отправки: {tg_id}") + logger.info(f"[HotLeads] Шаг 1 зафиксирован: {tg_id}") continue if not has_step_2: if (tg_id, "hot_lead_step_1") not in can_send_after_step1: continue - if not active_discount_tariffs: - logger.warning( - f"[HOT LEAD] Пропуск шага 2 для {tg_id}: нет активных тарифов со скидкой (discounts)" - ) + if not active_discounts: continue keyboard = build_hot_lead_kb() result = await send_notification(bot, tg_id, None, HOT_LEAD_MESSAGE, keyboard) if result: await add_notification(session, tg_id, "hot_lead_step_2") - logger.info(f"Шаг 2 — отправлено первое уведомление: {tg_id}") notified += 1 continue - if not has_step_3 and not has_expired_notification: + if not has_step_3 and not has_expired: if (tg_id, "hot_lead_step_2") in step2_expired_can_send: builder = InlineKeyboardBuilder() builder.row(InlineKeyboardButton(text=MAIN_MENU, callback_data="profile")) - result = await send_notification( - bot, - tg_id, - None, - HOT_LEAD_LOST_OPPORTUNITY, - builder.as_markup(), - ) + result = await send_notification(bot, tg_id, None, HOT_LEAD_LOST_OPPORTUNITY, builder.as_markup()) if result: await add_notification(session, tg_id, "hot_lead_step_2_expired") - logger.info(f"📭 Скидка упущена — отправлено уведомление: {tg_id}") continue if not has_step_3: if (tg_id, "hot_lead_step_2") not in can_send_after_step2: continue - if not active_discount_max_tariffs: - logger.warning( - f"[HOT LEAD] Пропуск шага 3 для {tg_id}: нет активных тарифов с максимальной скидкой (discounts_max)" - ) + if not active_max_discounts: continue keyboard = build_hot_lead_kb(final=True) result = await send_notification(bot, tg_id, None, HOT_LEAD_FINAL_MESSAGE, keyboard) if result: await add_notification(session, tg_id, "hot_lead_step_3") - logger.info(f"⚡ Шаг 3 — отправлено финальное уведомление: {tg_id}") notified += 1 - logger.info(f"Уведомления завершены. Отправлено: {notified}") + logger.info(f"[HotLeads] Отправлено: {notified}") except Exception as e: - logger.error(f"❌ Ошибка в notify_hot_leads: {e}") + logger.error(f"[HotLeads] Ошибка: {e}") diff --git a/handlers/notifications/processors/inactive_trial.py b/handlers/notifications/processors/inactive_trial.py new file mode 100644 index 00000000..8b7e6e41 --- /dev/null +++ b/handlers/notifications/processors/inactive_trial.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +from aiogram import Bot, types +from aiogram.utils.keyboard import InlineKeyboardBuilder +from sqlalchemy import update +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from config import NOTIFY_EXTRA_DAYS, NOTIFY_INACTIVE +from core.bootstrap import NOTIFICATIONS_CONFIG +from database import add_notification, check_notifications_bulk +from database.models import User +from database.tariffs import get_tariffs +from handlers.buttons import MAIN_MENU, TRIAL_BONUS +from handlers.notifications.sender import send_messages_with_limit +from handlers.texts import TRIAL_INACTIVE_BONUS_MSG, TRIAL_INACTIVE_FIRST_MSG +from handlers.utils import format_days +from logger import logger + + +async def process_inactive_trial( + bot: Bot, + session: AsyncSession, + *, + sessionmaker: async_sessionmaker | None = None, +): + inactive_hours = int(NOTIFICATIONS_CONFIG.get("INACTIVE_USER_ENABLED", NOTIFY_INACTIVE)) + extra_days = int(NOTIFICATIONS_CONFIG.get("EXTRA_DAYS_AFTER_EXPIRY", NOTIFY_EXTRA_DAYS)) + + if inactive_hours <= 0: + return + + users = await check_notifications_bulk(session, "inactive_trial", inactive_hours) + if not users: + return + + logger.info(f"[InactiveTrial] {len(users)} неактивных пользователей") + + trial_tariffs = await get_tariffs(session, group_code="trial") + if not trial_tariffs: + logger.error("[InactiveTrial] Триальный тариф не найден") + return + + trial_days = trial_tariffs[0]["duration_days"] + messages = [] + users_to_extend = [] + + for user in users: + tg_id = user["tg_id"] + display_name = user["username"] or user["first_name"] or user["last_name"] or "Пользователь" + + builder = InlineKeyboardBuilder() + 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() + + trial_extended = user["last_notification_time"] is not None + + if trial_extended and extra_days > 0: + total_days = extra_days + trial_days + message = TRIAL_INACTIVE_BONUS_MSG.format( + display_name=display_name, + extra_days_formatted=format_days(extra_days), + total_days_formatted=format_days(total_days), + ) + users_to_extend.append(tg_id) + else: + message = TRIAL_INACTIVE_FIRST_MSG.format( + display_name=display_name, + trial_time_formatted=format_days(trial_days), + ) + + messages.append({ + "tg_id": tg_id, + "text": message, + "keyboard": keyboard, + "notification_id": "inactive_trial", + }) + + if users_to_extend: + await session.execute(update(User).where(User.tg_id.in_(users_to_extend)).values(trial=-1)) + await session.commit() + logger.info(f"[InactiveTrial] {len(users_to_extend)} пользователей с расширенным триалом") + + if messages: + results = await send_messages_with_limit(bot, messages, messages_per_second=25) + + sent_tg_ids = [msg["tg_id"] for msg, result in zip(messages, results, strict=False) if result] + + if sent_tg_ids: + if sessionmaker is not None: + async with sessionmaker() as fresh_session: + for tg_id in sent_tg_ids: + await add_notification(fresh_session, tg_id, "inactive_trial") + await fresh_session.commit() + else: + for tg_id in sent_tg_ids: + await add_notification(session, tg_id, "inactive_trial") + logger.info(f"[InactiveTrial] Отправлено {len(sent_tg_ids)} уведомлений") diff --git a/handlers/notifications/processors/zero_traffic.py b/handlers/notifications/processors/zero_traffic.py new file mode 100644 index 00000000..ba35d3a5 --- /dev/null +++ b/handlers/notifications/processors/zero_traffic.py @@ -0,0 +1,141 @@ +from __future__ import annotations + +from datetime import datetime, timedelta + +import pytz + +from aiogram import Bot +from aiogram.types import InlineKeyboardButton, WebAppInfo +from aiogram.utils.keyboard import InlineKeyboardBuilder +from sqlalchemy import update +from sqlalchemy.ext.asyncio import AsyncSession + +from config import NOTIFY_INACTIVE_TRAFFIC, REMNAWAVE_WEBAPP, REMNAWAVE_WEBAPP_OPEN_IN_BROWSER, SUPPORT_CHAT_URL +from core.bootstrap import MODES_CONFIG, NOTIFICATIONS_CONFIG +from database.models import Key +from database.tariffs import get_tariffs +from handlers.buttons import CONNECT_DEVICE, MAIN_MENU, SUPPORT +from handlers.keys.utils import build_key_callback +from handlers.notifications.sender import send_messages_with_limit +from handlers.texts import ZERO_TRAFFIC_MSG +from handlers.utils import is_full_remnawave_cluster +from hooks.hook_buttons import insert_hook_buttons +from hooks.hooks import run_hooks +from logger import logger +from panels.remnawave_runtime import fetch_all_remnawave_traffic + +moscow_tz = pytz.timezone("Europe/Moscow") + + +async def process_zero_traffic( + bot: Bot, + session: AsyncSession, + current_time: int, + keys: list, +): + inactive_traffic_hours = int(NOTIFICATIONS_CONFIG.get("INACTIVE_TRAFFIC_ENABLED", NOTIFY_INACTIVE_TRAFFIC)) + if inactive_traffic_hours <= 0: + return + + trial_tariffs = await get_tariffs(session, group_code="trial") + trial_tariff_ids = {t["id"] for t in trial_tariffs} if trial_tariffs else set() + if not trial_tariff_ids: + return + + current_dt = datetime.fromtimestamp(current_time / 1000, tz=moscow_tz) + remnawave_webapp_enabled = bool(MODES_CONFIG.get("REMNAWAVE_WEBAPP_ENABLED", REMNAWAVE_WEBAPP)) + open_in_browser = bool(MODES_CONFIG.get("REMNAWAVE_WEBAPP_OPEN_IN_BROWSER", REMNAWAVE_WEBAPP_OPEN_IN_BROWSER)) + + candidate_keys = [] + for key in keys: + if key.tariff_id not in trial_tariff_ids: + continue + if key.created_at is None or key.notified: + continue + created_at_dt = pytz.utc.localize(datetime.fromtimestamp(key.created_at / 1000)).astimezone(moscow_tz) + if current_dt < created_at_dt + timedelta(hours=inactive_traffic_hours): + continue + if key.expiry_time: + expiry_dt = pytz.utc.localize(datetime.fromtimestamp(key.expiry_time / 1000)).astimezone(moscow_tz) + if current_dt > expiry_dt: + continue + candidate_keys.append(key) + + if not candidate_keys: + return + + needed_uuids = {k.client_id for k in candidate_keys if k.client_id} + logger.info(f"[ZeroTraffic] Кандидатов: {len(candidate_keys)}, UUID: {len(needed_uuids)}") + + try: + traffic_map = await fetch_all_remnawave_traffic(session, needed_uuids=needed_uuids) + except Exception as e: + logger.error(f"[ZeroTraffic] Ошибка получения трафика: {e}") + return + + messages = [] + keys_to_mark = [k.client_id for k in candidate_keys] + + for key in candidate_keys: + tg_id = key.tg_id + email = key.email + client_id = key.client_id + + used_bytes = traffic_map.get(client_id) + if used_bytes is None or used_bytes > 0: + continue + + builder = InlineKeyboardBuilder() + server_id = key.server_id + try: + is_full_remnawave = await is_full_remnawave_cluster(server_id, session) + final_link = key.key or key.remnawave_link + + if is_full_remnawave and final_link and remnawave_webapp_enabled: + if open_in_browser: + builder.row(InlineKeyboardButton(text=CONNECT_DEVICE, url=final_link)) + else: + builder.row(InlineKeyboardButton(text=CONNECT_DEVICE, web_app=WebAppInfo(url=final_link))) + else: + builder.row(InlineKeyboardButton( + text=CONNECT_DEVICE, + callback_data=build_key_callback("connect_device", client_id, email), + )) + except Exception as e: + logger.error(f"Ошибка типа панели для {email}: {e}") + builder.row(InlineKeyboardButton( + text=CONNECT_DEVICE, + callback_data=build_key_callback("connect_device", client_id, email), + )) + + builder.row(InlineKeyboardButton(text=SUPPORT, url=SUPPORT_CHAT_URL)) + builder.row(InlineKeyboardButton(text=MAIN_MENU, callback_data="profile")) + + try: + hook_commands = await run_hooks( + "zero_traffic_notification", chat_id=tg_id, admin=False, session=session, email=email, + ) + if hook_commands: + builder = insert_hook_buttons(builder, hook_commands) + except Exception as e: + logger.warning(f"[ZeroTraffic] Ошибка хуков: {e}") + + messages.append({ + "tg_id": tg_id, + "text": ZERO_TRAFFIC_MSG.format(email=email), + "keyboard": builder.as_markup(), + "client_id": client_id, + }) + + if keys_to_mark: + try: + await session.execute(update(Key).where(Key.client_id.in_(keys_to_mark)).values(notified=True)) + await session.commit() + logger.info(f"[ZeroTraffic] Отмечено {len(keys_to_mark)} ключей как notified") + except Exception as e: + logger.error(f"[ZeroTraffic] Ошибка обновления notified: {e}") + + if messages: + results = await send_messages_with_limit(bot, messages, messages_per_second=25) + sent_count = sum(1 for r in results if r) + logger.info(f"[ZeroTraffic] Отправлено {sent_count} уведомлений") diff --git a/handlers/notifications/renewal.py b/handlers/notifications/renewal.py new file mode 100644 index 00000000..cbe8da67 --- /dev/null +++ b/handlers/notifications/renewal.py @@ -0,0 +1,158 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from enum import Enum, auto +from typing import Any, NamedTuple + +from sqlalchemy import update + +from database import ( + add_notification, + check_notification_time, + get_balance, + update_balance, + update_key_expiry, + update_key_tariff, +) +from database.models import Key +from database.tariffs import check_tariff_exists, get_tariff_by_id, get_tariffs_for_cluster +from handlers.notifications.context import NotificationContext +from hooks.hooks import run_hooks +from logger import logger +from middlewares.session import release_session_early +from services.operations import renew_key_in_cluster +from services.tariffs.tariff_display import GB, get_effective_limits_for_key, resolve_price_to_charge + + +class RenewalStatus(Enum): + SUCCESS = auto() + FORBIDDEN_TARIFF = auto() + NO_BALANCE = auto() + NO_TARIFF = auto() + COOLDOWN = auto() + + +class RenewalResult(NamedTuple): + status: RenewalStatus + tariff: dict | None = None + new_expiry_time: int | None = None + + +FORBIDDEN_GROUPS = ["discounts", "discounts_max", "gifts", "trial"] + + +async def try_auto_renew(ctx: NotificationContext, key) -> RenewalResult: + 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: + return RenewalResult(RenewalStatus.COOLDOWN) + + server_id = key.server_id + tariff_id = key.tariff_id + + tariffs = await get_tariffs_for_cluster(ctx.session, server_id) + if not tariffs: + return RenewalResult(RenewalStatus.NO_TARIFF) + + 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 RenewalResult(RenewalStatus.NO_TARIFF) + + forbidden = list(FORBIDDEN_GROUPS) + try: + hook_results = await run_hooks("renewal_forbidden_groups", chat_id=tg_id, admin=False, session=ctx.session) + for hr in hook_results: + forbidden.extend(hr.get("additional_groups", [])) + except Exception as error: + logger.warning(f"[RENEW] Ошибка хуков forbidden_groups: {error}") + + if current_tariff["group_code"] in forbidden: + return RenewalResult(RenewalStatus.FORBIDDEN_TARIFF) + + 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) + + 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": None, + }, + ) + + if renewal_cost is None or balance < renewal_cost: + return RenewalResult(RenewalStatus.NO_BALANCE) + + 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 + + now_ms = datetime.now(timezone.utc).timestamp() * 1000 + base_expiry = current_expiry if current_expiry > now_ms else now_ms + new_expiry_time = int(base_expiry + duration_days * 24 * 60 * 60 * 1000) + + logger.info( + f"Продление {email} на {duration_days}д для {tg_id}. Баланс: {balance}, списываем: {renewal_cost}" + ) + + key_subgroup = current_tariff.get("subgroup_title") + + await release_session_early(ctx.session) + await renew_key_in_cluster( + cluster_id=server_id, + email=email, + client_id=client_id, + new_expiry_time=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"], + ) + + await ctx.session.execute( + update(Key).where(Key.client_id == client_id).values( + current_device_limit=selected_device_limit, + current_traffic_limit=selected_traffic_limit, + selected_price_rub=renewal_cost, + ) + ) + + if ctx.bulk_updates is not None: + bc = ctx.bulk_updates["balance_changes"] + bc[tg_id] = bc.get(tg_id, 0) - renewal_cost + ctx.bulk_updates["key_expiry_updates"].append((client_id, 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, 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 RenewalResult(RenewalStatus.SUCCESS, current_tariff, new_expiry_time) diff --git a/handlers/notifications/notify_utils.py b/handlers/notifications/sender.py similarity index 62% rename from handlers/notifications/notify_utils.py rename to handlers/notifications/sender.py index 51094bc3..917958ef 100644 --- a/handlers/notifications/notify_utils.py +++ b/handlers/notifications/sender.py @@ -1,33 +1,34 @@ +from __future__ import annotations + import asyncio import os import time from collections import OrderedDict, deque -from datetime import datetime import aiofiles -import pytz from aiogram import Bot -from aiogram.exceptions import ( - TelegramBadRequest, - TelegramForbiddenError, - TelegramRetryAfter, -) +from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError, TelegramRetryAfter from aiogram.types import BufferedInputFile, InlineKeyboardMarkup -from sqlalchemy.ext.asyncio import AsyncSession -from database import async_session_maker, create_blocked_user +from datetime import datetime + +import pytz + +from database import async_session_maker +from database.bans import save_blocked_user_ids from handlers.utils import format_hours, format_minutes, get_russian_month from logger import logger from services.tariffs.tariff_display import get_key_tariff_display - moscow_tz = pytz.timezone("Europe/Moscow") + _photo_cache: OrderedDict[str, str] = OrderedDict() _photo_cache_lock = asyncio.Lock() _PHOTO_CACHE_MAX = 64 +_SUPPORTED_EXTENSIONS = (".jpg", ".jpeg", ".png", ".webp", ".gif") async def _get_cached_file_id(photo_path: str) -> str | None: @@ -46,9 +47,6 @@ async def _set_cached_file_id(photo_path: str, file_id: str) -> None: _photo_cache.popitem(last=False) -_SUPPORTED_EXTENSIONS = (".jpg", ".jpeg", ".png", ".webp", ".gif") - - def _find_photo_file(photo_path: str) -> str | None: if os.path.isfile(photo_path): return photo_path @@ -64,106 +62,181 @@ 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.send_times: deque = 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) + cutoff = now - self.window + while self.send_times and self.send_times[0] <= cutoff: + self.send_times.popleft() 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 + time_to_wait = (self.send_times[0] + 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 +def rate_limited_send(func): + async def wrapper(*args, **kwargs): + while True: + try: + return await func(*args, **kwargs) + except TelegramRetryAfter as e: + await asyncio.sleep(int(e.retry_after) + 1) + except TelegramForbiddenError: + return False + except TelegramBadRequest: + return False + except Exception as e: + tg_id = kwargs.get("tg_id") or (args[1] if len(args) > 1 else "?") + logger.error(f"Ошибка отправки пользователю {tg_id}: {e}") + return False + return wrapper + + +async def send_notification( + bot: Bot, + tg_id: int, + image_filename: str | None, + caption: str, + keyboard: InlineKeyboardMarkup | None = None, +) -> bool: + if image_filename is None: + return await _send_text(bot, tg_id, caption, keyboard) + + photo_path = os.path.join("img", image_filename) + cached_id = await _get_cached_file_id(photo_path) + if cached_id: + return await _send_photo(bot, tg_id, photo_path, image_filename, caption, keyboard, cached_id) + + actual_path = _find_photo_file(photo_path) + if actual_path: + return await _send_photo(bot, tg_id, actual_path, image_filename, caption, keyboard) + else: + logger.warning(f"Файл изображения не найден: {photo_path}") + return await _send_text(bot, tg_id, caption, keyboard) + + +@rate_limited_send +async def _send_photo( + bot: Bot, + tg_id: int, + photo_path: str, + image_filename: str, + caption: str, + keyboard: InlineKeyboardMarkup | None = None, + cached_file_id: str | None = None, +) -> bool: + try: + if cached_file_id: + await bot.send_photo(tg_id, cached_file_id, caption=caption, reply_markup=keyboard) + return True + async with aiofiles.open(photo_path, "rb") as f: + image_data = await f.read() + buffered = BufferedInputFile(image_data, filename=image_filename) + result = await bot.send_photo(tg_id, buffered, caption=caption, reply_markup=keyboard) + if result and hasattr(result, "photo") and result.photo: + await _set_cached_file_id(os.path.join("img", image_filename), result.photo[-1].file_id) + return True + except (TelegramForbiddenError, TelegramBadRequest): + return False + except Exception as e: + logger.error(f"Ошибка отправки фото пользователю {tg_id}: {e}") + return await _send_text(bot, tg_id, caption, keyboard) + + +@rate_limited_send +async def _send_text( + bot: Bot, + tg_id: int, + caption: str, + keyboard: InlineKeyboardMarkup | None = None, +) -> bool: + try: + await bot.send_message(tg_id, caption, reply_markup=keyboard) + return True + except (TelegramForbiddenError, TelegramBadRequest): + return False + except Exception as e: + logger.error(f"Ошибка отправки текста пользователю {tg_id}: {e}") + return False class FastNotificationSender: - def __init__(self, bot: Bot, session: AsyncSession | None, messages_per_second: int = 35) -> None: + def __init__(self, bot: Bot, 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.blocked_users: set[int] = set() + self.queue: asyncio.Queue = asyncio.Queue() + self.delayed_queue: asyncio.Queue = asyncio.Queue() + self.results: list[bool] = [] self.total_sent = 0 self.is_running = False - async def _send_single_message(self, msg: NotificationMessage) -> bool: + async def _send_one(self, msg: dict) -> bool: + tg_id = msg["tg_id"] try: await self.rate_limiter.acquire() - if msg.photo: - photo_path = os.path.join("img", msg.photo) + if msg.get("photo"): + photo_path = os.path.join("img", msg["photo"]) cached_id = await _get_cached_file_id(photo_path) if cached_id: await self.bot.send_photo( - chat_id=msg.tg_id, photo=cached_id, caption=msg.text, reply_markup=msg.keyboard + chat_id=tg_id, photo=cached_id, + caption=msg["text"], reply_markup=msg.get("keyboard"), ) else: actual_path = _find_photo_file(photo_path) if actual_path: async with aiofiles.open(actual_path, "rb") as f: image_data = await f.read() - buffered_photo = BufferedInputFile(image_data, filename=os.path.basename(actual_path)) + buffered = BufferedInputFile(image_data, filename=os.path.basename(actual_path)) result = await self.bot.send_photo( - chat_id=msg.tg_id, photo=buffered_photo, caption=msg.text, reply_markup=msg.keyboard + chat_id=tg_id, photo=buffered, + caption=msg["text"], reply_markup=msg.get("keyboard"), ) if result and hasattr(result, "photo") and result.photo: await _set_cached_file_id(photo_path, result.photo[-1].file_id) else: - await self.bot.send_message(chat_id=msg.tg_id, text=msg.text, reply_markup=msg.keyboard) + await self.bot.send_message( + chat_id=tg_id, text=msg["text"], reply_markup=msg.get("keyboard"), + ) else: - await self.bot.send_message(chat_id=msg.tg_id, text=msg.text, reply_markup=msg.keyboard) + await self.bot.send_message( + chat_id=tg_id, text=msg["text"], reply_markup=msg.get("keyboard"), + ) return True except TelegramRetryAfter as e: - msg.retry_after = e.retry_after - msg.attempts += 1 + msg["_retry_after"] = e.retry_after + msg["_attempts"] = msg.get("_attempts", 0) + 1 await self.delayed_queue.put(msg) return False - except TelegramForbiddenError: - self.blocked_users.add(msg.tg_id) + self.blocked_users.add(tg_id) return False - except TelegramBadRequest as e: if "chat not found" in str(e).lower(): - self.blocked_users.add(msg.tg_id) + self.blocked_users.add(tg_id) return False - except Exception: return False - async def _process_delayed_messages(self): + async def _process_delayed(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: + if msg.get("_retry_after"): + await asyncio.sleep(msg["_retry_after"]) + msg["_retry_after"] = None + if msg.get("_attempts", 0) < 3: await self.queue.put(msg) else: self.results.append(False) @@ -178,11 +251,11 @@ class FastNotificationSender: while self.is_running: try: msg = await asyncio.wait_for(self.queue.get(), timeout=0.1) - success = await self._send_single_message(msg) + success = await self._send_one(msg) if success: self.total_sent += 1 self.results.append(True) - elif msg.attempts == 0: + elif msg.get("_attempts", 0) == 0: self.results.append(False) self.queue.task_done() except TimeoutError: @@ -194,14 +267,12 @@ class FastNotificationSender: if not self.blocked_users: return try: - from database.bans import save_blocked_user_ids - async with async_session_maker() as session: await save_blocked_user_ids(session, list(self.blocked_users)) await session.commit() - logger.info(f"📝 Добавлено до {len(self.blocked_users)} пользователей в blocked_users") + logger.info(f"Добавлено до {len(self.blocked_users)} пользователей в blocked_users") except Exception as e: - logger.error(f"❌ Ошибка при сохранении заблокированных пользователей: {e}") + logger.error(f"Ошибка сохранения заблокированных: {e}") async def send_all(self, messages: list[dict], workers: int = 15) -> list[bool]: if not messages: @@ -211,38 +282,29 @@ class FastNotificationSender: self.results = [] self.total_sent = 0 self.blocked_users = set() - start_time = time.time() + start = 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"), - ) + for msg in messages: 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()) + delayed_task = asyncio.create_task(self._process_delayed()) 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 + duration = time.time() - start 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)") + logger.info(f"Уведомления: {self.total_sent}/{len(messages)} за {duration:.1f}s ({speed:.1f} msg/s)") return self.results @@ -250,120 +312,13 @@ class FastNotificationSender: async def send_messages_with_limit( bot: Bot, messages: list[dict], - session: AsyncSession = None, - source_file: str = None, messages_per_second: int = 35, -): - sender = FastNotificationSender(bot, session, messages_per_second) +) -> list[bool]: + sender = FastNotificationSender(bot, messages_per_second) return await sender.send_all(messages) -async def try_add_blocked_user(tg_id: int, session: AsyncSession, source_file: str | None): - if source_file == "special_notifications" and session: - try: - await create_blocked_user(session, tg_id) - logger.info(f"Пользователь {tg_id} добавлен в blocked_users.") - except Exception as e: - logger.warning(f"Не удалось добавить {tg_id} в blocked_users: {e}") - - -def rate_limited_send(func): - async def wrapper(*args, **kwargs): - while True: - try: - return await func(*args, **kwargs) - except TelegramRetryAfter as e: - retry_in = int(e.retry_after) + 1 - logger.warning(f"⚠️ Flood control: повтор через {retry_in} сек.") - await asyncio.sleep(retry_in) - except TelegramForbiddenError: - tg_id = kwargs.get("tg_id") or args[1] - logger.warning(f"🚫 Бот заблокирован пользователем {tg_id}.") - return False - except TelegramBadRequest: - tg_id = kwargs.get("tg_id") or args[1] - logger.warning(f"🚫 Чат не найден для пользователя {tg_id}.") - return False - except Exception as e: - tg_id = kwargs.get("tg_id") or args[1] - logger.error(f"❌ Ошибка отправки сообщения пользователю {tg_id}: {e}") - return False - - return wrapper - - -async def send_notification( - bot: Bot, - tg_id: int, - image_filename: str | None, - caption: str, - keyboard: InlineKeyboardMarkup | None = None, -) -> bool: - if image_filename is None: - return await _send_text_notification(bot, tg_id, caption, keyboard) - - photo_path = os.path.join("img", image_filename) - cached_id = await _get_cached_file_id(photo_path) - if cached_id: - return await _send_photo_notification(bot, tg_id, photo_path, image_filename, caption, keyboard, cached_id) - - actual_path = _find_photo_file(photo_path) - if actual_path: - return await _send_photo_notification(bot, tg_id, actual_path, image_filename, caption, keyboard) - else: - logger.warning(f"Файл с изображением не найден: {photo_path}") - return await _send_text_notification(bot, tg_id, caption, keyboard) - - -@rate_limited_send -async def _send_photo_notification( - bot: Bot, - tg_id: int, - photo_path: str, - image_filename: str, - caption: str, - keyboard: InlineKeyboardMarkup | None = None, - cached_file_id: str | None = None, -) -> bool: - try: - if cached_file_id: - await bot.send_photo(tg_id, cached_file_id, caption=caption, reply_markup=keyboard) - return True - async with aiofiles.open(photo_path, "rb") as image_file: - image_data = await image_file.read() - buffered_photo = BufferedInputFile(image_data, filename=image_filename) - result = await bot.send_photo(tg_id, buffered_photo, caption=caption, reply_markup=keyboard) - if result and hasattr(result, "photo") and result.photo: - await _set_cached_file_id( - os.path.join("img", image_filename), result.photo[-1].file_id - ) - return True - except (TelegramForbiddenError, TelegramBadRequest): - return False - except Exception as e: - logger.error(f"Ошибка отправки фото для пользователя {tg_id}: {e}") - return await _send_text_notification(bot, tg_id, caption, keyboard) - - -@rate_limited_send -async def _send_text_notification( - bot: Bot, - tg_id: int, - caption: str, - keyboard: InlineKeyboardMarkup | None = None, -) -> bool: - try: - await bot.send_message(tg_id, caption, reply_markup=keyboard) - return True - except (TelegramForbiddenError, TelegramBadRequest): - return False - except Exception as e: - logger.error(f"Неизвестная ошибка при отправке сообщения для пользователя {tg_id}: {e}") - return False - - -async def prepare_key_expiry_data(key, session: AsyncSession, current_time: int) -> dict: - """Готовит данные об истечении подписки для уведомлений.""" +async def prepare_key_expiry_data(key, session, current_time: int) -> dict: if isinstance(key, dict): expiry_timestamp = key.get("expiry_time") email = key.get("email") or "" @@ -419,7 +374,7 @@ async def prepare_key_expiry_data(key, session: AsyncSession, current_time: int) if name: tariff_name = name except Exception as error: - logger.warning(f"[NOTIFY] Ошибка при получении тарифных лимитов для {email}: {error}") + logger.warning(f"[NOTIFY] Ошибка тарифных лимитов для {email}: {error}") traffic_text = "безлимит" if traffic_limit_gb == 0 else f"{traffic_limit_gb} ГБ" devices_text = "безлимит" if device_limit == 0 else str(device_limit) diff --git a/handlers/notifications/special_notifications.py b/handlers/notifications/special_notifications.py deleted file mode 100644 index 8e847b5c..00000000 --- a/handlers/notifications/special_notifications.py +++ /dev/null @@ -1,277 +0,0 @@ -from datetime import datetime, timedelta - -import pytz - -from aiogram import Bot, Router, types -from aiogram.types import InlineKeyboardButton, WebAppInfo -from aiogram.utils.keyboard import InlineKeyboardBuilder -from sqlalchemy import update -from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker - -from config import ( - NOTIFY_EXTRA_DAYS, - NOTIFY_INACTIVE, - NOTIFY_INACTIVE_TRAFFIC, - REMNAWAVE_WEBAPP, - REMNAWAVE_WEBAPP_OPEN_IN_BROWSER, - SUPPORT_CHAT_URL, -) -from core.bootstrap import MODES_CONFIG, NOTIFICATIONS_CONFIG -from database import add_notification, check_notifications_bulk -from database.models import Key, User -from database.tariffs import get_tariffs -from handlers.buttons import CONNECT_DEVICE, MAIN_MENU, SUPPORT, TRIAL_BONUS -from handlers.keys.utils import build_key_callback -from panels.remnawave_runtime import fetch_all_remnawave_traffic -from handlers.notifications.notify_utils import send_messages_with_limit -from handlers.texts import ( - TRIAL_INACTIVE_BONUS_MSG, - TRIAL_INACTIVE_FIRST_MSG, - ZERO_TRAFFIC_MSG, -) -from handlers.utils import format_days, is_full_remnawave_cluster -from hooks.hook_buttons import insert_hook_buttons -from hooks.hooks import run_hooks -from logger import logger -from services.operations import get_user_traffic - - -router = Router() -moscow_tz = pytz.timezone("Europe/Moscow") - - -async def notify_inactive_trial_users( - bot: Bot, session: AsyncSession, *, sessionmaker: async_sessionmaker | None = None -): - logger.info("Проверка пользователей, не активировавших пробный период...") - - inactive_hours = int(NOTIFICATIONS_CONFIG.get("INACTIVE_USER_ENABLED", NOTIFY_INACTIVE)) - extra_days = int(NOTIFICATIONS_CONFIG.get("EXTRA_DAYS_AFTER_EXPIRY", NOTIFY_EXTRA_DAYS)) - - if inactive_hours <= 0: - logger.info("INACTIVE_USER_ENABLED <= 0, уведомления для неактивных триалов отключены.") - return - - users = await check_notifications_bulk(session, "inactive_trial", inactive_hours) - logger.info(f"Найдено {len(users)} неактивных пользователей для уведомления.") - - if not users: - logger.info("Проверка пользователей с неактивным пробным периодом завершена.") - return - - trial_tariffs = await get_tariffs(session, group_code="trial") - if not trial_tariffs: - logger.error("[Notifications] Триальный тариф не найден") - return - - trial_days = trial_tariffs[0]["duration_days"] - messages = [] - users_to_extend = [] - - for user in users: - tg_id = user["tg_id"] - username = user["username"] - first_name = user["first_name"] - last_name = user["last_name"] - display_name = username or first_name or last_name or "Пользователь" - - builder = InlineKeyboardBuilder() - 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() - - trial_extended = user["last_notification_time"] is not None - - if trial_extended and extra_days > 0: - total_days = extra_days + trial_days - message = TRIAL_INACTIVE_BONUS_MSG.format( - display_name=display_name, - extra_days_formatted=format_days(extra_days), - total_days_formatted=format_days(total_days), - ) - users_to_extend.append(tg_id) - else: - message = TRIAL_INACTIVE_FIRST_MSG.format( - display_name=display_name, - trial_time_formatted=format_days(trial_days), - ) - - messages.append({ - "tg_id": tg_id, - "text": message, - "keyboard": keyboard, - "notification_id": "inactive_trial", - }) - - if users_to_extend: - await session.execute(update(User).where(User.tg_id.in_(users_to_extend)).values(trial=-1)) - await session.commit() - logger.info(f"Bulk: отмечено {len(users_to_extend)} пользователей с расширенным триалом") - - if messages: - results = await send_messages_with_limit( - bot, - messages, - session=session, - source_file="special_notifications", - messages_per_second=25, - ) - - sent_tg_ids = [] - for msg, result in zip(messages, results, strict=False): - if result: - sent_tg_ids.append(msg["tg_id"]) - - if sent_tg_ids: - if sessionmaker is not None: - async with sessionmaker() as fresh_session: - for tg_id in sent_tg_ids: - await add_notification(fresh_session, tg_id, "inactive_trial") - await fresh_session.commit() - else: - for tg_id in sent_tg_ids: - await add_notification(session, tg_id, "inactive_trial") - logger.info(f"Отправлено {len(sent_tg_ids)} уведомлений неактивным пользователям.") - - logger.info("Проверка пользователей с неактивным пробным периодом завершена.") - - -async def notify_users_no_traffic(bot: Bot, session: AsyncSession, current_time: int, keys: list): - logger.info("Проверка пользователей с нулевым трафиком...") - current_dt = datetime.fromtimestamp(current_time / 1000, tz=moscow_tz) - - inactive_traffic_hours = int(NOTIFICATIONS_CONFIG.get("INACTIVE_TRAFFIC_ENABLED", NOTIFY_INACTIVE_TRAFFIC)) - if inactive_traffic_hours <= 0: - logger.info("INACTIVE_TRAFFIC_ENABLED <= 0, уведомления о нулевом трафике отключены.") - return - - trial_tariffs = await get_tariffs(session, group_code="trial") - trial_tariff_ids = {t["id"] for t in trial_tariffs} if trial_tariffs else set() - - if not trial_tariff_ids: - return - - remnawave_webapp_enabled = bool(MODES_CONFIG.get("REMNAWAVE_WEBAPP_ENABLED", REMNAWAVE_WEBAPP)) - open_in_browser = bool(MODES_CONFIG.get("REMNAWAVE_WEBAPP_OPEN_IN_BROWSER", REMNAWAVE_WEBAPP_OPEN_IN_BROWSER)) - - candidate_keys = [] - for key in keys: - if key.tariff_id not in trial_tariff_ids: - continue - if key.created_at is None or key.notified: - continue - created_at_dt = pytz.utc.localize(datetime.fromtimestamp(key.created_at / 1000)).astimezone(moscow_tz) - if current_dt < created_at_dt + timedelta(hours=inactive_traffic_hours): - continue - if key.expiry_time: - expiry_dt = pytz.utc.localize(datetime.fromtimestamp(key.expiry_time / 1000)).astimezone(moscow_tz) - if current_dt > expiry_dt: - continue - candidate_keys.append(key) - - if not candidate_keys: - logger.info("Нет кандидатов для проверки нулевого трафика.") - return - - needed_uuids = {key.client_id for key in candidate_keys if key.client_id} - logger.info(f"[Zero Traffic] Кандидатов: {len(candidate_keys)}, уникальных UUID: {len(needed_uuids)}") - - try: - traffic_map = await fetch_all_remnawave_traffic(session, needed_uuids=needed_uuids) - except Exception as error: - logger.error(f"[Zero Traffic] Ошибка bulk-получения трафика: {error}") - return - - messages = [] - keys_to_mark_notified = [] - - for key in candidate_keys: - tg_id = key.tg_id - email = key.email - client_id = key.client_id - - keys_to_mark_notified.append(client_id) - - used_bytes = traffic_map.get(client_id) - if used_bytes is None: - logger.warning(f"[Zero Traffic] UUID {client_id} ({email}) не найден в bulk-данных, пропуск") - continue - - if used_bytes > 0: - continue - - if used_bytes == 0: - logger.info(f"У пользователя {tg_id} ({email}) 0 ГБ трафика. Отправляем уведомление.") - builder = InlineKeyboardBuilder() - - server_id = key.server_id - try: - is_full_remnawave = await is_full_remnawave_cluster(server_id, session) - final_link = key.key or key.remnawave_link - - if is_full_remnawave and final_link and remnawave_webapp_enabled: - if open_in_browser: - builder.row(InlineKeyboardButton(text=CONNECT_DEVICE, url=final_link)) - else: - builder.row(InlineKeyboardButton(text=CONNECT_DEVICE, web_app=WebAppInfo(url=final_link))) - else: - builder.row( - InlineKeyboardButton( - text=CONNECT_DEVICE, - callback_data=build_key_callback("connect_device", key.client_id, email), - ) - ) - except Exception as error: - logger.error(f"Ошибка при определении типа панели для {email}: {error}") - builder.row( - InlineKeyboardButton( - text=CONNECT_DEVICE, - callback_data=build_key_callback("connect_device", key.client_id, email), - ) - ) - - builder.row(InlineKeyboardButton(text=SUPPORT, url=SUPPORT_CHAT_URL)) - builder.row(InlineKeyboardButton(text=MAIN_MENU, callback_data="profile")) - - try: - hook_commands = await run_hooks( - "zero_traffic_notification", - chat_id=tg_id, - admin=False, - session=session, - email=email, - ) - if hook_commands: - builder = insert_hook_buttons(builder, hook_commands) - except Exception as error: - logger.warning(f"[ZERO_TRAFFIC_NOTIFICATION] Ошибка при применении хуков: {error}") - - keyboard = builder.as_markup() - message = ZERO_TRAFFIC_MSG.format(email=email) - messages.append({ - "tg_id": tg_id, - "text": message, - "keyboard": keyboard, - "client_id": client_id, - }) - - if keys_to_mark_notified: - try: - await session.execute(update(Key).where(Key.client_id.in_(keys_to_mark_notified)).values(notified=True)) - await session.commit() - logger.info(f"Bulk: отмечено {len(keys_to_mark_notified)} ключей как notified") - except Exception as error: - logger.error(f"Ошибка bulk-обновления notified: {error}") - - if messages: - results = await send_messages_with_limit( - bot, - messages, - session=session, - source_file="special_notifications", - messages_per_second=25, - ) - sent_count = sum(result for result in results if result) - logger.info(f"Отправлено {sent_count} уведомлений о нулевом трафике.") - - logger.info("Обработка пользователей с нулевым трафиком завершена.") diff --git a/main.py b/main.py old mode 100644 new mode 100755 diff --git a/services/operations/deletion.py b/services/operations/deletion.py index 4e8b8130..593f5a5a 100644 --- a/services/operations/deletion.py +++ b/services/operations/deletion.py @@ -10,13 +10,12 @@ from logger import ( PANEL_XUI, ) from panels._3xui import delete_client, get_xui_instance +from panels.remnawave import RemnawaveAPI from .utils import unique_by_api_url async def delete_key_from_cluster(cluster_id: str, email: str, client_id: str, session: AsyncSession): - from panels.remnawave import RemnawaveAPI - try: servers = await get_servers(session) cluster = servers.get(cluster_id) diff --git a/services/tariffs/tariff_display.py b/services/tariffs/tariff_display.py index 784a73c6..294b114d 100644 --- a/services/tariffs/tariff_display.py +++ b/services/tariffs/tariff_display.py @@ -102,17 +102,17 @@ async def resolve_price_to_charge(session: AsyncSession, state_data: dict[str, A traffic_target_gb = base_traffic_gb try: - device_step_rub = int(cfg.get("device_step_rub") or 0) + device_step_rub = int(tariff.get("device_step_rub") or 0) except (TypeError, ValueError): device_step_rub = 0 try: - traffic_step_rub = int(cfg.get("traffic_step_rub") or 0) + traffic_step_rub = int(tariff.get("traffic_step_rub") or 0) except (TypeError, ValueError): traffic_step_rub = 0 - device_overrides = cfg.get("device_overrides") or {} - traffic_overrides = cfg.get("traffic_overrides") or {} + device_overrides = tariff.get("device_overrides") or {} + traffic_overrides = tariff.get("traffic_overrides") or {} device_add_rub = 0 if device_target > base_device_limit: