houseassassin/Solo_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

servers in database

Browse Source
This commit is contained in:
Vladless
2024-12-01 23:24:53 +03:00
parent d9064ff016
commit 0e5961bdc5
8 changed files with 149 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
database.py
+47 -1
View File
@@ -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
handlers/keys/key_management.py
+1 -2
View File
@@ -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")
handlers/keys/key_utils.py
+34 -25
View File
@@ -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(
handlers/keys/keys.py
+8 -13
View File
@@ -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,
handlers/keys/subscriptions.py
+12 -9
View File
@@ -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
handlers/keys/trial_key.py
+31 -23
View File
@@ -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,
handlers/notifications.py
+15 -11
View File
@@ -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}")
handlers/payments/stars_pay.py
+1 -1
View File
@@ -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}")
logger.error(f"Ошибка конвертации user_id или amount: {e}")