292 lines
11 KiB
Python
292 lines
11 KiB
Python
import json
|
||
import logging
|
||
import uuid
|
||
import asyncio
|
||
import aiohttp
|
||
import os
|
||
|
||
from aiogram import BaseFilter
|
||
from aiogram.types import Message
|
||
from aiogram import Bot, Dispatcher, types
|
||
from aiogram.filters import Command
|
||
from aiogram.enums import ParseMode
|
||
from aiogram.client.default import DefaultBotProperties
|
||
|
||
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"]))
|
||
async def cmd_start(message: types.Message):
|
||
await message.answer(
|
||
"Привет! Чтобы создать профиль, отправь команду:\n"
|
||
"<code>/create telegram_id_or_username</code>\n\n"
|
||
"Например:\n"
|
||
"<code>/create udochudo</code>"
|
||
)
|
||
|
||
|
||
@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<code>/info udochudo</code>"
|
||
)
|
||
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"❌ Клиент <b>{name}</b> не найден.\n"
|
||
json_info = json.dumps(client, ensure_ascii=False, indent=2)
|
||
return f"🔹 <b>{name}</b>:\n<pre>{json_info}</pre>"
|
||
|
||
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<code>/create udochudo</code>"
|
||
)
|
||
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<code>{subscription_link}</code>"
|
||
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("❌ Произошла ошибка при создании профиля. Попробуйте позже.")
|
||
|
||
|
||
async def main():
|
||
logger.info("Запуск бота...")
|
||
try:
|
||
await dp.start_polling(bot)
|
||
finally:
|
||
await bot.session.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|