- FastAPI REST API with JWT auth - aiogram 3 Telegram bot with admin middleware - APScheduler daily tasks (expiry, reminders, revoke, sync) - SQLAlchemy 2 async ORM with Alembic migrations - Jinja2 admin panel (Dashboard, Users, Payments, Servers, Tariffs) - VPN provider abstraction with MockProvider - Stats service with revenue/subscription analytics - Docker Compose (PostgreSQL + Redis + app) - Healthcheck endpoint
107 lines
3.2 KiB
Python
107 lines
3.2 KiB
Python
START = (
|
|
"🔐 VPN Control Panel Bot\n\n"
|
|
"Manage users, subscriptions, and servers.\n"
|
|
"Use /help to see available commands."
|
|
)
|
|
|
|
HELP = (
|
|
"📋 Available commands:\n\n"
|
|
"👤 Users\n"
|
|
"/users - list all active users\n"
|
|
"/user <id> - show user details\n"
|
|
"/delete <id> - deactivate user\n\n"
|
|
"📅 Subscriptions\n"
|
|
"/renew <id> - check subscription info\n"
|
|
"/expired - list expired subscriptions\n"
|
|
"/expiring [days] - list expiring in N days\n\n"
|
|
"📊 System\n"
|
|
"/stats - system statistics\n"
|
|
"/help - this message"
|
|
)
|
|
|
|
USER_NOT_FOUND = "❌ User <code>{}</code> not found."
|
|
ADMIN_ONLY = "⛔ Access denied. Admins only."
|
|
INVALID_ARGS = "⚠️ Invalid arguments. Usage: <code>{}</code>"
|
|
ERROR = "❌ Error: {}"
|
|
NO_DATA = "📭 No data."
|
|
|
|
|
|
def user_info(
|
|
user_id: int,
|
|
telegram_id: int,
|
|
username: str | None,
|
|
full_name: str,
|
|
is_active: bool,
|
|
created_at: str,
|
|
subscription_active: bool,
|
|
tariff_name: str | None,
|
|
remaining_days: int,
|
|
) -> str:
|
|
status = "✅ Active" if is_active else "❌ Deactivated"
|
|
sub_status = "✅ Active" if subscription_active else "❌ Inactive"
|
|
lines = [
|
|
f"👤 <b>User #{user_id}</b>",
|
|
f"📍 Telegram ID: <code>{telegram_id}</code>",
|
|
f"👋 Username: @{username or '—'}",
|
|
f"📛 Name: {full_name}",
|
|
f"🔵 Status: {status}",
|
|
f"📅 Registered: {created_at}",
|
|
"",
|
|
f"📦 <b>Subscription</b>",
|
|
f"Status: {sub_status}",
|
|
]
|
|
if tariff_name:
|
|
lines.append(f"Plan: {tariff_name}")
|
|
if subscription_active:
|
|
lines.append(f"Days left: {remaining_days}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def user_short(user_id: int, telegram_id: int, username: str | None) -> str:
|
|
name = f"@{username}" if username else f"<code>{telegram_id}</code>"
|
|
return f" #{user_id} {name}"
|
|
|
|
|
|
def expired_list(items: list[dict]) -> str:
|
|
if not items:
|
|
return "✅ No expired subscriptions."
|
|
lines = ["⚠️ <b>Expired subscriptions:</b>", ""]
|
|
for item in items:
|
|
lines.append(
|
|
f" #{item['user_id']} tg:{item['telegram_id']} — "
|
|
f"expired {item['tariff_name']}"
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
|
|
def expiring_list(items: list[dict], days: int) -> str:
|
|
if not items:
|
|
return f"✅ No subscriptions expiring within {days} days."
|
|
lines = [f"⚠️ <b>Expiring within {days} days:</b>", ""]
|
|
for item in items:
|
|
lines.append(
|
|
f" #{item['user_id']} tg:{item['telegram_id']} — "
|
|
f"{item['remaining_days']}d left ({item['tariff_name']})"
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
|
|
def stats_text(
|
|
total_users: int,
|
|
active_users: int,
|
|
total_tariffs: int,
|
|
total_servers: int,
|
|
active_servers: int,
|
|
total_payments: int,
|
|
confirmed_payments: int,
|
|
total_revenue: float,
|
|
) -> str:
|
|
return (
|
|
"📊 <b>System Statistics</b>\n\n"
|
|
f"👥 Users: {active_users} / {total_users} active\n"
|
|
f"📦 Tariffs: {total_tariffs}\n"
|
|
f"🖥️ Servers: {active_servers} / {total_servers} active\n"
|
|
f"💳 Payments: {confirmed_payments} / {total_payments} confirmed\n"
|
|
f"💰 Revenue: {total_revenue:.2f} RUB"
|
|
)
|