import json import uuid from typing import Tuple, Optional, Dict, Any import aiohttp from bot.config import ( XUI_FULL_ADDRESS, INBOUND_VLESS_ID, INBOUND_SS_ID, SUB_BASE_URL ) from bot.services.xui import XUIService from bot.utils.logging import logger class ClientService: """Сервис для работы с клиентами VPN.""" def __init__(self, xui_service: XUIService): self.xui_service = xui_service def generate_client_credentials(self, telegram_id: str) -> Tuple[str, str, str, str]: """Генерация учетных данных для клиента.""" vless_email = f"{telegram_id}_vl_ssl" ss_email = f"{telegram_id}_ss" vless_uuid = str(uuid.uuid4()) ss_password = str(uuid.uuid4()) return vless_email, ss_email, vless_uuid, ss_password def get_subscription_link(self, telegram_id: str) -> str: """Получение ссылки на подписку.""" return f"{SUB_BASE_URL}{telegram_id}?name={telegram_id}" async def create_shadowsocks_client(self, telegram_id: str, password: str) -> bool: """Создание Shadowsocks клиента через API.""" try: self.xui_service.login() except Exception as e: logger.error(f"Ошибка авторизации в XUI: {e}") return False ss_email = f"{telegram_id}_ss" settings_str = json.dumps({ "clients": [{ "method": "chacha20-ietf-poly1305", "password": password, "email": ss_email, "totalGB": 0, "expiryTime": 0, "enable": True, "tgId": telegram_id, "subId": telegram_id, "reset": 0 }] }, separators=(',', ':')) payload = { "id": INBOUND_SS_ID, "settings": settings_str } api_url = f"{XUI_FULL_ADDRESS}/xui/API/inbounds/addClient" cookies = self.xui_service.get_cookies() async with aiohttp.ClientSession() as session: async def do_request(): async with session.post( api_url, json=payload, cookies=cookies, headers={ "Content-Type": "application/json", "User-Agent": "Mozilla/5.0" }, ssl=True ) as resp: text = await resp.text() logger.debug(f"Response status: {resp.status}") logger.debug(f"Response text: {text}") return resp.status, text status, text = await do_request() # Если сессия истекла, пробуем повторно залогиниться if status == 401 or (status == 200 and "login" in text.lower()): logger.info("Сессия XUI истекла, повторный вход...") try: self.xui_service.reset_session() self.xui_service.login() cookies = self.xui_service.get_cookies() except Exception as e: logger.error(f"Ошибка повторной авторизации в XUI: {e}") return False status, text = await do_request() if status == 200: if "successfully" in text.lower() or "success" in text.lower(): logger.info(f"Shadowsocks клиент успешно создан: {ss_email}") return True elif not text.strip(): logger.warning(f"Пустой ответ при статусе 200. Проверяем наличие клиента: {ss_email}") return await self.verify_client_created(telegram_id, ss_email) else: logger.error(f"Неожиданный ответ при создании SS клиента: {text}") return False else: logger.error(f"Ошибка создания Shadowsocks клиента: {status} {text}") return False async def verify_client_created(self, telegram_id: str, ss_email: str) -> bool: """Проверка, что клиент существует в inbound SS.""" try: inbound_info = self.xui_service.get_inbound(INBOUND_SS_ID) if not inbound_info: logger.error(f"Inbound {INBOUND_SS_ID} не найден.") return False settings = json.loads(inbound_info.get('settings', '{}')) clients = settings.get('clients', []) for client in clients: if client.get('email') == ss_email: logger.info(f"Клиент {ss_email} найден.") return True logger.warning(f"Клиент {ss_email} не найден.") return False except Exception as e: logger.error(f"Ошибка при проверке клиента: {e}") return False async def create_client_profile(self, telegram_id: str) -> Tuple[bool, str]: """Создание полного профиля клиента (VLESS + Shadowsocks) с проверкой существующего.""" try: # Сначала проверяем, есть ли уже клиенты 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 клиента 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, "Произошла ошибка при создании профиля" # 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" ss_email = f"{telegram_id}_ss" 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 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)