Files
RemnaWaveBOT/app/bot/ui/panel.py
2026-07-09 16:45:17 +03:00

640 lines
25 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
from __future__ import annotations
import html
import math
from dataclasses import dataclass
from datetime import datetime
from aiogram.types import InlineKeyboardMarkup
from aiogram.utils.keyboard import InlineKeyboardBuilder
from app.services.sync_service import CachedUserView
from app.utils.formatters import format_bytes, format_datetime, format_payment_plan, format_traffic_limit
PANEL_SECTION_HOME = "home"
PANEL_SECTION_ADMIN = "admin"
PANEL_SECTION_SUBSCRIPTION = "subscription"
PANEL_SECTION_REFERRAL = "referral"
PANEL_SECTION_SUPPORT = "support"
PANEL_SECTION_SUPPORT_TICKET = "support_ticket"
PANEL_SECTION_TERMS = "terms"
@dataclass(slots=True)
class PanelContext:
telegram_id: int
display_name: str
username: str | None
language_code: str | None
is_admin: bool
users: list[CachedUserView]
referral_enabled: bool = True
referral_code: str = ""
referral_count: int = 0
applied_referral_code: str = ""
referral_link: str = ""
referral_share_url: str = ""
referral_recent_names: list[str] | None = None
referral_discount_percent: int = 0
referral_bonus_days: int = 0
admin_total_telegram_users: int = 0
admin_total_cached_users: int = 0
admin_active_cached_users: int = 0
admin_pending_orders: int = 0
admin_review_orders: int = 0
admin_open_tickets: int = 0
admin_pending_referral_bonuses: int = 0
payment_enabled: bool = False
payment_plan_title: str = ""
payment_plan_description: str = ""
payment_plan_price_stars: int = 0
payment_plan_duration_days: int = 0
payment_plan_traffic_limit_bytes: int = 0
api_error: str | None = None
def _callback(section: str, *, refresh: bool = False) -> str:
return f"panel:{section}:refresh" if refresh else f"panel:{section}"
def _ticket_callback(action: str) -> str:
return f"ticket:{action}"
def _referral_callback(action: str) -> str:
return f"referral:{action}"
def _payment_callback(action: str) -> str:
return f"payment:{action}"
def _admin_callback(action: str) -> str:
return f"admin:{action}"
def _status_badge(status: str) -> str:
mapping = {
"ACTIVE": "🟢 Активна",
"LIMITED": "🟡 Ограничена",
"DISABLED": "⛔ Отключена",
"EXPIRED": "🔴 Истекла",
}
return mapping.get(status.upper(), f"{html.escape(status)}")
def _active_users_count(users: list[CachedUserView]) -> int:
return sum(1 for user in users if user.record.status.upper() == "ACTIVE")
def _primary_user(users: list[CachedUserView]) -> CachedUserView | None:
for user in users:
if user.record.status.upper() == "ACTIVE":
return user
return users[0] if users else None
def _home_subscription_status(users: list[CachedUserView]) -> str:
return "Активна" if _active_users_count(users) > 0 else "Не активна"
def _days_left_text(user: CachedUserView | None) -> str:
if user is None or user.record.expire_at is None:
return ""
seconds_left = (user.record.expire_at - datetime.now()).total_seconds()
if seconds_left <= 0:
return "0 дней"
days_left = math.ceil(seconds_left / 86400)
return f"{days_left} дней"
def _traffic_summary_text(user: CachedUserView | None) -> str:
if user is None:
return ""
return (
f"{format_bytes(user.record.used_traffic_bytes)} / "
f"{format_traffic_limit(user.record.traffic_limit_bytes)}"
)
def build_panel_keyboard(
*,
section: str,
context: PanelContext,
support_url: str = "",
terms_url: str = "",
payment_review_url: str = "",
support_ticket_url: str = "",
) -> InlineKeyboardMarkup:
builder = InlineKeyboardBuilder()
primary_user = _primary_user(context.users)
if section == PANEL_SECTION_HOME:
has_sub_link = bool(
primary_user
and primary_user.record.subscription_url.startswith(("http://", "https://"))
)
if has_sub_link:
builder.button(
text="🔗 Открыть подписку",
url=primary_user.record.subscription_url,
)
builder.button(
text="💳 Подписка",
callback_data=_payment_callback("menu"),
)
has_active = _active_users_count(context.users) > 0
if has_active:
builder.button(
text="📚 Гайды",
callback_data="guides:menu",
)
builder.button(
text="📱 Устройства",
callback_data="devices:list",
)
if context.is_admin:
builder.button(
text="🛠 Админ-панель",
callback_data=_callback(PANEL_SECTION_ADMIN),
)
if context.referral_enabled:
builder.button(
text="👥 Реферальная система",
callback_data=_callback(PANEL_SECTION_REFERRAL),
)
builder.button(
text="🆘 Поддержка",
callback_data=_callback(PANEL_SECTION_SUPPORT),
)
if terms_url:
builder.button(text="📜 Условия использования", url=terms_url)
else:
builder.button(
text="📜 Условия использования",
callback_data=_callback(PANEL_SECTION_TERMS),
)
if has_sub_link:
if has_active:
if context.is_admin and context.referral_enabled:
builder.adjust(1, 1, 2, 1, 2, 1)
elif context.is_admin:
builder.adjust(1, 1, 2, 1, 1, 1)
elif context.referral_enabled:
builder.adjust(1, 1, 2, 2, 1)
else:
builder.adjust(1, 1, 2, 1, 1)
else:
if context.is_admin and context.referral_enabled:
builder.adjust(1, 1, 1, 2, 1)
elif context.is_admin:
builder.adjust(1, 1, 1, 1, 1)
elif context.referral_enabled:
builder.adjust(1, 1, 2, 1)
else:
builder.adjust(1, 1, 1, 1)
else:
if has_active:
if context.is_admin and context.referral_enabled:
builder.adjust(1, 2, 1, 2, 1)
elif context.is_admin:
builder.adjust(1, 2, 1, 1, 1)
elif context.referral_enabled:
builder.adjust(1, 2, 2, 1)
else:
builder.adjust(1, 2, 1, 1)
else:
if context.is_admin and context.referral_enabled:
builder.adjust(1, 1, 2, 1)
elif context.is_admin:
builder.adjust(1, 1, 1, 1)
elif context.referral_enabled:
builder.adjust(1, 2, 1)
else:
builder.adjust(1, 1, 1)
return builder.as_markup()
if section == PANEL_SECTION_ADMIN:
if not context.is_admin:
builder.button(text="🏠 Главное меню", callback_data=_callback(PANEL_SECTION_HOME))
builder.adjust(1)
return builder.as_markup()
builder.button(
text="🔍 Найти пользователя",
callback_data=_admin_callback("lookup"),
)
builder.button(
text="👤 Пользователи",
callback_data=_admin_callback("users"),
)
builder.button(
text="🗂 История тикетов",
callback_data=_admin_callback("tickets"),
)
builder.button(
text="🎟 Промокоды",
callback_data=_admin_callback("promos"),
)
builder.button(
text="📚 Гайды",
callback_data=_admin_callback("guides"),
)
builder.button(
text="📣 Рассылка",
callback_data=_admin_callback("broadcast"),
)
builder.button(
text="🔄 Полная синхронизация",
callback_data=_admin_callback("sync_all"),
)
builder.button(
text="⚙️ Настройки",
callback_data=_admin_callback("settings"),
)
if payment_review_url.strip():
builder.button(text="💸 Очередь оплат", url=payment_review_url.strip())
if support_ticket_url.strip():
builder.button(text="🎫 Чат тикетов", url=support_ticket_url.strip())
builder.button(
text="↻ Обновить",
callback_data=_callback(PANEL_SECTION_ADMIN, refresh=True),
)
builder.button(text="🏠 Главное меню", callback_data=_callback(PANEL_SECTION_HOME))
builder.adjust(2, 2, 2, 2, 2, 1)
return builder.as_markup()
if section == PANEL_SECTION_SUBSCRIPTION:
if context.payment_enabled:
builder.button(
text="💳 Продлить подписку" if context.users else "💳 Купить подписку",
callback_data=_payment_callback("buy"),
)
if primary_user and primary_user.record.subscription_url.startswith(("http://", "https://")):
builder.button(
text="🔗 Открыть подписку",
url=primary_user.record.subscription_url,
)
builder.button(text="🏠 Главное меню", callback_data=_callback(PANEL_SECTION_HOME))
builder.adjust(1)
return builder.as_markup()
if section == PANEL_SECTION_REFERRAL:
if not context.referral_enabled:
builder.button(text="🏠 Главное меню", callback_data=_callback(PANEL_SECTION_HOME))
builder.adjust(1)
return builder.as_markup()
if context.referral_share_url:
builder.button(
text="📨 Поделиться ссылкой",
url=context.referral_share_url,
)
builder.button(
text="🎟 Ввести код скидки",
callback_data=_referral_callback("apply"),
)
builder.button(text="🏠 Главное меню", callback_data=_callback(PANEL_SECTION_HOME))
builder.adjust(1)
return builder.as_markup()
if section == PANEL_SECTION_SUPPORT:
if support_url.strip():
builder.button(text="Написать в поддержку", url=support_url.strip())
builder.button(
text="🎫 Создать тикет",
callback_data=_ticket_callback("create"),
)
builder.button(text="🏠 Главное меню", callback_data=_callback(PANEL_SECTION_HOME))
builder.adjust(1)
return builder.as_markup()
if section == PANEL_SECTION_SUPPORT_TICKET:
builder.button(
text="🎫 Создать тикет",
callback_data=_ticket_callback("create"),
)
builder.button(text="🆘 Поддержка", callback_data=_callback(PANEL_SECTION_SUPPORT))
builder.button(text="🏠 Главное меню", callback_data=_callback(PANEL_SECTION_HOME))
builder.adjust(1, 2)
return builder.as_markup()
if section == PANEL_SECTION_TERMS:
if terms_url:
builder.button(text="📜 Открыть Terms of Use", url=terms_url)
builder.button(text="🏠 Главное меню", callback_data=_callback(PANEL_SECTION_HOME))
builder.adjust(1)
return builder.as_markup()
builder.button(text="🏠 Главное меню", callback_data=_callback(PANEL_SECTION_HOME))
return builder.as_markup()
def render_panel_caption(*, section: str, context: PanelContext, brand_name: str) -> str:
if section == PANEL_SECTION_ADMIN:
return _render_admin_caption(context=context, brand_name=brand_name)
if section == PANEL_SECTION_SUBSCRIPTION:
return _render_subscription_caption(context=context)
if section == PANEL_SECTION_REFERRAL:
return _render_referral_caption(context=context)
if section == PANEL_SECTION_SUPPORT:
return _render_support_caption()
if section == PANEL_SECTION_SUPPORT_TICKET:
return _render_support_ticket_caption()
if section == PANEL_SECTION_TERMS:
return _render_terms_caption()
return _render_home_caption(context=context, brand_name=brand_name)
def _render_home_caption(*, context: PanelContext, brand_name: str) -> str:
primary_user = _primary_user(context.users)
sub_lines = [
f"Статус: <b>{_home_subscription_status(context.users)}</b>",
f"Срок: <b>{_days_left_text(primary_user)}</b>",
f"Трафик: <b>{_traffic_summary_text(primary_user)}</b>",
]
if primary_user and primary_user.record.subscription_url:
sub_lines.append(f"Ссылка: <code>{html.escape(primary_user.record.subscription_url)}</code>")
lines = [
f"Добро пожаловать в <b>{html.escape(brand_name)}</b>",
"",
"<b>Подписка</b>",
"<blockquote>" + "\n".join(sub_lines) + "</blockquote>",
]
if context.api_error:
lines.extend(
[
"",
"<i>Remnawave временно недоступен. Показаны данные из локального кэша.</i>",
]
)
if context.is_admin:
lines.extend(
[
"",
"<i>Для управления ботом используйте кнопку «Админ-панель» ниже.</i>",
]
)
return "\n".join(lines)
def _render_admin_caption(*, context: PanelContext, brand_name: str) -> str:
if not context.is_admin:
return "\n".join(
[
"<b>🛠 Админ-панель</b>",
"",
"<blockquote>Недостаточно прав для доступа к этому разделу.</blockquote>",
]
)
return "\n".join(
[
f"<b>🛠 Админ-панель {html.escape(brand_name)}</b>",
"",
(
"<blockquote>"
f"Пользователей в боте: <b>{context.admin_total_telegram_users}</b>\n"
f"Доступов в кэше: <b>{context.admin_total_cached_users}</b>\n"
f"Активных доступов: <b>{context.admin_active_cached_users}</b>\n"
f"Заказов без чека: <b>{context.admin_pending_orders}</b>\n"
f"Чеков на проверке: <b>{context.admin_review_orders}</b>\n"
f"Открытых тикетов: <b>{context.admin_open_tickets}</b>\n"
f"Бонусов в ожидании: <b>{context.admin_pending_referral_bonuses}</b>"
"</blockquote>"
),
"",
"<i>Используйте кнопки ниже для поиска пользователей, синхронизации и перехода в рабочие чаты.</i>",
]
)
def _render_subscription_caption(*, context: PanelContext) -> str:
lines = ["<b>💳 Подписка</b>"]
if not context.users:
if context.payment_enabled:
lines.extend(
[
"",
"<blockquote>У вас пока нет активного доступа.</blockquote>",
"",
format_payment_plan(
title=context.payment_plan_title,
description=context.payment_plan_description,
price_stars=context.payment_plan_price_stars,
duration_days=context.payment_plan_duration_days,
traffic_limit_bytes=context.payment_plan_traffic_limit_bytes,
),
"",
(
"<i>После выбора тарифа бот покажет реквизиты. "
"Затем нужно отправить чек, а администратор подтвердит оплату и выдаст доступ.</i>"
),
]
)
else:
lines.extend(
[
(
"<blockquote>"
"У вас пока нет активного доступа.\n\n"
"Если у вас уже есть подписка, откройте поддержку "
"и отправьте short UUID из ссылки подписки."
"</blockquote>"
),
"",
"<i>Покупка пока не настроена в .env. Нужны параметры PAYMENT_*.</i>",
]
)
return "\n".join(lines)
lines.append("<b>Ваши доступы</b>")
for index, user in enumerate(context.users[:2], start=1):
squad_text = ", ".join(user.internal_squads[:2]) if user.internal_squads else "не назначены"
if len(squad_text) > 80:
squad_text = f"{squad_text[:77]}..."
lines.append(
(
f"{index}. <b>{html.escape(user.record.username)}</b> — "
f"{_status_badge(user.record.status)}\n"
"<blockquote>"
f"Истекает: <b>{format_datetime(user.record.expire_at)}</b>\n"
f"Трафик: <b>{format_bytes(user.record.used_traffic_bytes)}</b> / "
f"<b>{format_traffic_limit(user.record.traffic_limit_bytes)}</b>\n"
f"Short UUID: <code>{html.escape(user.record.short_uuid)}</code>\n"
f"Группы: {html.escape(squad_text)}"
"</blockquote>"
)
)
if len(context.users) > 2:
lines.append(f"<i>И ещё аккаунтов: {len(context.users) - 2}</i>")
if context.payment_enabled:
lines.extend(
[
"",
(
"<i>"
"Чтобы продлить текущую подписку, используйте кнопку "
"«Продлить подписку». После подтверждения оплаты бот "
"автоматически добавит выбранный срок к вашему доступу."
"</i>"
),
]
)
if context.api_error:
lines.append("")
lines.append("<i>Последнее обновление пришло из кэша.</i>")
return "\n".join(lines)
def _render_referral_caption(*, context: PanelContext) -> str:
if not context.referral_enabled:
return "\n".join(
[
"<b>👥 Реферальная система</b>",
"",
"<blockquote>Реферальная система сейчас отключена администратором.</blockquote>",
]
)
lines = [
"<b>👥 Реферальная система</b>",
(
"<blockquote>"
f"Ваш реферальный код: <code>{html.escape(context.referral_code or '')}</code>\n"
f"Приглашено пользователей: <b>{context.referral_count}</b>"
"</blockquote>"
),
]
if context.referral_bonus_days > 0:
lines.extend(
[
"",
(
"<blockquote>"
f"🎁 Бонус за подтверждённую оплату по вашему коду: "
f"<b>+{context.referral_bonus_days} дней</b>"
"</blockquote>"
),
]
)
if context.applied_referral_code:
lines.extend(
[
"",
"<b>Активированная скидка</b>",
(
"<blockquote>"
f"Код: <code>{html.escape(context.applied_referral_code)}</code>\n"
f"Скидка перед оплатой: <b>{max(context.referral_discount_percent, 0)}%</b>"
"</blockquote>"
),
]
)
else:
lines.extend(
[
"",
(
"<i>"
f"Чтобы получить скидку {max(context.referral_discount_percent, 0)}%, "
"код нужно ввести вручную до оплаты."
"</i>"
),
]
)
if context.referral_link:
lines.extend(
[
"",
"<b>Ваша ссылка</b>",
f"<blockquote><code>{html.escape(context.referral_link)}</code></blockquote>",
]
)
else:
lines.extend(
[
"",
"<i>Для генерации ссылки укажите BOT_PUBLIC_USERNAME в .env.</i>",
]
)
if context.referral_recent_names:
lines.extend(
[
"",
"<b>Последние приглашения</b>",
"<blockquote>" + "\n".join(html.escape(name) for name in context.referral_recent_names) + "</blockquote>",
]
)
lines.extend(
[
"",
"<i>Переход по ссылке сам по себе не активирует скидку. Код нужно ввести вручную перед оплатой.</i>",
]
)
return "\n".join(lines)
def _render_terms_caption() -> str:
return "\n".join(
[
"<b>📜 Правила VPN</b>",
(
"<blockquote>"
"1. Используя сервис и бота, вы подтверждаете согласие с правилами платформы и обязуетесь использовать VPN только в законных целях.\n"
"2. VPN предназначен для защиты данных, приватности и безопасного доступа к интернет-ресурсам.\n"
"3. Запрещены передача доступа третьим лицам, спам, взлом, мошенничество, распространение вредоносного ПО, атаки и любой незаконный контент.\n"
"4. Пользователь самостоятельно несёт ответственность за свои действия при использовании сервиса.\n"
"5. Для стабильной работы могут обрабатываться технические данные подключения без передачи третьим лицам, кроме случаев, предусмотренных законодательством.\n"
"6. При нарушении правил доступ может быть ограничен или приостановлен без возврата средств.\n\n"
"⚠️ VPN повышает уровень конфиденциальности, но не гарантирует полной анонимности."
"</blockquote>"
),
]
)
def _render_support_caption() -> str:
return "\n".join(
[
"<b>🆘 Помощь и сопровождение</b>",
(
"<blockquote>"
"Здесь вы можете обратиться в поддержку по вопросам доступа, оплаты, подключения и работы сервиса.\n\n"
"Если нужна помощь, откройте тикет через кнопку ниже."
"</blockquote>"
),
]
)
def _render_support_ticket_caption() -> str:
return "\n".join(
[
"<b>🎫 Создание тикета</b>",
(
"<blockquote>"
"Опишите проблему как можно подробнее: устройство, приложение, текст ошибки и что именно не работает.\n\n"
"После нажатия кнопки ниже бот переведёт вас в режим создания тикета, и следующее сообщение будет отправлено в поддержку."
"</blockquote>"
),
]
)