340 подписчиков

3x-ui API подключение на Python

18 сентября 202318 сен 2023

3131

6 мин

Недавно у меня заказали Чат Бота в Телеграм который должен продавать подписки на VPN сервер 3x-ui.

В документации достаточно подробно описано подключение к серверу по API но есть моменты с ошибка и пришлось немного повозиться.

В итоге я хочу поделиться готовыми методами подключения и взаимодействия.

Итак как всегда я начну сразу с кода:

import datetime
import json

import requests
import uuid

class X3:
login = <Ваш логин>
password = <Ваш пароль>
host = <Адрес Вашего сервера>
header = []
data = {"username": login, "password": password}
ses = requests.Session()
# Тестовое соединение
def test_connect(self):
response = self.ses.post(f"{self.host}/login", data=self.data)
return response
# Список клиентов
def list(self):
resource = self.ses.get(f'{self.host}/panel/api/inbounds/list', json=self.data).json()
return resource
# Добавление клиентов
def addClient(self, day, tg_id, user_id):
epoch = datetime.datetime.utcfromtimestamp(0)
x_time = int((datetime.datetime.now() - epoch).total_seconds() * 1000.0)
x_time += 86400000 * day - 10800000
header = {"Accept": "application/json"}
data1 = {
"id": 1,
"settings":
"{\"clients\":"
"[{\"id\":\"" + str(uuid.uuid1()) + "\","
"\"alterId\":90,\"email\":\"" + str(user_id) + "\","
"\"limitIp\":3,\"totalGB\":0,"
"\"expiryTime\":" + str(x_time) + ",\"enable\":true,\"tgId\":\"" + str(tg_id) + "\",\"subId\":\"\"}]}"
}
resource = self.ses.post(f'{self.host}/panel/api/inbounds/addClient', headers=header, json=data1)
return resource
# Изменение клиентов
def updateClient(self, day, user_id):
dict_x = self.time_active(user_id)
for key, val in dict_x.items():
if key != '0' and val != '0':
val /= 1000
val += 10800
date_x = datetime.datetime.utcfromtimestamp(val)
# date_x += datetime.timedelta(day)
epoch = datetime.datetime.utcfromtimestamp(0)
x_time = int((date_x - epoch).total_seconds() * 1000.0)
x_time += 86400000 * day - 10800000
header = {"Accept": "application/json"}
data1 = {
"id": 1,
"settings":
"{\"clients\":"
"[{\"id\":\"" + str(key) + "\","
"\"alterId\":90,\"email\":\"" + str(user_id) + "\","
"\"limitIp\":3,\"totalGB\":0,"
"\"expiryTime\":" + str(x_time) + ",\"enable\":true,\"tgId\":\"" + str(user_id) + "\",\"subId\":\"\"}]}"
}
resource = self.ses.post(f'{self.host}/panel/api/inbounds/updateClient/{key}', headers=header, json=data1)
return resource
else:
epoch = datetime.datetime.utcfromtimestamp(0)
x_time = int((datetime.datetime.now() - epoch).total_seconds() * 1000.0)
x_time += 86400000 * day - 10800000
header = {"Accept": "application/json"}
data1 = {
"id": 1,
"settings":
"{\"clients\":"
"[{\"id\":\"" + str(key) + "\","
"\"alterId\":90,\"email\":\"" + str(user_id) + "\","
"\"limitIp\":3,\"totalGB\":0,"
"\"expiryTime\":" + str(
x_time) + ",\"enable\":true,\"tgId\":\"" + str(user_id) + "\",\"subId\":\"\"}]}"
}
resource = self.ses.post(f'{self.host}/panel/api/inbounds/updateClient/{key}', headers=header,
json=data1)
return resource
# Получение ссылки ключа
def link(self, user_id: str):
"""
Получение ссылки!
:param user_id: str
:return: str
"""
id = ''
y = json.loads(self.list()['obj'][0]['settings'])
for i in y["clients"]:
if i['email'] == user_id:
id = i["id"]
x = json.loads(self.list()['obj'][0]['streamSettings'])
tcp = x['network']
reality = x['security']
val = f"vless://{id}@vpn-x3.ru:52687/?type={tcp}&security={reality}&fp=chrome&pbk=T_95HnSovtH9WNr_XfaJ9iL7xnwp96p8E2A8Q3_t_xk&sni=microsoft.com&sid=24705084&spx=%2F#VPN-X3-{user_id}"
return val
# Проверка времени активной подписки
def time_active(self, user_id: str):
dict_x = {}
epoch = datetime.datetime.utcfromtimestamp(0)
x_time = int((datetime.datetime.now() - epoch).total_seconds() * 1000.0)
y = json.loads(self.list()['obj'][0]['settings'])
for i in y["clients"]:
if i['email'] == user_id:
if i['enable'] and i['expiryTime'] > x_time:
dict_x[i['id']] = i['expiryTime']
return dict_x
else:
dict_x[i['id']] = '0'
return dict_x
if len(dict_x) == 0:
dict_x['0'] = '0'
return dict_x
# Проверка активности
def activ(self, user_id: str):
"""
Проверка активности подписки
:param user_id: str
:return: str
"""
dict_x = {}
epoch = datetime.datetime.utcfromtimestamp(0)
x_time = int((datetime.datetime.now() - epoch).total_seconds() * 1000.0)
y = json.loads(self.list()['obj'][0]['settings'])
for i in y["clients"]:
if i['email'] == user_id:
if i['enable'] and i['expiryTime'] > x_time:
print(i)
print(i['enable'])
dict_x['activ'] = 'Активен'
ts = i['expiryTime']
ts /= 1000
ts += 10800
dict_x['time'] = datetime.datetime.utcfromtimestamp(ts).strftime('%d-%m-%Y %H:%M') + ' МСК'
return dict_x
else:
print(i)
print(i['enable'])
dict_x['activ'] = 'Не Активен'
ts = i['expiryTime']
ts /= 1000
ts += 10800
dict_x['time'] = datetime.datetime.utcfromtimestamp(ts).strftime('%d-%m-%Y %H:%M') + ' МСК'
return dict_x
else:
dict_x['activ'] = 'Не зарегистрирован'
dict_x['time'] = '-'
return dict_x
# Проверка активных пользователей с подпиской
def activ_list(self):
"""
Проверка активности подписки
:param user_id: str
:return: str
"""
dict_x = {}
y = json.loads(self.list()['obj'][0]['settings'])
for i in y["clients"]:
ts = i['expiryTime']
ts /= 1000
ts += 10800
x = datetime.datetime.now()
y = datetime.datetime.utcfromtimestamp(ts)
z = y - x
dict_x[i['email']] = z.days
return dict_x