UdoChudo
c2f7d2a88e
feat: /create command now check profile existed in vless and ss inbounds and if yes just send sub link
154 lines
5.4 KiB
Python
154 lines
5.4 KiB
Python
import json
|
||
from typing import Optional, Dict, Any
|
||
|
||
from pyxui import XUI
|
||
from pyxui.errors import BadLogin
|
||
|
||
from bot.config import (
|
||
XUI_FULL_ADDRESS,
|
||
XUI_PANEL_NAME,
|
||
XUI_USERNAME,
|
||
XUI_PASSWORD
|
||
)
|
||
from bot.utils.logging import logger
|
||
|
||
|
||
class XUIService:
|
||
"""Сервис для работы с XUI панелью."""
|
||
|
||
def __init__(self):
|
||
self.xui = XUI(
|
||
full_address=XUI_FULL_ADDRESS,
|
||
panel=XUI_PANEL_NAME,
|
||
https=True
|
||
)
|
||
|
||
def login(self) -> None:
|
||
"""Авторизация в XUI панели."""
|
||
if self.xui.session_string:
|
||
logger.debug("Уже залогинен в XUI.")
|
||
return
|
||
|
||
try:
|
||
self.xui.login(XUI_USERNAME, XUI_PASSWORD)
|
||
logger.info("Успешный вход в XUI.")
|
||
except BadLogin:
|
||
logger.error("Неверный логин или пароль XUI.")
|
||
raise
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при входе в XUI: {e}")
|
||
raise
|
||
|
||
def get_client(self, inbound_id: int, email: str) -> Optional[Dict[str, Any]]:
|
||
"""Получение информации о клиенте (работает и для SS, и для VLESS)."""
|
||
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
|
||
|
||
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,
|
||
email: str,
|
||
uuid: str,
|
||
telegram_id: str,
|
||
enable: bool = True,
|
||
flow: str = "xtls-rprx-vision",
|
||
limit_ip: int = 0,
|
||
total_gb: int = 0,
|
||
expire_time: int = 0
|
||
) -> bool:
|
||
"""Добавление VLESS клиента."""
|
||
try:
|
||
self.login()
|
||
self.xui.add_client(
|
||
inbound_id=inbound_id,
|
||
email=email,
|
||
uuid=uuid,
|
||
enable=enable,
|
||
flow=flow,
|
||
limit_ip=limit_ip,
|
||
total_gb=total_gb,
|
||
expire_time=expire_time,
|
||
telegram_id=telegram_id,
|
||
subscription_id=telegram_id
|
||
)
|
||
logger.info(f"VLESS клиент создан: {email}")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"Ошибка создания VLESS клиента {email}: {e}")
|
||
return False
|
||
|
||
def get_inbound(self, inbound_id: int) -> Optional[Dict[str, Any]]:
|
||
"""Получение информации об inbound."""
|
||
try:
|
||
self.login()
|
||
return self.xui.get_inbound(inbound_id)
|
||
except Exception as e:
|
||
logger.error(f"Ошибка получения inbound {inbound_id}: {e}")
|
||
return None
|
||
|
||
def get_cookies(self) -> Dict[str, str]:
|
||
"""Получение cookies для API запросов."""
|
||
cookies = {}
|
||
|
||
if hasattr(self.xui, 'session') and self.xui.session and hasattr(self.xui.session, 'cookies'):
|
||
for cookie in self.xui.session.cookies:
|
||
if cookie.name == 'x-ui':
|
||
cookies['x-ui'] = cookie.value
|
||
break
|
||
|
||
if self.xui.session_string:
|
||
cookies['x-ui'] = self.xui.session_string
|
||
|
||
return cookies
|
||
|
||
def reset_session(self) -> None:
|
||
"""Сброс сессии для повторной авторизации."""
|
||
self.xui.session_string = None |