diff --git a/database.py b/database.py index 1939871a..c3fb4741 100644 --- a/database.py +++ b/database.py @@ -545,7 +545,7 @@ async def update_balance(tg_id: int, amount: float, session: Any = None): Args: tg_id (int): Telegram ID пользователя. amount (float): Сумма для обновления баланса. - session (Any, optional): Сессия базы данных. Если не передана, создается новая. + session (Any, optional): Сессия базы данных. Если не передана, создается новая. Raises: Exception: В случае ошибки при подключении к базе данных или обновлении баланса. """ diff --git a/handlers/keys/subscriptions.py b/handlers/keys/subscriptions.py index ca2bea1a..6221549f 100644 --- a/handlers/keys/subscriptions.py +++ b/handlers/keys/subscriptions.py @@ -6,24 +6,22 @@ import aiohttp import asyncpg from aiohttp import web -from config import DATABASE_URL, PROJECT_NAME, SUB_MESSAGE, SUPERNODE, TRANSITION_DATE_STR +from config import DATABASE_URL, PROJECT_NAME, SUB_MESSAGE, SUPERNODE, TRANSITION_DATE_STR, USE_COUNTRY_SELECTION from database import get_key_details, get_servers from logger import logger -# Глобальная переменная для пула соединений db_pool = None async def init_db_pool(): - """ - Инициализация пула соединений, если он ещё не создан. - """ + """Инициализация пула соединений, если он ещё не создан.""" global db_pool if not db_pool: db_pool = await asyncpg.create_pool(dsn=DATABASE_URL, min_size=5, max_size=20) async def fetch_url_content(url, tg_id): + """Получает содержимое подписки по URL и декодирует его.""" try: logger.info(f"Получение URL: {url} для tg_id: {tg_id}") timeout = aiohttp.ClientTimeout(total=5) @@ -45,6 +43,7 @@ async def fetch_url_content(url, tg_id): async def combine_unique_lines(urls, tg_id, query_string): + """Объединяет строки подписки, удаляя дубликаты.""" if SUPERNODE: logger.info(f"Режим SUPERNODE активен. Возвращаем первую ссылку для tg_id: {tg_id}") if not urls: @@ -76,88 +75,19 @@ transition_timestamp_ms_adjusted = transition_timestamp_ms - (3 * 60 * 60 * 1000 logger.info(f"Время перехода (с поправкой на часовой пояс): {transition_timestamp_ms_adjusted}") -async def handle_old_subscription(request): +async def handle_subscription(request, old_subscription=False): + """Обрабатывает запрос на подписку (старую или новую).""" email = request.match_info.get("email") + tg_id = request.match_info.get("tg_id") if not old_subscription else None - if not email: - logger.warning("Получен запрос без email") - return web.Response( - text="❌ Неверные параметры запроса. Требуется email.", - status=400, - ) + if not email or (not old_subscription and not tg_id): + logger.warning("Получен запрос с отсутствующими параметрами") + return web.Response(text="❌ Неверные параметры запроса.", status=400) - logger.info(f"Обработка запроса для старого клиента с email: {email}") + logger.info( + f"Обработка запроса для {'старого' if old_subscription else 'нового'} клиента: email={email}, tg_id={tg_id}" + ) - # Инициализируем пул соединений - await init_db_pool() - - async with db_pool.acquire() as conn: - key_info = await get_key_details(email, conn) - - if not key_info: - logger.warning(f"Клиент с email {email} не найден в базе.") - return web.Response( - text="❌ Клиент с таким email не найден.", - status=404, - ) - - created_at_ms = key_info["created_at"] - cluster_name = key_info.get("server_id") - if not cluster_name: - logger.warning(f"У клиента с email {email} отсутствует cluster_name.") - return web.Response( - text="❌ Устаревшие данные. Обратитесь в поддержку.", - status=400, - ) - - logger.info(f"Значение created_at для клиента с email {email}: {created_at_ms}, кластер: {cluster_name}") - - created_at_datetime = datetime.utcfromtimestamp(created_at_ms / 1000) - logger.info(f"Время создания клиента в формате datetime (UTC): {created_at_datetime}") - - if created_at_ms >= transition_timestamp_ms_adjusted: - logger.info(f"Клиент с email {email} является новым.") - return web.Response( - text="❌ Эта ссылка устарела. Пожалуйста, обновите ссылку.", - status=400, - ) - - servers = await get_servers() - cluster_servers = servers.get(cluster_name, []) - logger.info(f"Сервера в кластере: {cluster_servers}") - - urls = [f"{server['subscription_url']}/{email}" for server in cluster_servers] - - combined_subscriptions = await combine_unique_lines(urls, email, "") - - base64_encoded = base64.b64encode("\n".join(combined_subscriptions).encode("utf-8")).decode("utf-8") - - encoded_project_name = f"{PROJECT_NAME} - {SUB_MESSAGE}" - headers = { - "Content-Type": "text/plain; charset=utf-8", - "Content-Disposition": "inline", - "profile-update-interval": "7", - "profile-title": "base64:" + base64.b64encode(encoded_project_name.encode("utf-8")).decode("utf-8"), - } - - logger.info(f"Возвращаем объединенные подписки для email: {email}") - return web.Response(text=base64_encoded, headers=headers) - - -async def handle_new_subscription(request): - email = request.match_info.get("email") - tg_id = request.match_info.get("tg_id") - - if not email or not tg_id: - logger.warning("Получен запрос с отсутствующими параметрами email или tg_id") - return web.Response( - text="❌ Неверные параметры запроса. Требуются email и tg_id.", - status=400, - ) - - logger.info(f"Обработка запроса для нового клиента: email={email}, tg_id={tg_id}") - - # Инициализируем пул соединений await init_db_pool() async with db_pool.acquire() as conn: @@ -165,41 +95,72 @@ async def handle_new_subscription(request): if not client_data: logger.warning(f"Клиент с email {email} не найден в базе.") - return web.Response( - text="❌ Клиент с таким email не найден.", - status=404, - ) + return web.Response(text="❌ Клиент с таким email не найден.", status=404) - stored_tg_id = client_data["tg_id"] - cluster_name = client_data["server_id"] + stored_tg_id = client_data.get("tg_id") + server_id = client_data["server_id"] # В режиме выбора стран — это server_name, иначе — cluster_name - if str(tg_id) != str(stored_tg_id): + if not old_subscription and str(tg_id) != str(stored_tg_id): logger.warning(f"Неверный tg_id для клиента с email {email}.") - return web.Response( - text="❌ Неверные данные. Получите свой ключ в боте.", - status=403, - ) + return web.Response(text="❌ Неверные данные. Получите свой ключ в боте.", status=403) - servers = await get_servers() - cluster_servers = servers.get(cluster_name, []) + if old_subscription: + created_at_ms = client_data["created_at"] + created_at_datetime = datetime.utcfromtimestamp(created_at_ms / 1000) - urls = [f"{server['subscription_url']}/{email}" for server in cluster_servers] + logger.info(f"created_at для {email}: {created_at_datetime}, server_id: {server_id}") - query_string = request.query_string - logger.info(f"Извлечен query string: {query_string}") + if created_at_ms >= transition_timestamp_ms_adjusted: + logger.info(f"Клиент с email {email} является новым.") + return web.Response(text="❌ Эта ссылка устарела. Пожалуйста, обновите ссылку.", status=400) - combined_subscriptions = await combine_unique_lines(urls, tg_id, query_string) + urls = [] - base64_encoded = base64.b64encode("\n".join(combined_subscriptions).encode("utf-8")).decode("utf-8") + if USE_COUNTRY_SELECTION: + logger.info(f"Режим выбора страны активен. Ищем сервер {server_id} в БД.") + server_data = await conn.fetchrow("SELECT subscription_url FROM servers WHERE server_name = $1", server_id) - encoded_project_name = f"{PROJECT_NAME} - {SUB_MESSAGE}" + if not server_data: + logger.warning(f"Не найден сервер {server_id} в БД!") + return web.Response(text="❌ Сервер не найден.", status=404) - headers = { - "Content-Type": "text/plain; charset=utf-8", - "Content-Disposition": "inline", - "profile-update-interval": "7", - "profile-title": "base64:" + base64.b64encode(encoded_project_name.encode("utf-8")).decode("utf-8"), - } + subscription_url = server_data["subscription_url"] + urls = [f"{subscription_url}/{email}"] + logger.info(f"Используем подписку {urls[0]}") - logger.info(f"Возвращаем объединенные подписки для email: {email}") - return web.Response(text=base64_encoded, headers=headers) + else: + servers = await get_servers() + logger.info(f"Режим выбора страны отключен. Используем кластер {server_id}.") + cluster_servers = servers.get(server_id, []) + + if not cluster_servers: + logger.warning(f"Не найдены сервера для {server_id}") + return web.Response(text="❌ Сервер не найден.", status=404) + + urls = [f"{server['subscription_url']}/{email}" for server in cluster_servers] + + query_string = request.query_string if not old_subscription else "" + combined_subscriptions = await combine_unique_lines(urls, tg_id or email, query_string) + + base64_encoded = base64.b64encode("\n".join(combined_subscriptions).encode("utf-8")).decode("utf-8") + encoded_project_name = f"{PROJECT_NAME} - {SUB_MESSAGE}" + + headers = { + "Content-Type": "text/plain; charset=utf-8", + "Content-Disposition": "inline", + "profile-update-interval": "7", + "profile-title": "base64:" + base64.b64encode(encoded_project_name.encode("utf-8")).decode("utf-8"), + } + + logger.info(f"Возвращаем объединенные подписки для email: {email}") + return web.Response(text=base64_encoded, headers=headers) + + +async def handle_old_subscription(request): + """Обработка запроса для старых клиентов.""" + return await handle_subscription(request, old_subscription=True) + + +async def handle_new_subscription(request): + """Обработка запроса для новых клиентов.""" + return await handle_subscription(request, old_subscription=False) diff --git a/handlers/notifications.py b/handlers/notifications.py index aca5cf67..6f53c5d8 100644 --- a/handlers/notifications.py +++ b/handlers/notifications.py @@ -16,12 +16,12 @@ from config import ( ADMIN_USERNAME, AUTO_DELETE_EXPIRED_KEYS, AUTO_RENEW_KEYS, - SUPPORT_CHAT_URL, DATABASE_URL, DELETE_KEYS_DELAY, DEV_MODE, EXPIRED_KEYS_CHECK_INTERVAL, RENEWAL_PLANS, + SUPPORT_CHAT_URL, TOTAL_GB, TRIAL_TIME, ) @@ -218,7 +218,9 @@ async def process_24h_record(record, bot, conn): days_left_message = ( "Ключ истек" if time_left.total_seconds() <= 0 - else f"{time_left.days}" if time_left.days > 0 else f"{time_left.seconds // 3600}" + else f"{time_left.days}" + if time_left.days > 0 + else f"{time_left.seconds // 3600}" ) message_24h = KEY_EXPIRY_24H.format( @@ -301,7 +303,6 @@ async def send_renewal_notification(bot, tg_id, email, message, conn, client_id, logger.error(f"Ошибка при отправке уведомления пользователю {tg_id}: {e}") - async def notify_inactive_trial_users(bot: Bot, conn: asyncpg.Connection): logger.info("Проверка пользователей, не активировавших пробный период...") @@ -368,7 +369,7 @@ async def handle_expired_keys(bot: Bot, conn: asyncpg.Connection, current_time: logger.info("Проверка подписок, срок действия которых скоро истекает...") threshold_time = int((datetime.utcnow() + timedelta(seconds=EXPIRED_KEYS_CHECK_INTERVAL * 1.5)).timestamp() * 1000) - + expiring_keys = await conn.fetch( """ SELECT tg_id, client_id, expiry_time, email, server_id FROM keys @@ -404,9 +405,7 @@ async def handle_expired_keys(bot: Bot, conn: asyncpg.Connection, current_time: await process_key(record, bot, conn, current_time) if time_since_expiry >= DELETE_KEYS_DELAY * 1000: await delete_key_from_cluster( - cluster_id=record["server_id"], - email=record["email"], - client_id=record["client_id"] + cluster_id=record["server_id"], email=record["email"], client_id=record["client_id"] ) await delete_key(record["client_id"], conn) logger.info(f"Подписка {record['client_id']} удалена") @@ -435,23 +434,22 @@ async def handle_expired_keys(bot: Bot, conn: asyncpg.Connection, current_time: reply_markup=keyboard.as_markup(), ) else: - await bot.send_message( - record["tg_id"], - text=message, - reply_markup=keyboard.as_markup() - ) - + await bot.send_message(record["tg_id"], text=message, reply_markup=keyboard.as_markup()) + logger.info(f"Уведомление об удалении отправлено пользователю {record['tg_id']}") else: remaining_time = (DELETE_KEYS_DELAY * 1000 - time_since_expiry) // 1000 - logger.info(f"Подписка {record['client_id']} не удалена. Осталось времени до удаления: {remaining_time} сек. (Удаление через {DELETE_KEYS_DELAY} сек после истечения)") + logger.info( + f"Подписка {record['client_id']} не удалена. Осталось времени до удаления: {remaining_time} сек. (Удаление через {DELETE_KEYS_DELAY} сек после истечения)" + ) except TelegramForbiddenError: logger.warning(f"Бот заблокирован пользователем {record['tg_id']}. Уведомление не отправлено.") except Exception as e: logger.error(f"Ошибка при удалении подписки {record['client_id']}: {e}") + async def process_key(record, bot, conn, current_time, renew=False): tg_id = record["tg_id"] client_id = record["client_id"] @@ -464,8 +462,7 @@ async def process_key(record, bot, conn, current_time, renew=False): current_date = datetime.now(moscow_tz) logger.info( - f"Время истечения подписки: {expiry_time_value} (МСК: {expiry_date}), " - f"Текущее время (МСК): {current_date}" + f"Время истечения подписки: {expiry_time_value} (МСК: {expiry_date}), Текущее время (МСК): {current_date}" ) current_time_utc = int(datetime.utcnow().timestamp() * 1000) @@ -480,25 +477,27 @@ async def process_key(record, bot, conn, current_time, renew=False): f"📅 Ваша подписка: {record['email']} истекла. Пополните баланс для продления.\n\n" ) remaining_time = (expiry_time_value + DELETE_KEYS_DELAY * 1000) - current_time_utc - + if remaining_time > 0: - message += f"⏳ Подписка будет удалена через {format_time_until_deletion(remaining_time//1000)}." - + message += ( + f"⏳ Подписка будет удалена через {format_time_until_deletion(remaining_time // 1000)}." + ) + await send_notification(bot, tg_id, message, "notify_expired.jpg", email) else: if (expiry_time_value - current_time_utc) <= (EXPIRED_KEYS_CHECK_INTERVAL * 1000): await send_notification( - bot, + bot, tg_id, f"Ваша подписка {email} скоро истечет. Пополните баланс для продления.", "notify_expiring.jpg", - email + email, ) elif renew and AUTO_RENEW_KEYS and balance >= RENEWAL_PLANS["1"]["price"]: await update_balance(tg_id, -RENEWAL_PLANS["1"]["price"], conn) new_expiry_time = int((datetime.now(moscow_tz) + timedelta(days=30)).timestamp() * 1000) - + await update_key_expiry(client_id, new_expiry_time, conn) servers = await get_servers(conn) @@ -509,34 +508,35 @@ async def process_key(record, bot, conn, current_time, renew=False): try: image_path = os.path.join("img", "notify_expired.jpg") caption = KEY_RENEWED.format(email=email) - + if os.path.isfile(image_path): async with aiofiles.open(image_path, "rb") as f: await bot.send_photo( tg_id, photo=BufferedInputFile(await f.read(), filename="notify_expired.jpg"), caption=caption, - reply_markup=InlineKeyboardBuilder().as_markup() + reply_markup=InlineKeyboardBuilder().as_markup(), ) else: await bot.send_message(tg_id, text=caption) - + logger.info(f"Уведомление о продлении отправлено {tg_id}") - + except Exception as e: logger.error(f"Ошибка отправки уведомления {tg_id}: {e}") except Exception as e: logger.error(f"Ошибка обработки подписки {tg_id}: {e}") + async def send_notification(bot, tg_id, message, image_name, email): keyboard = InlineKeyboardBuilder() if DELETE_KEYS_DELAY > 0: keyboard.row(types.InlineKeyboardButton(text="🔄 Продлить", callback_data=f"renew_key|{email}")) keyboard.row(types.InlineKeyboardButton(text="👤 Личный кабинет", callback_data="profile")) - + image_path = os.path.join("img", "notify_expired.jpg") - + try: if os.path.isfile(image_path): async with aiofiles.open(image_path, "rb") as f: @@ -544,10 +544,10 @@ async def send_notification(bot, tg_id, message, image_name, email): tg_id, photo=BufferedInputFile(await f.read(), filename="notify_expired.jpg"), caption=message, - reply_markup=keyboard.as_markup() + reply_markup=keyboard.as_markup(), ) else: await bot.send_message(tg_id, text=message, reply_markup=keyboard.as_markup()) - + except TelegramForbiddenError: logger.warning(f"Пользователь {tg_id} заблокировал бота") diff --git a/handlers/start.py b/handlers/start.py index 4cfb23cf..46f6cdb4 100644 --- a/handlers/start.py +++ b/handlers/start.py @@ -226,29 +226,33 @@ async def process_start_logic(message: Message, state: FSMContext, session: Any, async def check_subscription_callback(callback_query: CallbackQuery, state: FSMContext, session: Any, admin: bool): user_id = callback_query.from_user.id logger.info(f"[CALLBACK] Получен callback 'check_subscription' от пользователя {user_id}") - + try: logger.info(f"[CALLBACK] Запрос информации о подписке для пользователя {user_id} на канал {CHANNEL_ID}") member = await bot.get_chat_member(CHANNEL_ID, user_id) logger.info(f"[CALLBACK] Статус подписки пользователя {user_id}: {member.status}") if member.status not in ["member", "administrator", "creator"]: - logger.info(f"[CALLBACK] Пользователь {user_id} НЕ подписан на канал {CHANNEL_ID}. Текущий статус: {member.status}") + logger.info( + f"[CALLBACK] Пользователь {user_id} НЕ подписан на канал {CHANNEL_ID}. Текущий статус: {member.status}" + ) await callback_query.answer("Вы еще не подписаны на канал!", show_alert=True) builder = InlineKeyboardBuilder() builder.row(InlineKeyboardButton(text="✅ Я подписался", callback_data="check_subscription")) await callback_query.message.answer( f"Для использования бота, пожалуйста, подпишитесь на наш канал: {CHANNEL_URL}", - reply_markup=builder.as_markup() + reply_markup=builder.as_markup(), ) logger.info(f"[CALLBACK] Обновлено сообщение с приглашением подписаться для пользователя {user_id}") else: logger.info(f"[CALLBACK] Пользователь {user_id} подписан на канал {CHANNEL_ID} - подписка подтверждена") await callback_query.answer("Подписка подтверждена!") - logger.info(f"[CALLBACK] Перед вызовом process_start_logic для пользователя {user_id}. Текущее сообщение: {callback_query.message.text}") + logger.info( + f"[CALLBACK] Перед вызовом process_start_logic для пользователя {user_id}. Текущее сообщение: {callback_query.message.text}" + ) await process_start_logic(callback_query.message, state, session, admin) logger.info(f"[CALLBACK] Завершен вызов process_start_logic для пользователя {user_id}") - + except Exception as e: logger.error(f"[CALLBACK] Ошибка проверки подписки для пользователя {user_id}: {e}", exc_info=True) await callback_query.answer("Ошибка проверки подписки, повторите попытку", show_alert=True)