diff --git a/bot/__init__.py b/bot/__init__.py new file mode 100644 index 0000000..916e7be --- /dev/null +++ b/bot/__init__.py @@ -0,0 +1,3 @@ +""" +VPN Bot - модульный бот для управления VPN клиентами через XUI панель. +""" \ No newline at end of file diff --git a/bot/config.py b/bot/config.py new file mode 100644 index 0000000..3d6cd4d --- /dev/null +++ b/bot/config.py @@ -0,0 +1,12 @@ +import os + +XUI_FULL_ADDRESS = os.getenv("XUI_FULL_ADDRESS") +XUI_PANEL_NAME = os.getenv("XUI_PANEL_NAME") +XUI_USERNAME = os.getenv("XUI_USERNAME") +XUI_PASSWORD = os.getenv("XUI_PASSWORD") +INBOUND_VLESS_ID = int(os.getenv("INBOUND_VLESS_ID", 15)) +INBOUND_SS_ID = int(os.getenv("INBOUND_SS_ID", 2)) +SUBSCRIPTION_UUID = os.getenv("SUBSCRIPTION_UUID") +SUB_BASE_URL = f"https://udochudo.ru/{SUBSCRIPTION_UUID}/" +BOT_TOKEN = os.getenv("BOT_TOKEN") +ALLOWED_CHAT_IDS = set(map(int, os.getenv("ALLOWED_CHAT_IDS", "").split(","))) diff --git a/bot/core/__init__.py b/bot/core/__init__.py new file mode 100644 index 0000000..af2646d --- /dev/null +++ b/bot/core/__init__.py @@ -0,0 +1,3 @@ +""" +Ядро приложения - создание бота и диспетчера. +""" diff --git a/bot/core/bot.py b/bot/core/bot.py new file mode 100644 index 0000000..09ed783 --- /dev/null +++ b/bot/core/bot.py @@ -0,0 +1,23 @@ +from aiogram import Bot, Dispatcher +from aiogram.enums import ParseMode +from aiogram.client.default import DefaultBotProperties + +from bot.config import BOT_TOKEN +from bot.utils.logging import logger + + +def create_bot() -> Bot: + """Создание экземпляра бота.""" + bot = Bot( + token=BOT_TOKEN, + default=DefaultBotProperties(parse_mode=ParseMode.HTML) + ) + logger.info("Бот создан") + return bot + + +def create_dispatcher() -> Dispatcher: + """Создание диспетчера.""" + dp = Dispatcher() + logger.info("Диспетчер создан") + return dp \ No newline at end of file diff --git a/bot/filters/__init__.py b/bot/filters/__init__.py new file mode 100644 index 0000000..255b7d4 --- /dev/null +++ b/bot/filters/__init__.py @@ -0,0 +1,3 @@ +""" +Фильтры для обработки сообщений. +""" \ No newline at end of file diff --git a/bot/filters/access.py b/bot/filters/access.py new file mode 100644 index 0000000..960f5cb --- /dev/null +++ b/bot/filters/access.py @@ -0,0 +1,11 @@ +from aiogram.types import Message +from aiogram.filters import BaseFilter + +from bot.config import ALLOWED_CHAT_IDS + + +class AllowedUsersFilter(BaseFilter): + """Фильтр для проверки разрешенных пользователей.""" + + async def __call__(self, message: Message) -> bool: + return message.chat.id in ALLOWED_CHAT_IDS \ No newline at end of file diff --git a/bot/handlers/__init__.py b/bot/handlers/__init__.py new file mode 100644 index 0000000..349de3c --- /dev/null +++ b/bot/handlers/__init__.py @@ -0,0 +1,3 @@ +""" +Обработчики команд и сообщений. +""" \ No newline at end of file diff --git a/bot/handlers/base.py b/bot/handlers/base.py new file mode 100644 index 0000000..7548ae6 --- /dev/null +++ b/bot/handlers/base.py @@ -0,0 +1,29 @@ +from aiogram import types +from aiogram.filters import Command + +from bot.utils.logging import logger + + +async def cmd_start(message: types.Message): + """Обработчик команды /start.""" + await message.answer( + "Привет! Чтобы создать профиль, отправь команду:\n" + "/create Telegram id or username\n\n" + "Например:\n" + "/create udochudo", + reply_markup=None + ) + + +async def cmd_help(message: types.Message): + """Обработчик команды /help.""" + help_text = ( + "Доступные команды:\n\n" + "/start или /help - показать это сообщение\n" + "/create - создать профиль клиента\n" + "/info - получить информацию о клиенте\n\n" + "Примеры:\n" + "/create udochudo\n" + "/info udochudo" + ) + await message.answer(help_text) \ No newline at end of file diff --git a/bot/handlers/client.py b/bot/handlers/client.py new file mode 100644 index 0000000..1850bc0 --- /dev/null +++ b/bot/handlers/client.py @@ -0,0 +1,74 @@ +import json +from aiogram import types +from aiogram.enums import ParseMode +from pyxui.errors import BadLogin + +from bot.services.client import ClientService +from bot.utils.logging import logger + + +class ClientHandlers: + """Обработчики команд для работы с клиентами.""" + + def __init__(self, client_service: ClientService): + self.client_service = client_service + + async def cmd_create(self, message: types.Message): + """Обработчик команды /create.""" + args = (message.text or "").strip().split(maxsplit=1) + if len(args) < 2: + await message.answer( + "❌ Укажи Telegram ID или username после команды.\n" + "Пример:\n/create udochudo" + ) + return + + telegram_id = args[1].lstrip("@").strip() + + try: + success, result_message = await self.client_service.create_client_profile(telegram_id) + + if success: + await message.answer(result_message, parse_mode=ParseMode.HTML) + else: + await message.answer(f"❌ {result_message}") + + except BadLogin: + await message.answer("❌ Ошибка: неверный логин или пароль XUI.") + except Exception as e: + logger.error(f"Неожиданная ошибка при создании профиля: {e}") + await message.answer("❌ Произошла ошибка при создании профиля. Попробуйте позже.") + + async def cmd_info(self, message: types.Message): + """Обработчик команды /info.""" + args = (message.text or "").strip().split(maxsplit=1) + if len(args) < 2: + await message.answer( + "❌ Укажи Telegram ID или username после команды.\n" + "Пример:\n/info udochudo" + ) + return + + telegram_id = args[1].lstrip("@").strip() + + try: + vless_client, ss_client = self.client_service.get_client_info(telegram_id) + + def format_info(client, name): + if not client: + return f"❌ Клиент {name} не найден.\n" + json_info = json.dumps(client, ensure_ascii=False, indent=2) + return f"🔹 {name}:\n
{json_info}
" + + response = ( + format_info(vless_client, "VLESS") + + "\n\n" + + format_info(ss_client, "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 diff --git a/bot/main.py b/bot/main.py index 7ce9adc..d6b2f4e 100644 --- a/bot/main.py +++ b/bot/main.py @@ -1,291 +1,62 @@ -import json -import logging -import uuid import asyncio -import aiohttp -import os +from aiogram.filters import Command + +from bot.core.bot import create_bot, create_dispatcher +from bot.services.xui import XUIService +from bot.services.client import ClientService +from bot.handlers.base import cmd_start, cmd_help +from bot.handlers.client import ClientHandlers +from bot.filters.access import AllowedUsersFilter +from bot.utils.logging import logger -from aiogram.types import Message -from aiogram import Bot, Dispatcher, types -from aiogram.filters import Command, BaseFilter -from aiogram.enums import ParseMode -from aiogram.client.default import DefaultBotProperties +async def setup_handlers(dp, client_handlers: ClientHandlers): + """Настройка обработчиков команд.""" + # Базовые команды (доступны всем) + dp.message.register(cmd_start, Command(commands=["start"])) + dp.message.register(cmd_help, Command(commands=["help"])) -from pyxui import XUI # type: ignore -from pyxui.errors import BadLogin # type: ignore - - - -# --- Конфигурация и константы --- -XUI_FULL_ADDRESS = os.getenv("XUI_FULL_ADDRESS") -XUI_PANEL_NAME = os.getenv("XUI_PANEL_NAME") -XUI_USERNAME = os.getenv("XUI_USERNAME") -XUI_PASSWORD = os.getenv("XUI_PASSWORD") - -INBOUND_VLESS_ID = int(os.getenv("INBOUND_VLESS_ID", 15)) -INBOUND_SS_ID = int(os.getenv("INBOUND_SS_ID", 2)) - -SUBSCRIPTION_UUID = os.getenv("SUBSCRIPTION_UUID") -SUB_BASE_URL = f"https://udochudo.ru/{SUBSCRIPTION_UUID}/" - -BOT_TOKEN = os.getenv("BOT_TOKEN") - -ALLOWED_CHAT_IDS = set(map(int, os.getenv("ALLOWED_CHAT_IDS", "").split(","))) - -# --- Логирование --- -logging.basicConfig(level=logging.INFO) -logger = logging.getLogger(__name__) - -# --- Инициализация бота и диспетчера --- -bot = Bot(token=BOT_TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.HTML)) -dp = Dispatcher() - -# --- Экземпляр XUI --- -xui = XUI(full_address=XUI_FULL_ADDRESS, panel=XUI_PANEL_NAME, https=True) - - -def xui_login(): - if xui.session_string: - logger.debug("Уже залогинен в XUI.") - return - try: - 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 - - -async def create_shadowsocks_client_via_api(telegram_id: str, password: str) -> bool: - """Создание Shadowsocks клиента через API XUI с поддержкой повторного логина.""" - try: - xui_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" - - # Получаем куки из сессии XUI - cookies = {} - - if hasattr(xui, 'session') and xui.session and hasattr(xui.session, 'cookies'): - # Пробуем достать куки в нужном формате - for cookie in xui.session.cookies: - if cookie.name == 'x-ui': - cookies['x-ui'] = cookie.value - break - - # Если xui.session_string установлен, используем его как значение cookie x-ui - if xui.session_string: - cookies['x-ui'] = xui.session_string - - 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: - xui.session_string = None - xui_login() - if hasattr(xui, 'session') and xui.session and hasattr(xui.session, 'cookies'): - cookies = {cookie.name: cookie.value for cookie in xui.session.cookies} - if xui.session_string: - cookies['x-ui'] = xui.session_string - 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 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(telegram_id: str, ss_email: str) -> bool: - """Проверка, что клиент существует в inbound SS.""" - try: - inbound_info = xui.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 - -class AllowedUsersFilter(BaseFilter): - async def __call__(self, message: Message) -> bool: - return message.chat.id in ALLOWED_CHAT_IDS - -@dp.message(Command(commands=["start","help"])) -async def cmd_start(message: types.Message): - await message.answer( - "Привет! Чтобы создать профиль, отправь команду:\n" - "/create Telegram id or username\n\n" - "Например:\n" - "/create udochudo" - ,reply_markup=None) - - -@dp.message(Command(commands=["info"])) -async def cmd_info(message: types.Message): - args = (message.text or "").strip().split(maxsplit=1) - if len(args) < 2: - await message.answer( - "❌ Укажи Telegram ID или username после команды.\n" - "Пример:\n/info udochudo" - ) - return - - telegram_id = args[1].lstrip("@").strip() - vless_email = f"{telegram_id}_vl_ssl" - ss_email = f"{telegram_id}_ss" - - try: - xui_login() - vless_client = xui.get_client(inbound_id=INBOUND_VLESS_ID, email=vless_email) - ss_client = xui.get_client(inbound_id=INBOUND_SS_ID, email=ss_email) - - def format_info(client, name): - if not client: - return f"❌ Клиент {name} не найден.\n" - json_info = json.dumps(client, ensure_ascii=False, indent=2) - return f"🔹 {name}:\n
{json_info}
" - - response = format_info(vless_client, "VLESS") + "\n\n" + format_info(ss_client, "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("❌ Ошибка при получении информации. Проверь лог.") - - -@dp.message(Command(commands=["create"])) -async def cmd_create(message: types.Message): - args = (message.text or "").strip().split(maxsplit=1) - if len(args) < 2: - await message.answer( - "❌ Укажи Telegram ID или username после команды.\n" - "Пример:\n/create udochudo" - ) - return - - telegram_id = args[1].lstrip("@").strip() - - try: - xui_login() - - vless_email = f"{telegram_id}_vl_ssl" - vless_uuid = str(uuid.uuid4()) - ss_password = str(uuid.uuid4()) - - logger.info(f"Создаём клиентов для telegram_id={telegram_id} VLESS email={vless_email}, uuid={vless_uuid}") - - # Создаём VLESS клиента через pyxui - xui.add_client( - inbound_id=INBOUND_VLESS_ID, - email=vless_email, - uuid=vless_uuid, - enable=True, - flow="xtls-rprx-vision", - limit_ip=0, - total_gb=0, - expire_time=0, - telegram_id=telegram_id, - subscription_id=telegram_id - ) - logger.info(f"VLESS клиент создан с email={vless_email} и uuid={vless_uuid}") - - # Создаём Shadowsocks клиента через API - success_ss = await create_shadowsocks_client_via_api(telegram_id, ss_password) - if not success_ss: - await message.answer("❌ Ошибка при создании Shadowsocks клиента.") - return - - subscription_link = f"{SUB_BASE_URL}{telegram_id}?name={telegram_id}" - text = f"✅ Профиль для {telegram_id} успешно создан!\n🔗 Подписочная ссылка:\n{subscription_link}" - await message.answer(text,parse_mode=ParseMode.HTML) - - except BadLogin: - await message.answer("❌ Ошибка: неверный логин или пароль XUI.") - except Exception as e: - logger.error(f"Ошибка при создании профиля: {e}") - await message.answer("❌ Произошла ошибка при создании профиля. Попробуйте позже.") + # Команды для работы с клиентами (только для разрешенных пользователей) + dp.message.register( + client_handlers.cmd_create, + Command(commands=["create"]), + AllowedUsersFilter() + ) + dp.message.register( + client_handlers.cmd_info, + Command(commands=["info"]), + AllowedUsersFilter() + ) async def main(): + """Главная функция приложения.""" logger.info("Запуск бота...") + + # Создаем основные компоненты + bot = create_bot() + dp = create_dispatcher() + + # Создаем сервисы + xui_service = XUIService() + client_service = ClientService(xui_service) + + # Создаем обработчики + client_handlers = ClientHandlers(client_service) + + # Настраиваем обработчики + await setup_handlers(dp, client_handlers) + try: + logger.info("Бот запущен и готов к работе") await dp.start_polling(bot) + except Exception as e: + logger.error(f"Ошибка при запуске бота: {e}") + raise finally: await bot.session.close() + logger.info("Бот остановлен") if __name__ == "__main__": - asyncio.run(main()) + asyncio.run(main()) \ No newline at end of file diff --git a/bot/services/__init__.py b/bot/services/__init__.py new file mode 100644 index 0000000..28c98b7 --- /dev/null +++ b/bot/services/__init__.py @@ -0,0 +1,3 @@ +""" +Сервисы для бизнес-логики приложения. +""" \ No newline at end of file diff --git a/bot/services/client.py b/bot/services/client.py new file mode 100644 index 0000000..d514f21 --- /dev/null +++ b/bot/services/client.py @@ -0,0 +1,178 @@ +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_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 \ No newline at end of file diff --git a/bot/services/xui.py b/bot/services/xui.py new file mode 100644 index 0000000..ac3a578 --- /dev/null +++ b/bot/services/xui.py @@ -0,0 +1,110 @@ +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]]: + """Получение информации о клиенте.""" + try: + self.login() + return self.xui.get_client(inbound_id=inbound_id, email=email) + 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 \ No newline at end of file diff --git a/bot/utils/__init__.py b/bot/utils/__init__.py new file mode 100644 index 0000000..c2f97d0 --- /dev/null +++ b/bot/utils/__init__.py @@ -0,0 +1,3 @@ +""" +Утилиты и вспомогательные функции. +""" \ No newline at end of file diff --git a/bot/utils/logging.py b/bot/utils/logging.py new file mode 100644 index 0000000..7d91cca --- /dev/null +++ b/bot/utils/logging.py @@ -0,0 +1,15 @@ +import logging + + +def setup_logging(level: int = logging.INFO) -> logging.Logger: + """Настройка логирования для приложения.""" + logging.basicConfig( + level=level, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' + ) + logger = logging.getLogger(__name__) + return logger + + +# Создаем общий логгер для всего приложения +logger = setup_logging() \ No newline at end of file