UdoChudo
c2f7d2a88e
feat: /create command now check profile existed in vless and ss inbounds and if yes just send sub link
51 lines
2.0 KiB
Python
51 lines
2.0 KiB
Python
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
|