diff --git a/handlers/keys/subscriptions.py b/handlers/keys/subscriptions.py index 820a9023..c6ed0f83 100644 --- a/handlers/keys/subscriptions.py +++ b/handlers/keys/subscriptions.py @@ -12,10 +12,10 @@ from aiohttp import web from config import ( DATABASE_URL, + RANDOM_SUBSCRIPTIONS, PROJECT_NAME, SUPERNODE, SUPPORT_CHAT_URL, - TOTAL_GB, USERNAME_BOT, USE_COUNTRY_SELECTION, ) @@ -24,33 +24,43 @@ from handlers.utils import convert_to_bytes from logger import logger -async def fetch_url_content(url: str, identifier: str) -> list[str]: +async def fetch_url_content(url: str, identifier: str) -> tuple[list[str], dict[str, str]]: try: timeout = aiohttp.ClientTimeout(total=5) async with aiohttp.ClientSession(timeout=timeout) as session: async with session.get(url, ssl=False) as response: if response.status == 200: content = await response.text() - return base64.b64decode(content).decode("utf-8").split("\n") - return [] - except Exception: - return [] + lines = base64.b64decode(content).decode("utf-8").split("\n") + headers = {k.lower(): v for k, v in response.headers.items()} + logger.debug(f"Fetched {url}: {len(lines)} lines, headers: {headers}") + return lines, headers + return [], {} + except Exception as e: + logger.error(f"Error fetching URL {url}: {e}") + return [], {} -async def combine_unique_lines(urls: list[str], identifier: str, query_string: str) -> list[str]: +async def combine_unique_lines(urls: list[str], identifier: str, query_string: str) -> tuple[list[str], list[dict[str, str]]]: if SUPERNODE: + logger.info(f"Режим SUPERNODE активен. Возвращаем первую ссылку для идентификатора: {identifier}") if not urls: - return [] + return [], [] url_with_query = f"{urls[0]}?{query_string}" if query_string else urls[0] - return await fetch_url_content(url_with_query, identifier) + lines, headers = await fetch_url_content(url_with_query, identifier) + return lines, [headers] urls_with_query = [f"{url}?{query_string}" if query_string else url for url in urls] tasks = [fetch_url_content(url, identifier) for url in urls_with_query] results = await asyncio.gather(*tasks, return_exceptions=True) all_lines = set() - for lines in results: - all_lines.update(filter(None, lines)) - return list(all_lines) + all_headers = [] + for result in results: + if isinstance(result, tuple): + lines, headers = result + all_lines.update(filter(None, lines)) + all_headers.append(headers) + return list(all_lines), all_headers async def get_subscription_urls(server_id: str, email: str, conn, include_remnawave_key: str = None) -> list[str]: @@ -72,38 +82,59 @@ async def get_subscription_urls(server_id: str, email: str, conn, include_remnaw return urls -def calculate_traffic(cleaned_subscriptions: list[str], expiry_time_ms: int | None) -> str: +def calculate_traffic(cleaned_subscriptions: list[str], expiry_time_ms: int | None, headers_list: list[dict[str, str]]) -> str: + logger.debug(f"Calculating traffic with subscriptions: {cleaned_subscriptions}, headers: {headers_list}") expire_timestamp = int(expiry_time_ms / 1000) if expiry_time_ms else 0 - if TOTAL_GB != 0: - country_remaining = {} - for line in cleaned_subscriptions: - if "#" not in line: - continue - try: - _, meta = line.split("#", 1) - except ValueError: - continue - parts = meta.split("-") - country = parts[0].strip() - remaining_str = parts[1].strip() if len(parts) == 2 else "" - if remaining_str: - remaining_str = remaining_str.replace(",", ".") - m_total = re.search(r"([\d\.]+)\s*([GMKTB]B)", remaining_str, re.IGNORECASE) - if m_total: - value = float(m_total.group(1)) - unit = m_total.group(2).upper() - remaining_bytes = convert_to_bytes(value, unit) - country_remaining[country] = remaining_bytes - num_countries = len(country_remaining) - issued_per_country = TOTAL_GB - total_traffic_bytes = issued_per_country * num_countries - consumed_traffic_bytes = total_traffic_bytes - sum(country_remaining.values()) - if consumed_traffic_bytes < 0: - consumed_traffic_bytes = 0 - else: - consumed_traffic_bytes = 1 - total_traffic_bytes = 0 - return f"upload=0; download={consumed_traffic_bytes}; total={total_traffic_bytes}; expire={expire_timestamp}" + + upload = 0 + download = 0 + total = 0 + for headers in headers_list: + userinfo = headers.get("subscription-userinfo", "") + if userinfo: + parts = userinfo.split(";") + for part in parts: + part = part.strip() + if part.startswith("upload="): + upload += int(part.split("=")[1]) + elif part.startswith("download="): + download += int(part.split("=")[1]) + elif part.startswith("total="): + total += int(part.split("=")[1]) + logger.debug(f"Processed Subscription-Userinfo: {userinfo}") + + country_remaining = {} + for line in cleaned_subscriptions: + if "#" not in line: + continue + try: + _, meta = line.split("#", 1) + except ValueError: + continue + parts = meta.split("-") + country = parts[0].strip() + remaining_str = parts[1].strip() if len(parts) == 2 else "" + if remaining_str: + remaining_str = remaining_str.replace(",", ".") + m_total = re.search(r"([\d\.]+)\s*([GMKTB]B)", remaining_str, re.IGNORECASE) + if m_total: + value = float(m_total.group(1)) + unit = m_total.group(2).upper() + remaining_bytes = convert_to_bytes(value, unit) + country_remaining[country] = remaining_bytes + logger.debug(f"Found traffic: {value}{unit} for {country}") + + consumed_traffic_bytes = total - sum(country_remaining.values()) if country_remaining else download + if consumed_traffic_bytes < 0: + consumed_traffic_bytes = 0 + download = max(download, consumed_traffic_bytes) + + if download == 0 and total == 0 and not country_remaining: + download = 1 + + result = f"upload={upload}; download={download}; total={total}; expire={expire_timestamp}" + logger.debug(f"Calculated subscription-userinfo: {result}") + return result def clean_subscription_line(line: str) -> str: @@ -204,15 +235,16 @@ async def handle_subscription(request: web.Request) -> web.Response: time_left = format_time_left(expiry_time_ms) urls = await get_subscription_urls( - server_id, email, conn, include_remnawave_key=client_data.get("remnawave_link") + server_id, email, conn, include_remnawave_key=client_data.get("remnawave_key") ) if not urls: return web.Response(text="❌ Сервер не найден.", status=404) query_string = request.query_string - combined_subscriptions = await combine_unique_lines(urls, tg_id or email, query_string) - random.shuffle(combined_subscriptions) + combined_subscriptions, headers_list = await combine_unique_lines(urls, tg_id or email, query_string) + if RANDOM_SUBSCRIPTIONS: + random.shuffle(combined_subscriptions) cleaned_subscriptions = [clean_subscription_line(line) for line in combined_subscriptions] @@ -220,7 +252,7 @@ async def handle_subscription(request: web.Request) -> web.Response: subscription_info = f"📄 Подписка: {email} - {time_left}" user_agent = request.headers.get("User-Agent", "") - subscription_userinfo = calculate_traffic(cleaned_subscriptions, expiry_time_ms) + subscription_userinfo = calculate_traffic(cleaned_subscriptions, expiry_time_ms, headers_list) headers = prepare_headers(user_agent, PROJECT_NAME, subscription_info, subscription_userinfo) return web.Response(text=base64_encoded, headers=headers) diff --git a/handlers/notifications/general_notifications.py b/handlers/notifications/general_notifications.py index 4c26d80a..d6951065 100644 --- a/handlers/notifications/general_notifications.py +++ b/handlers/notifications/general_notifications.py @@ -1,11 +1,8 @@ import asyncio - from datetime import datetime, timedelta - import asyncpg -import pytz - from aiogram import Bot, Router +import pytz from config import ( DATABASE_URL, @@ -33,7 +30,10 @@ from database import ( update_key_expiry, ) from handlers.keys.key_utils import delete_key_from_cluster, renew_key_in_cluster -from handlers.notifications.notify_kb import build_notification_expired_kb, build_notification_kb +from handlers.notifications.notify_kb import ( + build_notification_expired_kb, + build_notification_kb, +) from handlers.texts import ( KEY_DELETED_MSG, KEY_EXPIRED_DELAY_HOURS_MINUTES_MSG, @@ -47,17 +47,13 @@ from handlers.texts import ( ) from handlers.utils import format_hours, format_minutes from logger import logger -from handlers.utils import format_hours, format_months, format_minutes from .notify_utils import 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() @@ -77,7 +73,6 @@ async def periodic_notifications(bot: Bot): try: conn = await asyncpg.connect(DATABASE_URL) current_time = int(datetime.now(moscow_tz).timestamp() * 1000) - threshold_time_10h = int((datetime.now(moscow_tz) + timedelta(hours=10)).timestamp() * 1000) threshold_time_24h = int((datetime.now(moscow_tz) + timedelta(days=1)).timestamp() * 1000) @@ -91,17 +86,35 @@ async def periodic_notifications(bot: Bot): keys = [] if not TRIAL_TIME_DISABLE: - await notify_inactive_trial_users(bot, conn) + try: + await notify_inactive_trial_users(bot, conn) + except Exception as e: + logger.error(f"Ошибка в notify_inactive_trial_users: {e}") await asyncio.sleep(0.5) - await notify_24h_keys(bot, conn, current_time, threshold_time_24h, keys) + try: + await notify_24h_keys(bot, conn, current_time, threshold_time_24h, keys) + except Exception as e: + logger.error(f"Ошибка в notify_24h_keys: {e}") await asyncio.sleep(1) - await notify_10h_keys(bot, conn, current_time, threshold_time_10h, keys) + + try: + await notify_10h_keys(bot, conn, current_time, threshold_time_10h, keys) + except Exception as e: + logger.error(f"Ошибка в notify_10h_keys: {e}") await asyncio.sleep(1) - await handle_expired_keys(bot, conn, current_time, keys) + + try: + await handle_expired_keys(bot, conn, current_time, keys) + except Exception as e: + logger.error(f"Ошибка в handle_expired_keys: {e}") await asyncio.sleep(0.5) + if NOTIFY_INACTIVE_TRAFFIC: - await notify_users_no_traffic(bot, conn, current_time, keys) + try: + await notify_users_no_traffic(bot, conn, current_time, keys) + except Exception as e: + logger.error(f"Ошибка в notify_users_no_traffic: {e}") await asyncio.sleep(0.5) logger.info("Завершена обработка уведомлений") @@ -162,7 +175,11 @@ async def notify_24h_keys(bot: Bot, conn: asyncpg.Connection, current_time: int, ) if NOTIFY_RENEW: - await process_auto_renew_or_notify(bot, conn, key, notification_id, 1, "notify_24h.jpg", notification_text) + try: + await process_auto_renew_or_notify(bot, conn, key, notification_id, 1, "notify_24h.jpg", notification_text) + except Exception as e: + logger.error(f"Ошибка авто-продления/уведомления для пользователя {tg_id}: {e}") + continue else: keyboard = build_notification_kb(email) messages.append({ @@ -170,12 +187,21 @@ async def notify_24h_keys(bot: Bot, conn: asyncpg.Connection, current_time: int, "text": notification_text, "photo": "notify_24h.jpg", "keyboard": keyboard, + "notification_id": notification_id, }) - await add_notification(tg_id, notification_id, session=conn) if messages: - await send_messages_with_limit(bot, messages) - logger.info(f"Отправлено {len(messages)} уведомлений об истечении подписки через 24 часа.") + results = await send_messages_with_limit(bot, messages, conn=conn) + sent_count = 0 + for msg, result in zip(messages, results): + tg_id = msg["tg_id"] + if result: + await add_notification(tg_id, msg["notification_id"], session=conn) + sent_count += 1 + logger.info(f"📢 Отправлено уведомление об истекающей подписке {msg['email']} пользователю {tg_id}.") + else: + logger.warning(f"📢 Не удалось отправить уведомление об истекающей подписке {msg['email']} пользователю {tg_id}.") + logger.info(f"Отправлено {sent_count} уведомлений об истечении подписки через 24 часа.") logger.info("Обработка всех уведомлений за 24 часа завершена.") await asyncio.sleep(1) @@ -233,6 +259,7 @@ async def notify_10h_keys(bot: Bot, conn: asyncpg.Connection, current_time: int, ) except Exception as e: logger.error(f"Ошибка авто-продления/уведомления для пользователя {tg_id}: {e}") + continue else: keyboard = build_notification_kb(email) messages.append({ @@ -240,12 +267,21 @@ async def notify_10h_keys(bot: Bot, conn: asyncpg.Connection, current_time: int, "text": notification_text, "photo": "notify_10h.jpg", "keyboard": keyboard, + "notification_id": notification_id, }) - await add_notification(tg_id, notification_id, session=conn) if messages: - await send_messages_with_limit(bot, messages) - logger.info(f"Отправлено {len(messages)} уведомлений об истечении подписки через 10 часов.") + results = await send_messages_with_limit(bot, messages, conn=conn) + sent_count = 0 + for msg, result in zip(messages, results): + tg_id = msg["tg_id"] + if result: + await add_notification(tg_id, msg["notification_id"], session=conn) + sent_count += 1 + logger.info(f"📢 Отправлено уведомление об истекающей подписке {msg['email']} пользователю {tg_id}.") + else: + logger.warning(f"📢 Не удалось отправить уведомление об истекающей подписке {msg['email']} пользователю {tg_id}.") + logger.info(f"Отправлено {sent_count} уведомлений об истечении подписки через 10 часов.") logger.info("Обработка всех уведомлений за 10 часов завершена.") await asyncio.sleep(1) @@ -317,6 +353,8 @@ async def handle_expired_keys(bot: Bot, conn: asyncpg.Connection, current_time: "text": KEY_DELETED_MSG.format(email=email), "photo": "notify_expired.jpg", "keyboard": keyboard, + "notification_id": notification_id, + "email": email, }) except Exception as e: logger.error(f"Ошибка удаления ключа {client_id} для пользователя {tg_id}: {e}") @@ -350,12 +388,23 @@ async def handle_expired_keys(bot: Bot, conn: asyncpg.Connection, current_time: "text": delay_message, "photo": "notify_expired.jpg", "keyboard": keyboard, + "notification_id": notification_id, + "email": email, }) - await add_notification(tg_id, notification_id, session=conn) if messages: - await send_messages_with_limit(bot, messages) - logger.info(f"Отправлено {len(messages)} уведомлений об истекших ключах.") + results = await send_messages_with_limit(bot, messages, conn=conn) + sent_count = 0 + for msg, result in zip(messages, results): + tg_id = msg["tg_id"] + email = msg["email"] + if result: + await add_notification(tg_id, msg["notification_id"], session=conn) + sent_count += 1 + logger.info(f"📢 Отправлено уведомление об истекшем ключе для подписки {email} пользователю {tg_id}.") + else: + logger.warning(f"📢 Не удалось отправить уведомление об истекшем ключе для подписки {email} пользователю {tg_id}.") + logger.info(f"Отправлено {sent_count} уведомлений об истекших ключах.") logger.info("Обработка истекших ключей завершена.") await asyncio.sleep(1) @@ -404,7 +453,7 @@ async def process_auto_renew_or_notify( total_gb = int(renewal_period_months * TOTAL_GB * 1024**3) logger.info( - f"[Автопродление] Продление подписки {email} на {renewal_period_months} мес. для пользователя {tg_id}. Баланс: {balance}, списываем: {renewal_cost}" + f"Продление подписки {email} на {renewal_period_months} мес. для пользователя {tg_id}. Баланс: {balance}, списываем: {renewal_cost}" ) try: @@ -420,9 +469,7 @@ async def process_auto_renew_or_notify( ) renewed_message = KEY_RENEWED.format( - email=email, - months_formatted=months_formatted, - expiry_date=formatted_expiry_date + email=email, months=renewal_period_months, expiry_date=formatted_expiry_date ) keyboard = build_notification_expired_kb() @@ -439,9 +486,9 @@ async def process_auto_renew_or_notify( logger.error(f"❌ Ошибка при продлении ключа {client_id} для пользователя {tg_id}: {e}") else: keyboard = build_notification_kb(email) + await add_notification(tg_id, notification_id, session=conn) result = await send_notification(bot, tg_id, standard_photo, standard_caption, keyboard) if result: logger.info(f"📢 Отправлено уведомление об истекающей подписке {email} пользователю {tg_id}.") - await add_notification(tg_id, notification_id, session=conn) else: logger.warning(f"📢 Не удалось отправить уведомление об истекающей подписке {email} пользователю {tg_id}.") diff --git a/handlers/notifications/notify_utils.py b/handlers/notifications/notify_utils.py index 679db645..7092c91b 100644 --- a/handlers/notifications/notify_utils.py +++ b/handlers/notifications/notify_utils.py @@ -2,29 +2,75 @@ import asyncio import os import aiofiles - +import asyncpg from aiogram import Bot -from aiogram.exceptions import TelegramForbiddenError, TelegramRetryAfter +from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError, TelegramRetryAfter from aiogram.types import BufferedInputFile, InlineKeyboardMarkup +from database import create_blocked_user from logger import logger -async def send_messages_with_limit(bot: Bot, messages: list[dict], messages_per_second: int = 25): +async def send_messages_with_limit( + bot: Bot, + messages: list[dict], + conn: asyncpg.Connection = None, + source_file: str = None, + messages_per_second: int = 25 +): """ Отправляет сообщения с ограничением по количеству сообщений в секунду. + Возвращает список результатов отправки (True для успеха, False для ошибки). """ batch_size = messages_per_second + results = [] for i in range(0, len(messages), batch_size): batch = messages[i : i + batch_size] tasks = [] for msg in batch: - tasks.append(send_notification(bot, msg["tg_id"], msg.get("photo"), msg["text"], msg.get("keyboard"))) - try: - await asyncio.gather(*tasks, return_exceptions=True) - except Exception as e: - logger.error(f"⚠ Ошибка при отправке сообщений в батче: {e}") + tasks.append(send_notification( + bot, + msg["tg_id"], + msg.get("photo"), + msg["text"], + msg.get("keyboard") + )) + batch_results = await asyncio.gather(*tasks, return_exceptions=True) + processed_results = [] + for msg, result in zip(batch, batch_results): + tg_id = msg["tg_id"] + if isinstance(result, bool) and result: + processed_results.append(True) + elif isinstance(result, TelegramForbiddenError): + logger.warning(f"🚫 Бот заблокирован пользователем {tg_id}.") + if source_file == "special_notifications" and conn: + try: + await create_blocked_user(tg_id, conn) + logger.info(f"Пользователь {tg_id} добавлен в blocked_users.") + except Exception: + pass + processed_results.append(False) + elif isinstance(result, TelegramBadRequest) and "chat not found" in str(result).lower(): + logger.warning(f"🚫 Чат не найден для пользователя {tg_id}.") + if source_file == "special_notifications" and conn: + try: + await create_blocked_user(tg_id, conn) + logger.info(f"Пользователь {tg_id} добавлен в blocked_users.") + except Exception: + pass + processed_results.append(False) + else: + logger.warning(f"📩 Не удалось отправить уведомление пользователю {tg_id}.") + if source_file == "special_notifications" and conn: + try: + await create_blocked_user(tg_id, conn) + logger.info(f"Пользователь {tg_id} добавлен в blocked_users.") + except Exception: + pass + processed_results.append(False) + results.extend(processed_results) await asyncio.sleep(1.0) + return results def rate_limited_send(func): @@ -38,13 +84,16 @@ def rate_limited_send(func): await asyncio.sleep(retry_in) except TelegramForbiddenError: tg_id = kwargs.get("tg_id") or args[1] - logger.warning(f"Пользователь {tg_id} заблокировал бота.") + 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 @@ -85,8 +134,7 @@ async def _send_photo_notification( buffered_photo = BufferedInputFile(image_data, filename=image_filename) await bot.send_photo(tg_id, buffered_photo, caption=caption, reply_markup=keyboard) return True - except TelegramForbiddenError: - logger.error(f"Пользователь {tg_id} заблокировал бота") + except (TelegramForbiddenError, TelegramBadRequest): return False except Exception as e: logger.error(f"Ошибка отправки фото для пользователя {tg_id}: {e}") @@ -104,8 +152,7 @@ async def _send_text_notification( try: await bot.send_message(tg_id, caption, reply_markup=keyboard) return True - except TelegramForbiddenError: - logger.error(f"Пользователь {tg_id} заблокировал бота") + except (TelegramForbiddenError, TelegramBadRequest): return False except Exception as e: logger.error(f"Неизвестная ошибка при отправке сообщения для пользователя {tg_id}: {e}") diff --git a/handlers/notifications/special_notifications.py b/handlers/notifications/special_notifications.py index d58a13e4..9f806027 100644 --- a/handlers/notifications/special_notifications.py +++ b/handlers/notifications/special_notifications.py @@ -1,32 +1,32 @@ import asyncio - from datetime import datetime, timedelta import asyncpg import pytz - from aiogram import Bot, Router, types -from aiogram.exceptions import TelegramForbiddenError from aiogram.utils.keyboard import InlineKeyboardBuilder -from config import NOTIFY_EXTRA_DAYS, NOTIFY_INACTIVE, NOTIFY_INACTIVE_TRAFFIC, SUPPORT_CHAT_URL, TRIAL_TIME -from database import ( - add_notification, - check_notifications_bulk, - create_blocked_user, +from config import ( + NOTIFY_EXTRA_DAYS, + NOTIFY_INACTIVE, + NOTIFY_INACTIVE_TRAFFIC, + SUPPORT_CHAT_URL, + TRIAL_TIME, ) +from database import add_notification, check_notifications_bulk, create_blocked_user from handlers.buttons import MAIN_MENU from handlers.keys.key_utils import get_user_traffic -from handlers.texts import TRIAL_INACTIVE_BONUS_MSG, TRIAL_INACTIVE_FIRST_MSG, ZERO_TRAFFIC_MSG +from handlers.texts import ( + TRIAL_INACTIVE_BONUS_MSG, + TRIAL_INACTIVE_FIRST_MSG, + ZERO_TRAFFIC_MSG, +) from handlers.utils import format_days from logger import logger -from handlers.utils import format_days - from .notify_utils import send_messages_with_limit, send_notification router = Router() - moscow_tz = pytz.timezone("Europe/Moscow") @@ -37,19 +37,15 @@ async def notify_inactive_trial_users(bot: Bot, conn: asyncpg.Connection): Если прошло 24 часа и триал не активирован, отправляется уведомление с бонусом +2 дня. """ logger.info("Проверка пользователей, не активировавших пробный период...") - users = await check_notifications_bulk("inactive_trial", NOTIFY_INACTIVE, conn) logger.info(f"Найдено {len(users)} неактивных пользователей для уведомления.") - messages = [] - 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( @@ -59,9 +55,7 @@ async def notify_inactive_trial_users(bot: Bot, conn: asyncpg.Connection): ) 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: total_days = NOTIFY_EXTRA_DAYS + TRIAL_TIME message = TRIAL_INACTIVE_BONUS_MSG.format( @@ -74,21 +68,30 @@ async def notify_inactive_trial_users(bot: Bot, conn: asyncpg.Connection): message = TRIAL_INACTIVE_FIRST_MSG.format( display_name=display_name, trial_time_formatted=format_days(TRIAL_TIME) ) - - try: - await bot.send_message(tg_id, message, reply_markup=keyboard) - logger.info(f"📩 Отправлено уведомление неактивному пользователю {tg_id}.") - await add_notification(tg_id, "inactive_trial", session=conn) - - except TelegramForbiddenError: - logger.warning(f"🚫 Бот заблокирован пользователем {tg_id}. Добавляем в blocked_users.") - await create_blocked_user(tg_id, conn) - - except Exception as e: - logger.error(f"⚠ Ошибка при отправке уведомления пользователю {tg_id}: {e}") - - await asyncio.sleep(1) - + messages.append({ + "tg_id": tg_id, + "text": message, + "keyboard": keyboard, + "notification_id": "inactive_trial", + }) + if messages: + results = await send_messages_with_limit( + bot, + messages, + conn=conn, + source_file="special_notifications", + messages_per_second=25 + ) + sent_count = 0 + for msg, result in zip(messages, results): + tg_id = msg["tg_id"] + if result: + await add_notification(tg_id, msg["notification_id"], session=conn) + sent_count += 1 + logger.info(f"📩 Отправлено уведомление неактивному пользователю {tg_id}.") + else: + logger.warning(f"📩 Не удалось отправить уведомление неактивному пользователю {tg_id}.") + logger.info(f"Отправлено {sent_count} уведомлений неактивным пользователям.") logger.info("✅ Проверка пользователей с неактивным пробным периодом завершена.") @@ -99,8 +102,8 @@ async def notify_users_no_traffic(bot: Bot, conn: asyncpg.Connection, current_ti но исключает пользователей, у которых подписка недавно продлилась. """ logger.info("Проверка пользователей с нулевым трафиком...") - current_dt = datetime.fromtimestamp(current_time / 1000, tz=moscow_tz) + messages = [] for key in keys: tg_id = key.get("tg_id") @@ -124,7 +127,6 @@ async def notify_users_no_traffic(bot: Bot, conn: asyncpg.Connection, current_ti expiry_dt = pytz.utc.localize(datetime.fromtimestamp(expiry_time / 1000)).astimezone(moscow_tz) renewal_threshold = expiry_dt - timedelta(days=30) renewal_recent = current_dt - renewal_threshold < timedelta(hours=NOTIFY_INACTIVE_TRAFFIC) - if renewal_recent: continue @@ -144,39 +146,45 @@ async def notify_users_no_traffic(bot: Bot, conn: asyncpg.Connection, current_ti total_traffic = sum( value if isinstance(value, int | float) else 0 for value in traffic_data.get("traffic", {}).values() ) - logger.info(f"Ключ для {email}: общий трафик: {total_traffic} ГБ") + + try: + await conn.execute( + "UPDATE keys SET notified = TRUE WHERE tg_id = $1 AND client_id = $2", tg_id, client_id + ) + except Exception as e: + logger.error(f"Ошибка обновления notified для пользователя {tg_id} (client_id: {client_id}): {e}") + continue if total_traffic == 0: logger.info(f"⚠ У пользователя {tg_id} ({email}) 0 ГБ трафика. Отправляем уведомление.") - builder = InlineKeyboardBuilder() builder.row(types.InlineKeyboardButton(text="🔧 Написать в поддержку", url=SUPPORT_CHAT_URL)) builder.row(types.InlineKeyboardButton(text=MAIN_MENU, callback_data="profile")) 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, + }) - try: - result = await send_notification(bot, tg_id, None, message, keyboard) - await conn.execute( - "UPDATE keys SET notified = TRUE WHERE tg_id = $1 AND client_id = $2", tg_id, client_id - ) - if result: - logger.info(f"📩 Отправлено уведомление пользователю {tg_id} о нулевом трафике.") - else: - logger.warning(f"📩 Не удалось отправить уведомление пользователю {tg_id} о нулевом трафике.") - except TelegramForbiddenError: - logger.warning(f"🚫 Бот заблокирован пользователем {tg_id}.") - await create_blocked_user(tg_id, conn) - except Exception as e: - logger.error(f"⚠ Ошибка при отправке уведомления пользователю {tg_id}: {e}") - else: - try: - await conn.execute( - "UPDATE keys SET notified = TRUE WHERE tg_id = $1 AND client_id = $2", tg_id, client_id - ) - logger.info(f"Ключ для {email} имеет трафик. Обновлено notified = TRUE.") - except Exception as e: - logger.error(f"Ошибка обновления notified для пользователя {tg_id}: {e}") + if messages: + results = await send_messages_with_limit( + bot, + messages, + conn=conn, + source_file="special_notifications", + messages_per_second=25 + ) + sent_count = 0 + for msg, result in zip(messages, results): + tg_id = msg["tg_id"] + if result: + sent_count += 1 + logger.info(f"📩 Отправлено уведомление пользователю {tg_id} о нулевом трафике.") + else: + logger.warning(f"📩 Не удалось отправить уведомление пользователю {tg_id} о нулевом трафике.") + logger.info(f"Отправлено {sent_count} уведомлений о нулевом трафике.") logger.info("✅ Обработка пользователей с нулевым трафиком завершена.")