houseassassin/Solo_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

hwid management in admin-panel and key menu

Browse Source
This commit is contained in:
Vladless
2025-05-12 23:16:10 +03:00
parent 2196bb53a0
commit 3757bb6082
5 changed files with 269 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files
handlers/admin/users/keyboard.py
+26 -1
View File
@@ -4,7 +4,7 @@ from aiogram.filters.callback_data import CallbackData
from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup
from aiogram.utils.keyboard import InlineKeyboardBuilder
from config import RENEWAL_PRICES, TOTAL_GB
from config import RENEWAL_PRICES, TOTAL_GB, HWID_RESET_BUTTON
from database import get_clusters
from handlers.buttons import BACK
@@ -189,11 +189,36 @@ def build_key_edit_kb(key_details: dict, email: str) -> InlineKeyboardMarkup:
action="users_reset_traffic", data=email, tg_id=key_details["tg_id"]
).pack(),
)
if HWID_RESET_BUTTON:
builder.button(
text="💻 HWID",
callback_data=AdminUserEditorCallback(
action="users_hwid_menu", data=email, tg_id=key_details["tg_id"]
).pack(),
)
builder.row(build_editor_back_btn(key_details["tg_id"], True))
builder.adjust(1)
return builder.as_markup()
def build_hwid_menu_kb(email: str, tg_id: int) -> InlineKeyboardMarkup:
builder = InlineKeyboardBuilder()
builder.button(
text="♻️ Сбросить HWID",
callback_data=AdminUserEditorCallback(
action="users_hwid_reset", data=email, tg_id=tg_id
).pack(),
)
builder.button(
text="🔙 Назад",
callback_data=AdminUserEditorCallback(
action="users_key_edit", data=email, tg_id=tg_id
).pack(),
)
builder.adjust(1)
return builder.as_markup()
def build_key_delete_kb(tg_id: int, email: str) -> InlineKeyboardMarkup:
builder = InlineKeyboardBuilder()
builder.row(
handlers/admin/users/users_handler.py
+84 -1
View File
@@ -14,7 +14,7 @@ from aiogram.fsm.state import State, StatesGroup
from aiogram.types import CallbackQuery, Message
from aiogram.utils.keyboard import InlineKeyboardBuilder
from config import RENEWAL_PRICES, TOTAL_GB, USE_COUNTRY_SELECTION
from config import RENEWAL_PRICES, TOTAL_GB, USE_COUNTRY_SELECTION, REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD
from database import (
delete_key,
delete_user_data,
@@ -39,6 +39,7 @@ from handlers.keys.key_utils import (
from handlers.utils import generate_random_email, sanitize_key_name
from logger import logger
from utils.csv_export import export_referrals_csv
from panels.remnawave import RemnawaveAPI
from ..panel.keyboard import AdminPanelCallback, build_admin_back_btn, build_admin_back_kb
from .keyboard import (
@@ -54,6 +55,7 @@ from .keyboard import (
build_users_balance_kb,
build_users_key_expiry_kb,
build_users_key_show_kb,
build_hwid_menu_kb
)
@@ -73,6 +75,87 @@ class UserEditorState(StatesGroup):
selecting_country = State()
@router.callback_query(AdminUserEditorCallback.filter(F.action == "users_hwid_menu"), IsAdminFilter())
async def handle_hwid_menu(callback_query: CallbackQuery, callback_data: AdminUserEditorCallback, session: Any):
email = callback_data.data
tg_id = callback_data.tg_id
client_id = await get_client_id_by_email(email)
if not client_id:
await callback_query.message.edit_text("🚫 Не удалось найти client_id по email.")
return
servers = await get_servers()
remna_server = None
for cluster_servers in servers.values():
for server in cluster_servers:
if server.get("panel_type", "") == "remnawave":
remna_server = server
break
if remna_server:
break
if not remna_server:
await callback_query.message.edit_text("🚫 Нет доступного сервера Remnawave.")
return
api = RemnawaveAPI(remna_server["api_url"])
if not await api.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD):
await callback_query.message.edit_text("❌ Ошибка авторизации в Remnawave.")
return
devices = await api.get_user_hwid_devices(client_id)
count = len(devices) if devices else 0
text = f"💻 <b>HWID устройства</b>\n\nПривязано: <b>{count}</b>"
await callback_query.message.edit_text(text, reply_markup=build_hwid_menu_kb(email, tg_id))
@router.callback_query(AdminUserEditorCallback.filter(F.action == "users_hwid_reset"), IsAdminFilter())
async def handle_hwid_reset(callback_query: CallbackQuery, callback_data: AdminUserEditorCallback, session: Any):
email = callback_data.data
tg_id = callback_data.tg_id
client_id = await get_client_id_by_email(email)
if not client_id:
await callback_query.message.edit_text("🚫 Не удалось найти client_id по email.")
return
servers = await get_servers()
remna_server = None
for cluster_servers in servers.values():
for server in cluster_servers:
if server.get("panel_type", "") == "remnawave":
remna_server = server
break
if remna_server:
break
if not remna_server:
await callback_query.message.edit_text("🚫 Нет доступного сервера Remnawave.")
return
api = RemnawaveAPI(remna_server["api_url"])
if not await api.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD):
await callback_query.message.edit_text("❌ Ошибка авторизации в Remnawave.")
return
devices = await api.get_user_hwid_devices(client_id)
if not devices:
await callback_query.message.edit_text("ℹ️ У пользователя нет привязанных устройств.", reply_markup=build_editor_kb(tg_id, True))
return
deleted = 0
for d in devices:
if await api.delete_user_hwid_device(client_id, d["hwid"]):
deleted += 1
await callback_query.message.edit_text(
f"✅ Удалено HWID-устройств: <b>{deleted}</b> из <b>{len(devices)}</b>.",
reply_markup=build_editor_kb(tg_id, True),
)
@router.callback_query(
AdminPanelCallback.filter(F.action == "search_user"),
IsAdminFilter(),
handlers/buttons.py
+1
View File
@@ -73,6 +73,7 @@ MANUAL_INSTRUCTIONS = "📖 Ручная установка"
RENEW_KEY = "🔄 Продлить подписку"
TV_CONTINUE = "▶ Продолжить"
TV_INSTRUCTIONS = "📖 Полная инструкция"
HWID_BUTTON = "♻️ Сбросить привязку"
# Кнопки касс
handlers/keys/key_view.py
+158 -128
View File
@@ -20,10 +20,12 @@ from config import (
QRCODE,
TOGGLE_CLIENT,
USE_COUNTRY_SELECTION,
HWID_RESET_BUTTON
)
from database import (
get_key_details,
get_keys,
get_servers
)
from handlers.buttons import (
ADD_SUB,
@@ -41,6 +43,7 @@ from handlers.buttons import (
RENEW_FULL,
TV_BUTTON,
UNFREEZE,
HWID_BUTTON
)
from handlers.texts import (
FROZEN_SUBSCRIPTION_MSG,
@@ -178,145 +181,172 @@ async def handle_new_alias_input(message: Message, state: FSMContext, session: A
@router.callback_query(F.data.startswith("view_key|"))
async def process_callback_view_key(callback_query: CallbackQuery, session: Any):
tg_id = callback_query.message.chat.id
key_name = callback_query.data.split("|")[1]
try:
record = await get_key_details(key_name, session)
if not record:
await callback_query.message.answer(text="<b>Информация о подписке не найдена.</b>")
return
image_path = os.path.join("img", "pic_view.jpg")
await render_key_info(callback_query.message, session, key_name, image_path)
is_frozen = record["is_frozen"]
builder = InlineKeyboardBuilder()
image_path = os.path.join("img", "pic_view.jpg")
if not os.path.isfile(image_path):
await callback_query.message.answer("Файл изображения не найден.")
return
async def render_key_info(message: Message, session: Any, key_name: str, image_path: str):
from config import REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD
from panels.remnawave import RemnawaveAPI
if is_frozen:
builder.row(
InlineKeyboardButton(
text=UNFREEZE,
callback_data=f"unfreeze_subscription|{key_name}",
)
)
builder.row(InlineKeyboardButton(text=BACK, callback_data="view_keys"))
builder.row(InlineKeyboardButton(text=MAIN_MENU, callback_data="profile"))
record = await get_key_details(key_name, session)
if not record:
await message.answer("<b>Информация о подписке не найдена.</b>")
return
await edit_or_send_message(
target_message=callback_query.message,
text=FROZEN_SUBSCRIPTION_MSG,
reply_markup=builder.as_markup(),
media_path=image_path,
)
return
is_frozen = record["is_frozen"]
email = record["email"]
client_id = record.get("client_id")
remnawave_link = record.get("remnawave_link")
key = record.get("key")
final_link = key or remnawave_link
key = record.get("key")
remnawave_link = record.get("remnawave_link")
final_link = key or remnawave_link
expiry_time = record["expiry_time"]
server_name = record["server_id"]
expiry_date = datetime.utcfromtimestamp(expiry_time / 1000)
time_left = expiry_date - datetime.utcnow()
if time_left.total_seconds() <= 0:
days_left_message = "<b>🕒 Статус подписки:</b>\n🔴 Истекла"
else:
total_seconds = int(time_left.total_seconds())
days = total_seconds // 86400
hours = (total_seconds % 86400) // 3600
minutes = (total_seconds % 3600) // 60
days_left_message = (
f"Осталось: <b>{format_days(days)}</b>, <b>{format_hours(hours)}</b>, <b>{format_minutes(minutes)}</b>"
)
formatted_expiry_date = (
f"{expiry_date.strftime('%d')} {get_russian_month(expiry_date)} {expiry_date.strftime('%Y')} года"
)
response_message = key_message(
final_link,
formatted_expiry_date,
days_left_message,
server_name,
server_name if USE_COUNTRY_SELECTION else None,
)
if ENABLE_UPDATE_SUBSCRIPTION_BUTTON:
builder.row(
InlineKeyboardButton(
text="🔄 Обновить подписку",
callback_data=f"update_subscription|{key_name}",
)
)
is_full_remnawave = await is_full_remnawave_cluster(server_name, session)
if is_full_remnawave and final_link:
builder.row(
InlineKeyboardButton(
text=CONNECT_DEVICE,
web_app=WebAppInfo(url=final_link),
)
)
else:
if CONNECT_PHONE_BUTTON:
builder.row(
InlineKeyboardButton(
text=CONNECT_PHONE,
callback_data=f"connect_phone|{key_name}",
)
)
builder.row(
InlineKeyboardButton(text=PC_BUTTON, callback_data=f"connect_pc|{key_name}"),
InlineKeyboardButton(text=TV_BUTTON, callback_data=f"connect_tv|{key_name}"),
)
else:
builder.row(
InlineKeyboardButton(
text=CONNECT_DEVICE,
callback_data=f"connect_device|{key_name}",
)
)
if QRCODE:
builder.row(
InlineKeyboardButton(
text=QR,
callback_data=f"show_qr|{key_name}",
)
)
if ENABLE_DELETE_KEY_BUTTON:
builder.row(
InlineKeyboardButton(text=RENEW, callback_data=f"renew_key|{key_name}"),
InlineKeyboardButton(text=DELETE, callback_data=f"delete_key|{key_name}"),
)
else:
builder.row(InlineKeyboardButton(text=RENEW_FULL, callback_data=f"renew_key|{key_name}"))
if USE_COUNTRY_SELECTION:
builder.row(InlineKeyboardButton(text=CHANGE_LOCATION, callback_data=f"change_location|{key_name}"))
if TOGGLE_CLIENT:
builder.row(
InlineKeyboardButton(
text=FREEZE,
callback_data=f"freeze_subscription|{key_name}",
)
)
builder = InlineKeyboardBuilder()
if is_frozen:
builder.row(InlineKeyboardButton(text=UNFREEZE, callback_data=f"unfreeze_subscription|{key_name}"))
builder.row(InlineKeyboardButton(text=BACK, callback_data="view_keys"))
builder.row(InlineKeyboardButton(text=MAIN_MENU, callback_data="profile"))
await edit_or_send_message(
target_message=callback_query.message,
text=response_message,
target_message=message,
text=FROZEN_SUBSCRIPTION_MSG,
reply_markup=builder.as_markup(),
media_path=image_path,
)
except Exception as e:
await handle_error(
tg_id,
callback_query,
f"Ошибка при получении информации о ключе: {e}",
return
expiry_time = record["expiry_time"]
server_name = record["server_id"]
expiry_date = datetime.utcfromtimestamp(expiry_time / 1000)
time_left = expiry_date - datetime.utcnow()
if time_left.total_seconds() <= 0:
days_left_message = "<b>🕒 Статус подписки:</b>\n🔴 Истекла"
else:
total_seconds = int(time_left.total_seconds())
days = total_seconds // 86400
hours = (total_seconds % 86400) // 3600
minutes = (total_seconds % 3600) // 60
days_left_message = (
f"Осталось: <b>{format_days(days)}</b>, <b>{format_hours(hours)}</b>, <b>{format_minutes(minutes)}</b>"
)
formatted_expiry_date = (
f"{expiry_date.strftime('%d')} {get_russian_month(expiry_date)} {expiry_date.strftime('%Y')} года"
)
hwid_count = 0
is_full_remnawave = await is_full_remnawave_cluster(server_name, session)
if is_full_remnawave and client_id:
servers = await get_servers()
remna_server = next(
(srv for cl in servers.values() for srv in cl if srv.get("panel_type") == "remnawave"), None
)
if remna_server:
api = RemnawaveAPI(remna_server["api_url"])
if await api.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD):
devices = await api.get_user_hwid_devices(client_id)
hwid_count = len(devices or [])
response_message = key_message(
final_link,
formatted_expiry_date,
days_left_message,
server_name,
server_name if USE_COUNTRY_SELECTION else None,
hwid_count=hwid_count,
)
if ENABLE_UPDATE_SUBSCRIPTION_BUTTON:
builder.row(InlineKeyboardButton(text="🔄 Обновить подписку", callback_data=f"update_subscription|{key_name}"))
if is_full_remnawave and final_link:
builder.row(InlineKeyboardButton(text=CONNECT_DEVICE, web_app=WebAppInfo(url=final_link)))
else:
if CONNECT_PHONE_BUTTON:
builder.row(InlineKeyboardButton(text=CONNECT_PHONE, callback_data=f"connect_phone|{key_name}"))
builder.row(
InlineKeyboardButton(text=PC_BUTTON, callback_data=f"connect_pc|{key_name}"),
InlineKeyboardButton(text=TV_BUTTON, callback_data=f"connect_tv|{key_name}"),
)
else:
builder.row(InlineKeyboardButton(text=CONNECT_DEVICE, callback_data=f"connect_device|{key_name}"))
if HWID_RESET_BUTTON and hwid_count > 0:
builder.row(
InlineKeyboardButton(
text=HWID_BUTTON,
callback_data=f"reset_hwid|{key_name}",
)
)
if QRCODE:
builder.row(InlineKeyboardButton(text=QR, callback_data=f"show_qr|{key_name}"))
if ENABLE_DELETE_KEY_BUTTON:
builder.row(
InlineKeyboardButton(text=RENEW, callback_data=f"renew_key|{key_name}"),
InlineKeyboardButton(text=DELETE, callback_data=f"delete_key|{key_name}"),
)
else:
builder.row(InlineKeyboardButton(text=RENEW_FULL, callback_data=f"renew_key|{key_name}"))
if USE_COUNTRY_SELECTION:
builder.row(InlineKeyboardButton(text=CHANGE_LOCATION, callback_data=f"change_location|{key_name}"))
if TOGGLE_CLIENT:
builder.row(InlineKeyboardButton(text=FREEZE, callback_data=f"freeze_subscription|{key_name}"))
builder.row(InlineKeyboardButton(text=BACK, callback_data="view_keys"))
builder.row(InlineKeyboardButton(text=MAIN_MENU, callback_data="profile"))
await edit_or_send_message(
target_message=message,
text=response_message,
reply_markup=builder.as_markup(),
media_path=image_path,
)
@router.callback_query(F.data.startswith("reset_hwid|"))
async def handle_reset_hwid(callback_query: CallbackQuery, session: Any):
from config import REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD
from panels.remnawave import RemnawaveAPI
key_name = callback_query.data.split("|")[1]
record = await get_key_details(key_name, session)
if not record:
await callback_query.answer("❌ Ключ не найден.", show_alert=True)
return
client_id = record.get("client_id")
if not client_id:
await callback_query.answer("❌ У ключа отсутствует client_id.", show_alert=True)
return
servers = await get_servers()
remna_server = next(
(srv for cl in servers.values() for srv in cl if srv.get("panel_type") == "remnawave"), None
)
if not remna_server:
await callback_query.answer("❌ Remnawave-сервер не найден.", show_alert=True)
return
api = RemnawaveAPI(remna_server["api_url"])
if not await api.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD):
await callback_query.answer("❌ Авторизация в Remnawave не удалась.", show_alert=True)
return
devices = await api.get_user_hwid_devices(client_id)
if not devices:
await callback_query.answer("✅ Устройства не были привязаны.", show_alert=True)
else:
deleted = 0
for device in devices:
if await api.delete_user_hwid_device(client_id, device["hwid"]):
deleted += 1
await callback_query.answer(f"✅ Устройства сброшены ({deleted})", show_alert=True)
image_path = os.path.join("img", "pic_view.jpg")
await render_key_info(callback_query.message, session, key_name, image_path)
panels/remnawave.cpython-312-x86_64-linux-gnu.so
BIN
View File
Binary file not shown.