From c2f7d2a88ea816c62dc45b299e1fbcd14aa5d3f4 Mon Sep 17 00:00:00 2001 From: UdoChudo Date: Sun, 31 Aug 2025 23:56:35 +0500 Subject: [PATCH] feat: /info command now show full stats and config in vless and ss inbounds feat: /create command now check profile existed in vless and ss inbounds and if yes just send sub link --- bot/handlers/client.py | 43 +++++++++++++++---- bot/services/client.py | 78 +++++++++++++++++++++++++++++++--- bot/services/xui.py | 48 ++++++++++++++++++++- bot/utils/XuiServiceWrapper.py | 50 ++++++++++++++++++++++ 4 files changed, 204 insertions(+), 15 deletions(-) create mode 100644 bot/utils/XuiServiceWrapper.py diff --git a/bot/handlers/client.py b/bot/handlers/client.py index 0897e9e..cffdcbc 100644 --- a/bot/handlers/client.py +++ b/bot/handlers/client.py @@ -40,7 +40,7 @@ class ClientHandlers: await message.answer(f"❌ Произошла ошибка при создании профиля. Попробуйте позже.\n {e}") async def cmd_info(self, message: types.Message): - """Обработчик команды /info.""" + """Обработчик команды /info с выводом конфига и статистики.""" args = (message.text or "").strip().split(maxsplit=1) if len(args) < 2: await message.answer( @@ -52,23 +52,50 @@ class ClientHandlers: telegram_id = args[1].lstrip("@").strip() try: - vless_client, ss_client = self.client_service.get_client_info(telegram_id) + (vless_client, vless_stats), (ss_client, ss_stats) = \ + self.client_service.get_client_info_with_stats(telegram_id) - def format_info(client, name): + def format_info(client, stats, name): if not client: return f"❌ Клиент {name} не найден.\n" + + # Конфиг клиента json_info = json.dumps(client, ensure_ascii=False, indent=2) - return f"🔹 {name}:\n
{json_info}
" + text = f"🔹 {name}:\n
{json_info}
" + + # Статистика + if stats: + up = stats.get("up", 0) + down = stats.get("down", 0) + total = up + down + + def human_size(num): + for unit in ["B", "KB", "MB", "GB", "TB"]: + if num < 1024: + return f"{num:.2f} {unit}" + num /= 1024 + return f"{num:.2f} PB" + + text += ( + f"\n📊 Трафик: ↑ {human_size(up)} / ↓ {human_size(down)}" + f" (Σ {human_size(total)})" + ) + else: + text += "\n📊 Статистика не найдена." + + return text response = ( - format_info(vless_client, "VLESS") + - "\n\n" + - format_info(ss_client, "Shadowsocks") + format_info(vless_client, vless_stats, "VLESS") + + "\n\n" + + format_info(ss_client, ss_stats, "Shadowsocks") ) + await message.answer(response, parse_mode=ParseMode.HTML) except BadLogin: await message.answer("❌ Ошибка входа в панель XUI.") except Exception as e: logger.error(f"Ошибка получения информации: {e}") - await message.answer("❌ Ошибка при получении информации. Проверь лог.") \ No newline at end of file + await message.answer("❌ Ошибка при получении информации. Проверь лог.") + diff --git a/bot/services/client.py b/bot/services/client.py index d514f21..17ea354 100644 --- a/bot/services/client.py +++ b/bot/services/client.py @@ -132,10 +132,20 @@ class ClientService: return False async def create_client_profile(self, telegram_id: str) -> Tuple[bool, str]: - """Создание полного профиля клиента (VLESS + Shadowsocks).""" + """Создание полного профиля клиента (VLESS + Shadowsocks) с проверкой существующего.""" try: - vless_email, ss_email, vless_uuid, ss_password = self.generate_client_credentials(telegram_id) + # Сначала проверяем, есть ли уже клиенты + vless_client, ss_client = self.get_client_info(telegram_id) + if vless_client or ss_client: + subscription_link = self.get_subscription_link(telegram_id) + logger.info(f"Клиент для {telegram_id} уже существует.") + return True, ( + f"ℹ Клиент для {telegram_id} уже существует.\n" + f"🔗 Подписочная ссылка:\n{subscription_link}" + ) + # Генерируем новые креды + vless_email, ss_email, vless_uuid, ss_password = self.generate_client_credentials(telegram_id) logger.info(f"Создаём клиентов для telegram_id={telegram_id}") # Создаём VLESS клиента @@ -151,13 +161,13 @@ class ClientService: # Создаём Shadowsocks клиента ss_success = await self.create_shadowsocks_client(telegram_id, ss_password) - if not ss_success: return False, "Ошибка при создании Shadowsocks клиента" + # Подписочная ссылка subscription_link = self.get_subscription_link(telegram_id) success_message = ( - f"✅ Профиль для {telegram_id} успешно создан!\n" + f"✅ Профиль для {telegram_id} успешно создан!\n" f"🔗 Подписочная ссылка:\n{subscription_link}" ) @@ -167,6 +177,43 @@ class ClientService: logger.error(f"Ошибка при создании профиля: {e}") return False, "Произошла ошибка при создании профиля" + + # async def create_client_profile(self, telegram_id: str) -> Tuple[bool, str]: + # """Создание полного профиля клиента (VLESS + Shadowsocks).""" + # try: + # vless_email, ss_email, vless_uuid, ss_password = self.generate_client_credentials(telegram_id) + # + # logger.info(f"Создаём клиентов для telegram_id={telegram_id}") + # + # # Создаём VLESS клиента + # vless_success = self.xui_service.add_vless_client( + # inbound_id=INBOUND_VLESS_ID, + # email=vless_email, + # uuid=vless_uuid, + # telegram_id=telegram_id + # ) + # + # if not vless_success: + # return False, "Ошибка при создании VLESS клиента" + # + # # Создаём Shadowsocks клиента + # ss_success = await self.create_shadowsocks_client(telegram_id, ss_password) + # + # if not ss_success: + # return False, "Ошибка при создании Shadowsocks клиента" + # + # subscription_link = self.get_subscription_link(telegram_id) + # success_message = ( + # f"✅ Профиль для {telegram_id} успешно создан!\n" + # f"🔗 Подписочная ссылка:\n{subscription_link}" + # ) + # + # return True, success_message + # + # except Exception as e: + # logger.error(f"Ошибка при создании профиля: {e}") + # return False, "Произошла ошибка при создании профиля" + def get_client_info(self, telegram_id: str) -> Tuple[Optional[Dict[str, Any]], Optional[Dict[str, Any]]]: """Получение информации о клиентах.""" vless_email = f"{telegram_id}_vl_ssl" @@ -175,4 +222,25 @@ class ClientService: vless_client = self.xui_service.get_client(INBOUND_VLESS_ID, vless_email) ss_client = self.xui_service.get_client(INBOUND_SS_ID, ss_email) - return vless_client, ss_client \ No newline at end of file + return vless_client, ss_client + + def get_client_info_with_stats(self, telegram_id: str): + """Возвращает клиентов и их статистику по Telegram ID.""" + vless_client, ss_client = self.get_client_info(telegram_id) + + vless_stats = None + ss_stats = None + + if vless_client: + vless_stats = self.xui_service.get_client_stats( + inbound_id=INBOUND_VLESS_ID, + email=vless_client["email"] + ) + + if ss_client: + ss_stats = self.xui_service.get_client_stats( + inbound_id=INBOUND_SS_ID, + email=ss_client["email"] + ) + + return (vless_client, vless_stats), (ss_client, ss_stats) \ No newline at end of file diff --git a/bot/services/xui.py b/bot/services/xui.py index ac3a578..ce95ff8 100644 --- a/bot/services/xui.py +++ b/bot/services/xui.py @@ -40,14 +40,58 @@ class XUIService: raise def get_client(self, inbound_id: int, email: str) -> Optional[Dict[str, Any]]: - """Получение информации о клиенте.""" + """Получение информации о клиенте (работает и для SS, и для VLESS).""" try: self.login() - return self.xui.get_client(inbound_id=inbound_id, email=email) + inbounds = self.xui.get_inbounds() + if "obj" not in inbounds: + logger.error("Некорректный ответ от XUI при запросе inbounds") + return None + + for inbound in inbounds["obj"]: + if inbound["id"] != inbound_id: + continue + + settings = json.loads(inbound["settings"]) + for client in settings.get("clients", []): + if client.get("email") == email: + # У SS-клиентов id нет — подставляем email + if "id" not in client: + client["id"] = client["email"] + return client + + logger.warning(f"Клиент {email} не найден в inbound {inbound_id}") + return None + except Exception as e: logger.error(f"Ошибка получения клиента {email}: {e}") return None + def get_client_stats(self, inbound_id: int, email: str) -> Optional[Dict[str, Any]]: + """Получение статистики клиента по inbound и email.""" + try: + self.login() + inbounds = self.xui.get_inbounds() + if "obj" not in inbounds: + logger.error("Некорректный ответ от XUI при запросе inbounds") + return None + + for inbound in inbounds["obj"]: + if inbound["id"] != inbound_id: + continue + + for client_stat in inbound.get("clientStats", []): + if client_stat.get("email") == email: + return client_stat + + logger.warning(f"Статистика по клиенту {email} не найдена в inbound {inbound_id}") + return None + + except Exception as e: + logger.error(f"Ошибка получения статистики клиента {email}: {e}") + return None + + def add_vless_client( self, inbound_id: int, diff --git a/bot/utils/XuiServiceWrapper.py b/bot/utils/XuiServiceWrapper.py new file mode 100644 index 0000000..4c37d93 --- /dev/null +++ b/bot/utils/XuiServiceWrapper.py @@ -0,0 +1,50 @@ +import json +import logging +from typing import Optional, Union +from pyxui import XUI, errors + +logger = logging.getLogger(__name__) + +class XuiServiceWrapper: + def __init__(self, host: str, username: str, password: str): + self.xui = XUI(host, username, password) + + def login(self): + """Авторизация в XUI""" + return self.xui.login() + + def get_client(self, inbound_id: int, email: Optional[str] = None, uuid: Optional[str] = None) -> Union[dict, None]: + """Безопасное получение клиента (работает и для SS, и для VLESS/VMess)""" + try: + inbounds = self.xui.get_inbounds() + for inbound in inbounds["obj"]: + if inbound["id"] != inbound_id: + continue + + settings = json.loads(inbound["settings"]) + for client in settings["clients"]: + if (email and client.get("email") == email) or (uuid and client.get("id") == uuid): + return client + + return None + except errors.NotFound: + logger.warning(f"Клиент не найден: inbound_id={inbound_id}, email={email}, uuid={uuid}") + return None + except Exception as e: + logger.error(f"Ошибка при поиске клиента {email or uuid}: {e}") + return None + + def get_client_stats(self, inbound_id: int, email: str) -> Union[dict, None]: + """Получение статистики клиента""" + try: + inbounds = self.xui.get_inbounds() + for inbound in inbounds["obj"]: + if inbound["id"] != inbound_id: + continue + for client in inbound.get("clientStats", []): + if client.get("email") == email: + return client + return None + except Exception as e: + logger.error(f"Ошибка при получении статистики клиента {email}: {e}") + return None