From 0e4dfbf369979a690b8238ce0485b5156817ed28 Mon Sep 17 00:00:00 2001 From: Vladless Date: Sat, 28 Feb 2026 14:12:30 +0300 Subject: [PATCH] fixed subscription renewal --- core/cache_config.py | 3 +- core/redis_cache.py | 23 +++++++ database/keys.py | 5 +- database/notifications.py | 123 ++++++++++++++++++++++++++++++------ middlewares/subscription.py | 50 ++------------- 5 files changed, 139 insertions(+), 65 deletions(-) diff --git a/core/cache_config.py b/core/cache_config.py index 979bc7ca..9728516f 100644 --- a/core/cache_config.py +++ b/core/cache_config.py @@ -4,9 +4,10 @@ CONCURRENCY_MAX_WAIT_SEC = 300 CONCURRENCY_LIMIT = 200 SUBSCRIPTION_CACHE_SUBSCRIBED_MAXSIZE = 200_000 -SUBSCRIPTION_CACHE_SUBSCRIBED_TTL_SEC = 300 +SUBSCRIPTION_CACHE_SUBSCRIBED_TTL_SEC = 600 SUBSCRIPTION_CACHE_UNSUBSCRIBED_MAXSIZE = 100_000 SUBSCRIPTION_CACHE_UNSUBSCRIBED_TTL_SEC = 60 +SUBSCRIPTION_GET_CHAT_MEMBER_TIMEOUT_SEC = 4.0 CONCURRENCY_REJECT_NOTICE_CACHE_MAXSIZE = 100_000 CONCURRENCY_REJECT_NOTICE_TTL_SEC = 5 diff --git a/core/redis_cache.py b/core/redis_cache.py index 4df2770c..aad1e36c 100644 --- a/core/redis_cache.py +++ b/core/redis_cache.py @@ -57,6 +57,29 @@ async def cache_get(key: str) -> Any | None: return None +async def cache_mget(keys: list[str]) -> list[Any | None]: + """Возвращает значения для ключей (None для отсутствующих). Один round-trip в Redis.""" + if not keys: + return [] + client = await _get_redis() + if client is None: + return [None] * len(keys) + try: + raw_list = await client.mget(keys) + result = [] + for raw in raw_list: + if raw is None: + result.append(None) + else: + try: + result.append(json.loads(raw)) + except Exception: + result.append(None) + return result + except Exception: + return [None] * len(keys) + + async def cache_set(key: str, value: Any, ttl_sec: float) -> bool: client = await _get_redis() if client is None: diff --git a/database/keys.py b/database/keys.py index 204537c6..889870d1 100644 --- a/database/keys.py +++ b/database/keys.py @@ -1,3 +1,4 @@ +import asyncio from datetime import datetime from types import SimpleNamespace @@ -111,7 +112,9 @@ async def store_key( current_device_limit=current_device_limit, current_traffic_limit=current_traffic_limit, ) - session.add(new_key) + add_result = session.add(new_key) + if asyncio.iscoroutine(add_result): + await add_result logger.info(f"[Store Key] Ключ создан: tg_id={tg_id}, client_id={client_id}, server_id={server_id}") await session.commit() diff --git a/database/notifications.py b/database/notifications.py index 0b9c8e59..ceec4029 100644 --- a/database/notifications.py +++ b/database/notifications.py @@ -70,6 +70,9 @@ async def bulk_add_notifications( logger.info(f"✅ Bulk: добавлено/обновлено {len(items)} уведомлений") +INACTIVE_TRIAL_REGISTERED_TYPE = "inactive_trial_registered" + + async def bulk_delete_notifications( session: AsyncSession, items: list[tuple[int, str]], *, commit: bool = False ) -> None: @@ -268,6 +271,100 @@ async def check_notifications_bulk( try: now = datetime.utcnow() + + if notification_type == "inactive_trial": + stmt_inactive = ( + select(User.tg_id) + .where( + and_( + User.trial.in_([0, -1]), + ~User.tg_id.in_(select(BlockedUser.tg_id)), + ~User.tg_id.in_(select(Key.tg_id.distinct())), + ) + ) + ) + result_inactive = await session.execute(stmt_inactive) + inactive_tg_ids = [r[0] for r in result_inactive.all()] + if inactive_tg_ids: + existing = await session.execute( + select(Notification.tg_id).where( + Notification.notification_type == INACTIVE_TRIAL_REGISTERED_TYPE, + Notification.tg_id.in_(inactive_tg_ids), + ) + ) + already = {r[0] for r in existing.all()} + to_register = [tid for tid in inactive_tg_ids if tid not in already] + if to_register: + await bulk_add_notifications( + session, + [(tid, INACTIVE_TRIAL_REGISTERED_TYPE) for tid in to_register], + commit=True, + ) + logger.info(f"Зарегистрировано как неактивные (шаг 1): {len(to_register)} пользователей.") + + subq_registered = ( + select( + Notification.tg_id, + func.max(Notification.last_notification_time).label("registered_time"), + ) + .where(Notification.notification_type == INACTIVE_TRIAL_REGISTERED_TYPE) + .group_by(Notification.tg_id) + .subquery() + ) + subq_sent = ( + select( + Notification.tg_id, + func.max(Notification.last_notification_time).label("last_notification_time"), + ) + .where(Notification.notification_type == notification_type) + .group_by(Notification.tg_id) + .subquery() + ) + stmt = ( + select( + User.tg_id, + Key.email, + User.username, + User.first_name, + User.last_name, + subq_registered.c.registered_time, + subq_sent.c.last_notification_time, + ) + .select_from(User) + .outerjoin(Key, Key.tg_id == User.tg_id) + .outerjoin(subq_registered, subq_registered.c.tg_id == User.tg_id) + .outerjoin(subq_sent, subq_sent.c.tg_id == User.tg_id) + .where( + and_( + User.trial.in_([0, -1]), + ~User.tg_id.in_(select(BlockedUser.tg_id)), + ~User.tg_id.in_(select(Key.tg_id.distinct())), + ) + ) + ) + result = await session.execute(stmt) + users = [] + for row in result: + registered_time = row.registered_time + last_sent_time = row.last_notification_time + first_ok = ( + registered_time is not None + and (now - registered_time) >= timedelta(hours=hours) + and last_sent_time is None + ) + second_ok = last_sent_time is not None and (now - last_sent_time) > timedelta(hours=hours) + if first_ok or second_ok: + users.append({ + "tg_id": row.tg_id, + "email": row.email, + "username": row.username, + "first_name": row.first_name, + "last_name": row.last_name, + "last_notification_time": int(last_sent_time.timestamp() * 1000) if last_sent_time else None, + }) + logger.info(f"Найдено {len(users)} пользователей, готовых к уведомлению типа {notification_type}") + return users + subq_last_notification = ( select(Notification.tg_id, func.max(Notification.last_notification_time).label("last_notification_time")) .where(Notification.notification_type == notification_type) @@ -289,20 +386,15 @@ async def check_notifications_bulk( .outerjoin(Key, Key.tg_id == User.tg_id) .outerjoin(subq_last_notification, subq_last_notification.c.tg_id == User.tg_id) ) - if notification_type == "inactive_trial": - stmt = stmt.where( - and_( - User.trial.in_([0, -1]), - ~User.tg_id.in_(select(BlockedUser.tg_id)), - ~User.tg_id.in_(select(Key.tg_id.distinct())), - ) - ) if tg_ids_batch: stmt = stmt.where(User.tg_id.in_(tg_ids_batch)) if emails_batch: stmt = stmt.where(Key.email.in_(emails_batch)) return stmt + def _can_notify(last_time): + return last_time is None or (now - last_time) > timedelta(hours=hours) + users: list[dict] = [] seen: set[tuple[int, str | None]] = set() @@ -316,8 +408,7 @@ async def check_notifications_bulk( continue seen.add(key) last_time = row.last_notification_time - can_notify = not last_time or (now - last_time > timedelta(hours=hours)) - if can_notify: + if _can_notify(last_time): users.append({ "tg_id": row.tg_id, "email": row.email, @@ -337,8 +428,7 @@ async def check_notifications_bulk( continue seen.add(key) last_time = row.last_notification_time - can_notify = not last_time or (now - last_time > timedelta(hours=hours)) - if can_notify: + if _can_notify(last_time): users.append({ "tg_id": row.tg_id, "email": row.email, @@ -357,8 +447,7 @@ async def check_notifications_bulk( continue seen.add(key) last_time = row.last_notification_time - can_notify = not last_time or (now - last_time > timedelta(hours=hours)) - if can_notify: + if _can_notify(last_time): users.append({ "tg_id": row.tg_id, "email": row.email, @@ -377,8 +466,7 @@ async def check_notifications_bulk( continue seen.add(key) last_time = row.last_notification_time - can_notify = not last_time or (now - last_time > timedelta(hours=hours)) - if can_notify: + if _can_notify(last_time): users.append({ "tg_id": row.tg_id, "email": row.email, @@ -392,8 +480,7 @@ async def check_notifications_bulk( result = await session.execute(stmt) for row in result: last_time = row.last_notification_time - can_notify = not last_time or (now - last_time > timedelta(hours=hours)) - if can_notify: + if _can_notify(last_time): users.append({ "tg_id": row.tg_id, "email": row.email, diff --git a/middlewares/subscription.py b/middlewares/subscription.py index bf480339..86545189 100644 --- a/middlewares/subscription.py +++ b/middlewares/subscription.py @@ -8,13 +8,9 @@ from aiogram.fsm.context import FSMContext from aiogram.types import InlineKeyboardButton, Message, Update from aiogram.utils.keyboard import InlineKeyboardBuilder +from bot import bot from config import CHANNEL_EXISTS, CHANNEL_ID, CHANNEL_REQUIRED, CHANNEL_URL from core.bootstrap import MODES_CONFIG -from core.cache_config import ( - SUBSCRIPTION_CACHE_SUBSCRIBED_TTL_SEC, - SUBSCRIPTION_CACHE_UNSUBSCRIBED_TTL_SEC, -) -from core.redis_cache import cache_delete, cache_get, cache_key, cache_set from handlers.buttons import SUB_CHANELL, SUB_CHANELL_DONE from handlers.texts import SUBSCRIPTION_REQUIRED_MSG from handlers.utils import edit_or_send_message @@ -22,10 +18,6 @@ from logger import logger class SubscriptionMiddleware(BaseMiddleware): - def __init__(self) -> None: - self._subscribed_ttl = SUBSCRIPTION_CACHE_SUBSCRIBED_TTL_SEC - self._unsubscribed_ttl = SUBSCRIPTION_CACHE_UNSUBSCRIBED_TTL_SEC - async def __call__( self, handler: Callable[[Update, dict[str, Any]], Awaitable[Any]], @@ -65,51 +57,19 @@ class SubscriptionMiddleware(BaseMiddleware): else: return await handler(event, data) - cached_status = await self._get_cached_status(tg_id) - if cached_status is False: - logger.info(f"[SubMiddleware] Пользователь {tg_id} не подписан (cache)") - await self._store_user_state(data, message, from_user) - return await self._ask_to_subscribe(message) - if cached_status is True: - return await handler(event, data) - - bot = data.get("bot") - if bot is None: - return await handler(event, data) - try: member = await bot.get_chat_member(CHANNEL_ID, tg_id) - is_subscribed = member.status in ("member", "administrator", "creator") - await self._cache_status(tg_id, is_subscribed) - if not is_subscribed: + if member.status not in ("member", "administrator", "creator"): logger.info(f"[SubMiddleware] Пользователь {tg_id} не подписан") await self._store_user_state(data, message, from_user) return await self._ask_to_subscribe(message) except (TelegramBadRequest, TelegramForbiddenError) as e: - logger.warning(f"[SubMiddleware] Ошибка проверки подписки {tg_id}, пропускаем: {e}") - return await handler(event, data) + logger.warning(f"[SubMiddleware] Ошибка при проверке подписки {tg_id}: {e}") + await self._store_user_state(data, message, from_user) + return await self._ask_to_subscribe(message) return await handler(event, data) - async def _get_cached_status(self, tg_id: int) -> bool | None: - subscribed_key = cache_key("subscribed", tg_id) - unsubscribed_key = cache_key("unsubscribed", tg_id) - if await cache_get(subscribed_key) is not None: - return True - if await cache_get(unsubscribed_key) is not None: - return False - return None - - async def _cache_status(self, tg_id: int, is_subscribed: bool) -> None: - subscribed_key = cache_key("subscribed", tg_id) - unsubscribed_key = cache_key("unsubscribed", tg_id) - if is_subscribed: - await cache_delete(unsubscribed_key) - await cache_set(subscribed_key, 1, self._subscribed_ttl) - else: - await cache_delete(subscribed_key) - await cache_set(unsubscribed_key, 1, self._unsubscribed_ttl) - async def _store_user_state(self, data: dict, message: Message, from_user): state: FSMContext = data.get("state") if not state or not from_user or from_user.is_bot: