diff --git a/database.py b/database.py index 43708f2f..be587d4c 100644 --- a/database.py +++ b/database.py @@ -118,6 +118,20 @@ async def init_db(): """ ) + await conn.execute( + """ + CREATE TABLE IF NOT EXISTS servers ( + id SERIAL PRIMARY KEY, + cluster_name TEXT NOT NULL, + server_name TEXT NOT NULL, + api_url TEXT NOT NULL, + subscription_url TEXT NOT NULL, + inbound_id TEXT NOT NULL, + UNIQUE(cluster_name, server_name) -- Уникальность по названию кластера и сервера + ) + """ + ) + await conn.close() @@ -1163,8 +1177,40 @@ async def check_notification_time(tg_id: int, notification_type: str, hours: int except Exception as e: logger.error(f"Ошибка при проверке времени уведомления для пользователя {tg_id}: {e}") - return False + return False finally: if conn is not None and session is None: await conn.close() + + +async def get_servers_from_db(): + conn = await asyncpg.connect(DATABASE_URL) + + # Запрос к базе данных для получения всех серверов + result = await conn.fetch( + """ + SELECT cluster_name, server_name, api_url, subscription_url, inbound_id + FROM servers + """ + ) + + await conn.close() + + # Преобразуем результат в удобный формат + servers = {} + for row in result: + cluster_name = row['cluster_name'] + if cluster_name not in servers: + servers[cluster_name] = [] + + servers[cluster_name].append( + { + 'server_name': row['server_name'], + 'api_url': row['api_url'], + 'subscription_url': row['subscription_url'], + 'inbound_id': row['inbound_id'], + } + ) + + return servers diff --git a/handlers/keys/key_management.py b/handlers/keys/key_management.py index 53b6c7a6..b2878b94 100644 --- a/handlers/keys/key_management.py +++ b/handlers/keys/key_management.py @@ -29,7 +29,7 @@ from database import ( use_trial, ) from handlers.keys.key_utils import create_key_on_cluster -from handlers.texts import KEY, KEY_TRIAL, NULL_BALANCE, key_message_success +from handlers.texts import KEY, NULL_BALANCE, key_message_success from handlers.utils import generate_random_email, get_least_loaded_cluster from logger import logger @@ -53,7 +53,6 @@ async def process_callback_create_key(callback_query: CallbackQuery, state: FSMC async def select_server(callback_query: CallbackQuery, state: FSMContext, session: Any): trial_status = await get_trial(callback_query.message.chat.id, session) if trial_status == 1: - # Если триал есть, бот предлагает подключить новое устройство builder = InlineKeyboardBuilder() builder.row( InlineKeyboardButton(text="✅ Да, подключить новое устройство", callback_data="confirm_create_new_key") diff --git a/handlers/keys/key_utils.py b/handlers/keys/key_utils.py index d6dab999..8d39c8b9 100644 --- a/handlers/keys/key_utils.py +++ b/handlers/keys/key_utils.py @@ -4,28 +4,32 @@ import asyncpg from py3xui import AsyncApi from client import add_client, delete_client, extend_client_key -from config import ADMIN_PASSWORD, ADMIN_USERNAME, CLUSTERS, DATABASE_URL, TOTAL_GB +from config import ADMIN_PASSWORD, ADMIN_USERNAME, DATABASE_URL, TOTAL_GB +from database import get_servers_from_db from logger import logger async def create_key_on_cluster(cluster_id, tg_id, client_id, email, expiry_timestamp): try: tasks = [] - cluster = CLUSTERS.get(cluster_id) + servers = await get_servers_from_db() + cluster = servers.get(cluster_id) if not cluster: raise ValueError(f"Кластер с ID {cluster_id} не найден.") - for server_info in cluster.values(): + for server_info in cluster: xui = AsyncApi( - server_info["API_URL"], + server_info["api_url"], username=ADMIN_USERNAME, password=ADMIN_PASSWORD, ) - inbound_id = server_info.get("INBOUND_ID") + inbound_id = server_info.get("inbound_id") if not inbound_id: - logger.warning(f"INBOUND_ID отсутствует для сервера {server_info.get('name', 'unknown')}. Пропуск.") + logger.warning( + f"INBOUND_ID отсутствует для сервера {server_info.get('server_name', 'unknown')}. Пропуск." + ) continue conn = await asyncpg.connect(DATABASE_URL) @@ -58,27 +62,26 @@ async def create_key_on_cluster(cluster_id, tg_id, client_id, email, expiry_time async def renew_key_in_cluster(cluster_id, email, client_id, new_expiry_time, total_gb): - """ - Функция для продления срока действия ключа на всех серверах в кластере - и обновления объема трафика. - """ try: - cluster = CLUSTERS.get(cluster_id) + servers = await get_servers_from_db() + cluster = servers.get(cluster_id) if not cluster: raise ValueError(f"Кластер с ID {cluster_id} не найден.") tasks = [] - for server_info in cluster.values(): + for server_info in cluster: xui = AsyncApi( - server_info["API_URL"], + server_info["api_url"], username=ADMIN_USERNAME, password=ADMIN_PASSWORD, ) - inbound_id = server_info.get("INBOUND_ID") + inbound_id = server_info.get("inbound_id") if not inbound_id: - logger.warning(f"INBOUND_ID отсутствует для сервера {server_info.get('name', 'unknown')}. Пропуск.") + logger.warning( + f"INBOUND_ID отсутствует для сервера {server_info.get('server_name', 'unknown')}. Пропуск." + ) continue tasks.append( @@ -109,22 +112,25 @@ async def delete_key_from_db(client_id, session): async def delete_key_from_cluster(cluster_id, email, client_id): """Удаление ключа с серверов в кластере""" try: - cluster = CLUSTERS.get(cluster_id) + servers = await get_servers_from_db() + cluster = servers.get(cluster_id) if not cluster: raise ValueError(f"Кластер с ID {cluster_id} не найден.") tasks = [] - for server_info in cluster.values(): + for server_info in cluster: xui = AsyncApi( - server_info["API_URL"], + server_info["api_url"], username=ADMIN_USERNAME, password=ADMIN_PASSWORD, ) - inbound_id = server_info.get("INBOUND_ID") + inbound_id = server_info.get("inbound_id") if not inbound_id: - logger.warning(f"INBOUND_ID отсутствует для сервера {server_info.get('name', 'unknown')}. Пропуск.") + logger.warning( + f"INBOUND_ID отсутствует для сервера {server_info.get('server_name', 'unknown')}. Пропуск." + ) continue tasks.append( @@ -145,22 +151,25 @@ async def delete_key_from_cluster(cluster_id, email, client_id): async def update_key_on_cluster(tg_id, client_id, email, expiry_time, cluster_id): try: - cluster = CLUSTERS.get(cluster_id) + servers = await get_servers_from_db() + cluster = servers.get(cluster_id) if not cluster: raise ValueError(f"Кластер с ID {cluster_id} не найден.") tasks = [] - for server_info in cluster.values(): + for server_info in cluster: xui = AsyncApi( - server_info["API_URL"], + server_info["api_url"], username=ADMIN_USERNAME, password=ADMIN_PASSWORD, ) - inbound_id = server_info.get("INBOUND_ID") + inbound_id = server_info.get("inbound_id") if not inbound_id: - logger.warning(f"INBOUND_ID отсутствует для сервера {server_info.get('name', 'unknown')}. Пропуск.") + logger.warning( + f"INBOUND_ID отсутствует для сервера {server_info.get('server_name', 'unknown')}. Пропуск." + ) continue tasks.append( diff --git a/handlers/keys/keys.py b/handlers/keys/keys.py index 5c507c8f..2fb994d2 100644 --- a/handlers/keys/keys.py +++ b/handlers/keys/keys.py @@ -8,17 +8,8 @@ from aiogram import F, Router, types from aiogram.types import BufferedInputFile, InlineKeyboardButton from aiogram.utils.keyboard import InlineKeyboardBuilder -from config import ( - CLUSTERS, - CONNECT_ANDROID, - CONNECT_IOS, - DOWNLOAD_ANDROID, - DOWNLOAD_IOS, - PUBLIC_LINK, - RENEWAL_PLANS, - TOTAL_GB, -) -from database import delete_key, get_balance, store_key, update_balance, update_key_expiry +from config import CONNECT_ANDROID, CONNECT_IOS, DOWNLOAD_ANDROID, DOWNLOAD_IOS, PUBLIC_LINK, RENEWAL_PLANS, TOTAL_GB +from database import delete_key, get_balance, get_servers_from_db, store_key, update_balance, update_key_expiry from handlers.keys.key_utils import ( delete_key_from_cluster, delete_key_from_db, @@ -362,10 +353,12 @@ async def process_callback_confirm_delete(callback_query: types.CallbackQuery, s reply_markup=keyboard, ) + servers = await get_servers_from_db() + async def delete_key_from_servers(): try: tasks = [] - for cluster_id, cluster in CLUSTERS.items(): + for cluster_id, cluster in servers.items(): tasks.append(delete_key_from_cluster(cluster_id, email, client_id)) await asyncio.gather(*tasks) @@ -438,9 +431,11 @@ async def process_callback_renew_plan(callback_query: types.CallbackQuery, sessi await callback_query.message.answer(response_message, reply_markup=builder.as_markup()) + servers = await get_servers_from_db() + async def renew_key_on_servers(): tasks = [] - for cluster_id in CLUSTERS: + for cluster_id in servers: task = asyncio.create_task( renew_key_in_cluster( cluster_id, diff --git a/handlers/keys/subscriptions.py b/handlers/keys/subscriptions.py index 70f2d6da..45b0d3fc 100644 --- a/handlers/keys/subscriptions.py +++ b/handlers/keys/subscriptions.py @@ -5,7 +5,8 @@ import aiohttp from aiohttp import web import asyncpg -from config import CLUSTERS, DATABASE_URL, TRANSITION_DATE_STR +from config import DATABASE_URL, TRANSITION_DATE_STR +from database import get_servers_from_db from logger import logger @@ -81,8 +82,6 @@ async def handle_old_subscription(request): created_at_datetime = datetime.utcfromtimestamp(created_at_ms / 1000) logger.info(f"Время создания клиента в формате datetime (UTC): {created_at_datetime}") - logger.info(f"Время перехода (с поправкой на часовой пояс): {transition_timestamp_ms_adjusted}") - if created_at_ms >= transition_timestamp_ms_adjusted: logger.info(f"Клиент с email {email} является новым.") return web.Response( @@ -90,10 +89,12 @@ async def handle_old_subscription(request): status=400, ) + servers = await get_servers_from_db() + urls = [] - for cluster in CLUSTERS.values(): - for server in cluster.values(): - server_subscription_url = f"{server['SUBSCRIPTION']}/{email}" + for cluster_name, cluster_servers in servers.items(): + for server in cluster_servers: + server_subscription_url = f"{server['subscription_url']}/{email}" urls.append(server_subscription_url) combined_subscriptions = await combine_unique_lines(urls, email, "") @@ -149,10 +150,12 @@ async def handle_new_subscription(request): finally: await conn.close() + servers = await get_servers_from_db() + urls = [] - for cluster in CLUSTERS.values(): - for server in cluster.values(): - server_subscription_url = f"{server['SUBSCRIPTION']}/{email}" + for cluster_name, cluster_servers in servers.items(): + for server in cluster_servers: + server_subscription_url = f"{server['subscription_url']}/{email}" urls.append(server_subscription_url) query_string = request.query_string diff --git a/handlers/keys/trial_key.py b/handlers/keys/trial_key.py index c8f81b3c..7117a435 100644 --- a/handlers/keys/trial_key.py +++ b/handlers/keys/trial_key.py @@ -1,3 +1,4 @@ +import asyncio from datetime import datetime, timedelta from typing import Any import uuid @@ -5,8 +6,8 @@ import uuid from py3xui import AsyncApi from client import add_client -from config import ADMIN_PASSWORD, ADMIN_USERNAME, CLUSTERS, PUBLIC_LINK, TOTAL_GB, TRIAL_TIME -from database import store_key, use_trial +from config import ADMIN_PASSWORD, ADMIN_USERNAME, PUBLIC_LINK, TOTAL_GB, TRIAL_TIME +from database import get_servers_from_db, store_key, use_trial from handlers.texts import INSTRUCTIONS from handlers.utils import generate_random_email, get_least_loaded_cluster @@ -21,31 +22,38 @@ async def create_trial_key(tg_id: int, session: Any): expiry_time = current_time + timedelta(days=TRIAL_TIME) expiry_timestamp = int(expiry_time.timestamp() * 1000) + clusters = await get_servers_from_db() + least_loaded_cluster = await get_least_loaded_cluster() - for server_id, server_info in CLUSTERS[least_loaded_cluster].items(): - xui = AsyncApi( - server_info["API_URL"], - username=ADMIN_USERNAME, - password=ADMIN_PASSWORD, + if least_loaded_cluster not in clusters: + raise ValueError(f"Кластер {least_loaded_cluster} не найден в базе данных.") + + servers_in_cluster = clusters[least_loaded_cluster] + + tasks = [] + + for server_info in servers_in_cluster: + tasks.append( + add_client( + AsyncApi( + server_info["api_url"], + username=ADMIN_USERNAME, + password=ADMIN_PASSWORD, + ), + client_id, + email, + tg_id, + limit_ip=1, + total_gb=TOTAL_GB, + expiry_time=expiry_timestamp, + enable=True, + flow="xtls-rprx-vision", + inbound_id=int(server_info["inbound_id"]), + ) ) - inbound_id = server_info.get("INBOUND_ID") - if not inbound_id: - raise ValueError(f"INBOUND_ID отсутствует для сервера {server_info.get('name', 'unknown')}") - - await add_client( - xui, - client_id, - email, - tg_id, - limit_ip=1, - total_gb=TOTAL_GB, - expiry_time=expiry_timestamp, - enable=True, - flow="xtls-rprx-vision", - inbound_id=int(inbound_id), - ) + await asyncio.gather(*tasks) await store_key( tg_id, diff --git a/handlers/notifications.py b/handlers/notifications.py index 80f58de4..7ac8ccf2 100644 --- a/handlers/notifications.py +++ b/handlers/notifications.py @@ -6,12 +6,13 @@ from aiogram.utils.keyboard import InlineKeyboardBuilder import asyncpg from py3xui import AsyncApi -from config import ADMIN_PASSWORD, ADMIN_USERNAME, CLUSTERS, DATABASE_URL, DEV_MODE, RENEWAL_PLANS, TOTAL_GB, TRIAL_TIME +from config import ADMIN_PASSWORD, ADMIN_USERNAME, DATABASE_URL, DEV_MODE, RENEWAL_PLANS, TOTAL_GB, TRIAL_TIME from database import ( add_notification, check_notification_time, delete_key, get_balance, + get_servers_from_db, update_balance, update_key_expiry, ) @@ -228,9 +229,7 @@ async def notify_inactive_trial_users(bot: Bot, conn: asyncpg.Connection): username = user.get('username', 'Пользователь') try: - can_notify = await check_notification_time( - tg_id, 'inactive_trial', hours=24, session=conn - ) + can_notify = await check_notification_time(tg_id, 'inactive_trial', hours=24, session=conn) if can_notify and not await is_bot_blocked(bot, tg_id): builder = InlineKeyboardBuilder() @@ -296,7 +295,9 @@ async def process_key(record, bot, conn): new_expiry_time = int((datetime.utcnow() + timedelta(days=30)).timestamp() * 1000) await update_key_expiry(client_id, new_expiry_time) - for cluster_id in CLUSTERS: + servers = await get_servers_from_db() + + for cluster_id in servers: await renew_key_in_cluster(cluster_id, email, client_id, new_expiry_time, TOTAL_GB) logger.info(f"Ключ для пользователя {tg_id} успешно продлен в кластере {cluster_id}.") @@ -324,9 +325,10 @@ async def process_key(record, bot, conn): except Exception as e: logger.error(f"Ошибка при отправке уведомления об истечении подписки пользователю {tg_id}: {e}") - for cluster_id in CLUSTERS: + servers = await get_servers_from_db() + + for cluster_id in servers: await delete_key_from_cluster(cluster_id, email, client_id) - # await xui.client.delete_depleted(-1) logger.info(f"Клиент {client_id} удален из кластера {cluster_id}.") await delete_key(client_id) @@ -337,14 +339,16 @@ async def process_key(record, bot, conn): async def check_online_users(): - for cluster_id, cluster in CLUSTERS.items(): - for server_id, server in cluster.items(): - xui = AsyncApi(server["API_URL"], username=ADMIN_USERNAME, password=ADMIN_PASSWORD) + servers = await get_servers_from_db() + + for cluster_id, cluster in servers.items(): + for server_id, server in enumerate(cluster): + xui = AsyncApi(server["api_url"], username=ADMIN_USERNAME, password=ADMIN_PASSWORD) await xui.login() try: online_users = len(await xui.client.online()) logger.info( - f"Сервер '{server['name']}' доступен, текущее количество активных пользователей: {online_users}." + f"Сервер '{server['server_name']}' доступен, текущее количество активных пользователей: {online_users}." ) except Exception as e: logger.error(f"Не удалось проверить пользователей на сервере {server_id}: {e}") diff --git a/handlers/payments/stars_pay.py b/handlers/payments/stars_pay.py index 0ed46279..3bb3136e 100644 --- a/handlers/payments/stars_pay.py +++ b/handlers/payments/stars_pay.py @@ -184,4 +184,4 @@ async def on_successful_payment( await update_balance(user_id, amount) await send_payment_success_notification(user_id, amount) except ValueError as e: - logger.error(f"Ошибка конвертации user_id или amount: {e}") \ No newline at end of file + logger.error(f"Ошибка конвертации user_id или amount: {e}")