From 1e9367ab9312ae354665cb5e085d23a214f9392d Mon Sep 17 00:00:00 2001 From: Boris Kovalskii <36034823+JustYay@users.noreply.github.com> Date: Wed, 15 Jan 2025 21:05:00 +1000 Subject: [PATCH 1/7] optimize subscriptions Get and update subscriptions faster --- handlers/keys/subscriptions.py | 101 +++++++++++++++++---------------- 1 file changed, 52 insertions(+), 49 deletions(-) diff --git a/handlers/keys/subscriptions.py b/handlers/keys/subscriptions.py index 3beba804..35b19f29 100644 --- a/handlers/keys/subscriptions.py +++ b/handlers/keys/subscriptions.py @@ -10,6 +10,16 @@ from config import DATABASE_URL, PROJECT_NAME, SUB_MESSAGE, SUPERNODE, TRANSITIO from database import get_servers_from_db 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): try: @@ -33,44 +43,42 @@ async def fetch_url_content(url, tg_id): logger.error(f"Ошибка при получении {url} для tg_id: {tg_id}: {e}") return [] - async def combine_unique_lines(urls, tg_id, query_string): if SUPERNODE: logger.info(f"Режим SUPERNODE активен. Возвращаем первую ссылку для tg_id: {tg_id}") - urls_with_query = [f"{urls[0]}?{query_string}"] if urls else [] - return await fetch_url_content(urls_with_query[0], tg_id) if urls_with_query else [] + if not urls: + return [] + url_with_query = f"{urls[0]}?{query_string}" if query_string else urls[0] + return await fetch_url_content(url_with_query, tg_id) - all_lines = [] logger.info( f"Начинаем объединение подписок для tg_id: {tg_id}, запрос: {query_string}" ) - urls_with_query = [f"{url}?{query_string}" for url in urls] + urls_with_query = [f"{url}?{query_string}" if query_string else url for url in urls] logger.info(f"Составлены URL-адреса: {urls_with_query}") - for url in urls_with_query: - lines = await fetch_url_content(url, tg_id) - all_lines.extend(lines) + tasks = [fetch_url_content(url, tg_id) for url in urls_with_query] + results = await asyncio.gather(*tasks) + + all_lines = set() + for lines in results: + all_lines.update(filter(None, lines)) - all_lines = list(set(filter(None, all_lines))) logger.info( f"Объединено {len(all_lines)} строк после фильтрации и удаления дубликатов для tg_id: {tg_id}" ) - return all_lines - + return list(all_lines) transition_date = datetime.strptime(TRANSITION_DATE_STR, "%Y-%m-%d %H:%M:%S") - transition_timestamp_ms = int(transition_date.timestamp() * 1000) - transition_timestamp_ms_adjusted = transition_timestamp_ms - (3 * 60 * 60 * 1000) logger.info( f"Время перехода (с поправкой на часовой пояс): {transition_timestamp_ms_adjusted}" ) - async def handle_old_subscription(request): email = request.match_info.get("email") @@ -83,8 +91,10 @@ async def handle_old_subscription(request): logger.info(f"Обработка запроса для старого клиента с email: {email}") - conn = await asyncpg.connect(DATABASE_URL) - try: + # Инициализируем пул соединений + await init_db_pool() + + async with db_pool.acquire() as conn: key_info = await conn.fetchrow( "SELECT created_at, server_id FROM keys WHERE email = $1", email ) @@ -121,35 +131,31 @@ async def handle_old_subscription(request): status=400, ) - servers = await get_servers_from_db() - cluster_servers = servers.get(cluster_name, []) - logger.info(f"Сервера в кластере: {cluster_servers}") + servers = await get_servers_from_db() + cluster_servers = servers.get(cluster_name, []) + logger.info(f"Сервера в кластере: {cluster_servers}") - urls = [] - for server in cluster_servers: - server_subscription_url = f"{server['subscription_url']}/{email}" - urls.append(server_subscription_url) + urls = [ + f"{server['subscription_url']}/{email}" for server in cluster_servers + ] - combined_subscriptions = await combine_unique_lines(urls, email, "") + combined_subscriptions = await combine_unique_lines(urls, email, "") - base64_encoded = base64.b64encode( - "\n".join(combined_subscriptions).encode("utf-8") - ).decode("utf-8") + 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"), - } + 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) - - finally: - await conn.close() + logger.info(f"Возвращаем объединенные подписки для email: {email}") + return web.Response(text=base64_encoded, headers=headers) async def handle_new_subscription(request): @@ -165,9 +171,10 @@ async def handle_new_subscription(request): logger.info(f"Обработка запроса для нового клиента: email={email}, tg_id={tg_id}") - conn = await asyncpg.connect(DATABASE_URL) - try: + # Инициализируем пул соединений + await init_db_pool() + async with db_pool.acquire() as conn: client_data = await conn.fetchrow( "SELECT tg_id, server_id FROM keys WHERE email = $1", email ) @@ -189,16 +196,12 @@ async def handle_new_subscription(request): status=403, ) - finally: - await conn.close() - servers = await get_servers_from_db() cluster_servers = servers.get(cluster_name, []) - urls = [] - for server in cluster_servers: - server_subscription_url = f"{server['subscription_url']}/{email}" - urls.append(server_subscription_url) + urls = [ + f"{server['subscription_url']}/{email}" for server in cluster_servers + ] query_string = request.query_string logger.info(f"Извлечен query string: {query_string}") From 17a0fa0b214db9fc1dcb74654d68afe420d579df Mon Sep 17 00:00:00 2001 From: Vladless Date: Wed, 15 Jan 2025 18:46:33 +0300 Subject: [PATCH 2/7] SuperNode/SyncTimeout/pytz --- handlers/keys/key_utils.py | 1 + 1 file changed, 1 insertion(+) diff --git a/handlers/keys/key_utils.py b/handlers/keys/key_utils.py index 3b330d64..47f9401c 100644 --- a/handlers/keys/key_utils.py +++ b/handlers/keys/key_utils.py @@ -212,6 +212,7 @@ async def update_key_on_cluster(tg_id, client_id, email, expiry_time, cluster_id enable=True, flow="xtls-rprx-vision", inbound_id=int(inbound_id), + sub_id=email ) ) From 1d433d2cfdd06b204180ea986163460715546c10 Mon Sep 17 00:00:00 2001 From: Vladless Date: Wed, 15 Jan 2025 20:16:52 +0300 Subject: [PATCH 3/7] fix trial_key for SuperNode --- handlers/keys/trial_key.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/handlers/keys/trial_key.py b/handlers/keys/trial_key.py index 9e803cf8..97821fb0 100644 --- a/handlers/keys/trial_key.py +++ b/handlers/keys/trial_key.py @@ -43,9 +43,11 @@ async def create_trial_key(tg_id: int, session: Any): servers_in_cluster = clusters[least_loaded_cluster] tasks = [] - for index, server_info in enumerate(servers_in_cluster): + for server_info in servers_in_cluster: + server_name = server_info.get("server_name", "unknown") + if SUPERNODE: - email = f"{base_email}{index}" + email = f"{base_email}_{server_name.lower()}" else: email = base_email From 981dc0db01c3edf61b3c884f90574b71fddb0eb0 Mon Sep 17 00:00:00 2001 From: Vladless Date: Thu, 16 Jan 2025 01:50:34 +0300 Subject: [PATCH 4/7] pytz for notifications --- handlers/notifications.py | 65 ++++++++++++++++++++++++++------------- 1 file changed, 43 insertions(+), 22 deletions(-) diff --git a/handlers/notifications.py b/handlers/notifications.py index fb79f513..a490edf1 100644 --- a/handlers/notifications.py +++ b/handlers/notifications.py @@ -1,5 +1,6 @@ import asyncio from datetime import datetime, timedelta +import pytz import asyncpg from aiogram import Bot, Router, types @@ -74,15 +75,15 @@ async def notify_expiring_keys(bot: Bot): logger.info("Начало обработки уведомлений.") await notify_inactive_trial_users(bot, conn) - await asyncio.sleep(1) + await asyncio.sleep(0,5) await check_online_users() - await asyncio.sleep(1) + await asyncio.sleep(0,5) await notify_10h_keys(bot, conn, current_time, threshold_time_10h) - await asyncio.sleep(1) + await asyncio.sleep(0,5) await notify_24h_keys(bot, conn, current_time, threshold_time_24h) - await asyncio.sleep(1) + await asyncio.sleep(0,5) await handle_expired_keys(bot, conn, current_time) - await asyncio.sleep(1) + await asyncio.sleep(0,5) except Exception as e: logger.error(f"Ошибка при отправке уведомлений: {e}") @@ -137,8 +138,10 @@ async def process_10h_record(record, bot, conn): email = record["email"] expiry_time = record["expiry_time"] - expiry_date = datetime.utcfromtimestamp(expiry_time / 1000) - current_date = datetime.utcnow() + moscow_tz = pytz.timezone("Europe/Moscow") + + expiry_date = datetime.fromtimestamp(expiry_time / 1000, tz=moscow_tz) + current_date = datetime.now(moscow_tz) time_left = expiry_date - current_date days_left_message = ( @@ -211,8 +214,10 @@ async def process_24h_record(record, bot, conn): email = record["email"] expiry_time = record["expiry_time"] - expiry_date = datetime.utcfromtimestamp(expiry_time / 1000) - current_date = datetime.utcnow() + moscow_tz = pytz.timezone("Europe/Moscow") + + expiry_date = datetime.fromtimestamp(expiry_time / 1000, tz=moscow_tz) + current_date = datetime.now(moscow_tz) time_left = expiry_date - current_date days_left_message = ( @@ -368,13 +373,15 @@ async def process_key(record, bot, conn): email = record["email"] balance = await get_balance(tg_id) expiry_time = record["expiry_time"] - expiry_date = datetime.utcfromtimestamp(expiry_time / 1000) - current_date = datetime.utcnow() + + moscow_tz = pytz.timezone("Europe/Moscow") + expiry_date = datetime.fromtimestamp(expiry_time / 1000, tz=moscow_tz) + current_date = datetime.now(moscow_tz) time_left = expiry_date - current_date logger.info( - f"Время истечения ключа: {expiry_time} (UTC: {expiry_date}), " - f"Текущее время (UTC): {current_date}, " + f"Время истечения ключа: {expiry_time} (МСК: {expiry_date}), " + f"Текущее время (МСК): {current_date}, " f"Оставшееся время: {time_left}" ) @@ -391,7 +398,8 @@ async def process_key(record, bot, conn): try: if AUTO_RENEW_KEYS and balance >= RENEWAL_PLANS["1"]["price"]: await update_balance(tg_id, -RENEWAL_PLANS["1"]["price"]) - new_expiry_time = int((datetime.utcnow() + timedelta(days=30)).timestamp() * 1000) + + new_expiry_time = int((datetime.now(moscow_tz) + timedelta(days=30)).timestamp() * 1000) await update_key_expiry(client_id, new_expiry_time) servers = await get_servers_from_db() @@ -410,23 +418,35 @@ async def process_key(record, bot, conn): ) logger.info(f"Флаги notified сброшены для клиента {client_id}.") - await bot.send_message(tg_id, text=KEY_RENEWED, reply_markup=keyboard) - logger.info(f"Уведомление об успешном продлении отправлено клиенту {tg_id}.") + try: + await bot.send_message(tg_id, text=KEY_RENEWED, reply_markup=keyboard) + logger.info(f"Уведомление об успешном продлении отправлено клиенту {tg_id}.") + except Exception as e: + logger.error(f"Не удалось отправить уведомление о продлении клиенту {tg_id}: {e}") else: message_expired = "Ваша подписка истекла. Пополните баланс для продления." - await bot.send_message(tg_id, text=message_expired, reply_markup=keyboard) - logger.info(f"Уведомление об истечении подписки отправлено пользователю {tg_id}.") + try: + await bot.send_message(tg_id, text=message_expired, reply_markup=keyboard) + logger.info(f"Уведомление об истечении подписки отправлено пользователю {tg_id}.") + except Exception as e: + logger.error(f"Не удалось отправить уведомление об истечении клиенту {tg_id}: {e}") if AUTO_DELETE_EXPIRED_KEYS: servers = await get_servers_from_db() for cluster_id in servers: - await delete_key_from_cluster(cluster_id, email, client_id) - logger.info(f"Клиент {client_id} удален из кластера {cluster_id}.") + try: + await delete_key_from_cluster(cluster_id, email, client_id) + logger.info(f"Клиент {client_id} удален из кластера {cluster_id}.") + except Exception as e: + logger.error(f"Ошибка при удалении клиента {client_id} из кластера {cluster_id}: {e}") - await delete_key(client_id) - logger.info(f"Ключ {client_id} удалён из базы данных.") + try: + await delete_key(client_id) + logger.info(f"Ключ {client_id} удалён из базы данных.") + except Exception as e: + logger.error(f"Ошибка при удалении ключа {client_id} из базы данных: {e}") else: logger.info(f"Ключ {client_id} НЕ был удалён (AUTO_DELETE_EXPIRED_KEYS=False).") @@ -434,6 +454,7 @@ async def process_key(record, bot, conn): logger.error(f"Ошибка при обработке ключа для клиента {tg_id}: {e}") + async def check_online_users(): servers = await get_servers_from_db() From b6184d4c90deed7162f618bdebefa44f3e0ee348 Mon Sep 17 00:00:00 2001 From: Vladless Date: Thu, 16 Jan 2025 03:03:37 +0300 Subject: [PATCH 5/7] fix cluster_name in admin_panel --- handlers/admin/admin_user_editor.py | 18 ++++++------------ 1 file changed, 6 insertions(+), 12 deletions(-) diff --git a/handlers/admin/admin_user_editor.py b/handlers/admin/admin_user_editor.py index ca1ebfbc..db2be7ba 100644 --- a/handlers/admin/admin_user_editor.py +++ b/handlers/admin/admin_user_editor.py @@ -318,6 +318,7 @@ async def handle_new_balance_input( async def get_key_details(email, session): + record = await session.fetchrow( """ SELECT k.key, k.expiry_time, k.server_id, c.tg_id, c.balance @@ -331,18 +332,11 @@ async def get_key_details(email, session): if not record: return None - servers = await get_servers_from_db() + cluster_name = record["server_id"] - cluster_name = "Неизвестный кластер" - for cluster_name, cluster_servers in servers.items(): - if any( - server["inbound_id"] == record["server_id"] for server in cluster_servers - ): - cluster_name = cluster_name - break - - expiry_date = datetime.utcfromtimestamp(record["expiry_time"] / 1000) - current_date = datetime.utcnow() + moscow_tz = pytz.timezone("Europe/Moscow") + expiry_date = datetime.fromtimestamp(record["expiry_time"] / 1000, tz=moscow_tz) + current_date = datetime.now(moscow_tz) time_left = expiry_date - current_date if time_left.total_seconds() <= 0: @@ -355,7 +349,7 @@ async def get_key_details(email, session): return { "key": record["key"], - "expiry_date": expiry_date.strftime("%d %B %Y года"), + "expiry_date": expiry_date.strftime("%d %B %Y года %H:%M"), "days_left_message": days_left_message, "server_name": cluster_name, "balance": record["balance"], From 9db8fdf79b516d72e175f5914850bb11da4e767a Mon Sep 17 00:00:00 2001 From: Vladless Date: Thu, 16 Jan 2025 04:03:53 +0300 Subject: [PATCH 6/7] expired_time --- handlers/notifications.py | 38 ++++++++++++++++++++++++++------------ 1 file changed, 26 insertions(+), 12 deletions(-) diff --git a/handlers/notifications.py b/handlers/notifications.py index a490edf1..d5862e9d 100644 --- a/handlers/notifications.py +++ b/handlers/notifications.py @@ -18,6 +18,7 @@ from config import ( RENEWAL_PLANS, TOTAL_GB, TRIAL_TIME, + EXPIRED_KEYS_CHECK_INTERVAL ) from database import ( add_blocked_user, @@ -57,6 +58,24 @@ async def check_users_and_update_blocked(bot: Bot): await conn.close() +async def periodic_expired_keys_check(bot: Bot): + """Периодическая проверка истекших ключей с кастомным интервалом.""" + while True: + conn = None + try: + conn = await asyncpg.connect(DATABASE_URL) + current_time = int(datetime.utcnow().timestamp() * 1000) + await handle_expired_keys(bot, conn, current_time) + logger.info("✅ Проверка истекших ключей выполнена.") + except Exception as e: + logger.error(f"❌ Ошибка в periodic_expired_keys_check: {e}") + finally: + if conn: + await conn.close() + + await asyncio.sleep(EXPIRED_KEYS_CHECK_INTERVAL) + + async def notify_expiring_keys(bot: Bot): conn = None @@ -65,25 +84,19 @@ async def notify_expiring_keys(bot: Bot): logger.info("Подключение к базе данных успешно.") current_time = int(datetime.utcnow().timestamp() * 1000) - threshold_time_10h = int( - (datetime.utcnow() + timedelta(hours=10)).timestamp() * 1000 - ) - threshold_time_24h = int( - (datetime.utcnow() + timedelta(days=1)).timestamp() * 1000 - ) + threshold_time_10h = int((datetime.utcnow() + timedelta(hours=10)).timestamp() * 1000) + threshold_time_24h = int((datetime.utcnow() + timedelta(days=1)).timestamp() * 1000) logger.info("Начало обработки уведомлений.") await notify_inactive_trial_users(bot, conn) - await asyncio.sleep(0,5) + await asyncio.sleep(0.5) await check_online_users() - await asyncio.sleep(0,5) + await asyncio.sleep(0.5) await notify_10h_keys(bot, conn, current_time, threshold_time_10h) - await asyncio.sleep(0,5) + await asyncio.sleep(0.5) await notify_24h_keys(bot, conn, current_time, threshold_time_24h) - await asyncio.sleep(0,5) - await handle_expired_keys(bot, conn, current_time) - await asyncio.sleep(0,5) + await asyncio.sleep(0.5) except Exception as e: logger.error(f"Ошибка при отправке уведомлений: {e}") @@ -93,6 +106,7 @@ async def notify_expiring_keys(bot: Bot): logger.info("Соединение с базой данных закрыто.") + async def is_bot_blocked(bot: Bot, chat_id: int) -> bool: if DEV_MODE: return False From c158c1339f65d765a445776b1179778115ad5d44 Mon Sep 17 00:00:00 2001 From: Vladless Date: Thu, 16 Jan 2025 04:15:17 +0300 Subject: [PATCH 7/7] backup_update --- backup.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/backup.py b/backup.py index c6811fa7..e44b45f1 100644 --- a/backup.py +++ b/backup.py @@ -4,7 +4,7 @@ from datetime import datetime from aiogram.types import BufferedInputFile -from config import ADMIN_ID, BACK_DIR, DB_NAME, DB_PASSWORD, DB_USER +from config import ADMIN_ID, BACK_DIR, DB_NAME, DB_PASSWORD, DB_USER, PG_HOST, PG_PORT from logger import logger @@ -21,7 +21,8 @@ async def backup_database(): def _create_database_backup(): USER = DB_USER - HOST = "localhost" + HOST = PG_HOST + PORT = PG_PORT BACKUP_DIR = BACK_DIR DATE = datetime.now().strftime("%Y-%m-%d-%H%M%S") BACKUP_FILE = f"{BACKUP_DIR}/{DB_NAME}-backup-{DATE}.sql" @@ -36,6 +37,8 @@ def _create_database_backup(): USER, "-h", HOST, + "-p", + PORT, "-F", "c", "-f",