diff --git a/api/routes/base_crud.py b/api/routes/base_crud.py index 7b00563d..ed6d7b08 100644 --- a/api/routes/base_crud.py +++ b/api/routes/base_crud.py @@ -1,5 +1,3 @@ -from typing import Any, Type, Union - from fastapi import APIRouter, Depends, HTTPException, Path, Query from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession diff --git a/api/routes/keys.py b/api/routes/keys.py index 729cc242..54614714 100644 --- a/api/routes/keys.py +++ b/api/routes/keys.py @@ -8,7 +8,7 @@ from api.depends import get_session, verify_admin_token from api.routes.base_crud import generate_crud_router from api.schemas.keys import KeyBase, KeyCreateRequest, KeyResponse, KeyUpdate from database.models import Admin, Key, Tariff -from handlers.keys.key_utils import create_key_on_cluster, delete_key_from_cluster, renew_key_in_cluster +from handlers.keys.operations import create_key_on_cluster, delete_key_from_cluster, renew_key_in_cluster from logger import logger diff --git a/api/routes/misc.py b/api/routes/misc.py index 9b6295da..61de5822 100644 --- a/api/routes/misc.py +++ b/api/routes/misc.py @@ -13,6 +13,7 @@ from api.schemas import ( TemporaryDataResponse, TrackingSourceResponse, ) +from database import get_tracking_source_stats from database.models import ( Admin, BlockedUser, @@ -133,9 +134,36 @@ router.include_router( schema_create=None, schema_update=None, identifier_field="id", - enabled_methods=["get_all", "get_one", "delete"], + enabled_methods=["get_all", "delete"], ), prefix="/tracking-sources", tags=["TrackingSources"], dependencies=[Depends(verify_admin_token)], ) + + +@router.get( + "/tracking-sources/{code}", response_model=TrackingSourceResponse, dependencies=[Depends(verify_admin_token)] +) +async def get_tracking_source_with_stats( + code: str, + session: AsyncSession = Depends(get_session), +): + result = await session.execute(select(TrackingSource).where(TrackingSource.code == code)) + source = result.scalar_one_or_none() + if not source: + raise HTTPException(status_code=404, detail="Tracking source not found") + + stats = await get_tracking_source_stats(session, code) + + return TrackingSourceResponse( + id=source.id, + name=source.name, + code=source.code, + type=source.type, + created_by=source.created_by, + created_at=source.created_at, + registrations=stats.get("registrations", 0), + trials=stats.get("trials", 0), + payments=stats.get("payments", 0), + ) diff --git a/api/routes/users.py b/api/routes/users.py index 19ab005c..befb130e 100644 --- a/api/routes/users.py +++ b/api/routes/users.py @@ -9,7 +9,7 @@ from api.routes.base_crud import generate_crud_router from api.schemas.users import UserBase, UserResponse, UserUpdate from database import delete_user_data, get_servers from database.models import Key, User -from handlers.keys.key_utils import delete_key_from_cluster +from handlers.keys.operations import delete_key_from_cluster from logger import logger diff --git a/api/schemas/coupons.py b/api/schemas/coupons.py index f50bc943..b7bbbfbf 100644 --- a/api/schemas/coupons.py +++ b/api/schemas/coupons.py @@ -1,5 +1,4 @@ from datetime import datetime -from typing import Optional from pydantic import BaseModel, Field, model_validator diff --git a/api/schemas/gifts.py b/api/schemas/gifts.py index e47ea6cc..b6260c2b 100644 --- a/api/schemas/gifts.py +++ b/api/schemas/gifts.py @@ -1,5 +1,4 @@ from datetime import datetime -from typing import Optional from pydantic import BaseModel diff --git a/api/schemas/keys.py b/api/schemas/keys.py index d6f35609..5bfd38b4 100644 --- a/api/schemas/keys.py +++ b/api/schemas/keys.py @@ -1,5 +1,3 @@ -from typing import Optional - from pydantic import BaseModel, Field diff --git a/api/schemas/misc.py b/api/schemas/misc.py index 23995d20..f24f716c 100644 --- a/api/schemas/misc.py +++ b/api/schemas/misc.py @@ -102,5 +102,9 @@ class TrackingSourceResponse(BaseModel): created_by: int created_at: datetime + registrations: int = 0 + trials: int = 0 + payments: int = 0 + class Config: from_attributes = True diff --git a/api/schemas/servers.py b/api/schemas/servers.py index 5c8a5f36..99cee5af 100644 --- a/api/schemas/servers.py +++ b/api/schemas/servers.py @@ -1,5 +1,3 @@ -from typing import Optional - from pydantic import BaseModel diff --git a/api/schemas/tariffs.py b/api/schemas/tariffs.py index 519f9ec0..e1585958 100644 --- a/api/schemas/tariffs.py +++ b/api/schemas/tariffs.py @@ -1,5 +1,4 @@ from datetime import datetime -from typing import Optional from pydantic import BaseModel diff --git a/api/schemas/users.py b/api/schemas/users.py index 659b550b..6638a4ed 100644 --- a/api/schemas/users.py +++ b/api/schemas/users.py @@ -1,5 +1,4 @@ from datetime import datetime -from typing import Optional from pydantic import BaseModel diff --git a/bot.py b/bot.py index b566a39e..11994a8c 100644 --- a/bot.py +++ b/bot.py @@ -44,7 +44,7 @@ async def errors_handler(event: ErrorEvent, bot: Bot) -> bool: ): logger.warning("Отправляем стартовое меню.") try: - from handlers.start import handle_start_callback_query, start_command + from handlers.start import start_entry if event.update.message: fsm_context = dp.fsm.get_context( @@ -52,8 +52,8 @@ async def errors_handler(event: ErrorEvent, bot: Bot) -> bool: chat_id=event.update.message.chat.id, user_id=event.update.message.from_user.id, ) - await start_command( - event.update.message, + await start_entry( + event=event.update.message, state=fsm_context, session=None, admin=False, @@ -65,15 +65,15 @@ async def errors_handler(event: ErrorEvent, bot: Bot) -> bool: chat_id=event.update.callback_query.message.chat.id, user_id=event.update.callback_query.from_user.id, ) - await handle_start_callback_query( - event.update.callback_query, + await start_entry( + event=event.update.callback_query, state=fsm_context, session=None, admin=False, captcha=False, ) except Exception as e: - logger.error(f"Ошибка при показе стартового меню после ошибки: {e}") + logger.error(f"Ошибка при показе стартового меню после ошибки: {e}", exc_info=True) return True @@ -93,7 +93,7 @@ async def errors_handler(event: ErrorEvent, bot: Bot) -> bool: caption=f"{hbold(type(event.exception).__name__)}: {str(event.exception)[:1021]}...", ) - from handlers.start import handle_start_callback_query, start_command + from handlers.start import start_entry if event.update.message: fsm_context = dp.fsm.get_context( @@ -101,8 +101,8 @@ async def errors_handler(event: ErrorEvent, bot: Bot) -> bool: chat_id=event.update.message.chat.id, user_id=event.update.message.from_user.id, ) - await start_command( - event.update.message, + await start_entry( + event=event.update.message, state=fsm_context, session=None, admin=False, @@ -114,8 +114,8 @@ async def errors_handler(event: ErrorEvent, bot: Bot) -> bool: chat_id=event.update.callback_query.message.chat.id, user_id=event.update.callback_query.from_user.id, ) - await handle_start_callback_query( - event.update.callback_query, + await start_entry( + event=event.update.callback_query, state=fsm_context, session=None, admin=False, diff --git a/handlers/admin/bans/bans_handler.py b/handlers/admin/bans/bans_handler.py index 055a0601..a9728ef5 100644 --- a/handlers/admin/bans/bans_handler.py +++ b/handlers/admin/bans/bans_handler.py @@ -151,7 +151,7 @@ async def handle_preemptive_ban_start(callback: CallbackQuery, state: FSMContext await callback.message.edit_text( "📥 Отправьте список Telegram ID (один на строке), которых нужно заранее забанить (теневой бан).\n\n" "Пример:\n123456789\n987654321", - reply_markup=build_admin_back_kb("bans") + reply_markup=build_admin_back_kb("bans"), ) diff --git a/handlers/admin/bans/keyboard.py b/handlers/admin/bans/keyboard.py index ab6a9d75..7c3323a3 100644 --- a/handlers/admin/bans/keyboard.py +++ b/handlers/admin/bans/keyboard.py @@ -1,7 +1,6 @@ -from aiogram.types import InlineKeyboardMarkup from aiogram.utils.keyboard import InlineKeyboardBuilder -from ..panel.keyboard import AdminPanelCallback, build_admin_back_btn +from ..panel.keyboard import AdminPanelCallback def build_bans_kb(): diff --git a/handlers/admin/clusters/clusters_handler.py b/handlers/admin/clusters/clusters_handler.py index b1a21665..b882f633 100644 --- a/handlers/admin/clusters/clusters_handler.py +++ b/handlers/admin/clusters/clusters_handler.py @@ -21,7 +21,7 @@ from config import ( from database import check_unique_server_name, get_servers, update_key_expiry from database.models import Key, Server, Tariff from filters.admin import IsAdminFilter -from handlers.keys.key_utils import ( +from handlers.keys.operations import ( create_client_on_server, create_key_on_cluster, delete_key_from_cluster, @@ -745,13 +745,15 @@ async def handle_days_input(message: Message, state: FSMContext, session: AsyncS device_limit = 0 if key.tariff_id: result = await session.execute( - select(Tariff.traffic_limit, Tariff.device_limit).where(Tariff.id == key.tariff_id, Tariff.is_active.is_(True)) + select(Tariff.traffic_limit, Tariff.device_limit).where( + Tariff.id == key.tariff_id, Tariff.is_active.is_(True) + ) ) tariff = result.first() if tariff: traffic_limit = int(tariff[0]) if tariff[0] is not None else 0 device_limit = int(tariff[1]) if tariff[1] is not None else 0 - + await renew_key_in_cluster( cluster_name, email=key.email, diff --git a/handlers/admin/coupons/keyboard.py b/handlers/admin/coupons/keyboard.py index 807b4f79..dc6b9743 100644 --- a/handlers/admin/coupons/keyboard.py +++ b/handlers/admin/coupons/keyboard.py @@ -1,5 +1,3 @@ -from typing import Optional - from aiogram.filters.callback_data import CallbackData from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup from aiogram.utils.keyboard import InlineKeyboardBuilder diff --git a/handlers/admin/gifts/keyboard.py b/handlers/admin/gifts/keyboard.py index 61f34608..ab95e6ce 100644 --- a/handlers/admin/gifts/keyboard.py +++ b/handlers/admin/gifts/keyboard.py @@ -3,7 +3,7 @@ from aiogram.utils.keyboard import InlineKeyboardBuilder from database.models import Gift, Tariff from handlers.buttons import BACK -from handlers.utils import format_days, format_months, get_plural_form +from handlers.utils import format_days, format_months from ..panel.keyboard import AdminPanelCallback diff --git a/handlers/admin/management/management_handler.py b/handlers/admin/management/management_handler.py index de53cdd4..b7aa138b 100644 --- a/handlers/admin/management/management_handler.py +++ b/handlers/admin/management/management_handler.py @@ -22,7 +22,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from config import DB_NAME, DB_PASSWORD, DB_USER, PG_HOST, PG_PORT, REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD from database.models import Admin, Key, Server, User from filters.admin import IsAdminFilter -from handlers.keys.key_utils import update_subscription +from handlers.keys.operations import update_subscription from logger import logger from middlewares import maintenance from panels.remnawave import RemnawaveAPI diff --git a/handlers/admin/panel/keyboard.py b/handlers/admin/panel/keyboard.py index ed57b1ab..4472cdf2 100644 --- a/handlers/admin/panel/keyboard.py +++ b/handlers/admin/panel/keyboard.py @@ -5,8 +5,8 @@ from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup from aiogram.utils.keyboard import InlineKeyboardBuilder from handlers.buttons import BACK, MAIN_MENU +from hooks.hook_buttons import insert_hook_buttons from hooks.hooks import run_hooks -from logger import logger class AdminPanelCallback(CallbackData, prefix="admin_panel"): @@ -21,6 +21,7 @@ class AdminPanelCallback(CallbackData, prefix="admin_panel"): async def build_panel_kb(admin_role: str) -> InlineKeyboardMarkup: builder = InlineKeyboardBuilder() + builder.button( text="👤 Поиск пользователя", callback_data=AdminPanelCallback(action="search_user").pack(), @@ -29,15 +30,18 @@ async def build_panel_kb(admin_role: str) -> InlineKeyboardMarkup: text="🔑 Поиск по подписке", callback_data=AdminPanelCallback(action="search_key").pack(), ) + if admin_role == "superadmin": builder.button( text="🖥️ Управление серверами", callback_data=AdminPanelCallback(action="clusters").pack(), ) + builder.row( InlineKeyboardButton(text="📢 Рассылка", callback_data=AdminPanelCallback(action="sender").pack()), InlineKeyboardButton(text="🎟️ Купоны", callback_data=AdminPanelCallback(action="coupons").pack()), ) + if admin_role == "superadmin": builder.row( InlineKeyboardButton(text="💸 Тарифы", callback_data=AdminPanelCallback(action="tariffs").pack()), @@ -48,14 +52,8 @@ async def build_panel_kb(admin_role: str) -> InlineKeyboardMarkup: callback_data=AdminPanelCallback(action="management").pack(), ) builder.row( - InlineKeyboardButton( - text="📊 Статистика", - callback_data=AdminPanelCallback(action="stats").pack(), - ), - InlineKeyboardButton( - text="📈 Аналитика", - callback_data=AdminPanelCallback(action="ads").pack(), - ), + InlineKeyboardButton(text="📊 Статистика", callback_data=AdminPanelCallback(action="stats").pack()), + InlineKeyboardButton(text="📈 Аналитика", callback_data=AdminPanelCallback(action="ads").pack()), ) else: builder.button( @@ -64,40 +62,7 @@ async def build_panel_kb(admin_role: str) -> InlineKeyboardMarkup: ) module_buttons = await run_hooks("admin_panel", admin_role=admin_role) - - for module_btn in module_buttons: - if isinstance(module_btn, dict) and "after" in module_btn: - after_callback = module_btn["after"] - insert_pos = -1 - - current_markup = builder.as_markup() - - for i, row in enumerate(current_markup.inline_keyboard): - for btn in row: - if btn.callback_data == after_callback: - insert_pos = i + 1 - break - if insert_pos > 0: - break - - if insert_pos > 0: - new_buttons = [] - for i, row in enumerate(current_markup.inline_keyboard): - if i == insert_pos: - new_buttons.append([module_btn["button"]]) - new_buttons.append(row) - - if insert_pos >= len(current_markup.inline_keyboard): - new_buttons.append([module_btn["button"]]) - - builder = InlineKeyboardBuilder.from_markup(InlineKeyboardMarkup(inline_keyboard=new_buttons)) - else: - builder.row(module_btn["button"]) - else: - if isinstance(module_btn, dict): - builder.row(module_btn["button"]) - else: - builder.row(module_btn) + builder = insert_hook_buttons(builder, module_buttons) builder.button( text=MAIN_MENU, @@ -108,7 +73,7 @@ async def build_panel_kb(admin_role: str) -> InlineKeyboardMarkup: builder.adjust(1, 1, 1, 1, 2, 2, 1, 2, 1) else: builder.adjust(1, 1, 1, 2, 1, 1) - + return builder.as_markup() diff --git a/handlers/admin/restart/restart_handler.py b/handlers/admin/restart/restart_handler.py index 218ee2f6..95dd6a4c 100644 --- a/handlers/admin/restart/restart_handler.py +++ b/handlers/admin/restart/restart_handler.py @@ -10,7 +10,7 @@ from aiogram.types import CallbackQuery from filters.admin import IsAdminFilter -from ..panel.keyboard import AdminPanelCallback, build_admin_back_kb, build_restart_kb +from ..panel.keyboard import AdminPanelCallback, build_admin_back_kb router = Router() diff --git a/handlers/admin/users/users_handler.py b/handlers/admin/users/users_handler.py index 9edc0679..622b14af 100644 --- a/handlers/admin/users/users_handler.py +++ b/handlers/admin/users/users_handler.py @@ -33,7 +33,7 @@ from database import ( ) from database.models import Key, ManualBan, Payment, Referral, Server, Tariff, User from filters.admin import IsAdminFilter -from handlers.keys.key_utils import ( +from handlers.keys.operations import ( create_key_on_cluster, delete_key_from_cluster, get_user_traffic, diff --git a/handlers/coupons.py b/handlers/coupons.py index ea030db2..b07b4f10 100644 --- a/handlers/coupons.py +++ b/handlers/coupons.py @@ -28,7 +28,7 @@ from database import ( update_key_expiry, ) from handlers.buttons import MAIN_MENU -from handlers.keys.key_utils import renew_key_in_cluster +from handlers.keys.operations import renew_key_in_cluster from handlers.profile import process_callback_view_profile from handlers.texts import ( COUPONS_DAYS_MESSAGE, diff --git a/handlers/keys/key_freeze.py b/handlers/keys/key_freeze.py index a0bad251..8a680a15 100644 --- a/handlers/keys/key_freeze.py +++ b/handlers/keys/key_freeze.py @@ -5,7 +5,6 @@ from typing import Any from aiogram import F, Router from aiogram.types import CallbackQuery, InlineKeyboardButton from aiogram.utils.keyboard import InlineKeyboardBuilder -from sqlalchemy import text from database import ( get_key_details, @@ -15,7 +14,7 @@ from database import ( mark_key_as_unfrozen, ) from handlers.buttons import APPLY, BACK, CANCEL -from handlers.keys.key_utils import renew_key_in_cluster, toggle_client_on_cluster +from handlers.keys.operations import renew_key_in_cluster, toggle_client_on_cluster from handlers.texts import ( FREEZE_SUBSCRIPTION_CONFIRM_MSG, SUBSCRIPTION_FROZEN_MSG, diff --git a/handlers/keys/key_mode/key_cluster_mode.py b/handlers/keys/key_mode/key_cluster_mode.py index 7a168e29..60aebc26 100644 --- a/handlers/keys/key_mode/key_cluster_mode.py +++ b/handlers/keys/key_mode/key_cluster_mode.py @@ -32,12 +32,10 @@ from handlers.buttons import ( SUPPORT, TV_BUTTON, ) -from handlers.keys.key_utils import create_key_on_cluster +from handlers.keys.operations import create_key_on_cluster from handlers.texts import key_message_success from handlers.utils import ( edit_or_send_message, - format_days, - format_months, generate_random_email, get_least_loaded_cluster, is_full_remnawave_cluster, diff --git a/handlers/keys/key_mode/key_country_mode.py b/handlers/keys/key_mode/key_country_mode.py index 3ea30328..b2c3713d 100644 --- a/handlers/keys/key_mode/key_country_mode.py +++ b/handlers/keys/key_mode/key_country_mode.py @@ -30,19 +30,16 @@ from database import ( check_server_name_by_cluster, check_user_exists, get_key_details, - get_tariff_by_id, get_trial, update_balance, update_trial, ) from database.models import Key, Server, Tariff from handlers.buttons import BACK, CONNECT_DEVICE, CONNECT_PHONE, MAIN_MENU, MY_SUB, PC_BUTTON, SUPPORT, TV_BUTTON -from handlers.keys.key_utils import create_client_on_server +from handlers.keys.operations import create_client_on_server from handlers.texts import SELECT_COUNTRY_MSG, key_message_success from handlers.utils import ( edit_or_send_message, - format_days, - format_months, generate_random_email, get_least_loaded_cluster, is_full_remnawave_cluster, diff --git a/handlers/keys/key_mode/key_create.py b/handlers/keys/key_mode/key_create.py index cce313fa..06b52532 100644 --- a/handlers/keys/key_mode/key_create.py +++ b/handlers/keys/key_mode/key_create.py @@ -10,7 +10,6 @@ from aiogram.fsm.context import FSMContext from aiogram.types import CallbackQuery, InlineKeyboardButton, Message from aiogram.utils.keyboard import InlineKeyboardBuilder from sqlalchemy import select -from sqlalchemy.ext.asyncio import AsyncSession from config import ( NOTIFY_EXTRA_DAYS, @@ -33,6 +32,7 @@ from handlers.admin.panel.keyboard import AdminPanelCallback from handlers.buttons import MAIN_MENU, PAYMENT from handlers.payments.robokassa_pay import handle_custom_amount_input from handlers.payments.stars_pay import process_custom_amount_input_stars +from handlers.payments.wata import handle_custom_amount_input as handle_custom_amount_input_wata from handlers.payments.yookassa_pay import process_custom_amount_input from handlers.payments.yoomoney_pay import process_custom_amount_input_yoomoney from handlers.texts import ( @@ -321,6 +321,10 @@ async def select_tariff_plan(callback_query: CallbackQuery, session: Any, state: await process_custom_amount_input_stars(callback_query, session) elif USE_NEW_PAYMENT_FLOW == "YOOMONEY": await process_custom_amount_input_yoomoney(callback_query, session) + elif USE_NEW_PAYMENT_FLOW == "WATA": + await state.update_data(wata_cassa="sbp", required_amount=required_amount) + await handle_custom_amount_input_wata(callback_query, state) + else: builder = InlineKeyboardBuilder() builder.row(InlineKeyboardButton(text=PAYMENT, callback_data="pay")) diff --git a/handlers/keys/key_renew.py b/handlers/keys/key_renew.py index b929912d..9ca625c3 100644 --- a/handlers/keys/key_renew.py +++ b/handlers/keys/key_renew.py @@ -28,9 +28,10 @@ from database import ( from database.models import Key, Server from database.tariffs import create_subgroup_hash, find_subgroup_by_hash from handlers.buttons import BACK, MAIN_MENU, MY_SUB, PAYMENT -from handlers.keys.key_utils import renew_key_in_cluster +from handlers.keys.operations import renew_key_in_cluster from handlers.payments.robokassa_pay import handle_custom_amount_input from handlers.payments.stars_pay import process_custom_amount_input_stars +from handlers.payments.wata import handle_custom_amount_input as handle_custom_amount_input_wata from handlers.payments.yookassa_pay import process_custom_amount_input from handlers.payments.yoomoney_pay import process_custom_amount_input_yoomoney from handlers.texts import ( @@ -292,6 +293,9 @@ async def process_callback_renew_plan(callback_query: CallbackQuery, state: FSMC await process_custom_amount_input_stars(callback_query, session) elif USE_NEW_PAYMENT_FLOW == "YOOMONEY": await process_custom_amount_input_yoomoney(callback_query, session) + elif USE_NEW_PAYMENT_FLOW == "WATA": + await state.update_data(wata_cassa="sbp", required_amount=required_amount) + await handle_custom_amount_input_wata(callback_query, state) else: builder = InlineKeyboardBuilder() builder.row(InlineKeyboardButton(text=PAYMENT, callback_data="pay")) diff --git a/handlers/keys/key_utils.py b/handlers/keys/key_utils.py deleted file mode 100644 index 6a0f7d19..00000000 --- a/handlers/keys/key_utils.py +++ /dev/null @@ -1,987 +0,0 @@ -import asyncio - -from datetime import datetime, timezone -from typing import Any - -from sqlalchemy import delete, select, update -from sqlalchemy.ext.asyncio import AsyncSession - -from config import PUBLIC_LINK, REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD, SUPERNODE -from database import delete_notification, get_servers, get_tariff_by_id, store_key -from database.models import Key, Server, Tariff -from handlers.utils import check_server_key_limit, get_least_loaded_cluster -from logger import logger -from panels.remnawave import RemnawaveAPI -from panels.three_xui import ( - ClientConfig, - add_client, - delete_client, - extend_client_key, - get_client_traffic, - get_xui_instance, - toggle_client, -) - - -async def create_key_on_cluster( - cluster_id: str, - tg_id: int, - client_id: str, - email: str, - expiry_timestamp: int, - plan: int = None, - session: AsyncSession = None, - remnawave_link: str = None, - hwid_limit: int = None, - traffic_limit_bytes: int = None, - is_trial: bool = False, -): - try: - servers = await get_servers(session) - cluster = servers.get(cluster_id) - server_id_to_store = cluster_id - - if not cluster: - found_servers = [] - for _key, server_list in servers.items(): - for server_info in server_list: - if server_info.get("server_name", "").lower() == cluster_id.lower(): - found_servers.append(server_info) - if found_servers: - cluster = found_servers - server_id_to_store = found_servers[0].get("server_name") - else: - raise ValueError(f"Кластер или сервер с ID/именем {cluster_id} не найден.") - - enabled_servers = [s for s in cluster if s.get("enabled", True)] - if not enabled_servers: - logger.warning(f"[Key Creation] Нет доступных серверов в кластере {cluster_id}") - return - - if plan is not None and traffic_limit_bytes is None: - tariff = await get_tariff_by_id(session, plan) - if not tariff: - raise ValueError(f"Тариф с id={plan} не найден.") - traffic_limit_bytes = int(tariff["traffic_limit"]) if tariff["traffic_limit"] else None - if hwid_limit is None and tariff.get("device_limit") is not None: - hwid_limit = int(tariff["device_limit"]) - - remnawave_servers = [ - s - for s in enabled_servers - if s.get("panel_type", "3x-ui").lower() == "remnawave" and await check_server_key_limit(s, session) - ] - xui_servers = [ - s - for s in enabled_servers - if s.get("panel_type", "3x-ui").lower() == "3x-ui" and await check_server_key_limit(s, session) - ] - - if not remnawave_servers and not xui_servers: - logger.warning(f"[Key Creation] Нет серверов с доступным лимитом в кластере {cluster_id}") - return - - semaphore = asyncio.Semaphore(2) - remnawave_created = False - remnawave_key = None - remnawave_client_id = None - - if remnawave_servers: - remna = RemnawaveAPI(remnawave_servers[0]["api_url"]) - logged_in = await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD) - if not logged_in: - logger.error("Не удалось войти в Remnawave API") - else: - expire_at = datetime.utcfromtimestamp(expiry_timestamp / 1000).isoformat() + "Z" - inbound_ids = [s.get("inbound_id") for s in remnawave_servers if s.get("inbound_id")] - - if not inbound_ids: - logger.warning("Нет inbound_id у серверов Remnawave") - else: - short_uuid = None - if remnawave_link and "/" in remnawave_link: - short_uuid = remnawave_link.rstrip("/").split("/")[-1] - - user_data = { - "username": email, - "trafficLimitStrategy": "NO_RESET", - "expireAt": expire_at, - "telegramId": tg_id, - "activeInternalSquads": inbound_ids, - } - - if traffic_limit_bytes and traffic_limit_bytes > 0: - user_data["trafficLimitBytes"] = traffic_limit_bytes * 1024 * 1024 * 1024 - - if short_uuid: - user_data["shortUuid"] = short_uuid - if hwid_limit is not None: - user_data["hwidDeviceLimit"] = hwid_limit - logger.info(f"[Key Creation] Данные для создания клиента в Remnawave: {user_data}") - - result = await remna.create_user(user_data) - if not result: - logger.error("Ошибка при создании пользователя в Remnawave") - else: - remnawave_created = True - remnawave_key = result.get("subscriptionUrl") - remnawave_client_id = result.get("uuid") - logger.info(f"[Key Creation] Пользователь создан в Remnawave: {result}") - - public_link = f"{PUBLIC_LINK}{email}/{tg_id}" if xui_servers else None - final_client_id = remnawave_client_id or client_id - logger.info(f"[Debug] 3x-ui servers для кластера {cluster_id}: {[s['server_name'] for s in xui_servers]}") - - if xui_servers: - if SUPERNODE: - for server_info in xui_servers: - await create_client_on_server( - server_info, - tg_id, - final_client_id, - email, - expiry_timestamp, - semaphore, - plan=plan, - session=session, - is_trial=is_trial, - ) - else: - await asyncio.gather( - *[ - create_client_on_server( - server, - tg_id, - final_client_id, - email, - expiry_timestamp, - semaphore, - plan=plan, - session=session, - is_trial=is_trial, - ) - for server in xui_servers - ], - return_exceptions=True, - ) - - if (remnawave_created and remnawave_client_id) or xui_servers: - await store_key( - session=session, - tg_id=tg_id, - client_id=final_client_id, - email=email, - expiry_time=expiry_timestamp, - key=public_link, - server_id=server_id_to_store, - remnawave_link=remnawave_key, - tariff_id=plan, - ) - - except Exception as e: - logger.error(f"Ошибка при создании ключа: {e}") - raise e - - -async def create_client_on_server( - server_info: dict, - tg_id: int, - client_id: str, - email: str, - expiry_timestamp: int, - semaphore: asyncio.Semaphore, - plan: int = None, - session=None, - is_trial: bool = False, -): - """ - Создает клиента на указанном 3x-ui сервере с лимитом по тарифу или триалу. - """ - logger.info( - f"[Client] Вход в create_client_on_server: сервер={server_info.get('server_name')}, план={plan}, is_trial={is_trial}" - ) - - async with semaphore: - xui = await get_xui_instance(server_info["api_url"]) - inbound_id = server_info.get("inbound_id") - server_name = server_info.get("server_name", "unknown") - - if not inbound_id: - logger.warning(f"[Client] INBOUND_ID отсутствует для сервера {server_name}. Пропуск.") - return - - if SUPERNODE: - unique_email = f"{email}_{server_name.lower()}" - sub_id = email - else: - unique_email = email - sub_id = unique_email - - total_gb_value = 0 - device_limit_value = 0 - - if plan is not None: - tariff = await get_tariff_by_id(session, plan) - logger.info(f"[Tariff Debug] Получен тариф: {tariff}") - if not tariff: - raise ValueError(f"Тариф с id={plan} не найден.") - - total_gb_value = int(tariff["traffic_limit"]) if tariff["traffic_limit"] else 0 - device_limit_value = int(tariff["device_limit"]) if tariff.get("device_limit") is not None else 0 - - try: - logger.info( - f"[Client] Вызов add_client: email={email}, client_id={client_id}, GB={total_gb_value}, Devices={device_limit_value}" - ) - traffic_limit_bytes = total_gb_value * 1024 * 1024 * 1024 - await add_client( - xui, - ClientConfig( - client_id=client_id, - email=unique_email, - tg_id=tg_id, - limit_ip=device_limit_value, - total_gb=traffic_limit_bytes, - expiry_time=expiry_timestamp, - enable=True, - flow="xtls-rprx-vision", - inbound_id=int(inbound_id), - sub_id=sub_id, - ), - ) - logger.info(f"[Client] Клиент успешно добавлен на сервер {server_name}") - except Exception as e: - logger.error(f"[Client Error] Не удалось создать клиента на {server_name}: {e}") - - if SUPERNODE: - await asyncio.sleep(0.7) - - -async def renew_key_in_cluster( - cluster_id: str, - email: str, - client_id: str, - new_expiry_time: int, - total_gb: int, - session: AsyncSession, - hwid_device_limit: int = 0, - reset_traffic: bool = True, -): - try: - servers = await get_servers(session) - cluster = servers.get(cluster_id) - - if not cluster: - found_servers = [] - for _key, server_list in servers.items(): - for server_info in server_list: - if server_info.get("server_name", "").lower() == cluster_id.lower(): - found_servers.append(server_info) - if found_servers: - cluster = found_servers - else: - raise ValueError(f"Кластер или сервер с ID/именем {cluster_id} не найден.") - - result = await session.execute(select(Key.tg_id, Key.server_id).where(Key.client_id == client_id).limit(1)) - row = result.first() - if not row: - logger.error(f"Не найден пользователь с client_id={client_id} в таблице keys.") - return False - - tg_id, server_id = row - - result = await session.execute(select(Server.tariff_group).where(Server.server_name == server_id)) - tariff_group_row = result.scalar_one_or_none() - - if tariff_group_row: - result = await session.execute( - select(Tariff) - .where(Tariff.group_code == tariff_group_row, Tariff.is_active.is_(True)) - .order_by(Tariff.duration_days.desc()) - .limit(1) - ) - tariff = result.scalar_one_or_none() - if tariff and tariff.device_limit is not None: - hwid_device_limit = int(tariff.device_limit) - - remnawave_inbound_ids = [] - tasks = [] - for server_info in cluster: - if server_info.get("panel_type", "3x-ui").lower() == "remnawave": - inbound_id = server_info.get("inbound_id") - if inbound_id: - remnawave_inbound_ids.append(inbound_id) - - if remnawave_inbound_ids: - remnawave_server = next( - ( - s - for s in cluster - if s.get("panel_type", "").lower() == "remnawave" and s.get("inbound_id") in remnawave_inbound_ids - ), - None, - ) - if remnawave_server: - remna = RemnawaveAPI(remnawave_server["api_url"]) - if await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD): - expire_iso = datetime.utcfromtimestamp(new_expiry_time // 1000).isoformat() + "Z" - traffic_limit_bytes = total_gb * 1024 * 1024 * 1024 if total_gb else 0 - updated = await remna.update_user( - uuid=client_id, - expire_at=expire_iso, - active_user_inbounds=remnawave_inbound_ids, - traffic_limit_bytes=traffic_limit_bytes, - hwid_device_limit=hwid_device_limit, - ) - if updated: - logger.info(f"Подписка Remnawave {client_id} успешно продлена") - if reset_traffic: - await remna.reset_user_traffic(client_id) - else: - logger.warning(f"Не удалось продлить подписку Remnawave {client_id}, пробуем создать") - result = await session.execute( - select(Key.remnawave_link, Key.key).where(Key.client_id == client_id) - ) - row = result.one_or_none() - remnawave_link = row[0] if row else None - row[1] if row else None - - user_data = { - "username": email, - "trafficLimitStrategy": "NO_RESET", - "expireAt": expire_iso, - "telegramId": tg_id, - "activeInternalSquads": remnawave_inbound_ids, - } - if remnawave_link and "/" in remnawave_link: - user_data["shortUuid"] = remnawave_link.rstrip("/").split("/")[-1] - if traffic_limit_bytes and traffic_limit_bytes > 0: - user_data["trafficLimitBytes"] = traffic_limit_bytes - if hwid_device_limit is not None: - user_data["hwidDeviceLimit"] = hwid_device_limit - - result = await remna.create_user(user_data) - if result: - new_client_id = result.get("uuid") - new_remnawave_link = result.get("subscriptionUrl") - logger.info(f"Пользователь Remnawave {client_id} успешно создан") - - await session.execute( - update(Key) - .where(Key.client_id == client_id) - .values(client_id=new_client_id, remnawave_link=new_remnawave_link) - ) - await session.commit() - else: - logger.error(f"Не удалось создать пользователя Remnawave {client_id}") - else: - logger.error("Не удалось войти в Remnawave API") - - for server_info in cluster: - if server_info.get("panel_type", "3x-ui").lower() != "3x-ui": - continue - - xui = await get_xui_instance(server_info["api_url"]) - inbound_id = server_info.get("inbound_id") - server_name = server_info.get("server_name", "unknown") - - if not inbound_id: - logger.warning(f"INBOUND_ID отсутствует для сервера {server_name}. Пропуск.") - continue - - if SUPERNODE: - unique_email = f"{email}_{server_name.lower()}" - sub_id = email - else: - unique_email = email - sub_id = unique_email - - traffic_bytes = total_gb * 1024 * 1024 * 1024 if total_gb else 0 - - async def update_or_create_client(xui, inbound_id, unique_email, sub_id, server_name): - updated = await extend_client_key( - xui=xui, - inbound_id=int(inbound_id), - email=unique_email, - new_expiry_time=new_expiry_time, - client_id=client_id, - total_gb=traffic_bytes, - sub_id=sub_id, - tg_id=tg_id, - limit_ip=hwid_device_limit, - ) - - if not updated: - logger.warning(f"Не удалось обновить клиента {unique_email}, пробуем создать") - config = ClientConfig( - client_id=client_id, - email=unique_email, - tg_id=tg_id, - limit_ip=hwid_device_limit if hwid_device_limit is not None else 0, - total_gb=traffic_bytes, - expiry_time=new_expiry_time, - enable=True, - flow="xtls-rprx-vision", - inbound_id=int(inbound_id), - sub_id=sub_id, - ) - await add_client(xui, config) - - tasks.append(update_or_create_client(xui, inbound_id, unique_email, sub_id, server_name)) - - await asyncio.gather(*tasks, return_exceptions=True) - - notification_prefixes = ["key_24h", "key_10h", "key_expired", "renew"] - for notif in notification_prefixes: - notification_id = f"{email}_{notif}" - await delete_notification(session, tg_id, notification_id) - logger.info(f"🧹 Уведомления для ключа {email} очищены при продлении.") - - except Exception as e: - logger.error(f"Не удалось продлить ключ {client_id} в кластере/на сервере {cluster_id}: {e}") - raise - - -async def delete_key_from_cluster(cluster_id: str, email: str, client_id: str, session: AsyncSession): - """Удаление ключа с серверов в кластере или с конкретного сервера""" - try: - servers = await get_servers(session) - cluster = servers.get(cluster_id) - - if not cluster: - found_servers = [] - for _, server_list in servers.items(): - for server_info in server_list: - if server_info.get("server_name", "").lower() == cluster_id.lower(): - found_servers.append(server_info) - - if found_servers: - cluster = found_servers - else: - raise ValueError(f"Кластер или сервер с ID/именем {cluster_id} не найден.") - - for server_info in cluster: - panel_type = server_info.get("panel_type", "3x-ui").lower() - server_name = server_info.get("server_name", "unknown") - - if panel_type == "remnawave": - remna = RemnawaveAPI(server_info["api_url"]) - if not await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD): - logger.error(f"[Remnawave] Не удалось войти на сервер {server_name}") - continue - - success = await remna.delete_user(client_id) - if success: - logger.info(f"[Remnawave] Клиент {client_id} успешно удалён с {server_name}") - else: - logger.warning(f"[Remnawave] Не удалось удалить клиента {client_id} с {server_name}") - - elif panel_type == "3x-ui": - xui = await get_xui_instance(server_info["api_url"]) - inbound_id = server_info.get("inbound_id") - - if not inbound_id: - logger.warning(f"[3x-ui] INBOUND_ID отсутствует на сервере {server_name}. Пропуск.") - continue - - await delete_client( - xui, - inbound_id=int(inbound_id), - email=email, - client_id=client_id, - ) - logger.info(f"[3x-ui] Клиент {client_id} удалён с сервера {server_name}") - - else: - logger.warning(f"[Unknown] Неизвестный тип панели '{panel_type}' для сервера {server_name}") - - except Exception as e: - logger.error(f"❌ Ошибка при удалении ключа {client_id} из кластера/сервера {cluster_id}: {e}") - raise - - -async def update_key_on_cluster( - tg_id: int, - client_id: str, - email: str, - expiry_time: int, - cluster_id: str, - session: AsyncSession, - traffic_limit: int = None, - device_limit: int = None, - remnawave_link: str = None, -): - """ - Пересоздаёт ключ на всех серверах указанного кластера (или сервера, если передано имя). - Работает с панелями 3x-ui и Remnawave. Возвращает кортеж: (новый client_id, remnawave ссылка или None). - """ - try: - servers = await get_servers(session) - cluster = servers.get(cluster_id) - - if not cluster: - found_servers = [] - for _key, server_list in servers.items(): - for server_info in server_list: - if server_info.get("server_name", "").lower() == cluster_id.lower(): - found_servers.append(server_info) - if found_servers: - cluster = found_servers - else: - raise ValueError(f"Кластер или сервер с ID/именем {cluster_id} не найден.") - - expire_iso = datetime.utcfromtimestamp(expiry_time / 1000).replace(tzinfo=timezone.utc).isoformat() - - remnawave_servers = [s for s in cluster if s.get("panel_type", "3x-ui").lower() == "remnawave"] - xui_servers = [s for s in cluster if s.get("panel_type", "3x-ui").lower() == "3x-ui"] - - remnawave_client_id = None - remnawave_key = None - - if remnawave_servers: - inbound_ids = [s["inbound_id"] for s in remnawave_servers if s.get("inbound_id")] - remna = RemnawaveAPI(remnawave_servers[0]["api_url"]) - if await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD): - await remna.delete_user(client_id) - - group_code = remnawave_servers[0].get("tariff_group") - if not group_code: - raise ValueError("У Remnawave-сервера отсутствует tariff_group") - - result = await session.execute( - select(Tariff) - .where(Tariff.group_code == group_code, Tariff.is_active.is_(True)) - .order_by(Tariff.duration_days.desc()) - .limit(1) - ) - result.scalar_one_or_none() - - short_uuid = None - if remnawave_link and "/" in remnawave_link: - short_uuid = remnawave_link.rstrip("/").split("/")[-1] - logger.info(f"[Update] Извлечен short_uuid из ссылки: {short_uuid}") - - user_data = { - "username": email, - "trafficLimitStrategy": "NO_RESET", - "expireAt": expire_iso, - "telegramId": tg_id, - "activeInternalSquads": inbound_ids, - } - if traffic_limit is not None: - user_data["trafficLimitBytes"] = traffic_limit * 1024**3 - if device_limit is not None: - user_data["hwidDeviceLimit"] = device_limit - if short_uuid: - user_data["shortUuid"] = short_uuid - logger.info(f"[Update] Добавлен short_uuid в user_data: {short_uuid}") - - result = await remna.create_user(user_data) - if result: - remnawave_client_id = result.get("uuid") - remnawave_key = result.get("subscriptionUrl") - logger.info(f"[Update] Remnawave: клиент заново создан, новый UUID: {remnawave_client_id}") - else: - logger.error("[Update] Ошибка создания Remnawave клиента") - else: - logger.error("[Update] Не удалось авторизоваться в Remnawave") - - if not remnawave_client_id: - logger.warning(f"[Update] Remnawave client_id не получен. Используется исходный: {client_id}") - remnawave_client_id = client_id - - tasks = [] - for server_info in xui_servers: - server_name = server_info.get("server_name", "unknown") - inbound_id = server_info.get("inbound_id") - - if not inbound_id: - logger.warning(f"[Update] INBOUND_ID отсутствует для сервера {server_name}. Пропуск.") - continue - - xui = await get_xui_instance(server_info["api_url"]) - - sub_id = email - unique_email = f"{email}_{server_name.lower()}" if SUPERNODE else email - - group_code = server_info.get("tariff_group") - if not group_code: - raise ValueError(f"У сервера {server_name} отсутствует tariff_group") - - result = await session.execute( - select(Tariff) - .where(Tariff.group_code == group_code, Tariff.is_active.is_(True)) - .order_by(Tariff.duration_days.desc()) - .limit(1) - ) - result.scalar_one_or_none() - - total_gb_bytes = int(traffic_limit * 1024**3) if traffic_limit is not None else 0 - device_limit_value = device_limit if device_limit is not None else 0 - - config = ClientConfig( - client_id=remnawave_client_id, - email=unique_email, - tg_id=tg_id, - limit_ip=device_limit_value, - total_gb=total_gb_bytes, - expiry_time=expiry_time, - enable=True, - flow="xtls-rprx-vision", - inbound_id=int(inbound_id), - sub_id=sub_id, - ) - - tasks.append(add_client(xui, config)) - - if tasks: - await asyncio.gather(*tasks, return_exceptions=True) - - logger.info(f"[Update] Ключ {remnawave_client_id} обновлён на всех серверах кластера {cluster_id}") - return remnawave_client_id, remnawave_key - - except Exception as e: - logger.error(f"[Update Error] Ошибка при обновлении ключа {client_id} на {cluster_id}: {e}") - raise - - -async def update_subscription( - tg_id: int, - email: str, - session: AsyncSession, - cluster_override: str = None, - country_override: str = None, - remnawave_link: str = None, -) -> None: - result = await session.execute(select(Key).where(Key.tg_id == tg_id, Key.email == email)) - record = result.scalar_one_or_none() - - if not record: - raise ValueError(f"The key {email} does not exist in database") - - expiry_time = record.expiry_time - client_id = record.client_id - old_cluster_id = record.server_id - tariff_id = record.tariff_id - alias = record.alias - remnawave_link = remnawave_link or record.remnawave_link - public_link = f"{PUBLIC_LINK}{email}/{tg_id}" - - traffic_limit = None - device_limit = None - if tariff_id: - result = await session.execute(select(Tariff).where(Tariff.id == tariff_id, Tariff.is_active.is_(True))) - tariff = result.scalar_one_or_none() - if tariff: - traffic_limit = int(tariff.traffic_limit) if tariff.traffic_limit is not None else None - device_limit = int(tariff.device_limit) if tariff.device_limit is not None else 0 - else: - logger.warning(f"[LOG] update_subscription: тариф с id={tariff_id} не найден!") - else: - logger.warning("[LOG] update_subscription: tariff_id отсутствует!") - - await delete_key_from_cluster(old_cluster_id, email, client_id, session=session) - await session.execute(delete(Key).where(Key.tg_id == tg_id, Key.email == email)) - await session.commit() - - if country_override or cluster_override: - new_cluster_id = country_override or cluster_override - else: - try: - new_cluster_id = await get_least_loaded_cluster(session) - except ValueError: - logger.warning("[Update] Нет доступных кластеров, оставляем на старом") - new_cluster_id = old_cluster_id - - new_client_id, remnawave_key = await update_key_on_cluster( - tg_id=tg_id, - client_id=client_id, - email=email, - expiry_time=expiry_time, - cluster_id=new_cluster_id, - session=session, - traffic_limit=traffic_limit, - device_limit=device_limit, - remnawave_link=remnawave_link, - ) - - servers = await get_servers(session) - cluster_servers = servers.get(new_cluster_id) - - if cluster_servers is None: - for server_list in servers.values(): - for server_info in server_list: - if server_info.get("server_name", "").lower() == new_cluster_id.lower(): - cluster_servers = [server_info] - break - if cluster_servers: - break - else: - cluster_servers = [] - - has_xui = any(s.get("panel_type", "").lower() == "3x-ui" for s in cluster_servers) - final_key_link = public_link if has_xui else None - - await store_key( - session=session, - tg_id=tg_id, - client_id=new_client_id, - email=email, - expiry_time=expiry_time, - key=final_key_link, - remnawave_link=remnawave_key, - server_id=new_cluster_id, - tariff_id=tariff_id, - alias=alias, - ) - - -async def get_user_traffic(session: AsyncSession, tg_id: int, email: str) -> dict[str, Any]: - """ - Получает трафик пользователя на всех серверах, где у него есть ключ (3x-ui и Remnawave). - Для Remnawave трафик считается один раз и отображается как "Remnawave (общий):". - """ - result = await session.execute(select(Key.client_id, Key.server_id).where(Key.tg_id == tg_id, Key.email == email)) - rows = result.all() - if not rows: - return {"status": "error", "message": "У пользователя нет активных ключей."} - - server_ids = {row.server_id for row in rows} - server_id = list(server_ids)[0] - - result = await session.execute( - select(Server) - .where(Server.enabled.is_(True)) - .where(Server.server_name.in_(server_ids) | Server.cluster_name.in_(server_ids)) - ) - server_rows = result.scalars().all() - if not server_rows: - logger.error(f"Не найдено серверов для: {server_ids}") - return { - "status": "error", - "message": f"Серверы не найдены: {', '.join(server_ids)}", - } - - servers_map = { - s.server_name: { - "server_name": s.server_name, - "cluster_name": s.cluster_name, - "api_url": s.api_url, - "panel_type": s.panel_type, - } - for s in server_rows - } - - user_traffic_data = {} - tasks = [] - - remnawave_client_id = None - remnawave_checked = False - remnawave_api_url = None - - async def fetch_traffic(server_info: dict, client_id: str) -> tuple[str, Any]: - server_name = server_info["server_name"] - api_url = server_info["api_url"] - panel_type = server_info.get("panel_type", "3x-ui").lower() - - try: - if panel_type == "3x-ui": - xui = await get_xui_instance(api_url) - traffic_info = await get_client_traffic(xui, client_id) - if traffic_info["status"] == "success" and traffic_info["traffic"]: - client_data = traffic_info["traffic"][0] - used_gb = (client_data.up + client_data.down) / 1073741824 - return server_name, round(used_gb, 2) - else: - return server_name, "Ошибка получения трафика" - else: - return server_name, f"Неизвестная панель: {panel_type}" - except Exception as e: - return server_name, f"Ошибка: {e}" - - for row in rows: - client_id = row.client_id - server_id = row.server_id - - matched_servers = [ - s for s in servers_map.values() if s["server_name"] == server_id or s["cluster_name"] == server_id - ] - for server_info in matched_servers: - panel_type = server_info.get("panel_type", "3x-ui").lower() - - if panel_type == "remnawave" and not remnawave_checked: - remnawave_client_id = client_id - remnawave_api_url = server_info["api_url"] - remnawave_checked = True - elif panel_type == "3x-ui": - tasks.append(fetch_traffic(server_info, client_id)) - - results = await asyncio.gather(*tasks, return_exceptions=True) - for server, result in results: - user_traffic_data[server] = result - - if remnawave_client_id and remnawave_api_url: - try: - remna = RemnawaveAPI(remnawave_api_url) - if not await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD): - user_traffic_data["Remnawave (общий)"] = "Не удалось авторизоваться" - else: - user_data = await remna.get_user_by_uuid(remnawave_client_id) - if not user_data: - user_traffic_data["Remnawave (общий)"] = "Клиент не найден" - else: - used_bytes = user_data.get("usedTrafficBytes", 0) - used_gb = round(used_bytes / 1073741824, 2) - user_traffic_data["Remnawave (общий)"] = used_gb - except Exception as e: - user_traffic_data["Remnawave (общий)"] = f"Ошибка: {e}" - - return {"status": "success", "traffic": user_traffic_data} - - -async def toggle_client_on_cluster( - cluster_id: str, - email: str, - client_id: str, - enable: bool = True, - session: AsyncSession = None, -) -> dict[str, Any]: - try: - if session is None: - raise ValueError("[Cluster Toggle] Не передан объект сессии для toggle_client_on_cluster") - servers = await get_servers(session) - cluster = servers.get(cluster_id) - - if not cluster: - found_servers = [] - for _, server_list in servers.items(): - for server_info in server_list: - if server_info.get("server_name", "").lower() == cluster_id.lower(): - found_servers.append(server_info) - if found_servers: - cluster = found_servers - else: - raise ValueError(f"Кластер или сервер с ID/именем '{cluster_id}' не найден.") - - results = {} - tasks = [] - - for server_info in cluster: - panel_type = server_info.get("panel_type", "3x-ui").lower() - server_name = server_info.get("server_name", "unknown") - - if panel_type == "3x-ui": - inbound_id = server_info.get("inbound_id") - if not inbound_id: - logger.warning(f"[3x-ui] INBOUND_ID отсутствует для сервера {server_name}. Пропуск.") - results[server_name] = False - continue - - xui = await get_xui_instance(server_info["api_url"]) - unique_email = f"{email}_{server_name.lower()}" if SUPERNODE else email - - tasks.append(toggle_client(xui, int(inbound_id), unique_email, client_id, enable)) - - elif panel_type == "remnawave": - remna = RemnawaveAPI(server_info["api_url"]) - if not await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD): - logger.error(f"[Remnawave] Авторизация не удалась на сервере {server_name}") - results[server_name] = False - continue - - func = remna.enable_user if enable else remna.disable_user - tasks.append(func(client_id)) - - else: - logger.warning( - f"[Cluster Toggle] Неизвестный тип панели '{panel_type}' на сервере {server_name}. Пропуск." - ) - results[server_name] = False - - task_results = await asyncio.gather(*tasks, return_exceptions=True) - - for server_info, result in zip(cluster, task_results, strict=False): - server_name = server_info.get("server_name", "unknown") - if isinstance(result, Exception): - logger.error(f"[Cluster Toggle] Ошибка на сервере {server_name}: {result}") - results[server_name] = False - else: - results[server_name] = result - - status = "включен" if enable else "отключен" - logger.info(f"[Cluster Toggle] Клиент {email} {status} на серверах кластера {cluster_id}") - logger.info(f"[Cluster Toggle DEBUG] Результаты: {results}") - - return { - "status": "success" if any(results.values()) else "error", - "results": results, - } - - except Exception as e: - logger.error(f"[Cluster Toggle] Ошибка при изменении состояния клиента {email} в кластере {cluster_id}: {e}") - return {"status": "error", "error": str(e)} - - -async def reset_traffic_in_cluster(cluster_id: str, email: str, session: AsyncSession) -> None: - try: - servers = await get_servers(session) - cluster = servers.get(cluster_id) - - if not cluster: - found_servers = [] - for _, server_list in servers.items(): - for server_info in server_list: - if server_info.get("server_name", "").lower() == cluster_id.lower(): - found_servers.append(server_info) - if found_servers: - cluster = found_servers - else: - raise ValueError(f"Кластер или сервер с ID/именем {cluster_id} не найден.") - - tasks = [] - remnawave_done = False - - for server_info in cluster: - panel_type = server_info.get("panel_type", "3x-ui").lower() - server_name = server_info.get("server_name", "unknown") - api_url = server_info.get("api_url") - inbound_id = server_info.get("inbound_id") - - if panel_type == "remnawave" and not remnawave_done: - result = await session.execute( - select(Key.client_id).where(Key.email == email, Key.server_id == cluster_id).limit(1) - ) - row = result.first() - - if not row: - logger.warning(f"[Remnawave Reset] client_id не найден для {email} на {server_name}") - continue - - client_id = row[0] - - remna = RemnawaveAPI(api_url) - if not await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD): - logger.warning(f"[Reset Traffic] Не удалось авторизоваться в Remnawave ({server_name})") - continue - - tasks.append(remna.reset_user_traffic(client_id)) - remnawave_done = True - continue - - if panel_type == "3x-ui": - if not inbound_id: - logger.warning(f"INBOUND_ID отсутствует для сервера {server_name}. Пропуск.") - continue - - xui = await get_xui_instance(api_url) - unique_email = f"{email}_{server_name.lower()}" if SUPERNODE else email - tasks.append(xui.client.reset_stats(int(inbound_id), unique_email)) - else: - logger.warning(f"[Reset Traffic] Неизвестный тип панели '{panel_type}' на {server_name}") - - await asyncio.gather(*tasks, return_exceptions=True) - logger.info(f"[Reset Traffic] Трафик клиента {email} успешно сброшен в кластере {cluster_id}") - - except Exception as e: - logger.error(f"[Reset Traffic] Ошибка при сбросе трафика клиента {email} в кластере {cluster_id}: {e}") - raise diff --git a/handlers/keys/keys.py b/handlers/keys/keys.py index 5c926bba..6e771690 100644 --- a/handlers/keys/keys.py +++ b/handlers/keys/keys.py @@ -1,5 +1,3 @@ -import asyncio - from aiogram import F, Router, types from aiogram.exceptions import TelegramBadRequest from aiogram.types import CallbackQuery @@ -7,8 +5,8 @@ from sqlalchemy.ext.asyncio import AsyncSession from database import delete_key, get_key_details from handlers.buttons import APPLY, BACK, CANCEL -from handlers.keys.key_utils import delete_key_from_cluster, update_subscription from handlers.keys.key_view import process_callback_view_key +from handlers.keys.operations import delete_key_from_cluster, update_subscription from handlers.texts import DELETE_KEY_CONFIRM_MSG, KEY_DELETED_MSG_SIMPLE from handlers.utils import edit_or_send_message, handle_error from logger import logger diff --git a/handlers/keys/operations/__init__.py b/handlers/keys/operations/__init__.py new file mode 100644 index 00000000..e9da8a9d --- /dev/null +++ b/handlers/keys/operations/__init__.py @@ -0,0 +1,21 @@ +# handlers/keys/operations/__init__.py + +from .creation import create_client_on_server, create_key_on_cluster +from .deletion import delete_key_from_cluster +from .renewal import renew_key_in_cluster +from .toggles import toggle_client_on_cluster +from .traffic import get_user_traffic, reset_traffic_in_cluster +from .update import update_key_on_cluster, update_subscription + + +__all__ = [ + "create_key_on_cluster", + "create_client_on_server", + "renew_key_in_cluster", + "update_key_on_cluster", + "update_subscription", + "delete_key_from_cluster", + "get_user_traffic", + "reset_traffic_in_cluster", + "toggle_client_on_cluster", +] diff --git a/handlers/keys/operations/creation.py b/handlers/keys/operations/creation.py new file mode 100644 index 00000000..8315e3f4 --- /dev/null +++ b/handlers/keys/operations/creation.py @@ -0,0 +1,255 @@ +import asyncio + +from datetime import datetime + +from sqlalchemy import update +from sqlalchemy.ext.asyncio import AsyncSession + +from config import PUBLIC_LINK, REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD, SUPERNODE +from database import get_servers, get_tariff_by_id, store_key +from database.models import User +from handlers.utils import check_server_key_limit +from logger import logger +from panels.remnawave import RemnawaveAPI +from panels.three_xui import ( + ClientConfig, + add_client, + get_xui_instance, +) + + +async def create_key_on_cluster( + cluster_id: str, + tg_id: int, + client_id: str, + email: str, + expiry_timestamp: int, + plan: int = None, + session: AsyncSession = None, + remnawave_link: str = None, + hwid_limit: int = None, + traffic_limit_bytes: int = None, + is_trial: bool = False, +): + try: + servers = await get_servers(session) + cluster = servers.get(cluster_id) + server_id_to_store = cluster_id + + if not cluster: + found_servers = [] + for _key, server_list in servers.items(): + for server_info in server_list: + if server_info.get("server_name", "").lower() == cluster_id.lower(): + found_servers.append(server_info) + if found_servers: + cluster = found_servers + server_id_to_store = found_servers[0].get("server_name") + else: + raise ValueError(f"Кластер или сервер с ID/именем {cluster_id} не найден.") + + enabled_servers = [s for s in cluster if s.get("enabled", True)] + if not enabled_servers: + logger.warning(f"[Key Creation] Нет доступных серверов в кластере {cluster_id}") + return + + if plan is not None and traffic_limit_bytes is None: + tariff = await get_tariff_by_id(session, plan) + if not tariff: + raise ValueError(f"Тариф с id={plan} не найден.") + traffic_limit_bytes = int(tariff["traffic_limit"]) if tariff["traffic_limit"] else None + if hwid_limit is None and tariff.get("device_limit") is not None: + hwid_limit = int(tariff["device_limit"]) + + remnawave_servers = [ + s + for s in enabled_servers + if s.get("panel_type", "3x-ui").lower() == "remnawave" and await check_server_key_limit(s, session) + ] + xui_servers = [ + s + for s in enabled_servers + if s.get("panel_type", "3x-ui").lower() == "3x-ui" and await check_server_key_limit(s, session) + ] + + if not remnawave_servers and not xui_servers: + logger.warning(f"[Key Creation] Нет серверов с доступным лимитом в кластере {cluster_id}") + return + + semaphore = asyncio.Semaphore(2) + remnawave_created = False + remnawave_key = None + remnawave_client_id = None + + if remnawave_servers: + remna = RemnawaveAPI(remnawave_servers[0]["api_url"]) + logged_in = await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD) + if not logged_in: + logger.error("Не удалось войти в Remnawave API") + else: + expire_at = datetime.utcfromtimestamp(expiry_timestamp / 1000).isoformat() + "Z" + inbound_ids = [s.get("inbound_id") for s in remnawave_servers if s.get("inbound_id")] + + if not inbound_ids: + logger.warning("Нет inbound_id у серверов Remnawave") + else: + short_uuid = None + if remnawave_link and "/" in remnawave_link: + short_uuid = remnawave_link.rstrip("/").split("/")[-1] + + user_data = { + "username": email, + "trafficLimitStrategy": "NO_RESET", + "expireAt": expire_at, + "telegramId": tg_id, + "activeInternalSquads": inbound_ids, + } + + if traffic_limit_bytes and traffic_limit_bytes > 0: + user_data["trafficLimitBytes"] = traffic_limit_bytes * 1024 * 1024 * 1024 + + if short_uuid: + user_data["shortUuid"] = short_uuid + if hwid_limit is not None: + user_data["hwidDeviceLimit"] = hwid_limit + logger.info(f"[Key Creation] Данные для создания клиента в Remnawave: {user_data}") + + result = await remna.create_user(user_data) + if not result: + logger.error("Ошибка при создании пользователя в Remnawave") + else: + remnawave_created = True + remnawave_key = result.get("subscriptionUrl") + remnawave_client_id = result.get("uuid") + logger.info(f"[Key Creation] Пользователь создан в Remnawave: {result}") + + public_link = f"{PUBLIC_LINK}{email}/{tg_id}" if xui_servers else None + final_client_id = remnawave_client_id or client_id + logger.info(f"[Debug] 3x-ui servers для кластера {cluster_id}: {[s['server_name'] for s in xui_servers]}") + + if xui_servers: + if SUPERNODE: + for server_info in xui_servers: + await create_client_on_server( + server_info, + tg_id, + final_client_id, + email, + expiry_timestamp, + semaphore, + plan=plan, + session=session, + is_trial=is_trial, + ) + else: + await asyncio.gather( + *[ + create_client_on_server( + server, + tg_id, + final_client_id, + email, + expiry_timestamp, + semaphore, + plan=plan, + session=session, + is_trial=is_trial, + ) + for server in xui_servers + ], + return_exceptions=True, + ) + + if (remnawave_created and remnawave_client_id) or xui_servers: + await store_key( + session=session, + tg_id=tg_id, + client_id=final_client_id, + email=email, + expiry_time=expiry_timestamp, + key=public_link, + server_id=server_id_to_store, + remnawave_link=remnawave_key, + tariff_id=plan, + ) + + await session.execute(update(User).where(User.tg_id == tg_id, User.trial.in_([0, -1])).values(trial=1)) + await session.commit() + + except Exception as e: + logger.error(f"Ошибка при создании ключа: {e}") + raise e + + +async def create_client_on_server( + server_info: dict, + tg_id: int, + client_id: str, + email: str, + expiry_timestamp: int, + semaphore: asyncio.Semaphore, + plan: int = None, + session=None, + is_trial: bool = False, +): + """ + Создает клиента на указанном 3x-ui сервере с лимитом по тарифу или триалу. + """ + logger.info( + f"[Client] Вход в create_client_on_server: сервер={server_info.get('server_name')}, план={plan}, is_trial={is_trial}" + ) + + async with semaphore: + xui = await get_xui_instance(server_info["api_url"]) + inbound_id = server_info.get("inbound_id") + server_name = server_info.get("server_name", "unknown") + + if not inbound_id: + logger.warning(f"[Client] INBOUND_ID отсутствует для сервера {server_name}. Пропуск.") + return + + if SUPERNODE: + unique_email = f"{email}_{server_name.lower()}" + sub_id = email + else: + unique_email = email + sub_id = unique_email + + total_gb_value = 0 + device_limit_value = 0 + + if plan is not None: + tariff = await get_tariff_by_id(session, plan) + logger.info(f"[Tariff Debug] Получен тариф: {tariff}") + if not tariff: + raise ValueError(f"Тариф с id={plan} не найден.") + + total_gb_value = int(tariff["traffic_limit"]) if tariff["traffic_limit"] else 0 + device_limit_value = int(tariff["device_limit"]) if tariff.get("device_limit") is not None else 0 + + try: + logger.info( + f"[Client] Вызов add_client: email={email}, client_id={client_id}, GB={total_gb_value}, Devices={device_limit_value}" + ) + traffic_limit_bytes = total_gb_value * 1024 * 1024 * 1024 + await add_client( + xui, + ClientConfig( + client_id=client_id, + email=unique_email, + tg_id=tg_id, + limit_ip=device_limit_value, + total_gb=traffic_limit_bytes, + expiry_time=expiry_timestamp, + enable=True, + flow="xtls-rprx-vision", + inbound_id=int(inbound_id), + sub_id=sub_id, + ), + ) + logger.info(f"[Client] Клиент успешно добавлен на сервер {server_name}") + except Exception as e: + logger.error(f"[Client Error] Не удалось создать клиента на {server_name}: {e}") + + if SUPERNODE: + await asyncio.sleep(0.7) diff --git a/handlers/keys/operations/deletion.py b/handlers/keys/operations/deletion.py new file mode 100644 index 00000000..f698c4a5 --- /dev/null +++ b/handlers/keys/operations/deletion.py @@ -0,0 +1,65 @@ +from sqlalchemy.ext.asyncio import AsyncSession + +from config import REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD +from database import get_servers +from logger import logger +from panels.remnawave import RemnawaveAPI +from panels.three_xui import delete_client, get_xui_instance + + +async def delete_key_from_cluster(cluster_id: str, email: str, client_id: str, session: AsyncSession): + """Удаление ключа с серверов в кластере или с конкретного сервера""" + try: + servers = await get_servers(session) + cluster = servers.get(cluster_id) + + if not cluster: + found_servers = [] + for _, server_list in servers.items(): + for server_info in server_list: + if server_info.get("server_name", "").lower() == cluster_id.lower(): + found_servers.append(server_info) + + if found_servers: + cluster = found_servers + else: + raise ValueError(f"Кластер или сервер с ID/именем {cluster_id} не найден.") + + for server_info in cluster: + panel_type = server_info.get("panel_type", "3x-ui").lower() + server_name = server_info.get("server_name", "unknown") + + if panel_type == "remnawave": + remna = RemnawaveAPI(server_info["api_url"]) + if not await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD): + logger.error(f"[Remnawave] Не удалось войти на сервер {server_name}") + continue + + success = await remna.delete_user(client_id) + if success: + logger.info(f"[Remnawave] Клиент {client_id} успешно удалён с {server_name}") + else: + logger.warning(f"[Remnawave] Не удалось удалить клиента {client_id} с {server_name}") + + elif panel_type == "3x-ui": + xui = await get_xui_instance(server_info["api_url"]) + inbound_id = server_info.get("inbound_id") + + if not inbound_id: + logger.warning(f"[3x-ui] INBOUND_ID отсутствует на сервере {server_name}. Пропуск.") + continue + + await delete_client( + xui, + inbound_id=int(inbound_id), + email=email, + client_id=client_id, + ) + logger.info(f"[3x-ui] Клиент {client_id} удалён с сервера {server_name}") + + else: + logger.warning(f"[Unknown] Неизвестный тип панели '{panel_type}' для сервера {server_name}") + + except Exception as e: + logger.error(f"❌ Ошибка при удалении ключа {client_id} из кластера/сервера {cluster_id}: {e}") + raise diff --git a/handlers/keys/operations/renewal.py b/handlers/keys/operations/renewal.py new file mode 100644 index 00000000..ef552b50 --- /dev/null +++ b/handlers/keys/operations/renewal.py @@ -0,0 +1,198 @@ +import asyncio + +from datetime import datetime + +from sqlalchemy import select, update +from sqlalchemy.ext.asyncio import AsyncSession + +from config import REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD, SUPERNODE +from database import delete_notification, get_servers +from database.models import Key, Server, Tariff +from logger import logger +from panels.remnawave import RemnawaveAPI +from panels.three_xui import ClientConfig, add_client, extend_client_key, get_xui_instance + + +async def renew_key_in_cluster( + cluster_id: str, + email: str, + client_id: str, + new_expiry_time: int, + total_gb: int, + session: AsyncSession, + hwid_device_limit: int = 0, + reset_traffic: bool = True, +): + try: + servers = await get_servers(session) + cluster = servers.get(cluster_id) + + if not cluster: + found_servers = [] + for _key, server_list in servers.items(): + for server_info in server_list: + if server_info.get("server_name", "").lower() == cluster_id.lower(): + found_servers.append(server_info) + if found_servers: + cluster = found_servers + else: + raise ValueError(f"Кластер или сервер с ID/именем {cluster_id} не найден.") + + result = await session.execute(select(Key.tg_id, Key.server_id).where(Key.client_id == client_id).limit(1)) + row = result.first() + if not row: + logger.error(f"Не найден пользователь с client_id={client_id} в таблице keys.") + return False + + tg_id, server_id = row + + result = await session.execute(select(Server.tariff_group).where(Server.server_name == server_id)) + tariff_group_row = result.scalar_one_or_none() + + if tariff_group_row: + result = await session.execute( + select(Tariff) + .where(Tariff.group_code == tariff_group_row, Tariff.is_active.is_(True)) + .order_by(Tariff.duration_days.desc()) + .limit(1) + ) + tariff = result.scalar_one_or_none() + if tariff and tariff.device_limit is not None: + hwid_device_limit = int(tariff.device_limit) + + remnawave_inbound_ids = [] + tasks = [] + for server_info in cluster: + if server_info.get("panel_type", "3x-ui").lower() == "remnawave": + inbound_id = server_info.get("inbound_id") + if inbound_id: + remnawave_inbound_ids.append(inbound_id) + + if remnawave_inbound_ids: + remnawave_server = next( + ( + s + for s in cluster + if s.get("panel_type", "").lower() == "remnawave" and s.get("inbound_id") in remnawave_inbound_ids + ), + None, + ) + if remnawave_server: + remna = RemnawaveAPI(remnawave_server["api_url"]) + if await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD): + expire_iso = datetime.utcfromtimestamp(new_expiry_time // 1000).isoformat() + "Z" + traffic_limit_bytes = total_gb * 1024 * 1024 * 1024 if total_gb else 0 + updated = await remna.update_user( + uuid=client_id, + expire_at=expire_iso, + active_user_inbounds=remnawave_inbound_ids, + traffic_limit_bytes=traffic_limit_bytes, + hwid_device_limit=hwid_device_limit, + ) + if updated: + logger.info(f"Подписка Remnawave {client_id} успешно продлена") + if reset_traffic: + await remna.reset_user_traffic(client_id) + else: + logger.warning(f"Не удалось продлить подписку Remnawave {client_id}, пробуем создать") + result = await session.execute( + select(Key.remnawave_link, Key.key).where(Key.client_id == client_id) + ) + row = result.one_or_none() + remnawave_link = row[0] if row else None + row[1] if row else None + + user_data = { + "username": email, + "trafficLimitStrategy": "NO_RESET", + "expireAt": expire_iso, + "telegramId": tg_id, + "activeInternalSquads": remnawave_inbound_ids, + } + if remnawave_link and "/" in remnawave_link: + user_data["shortUuid"] = remnawave_link.rstrip("/").split("/")[-1] + if traffic_limit_bytes and traffic_limit_bytes > 0: + user_data["trafficLimitBytes"] = traffic_limit_bytes + if hwid_device_limit is not None: + user_data["hwidDeviceLimit"] = hwid_device_limit + + result = await remna.create_user(user_data) + if result: + new_client_id = result.get("uuid") + new_remnawave_link = result.get("subscriptionUrl") + logger.info(f"Пользователь Remnawave {client_id} успешно создан") + + await session.execute( + update(Key) + .where(Key.client_id == client_id) + .values(client_id=new_client_id, remnawave_link=new_remnawave_link) + ) + await session.commit() + else: + logger.error(f"Не удалось создать пользователя Remnawave {client_id}") + else: + logger.error("Не удалось войти в Remnawave API") + + for server_info in cluster: + if server_info.get("panel_type", "3x-ui").lower() != "3x-ui": + continue + + xui = await get_xui_instance(server_info["api_url"]) + inbound_id = server_info.get("inbound_id") + server_name = server_info.get("server_name", "unknown") + + if not inbound_id: + logger.warning(f"INBOUND_ID отсутствует для сервера {server_name}. Пропуск.") + continue + + if SUPERNODE: + unique_email = f"{email}_{server_name.lower()}" + sub_id = email + else: + unique_email = email + sub_id = unique_email + + traffic_bytes = total_gb * 1024 * 1024 * 1024 if total_gb else 0 + + async def update_or_create_client(xui, inbound_id, unique_email, sub_id, server_name): + updated = await extend_client_key( + xui=xui, + inbound_id=int(inbound_id), + email=unique_email, + new_expiry_time=new_expiry_time, + client_id=client_id, + total_gb=traffic_bytes, + sub_id=sub_id, + tg_id=tg_id, + limit_ip=hwid_device_limit, + ) + + if not updated: + logger.warning(f"Не удалось обновить клиента {unique_email}, пробуем создать") + config = ClientConfig( + client_id=client_id, + email=unique_email, + tg_id=tg_id, + limit_ip=hwid_device_limit if hwid_device_limit is not None else 0, + total_gb=traffic_bytes, + expiry_time=new_expiry_time, + enable=True, + flow="xtls-rprx-vision", + inbound_id=int(inbound_id), + sub_id=sub_id, + ) + await add_client(xui, config) + + tasks.append(update_or_create_client(xui, inbound_id, unique_email, sub_id, server_name)) + + await asyncio.gather(*tasks, return_exceptions=True) + + notification_prefixes = ["key_24h", "key_10h", "key_expired", "renew"] + for notif in notification_prefixes: + notification_id = f"{email}_{notif}" + await delete_notification(session, tg_id, notification_id) + logger.info(f"🧹 Уведомления для ключа {email} очищены при продлении.") + + except Exception as e: + logger.error(f"Не удалось продлить ключ {client_id} в кластере/на сервере {cluster_id}: {e}") + raise diff --git a/handlers/keys/operations/toggles.py b/handlers/keys/operations/toggles.py new file mode 100644 index 00000000..9587fb47 --- /dev/null +++ b/handlers/keys/operations/toggles.py @@ -0,0 +1,94 @@ +import asyncio + +from typing import Any + +from sqlalchemy.ext.asyncio import AsyncSession + +from config import REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD, SUPERNODE +from database import get_servers +from logger import logger +from panels.remnawave import RemnawaveAPI +from panels.three_xui import get_xui_instance, toggle_client + + +async def toggle_client_on_cluster( + cluster_id: str, + email: str, + client_id: str, + enable: bool = True, + session: AsyncSession = None, +) -> dict[str, Any]: + try: + if session is None: + raise ValueError("[Cluster Toggle] Не передан объект сессии для toggle_client_on_cluster") + servers = await get_servers(session) + cluster = servers.get(cluster_id) + + if not cluster: + found_servers = [] + for _, server_list in servers.items(): + for server_info in server_list: + if server_info.get("server_name", "").lower() == cluster_id.lower(): + found_servers.append(server_info) + if found_servers: + cluster = found_servers + else: + raise ValueError(f"Кластер или сервер с ID/именем '{cluster_id}' не найден.") + + results = {} + tasks = [] + + for server_info in cluster: + panel_type = server_info.get("panel_type", "3x-ui").lower() + server_name = server_info.get("server_name", "unknown") + + if panel_type == "3x-ui": + inbound_id = server_info.get("inbound_id") + if not inbound_id: + logger.warning(f"[3x-ui] INBOUND_ID отсутствует для сервера {server_name}. Пропуск.") + results[server_name] = False + continue + + xui = await get_xui_instance(server_info["api_url"]) + unique_email = f"{email}_{server_name.lower()}" if SUPERNODE else email + + tasks.append(toggle_client(xui, int(inbound_id), unique_email, client_id, enable)) + + elif panel_type == "remnawave": + remna = RemnawaveAPI(server_info["api_url"]) + if not await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD): + logger.error(f"[Remnawave] Авторизация не удалась на сервере {server_name}") + results[server_name] = False + continue + + func = remna.enable_user if enable else remna.disable_user + tasks.append(func(client_id)) + + else: + logger.warning( + f"[Cluster Toggle] Неизвестный тип панели '{panel_type}' на сервере {server_name}. Пропуск." + ) + results[server_name] = False + + task_results = await asyncio.gather(*tasks, return_exceptions=True) + + for server_info, result in zip(cluster, task_results, strict=False): + server_name = server_info.get("server_name", "unknown") + if isinstance(result, Exception): + logger.error(f"[Cluster Toggle] Ошибка на сервере {server_name}: {result}") + results[server_name] = False + else: + results[server_name] = result + + status = "включен" if enable else "отключен" + logger.info(f"[Cluster Toggle] Клиент {email} {status} на серверах кластера {cluster_id}") + logger.info(f"[Cluster Toggle DEBUG] Результаты: {results}") + + return { + "status": "success" if any(results.values()) else "error", + "results": results, + } + + except Exception as e: + logger.error(f"[Cluster Toggle] Ошибка при изменении состояния клиента {email} в кластере {cluster_id}: {e}") + return {"status": "error", "error": str(e)} diff --git a/handlers/keys/operations/traffic.py b/handlers/keys/operations/traffic.py new file mode 100644 index 00000000..66702b95 --- /dev/null +++ b/handlers/keys/operations/traffic.py @@ -0,0 +1,181 @@ +import asyncio + +from typing import Any + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from config import REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD, SUPERNODE +from database import get_servers +from database.models import Key, Server +from logger import logger +from panels.remnawave import RemnawaveAPI +from panels.three_xui import get_client_traffic, get_xui_instance + + +async def get_user_traffic(session: AsyncSession, tg_id: int, email: str) -> dict[str, Any]: + """ + Получает трафик пользователя на всех серверах, где у него есть ключ (3x-ui и Remnawave). + Для Remnawave трафик считается один раз и отображается как "Remnawave (общий):". + """ + result = await session.execute(select(Key.client_id, Key.server_id).where(Key.tg_id == tg_id, Key.email == email)) + rows = result.all() + if not rows: + return {"status": "error", "message": "У пользователя нет активных ключей."} + + server_ids = {row.server_id for row in rows} + server_id = list(server_ids)[0] + + result = await session.execute( + select(Server) + .where(Server.enabled.is_(True)) + .where(Server.server_name.in_(server_ids) | Server.cluster_name.in_(server_ids)) + ) + server_rows = result.scalars().all() + if not server_rows: + logger.error(f"Не найдено серверов для: {server_ids}") + return { + "status": "error", + "message": f"Серверы не найдены: {', '.join(server_ids)}", + } + + servers_map = { + s.server_name: { + "server_name": s.server_name, + "cluster_name": s.cluster_name, + "api_url": s.api_url, + "panel_type": s.panel_type, + } + for s in server_rows + } + + user_traffic_data = {} + tasks = [] + + remnawave_client_id = None + remnawave_checked = False + remnawave_api_url = None + + async def fetch_traffic(server_info: dict, client_id: str) -> tuple[str, Any]: + server_name = server_info["server_name"] + api_url = server_info["api_url"] + panel_type = server_info.get("panel_type", "3x-ui").lower() + + try: + if panel_type == "3x-ui": + xui = await get_xui_instance(api_url) + traffic_info = await get_client_traffic(xui, client_id) + if traffic_info["status"] == "success" and traffic_info["traffic"]: + client_data = traffic_info["traffic"][0] + used_gb = (client_data.up + client_data.down) / 1073741824 + return server_name, round(used_gb, 2) + else: + return server_name, "Ошибка получения трафика" + else: + return server_name, f"Неизвестная панель: {panel_type}" + except Exception as e: + return server_name, f"Ошибка: {e}" + + for row in rows: + client_id = row.client_id + server_id = row.server_id + + matched_servers = [ + s for s in servers_map.values() if s["server_name"] == server_id or s["cluster_name"] == server_id + ] + for server_info in matched_servers: + panel_type = server_info.get("panel_type", "3x-ui").lower() + + if panel_type == "remnawave" and not remnawave_checked: + remnawave_client_id = client_id + remnawave_api_url = server_info["api_url"] + remnawave_checked = True + elif panel_type == "3x-ui": + tasks.append(fetch_traffic(server_info, client_id)) + + results = await asyncio.gather(*tasks, return_exceptions=True) + for server, result in results: + user_traffic_data[server] = result + + if remnawave_client_id and remnawave_api_url: + try: + remna = RemnawaveAPI(remnawave_api_url) + if not await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD): + user_traffic_data["Remnawave (общий)"] = "Не удалось авторизоваться" + else: + user_data = await remna.get_user_by_uuid(remnawave_client_id) + if not user_data: + user_traffic_data["Remnawave (общий)"] = "Клиент не найден" + else: + used_bytes = user_data.get("usedTrafficBytes", 0) + used_gb = round(used_bytes / 1073741824, 2) + user_traffic_data["Remnawave (общий)"] = used_gb + except Exception as e: + user_traffic_data["Remnawave (общий)"] = f"Ошибка: {e}" + + return {"status": "success", "traffic": user_traffic_data} + + +async def reset_traffic_in_cluster(cluster_id: str, email: str, session: AsyncSession) -> None: + try: + servers = await get_servers(session) + cluster = servers.get(cluster_id) + + if not cluster: + found_servers = [] + for _, server_list in servers.items(): + for server_info in server_list: + if server_info.get("server_name", "").lower() == cluster_id.lower(): + found_servers.append(server_info) + if found_servers: + cluster = found_servers + else: + raise ValueError(f"Кластер или сервер с ID/именем {cluster_id} не найден.") + + tasks = [] + remnawave_done = False + + for server_info in cluster: + panel_type = server_info.get("panel_type", "3x-ui").lower() + server_name = server_info.get("server_name", "unknown") + api_url = server_info.get("api_url") + inbound_id = server_info.get("inbound_id") + + if panel_type == "remnawave" and not remnawave_done: + result = await session.execute( + select(Key.client_id).where(Key.email == email, Key.server_id == cluster_id).limit(1) + ) + row = result.first() + + if not row: + logger.warning(f"[Remnawave Reset] client_id не найден для {email} на {server_name}") + continue + + client_id = row[0] + + remna = RemnawaveAPI(api_url) + if not await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD): + logger.warning(f"[Reset Traffic] Не удалось авторизоваться в Remnawave ({server_name})") + continue + + tasks.append(remna.reset_user_traffic(client_id)) + remnawave_done = True + continue + + if panel_type == "3x-ui": + if not inbound_id: + logger.warning(f"INBOUND_ID отсутствует для сервера {server_name}. Пропуск.") + continue + + xui = await get_xui_instance(api_url) + unique_email = f"{email}_{server_name.lower()}" if SUPERNODE else email + tasks.append(xui.client.reset_stats(int(inbound_id), unique_email)) + else: + logger.warning(f"[Reset Traffic] Неизвестный тип панели '{panel_type}' на {server_name}") + + await asyncio.gather(*tasks, return_exceptions=True) + logger.info(f"[Reset Traffic] Трафик клиента {email} успешно сброшен в кластере {cluster_id}") + + except Exception as e: + logger.error(f"[Reset Traffic] Ошибка при сбросе трафика клиента {email} в кластере {cluster_id}: {e}") + raise diff --git a/handlers/keys/operations/update.py b/handlers/keys/operations/update.py new file mode 100644 index 00000000..58185f6a --- /dev/null +++ b/handlers/keys/operations/update.py @@ -0,0 +1,252 @@ +import asyncio + +from datetime import datetime, timezone + +from sqlalchemy import delete, select +from sqlalchemy.ext.asyncio import AsyncSession + +from config import PUBLIC_LINK, REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD, SUPERNODE +from database import get_servers, store_key +from database.models import Key, Tariff +from handlers.utils import get_least_loaded_cluster +from logger import logger +from panels.remnawave import RemnawaveAPI +from panels.three_xui import ClientConfig, add_client, get_xui_instance + +from .deletion import delete_key_from_cluster + + +async def update_key_on_cluster( + tg_id: int, + client_id: str, + email: str, + expiry_time: int, + cluster_id: str, + session: AsyncSession, + traffic_limit: int = None, + device_limit: int = None, + remnawave_link: str = None, +): + """ + Пересоздаёт ключ на всех серверах указанного кластера (или сервера, если передано имя). + Работает с панелями 3x-ui и Remnawave. Возвращает кортеж: (новый client_id, remnawave ссылка или None). + """ + try: + servers = await get_servers(session) + cluster = servers.get(cluster_id) + + if not cluster: + found_servers = [] + for _key, server_list in servers.items(): + for server_info in server_list: + if server_info.get("server_name", "").lower() == cluster_id.lower(): + found_servers.append(server_info) + if found_servers: + cluster = found_servers + else: + raise ValueError(f"Кластер или сервер с ID/именем {cluster_id} не найден.") + + expire_iso = datetime.utcfromtimestamp(expiry_time / 1000).replace(tzinfo=timezone.utc).isoformat() + + remnawave_servers = [s for s in cluster if s.get("panel_type", "3x-ui").lower() == "remnawave"] + xui_servers = [s for s in cluster if s.get("panel_type", "3x-ui").lower() == "3x-ui"] + + remnawave_client_id = None + remnawave_key = None + + if remnawave_servers: + inbound_ids = [s["inbound_id"] for s in remnawave_servers if s.get("inbound_id")] + remna = RemnawaveAPI(remnawave_servers[0]["api_url"]) + if await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD): + await remna.delete_user(client_id) + + group_code = remnawave_servers[0].get("tariff_group") + if not group_code: + raise ValueError("У Remnawave-сервера отсутствует tariff_group") + + result = await session.execute( + select(Tariff) + .where(Tariff.group_code == group_code, Tariff.is_active.is_(True)) + .order_by(Tariff.duration_days.desc()) + .limit(1) + ) + result.scalar_one_or_none() + + short_uuid = None + if remnawave_link and "/" in remnawave_link: + short_uuid = remnawave_link.rstrip("/").split("/")[-1] + logger.info(f"[Update] Извлечен short_uuid из ссылки: {short_uuid}") + + user_data = { + "username": email, + "trafficLimitStrategy": "NO_RESET", + "expireAt": expire_iso, + "telegramId": tg_id, + "activeInternalSquads": inbound_ids, + } + if traffic_limit is not None: + user_data["trafficLimitBytes"] = traffic_limit * 1024**3 + if device_limit is not None: + user_data["hwidDeviceLimit"] = device_limit + if short_uuid: + user_data["shortUuid"] = short_uuid + logger.info(f"[Update] Добавлен short_uuid в user_data: {short_uuid}") + + result = await remna.create_user(user_data) + if result: + remnawave_client_id = result.get("uuid") + remnawave_key = result.get("subscriptionUrl") + logger.info(f"[Update] Remnawave: клиент заново создан, новый UUID: {remnawave_client_id}") + else: + logger.error("[Update] Ошибка создания Remnawave клиента") + else: + logger.error("[Update] Не удалось авторизоваться в Remnawave") + + if not remnawave_client_id: + logger.warning(f"[Update] Remnawave client_id не получен. Используется исходный: {client_id}") + remnawave_client_id = client_id + + tasks = [] + for server_info in xui_servers: + server_name = server_info.get("server_name", "unknown") + inbound_id = server_info.get("inbound_id") + + if not inbound_id: + logger.warning(f"[Update] INBOUND_ID отсутствует для сервера {server_name}. Пропуск.") + continue + + xui = await get_xui_instance(server_info["api_url"]) + + sub_id = email + unique_email = f"{email}_{server_name.lower()}" if SUPERNODE else email + + group_code = server_info.get("tariff_group") + if not group_code: + raise ValueError(f"У сервера {server_name} отсутствует tariff_group") + + result = await session.execute( + select(Tariff) + .where(Tariff.group_code == group_code, Tariff.is_active.is_(True)) + .order_by(Tariff.duration_days.desc()) + .limit(1) + ) + result.scalar_one_or_none() + + total_gb_bytes = int(traffic_limit * 1024**3) if traffic_limit is not None else 0 + device_limit_value = device_limit if device_limit is not None else 0 + + config = ClientConfig( + client_id=remnawave_client_id, + email=unique_email, + tg_id=tg_id, + limit_ip=device_limit_value, + total_gb=total_gb_bytes, + expiry_time=expiry_time, + enable=True, + flow="xtls-rprx-vision", + inbound_id=int(inbound_id), + sub_id=sub_id, + ) + + tasks.append(add_client(xui, config)) + + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + + logger.info(f"[Update] Ключ {remnawave_client_id} обновлён на всех серверах кластера {cluster_id}") + return remnawave_client_id, remnawave_key + + except Exception as e: + logger.error(f"[Update Error] Ошибка при обновлении ключа {client_id} на {cluster_id}: {e}") + raise + + +async def update_subscription( + tg_id: int, + email: str, + session: AsyncSession, + cluster_override: str = None, + country_override: str = None, + remnawave_link: str = None, +) -> None: + result = await session.execute(select(Key).where(Key.tg_id == tg_id, Key.email == email)) + record = result.scalar_one_or_none() + + if not record: + raise ValueError(f"The key {email} does not exist in database") + + expiry_time = record.expiry_time + client_id = record.client_id + old_cluster_id = record.server_id + tariff_id = record.tariff_id + alias = record.alias + remnawave_link = remnawave_link or record.remnawave_link + public_link = f"{PUBLIC_LINK}{email}/{tg_id}" + + traffic_limit = None + device_limit = None + if tariff_id: + result = await session.execute(select(Tariff).where(Tariff.id == tariff_id, Tariff.is_active.is_(True))) + tariff = result.scalar_one_or_none() + if tariff: + traffic_limit = int(tariff.traffic_limit) if tariff.traffic_limit is not None else None + device_limit = int(tariff.device_limit) if tariff.device_limit is not None else 0 + else: + logger.warning(f"[LOG] update_subscription: тариф с id={tariff_id} не найден!") + else: + logger.warning("[LOG] update_subscription: tariff_id отсутствует!") + + await delete_key_from_cluster(old_cluster_id, email, client_id, session=session) + await session.execute(delete(Key).where(Key.tg_id == tg_id, Key.email == email)) + await session.commit() + + if country_override or cluster_override: + new_cluster_id = country_override or cluster_override + else: + try: + new_cluster_id = await get_least_loaded_cluster(session) + except ValueError: + logger.warning("[Update] Нет доступных кластеров, оставляем на старом") + new_cluster_id = old_cluster_id + + new_client_id, remnawave_key = await update_key_on_cluster( + tg_id=tg_id, + client_id=client_id, + email=email, + expiry_time=expiry_time, + cluster_id=new_cluster_id, + session=session, + traffic_limit=traffic_limit, + device_limit=device_limit, + remnawave_link=remnawave_link, + ) + + servers = await get_servers(session) + cluster_servers = servers.get(new_cluster_id) + + if cluster_servers is None: + for server_list in servers.values(): + for server_info in server_list: + if server_info.get("server_name", "").lower() == new_cluster_id.lower(): + cluster_servers = [server_info] + break + if cluster_servers: + break + else: + cluster_servers = [] + + has_xui = any(s.get("panel_type", "").lower() == "3x-ui" for s in cluster_servers) + final_key_link = public_link if has_xui else None + + await store_key( + session=session, + tg_id=tg_id, + client_id=new_client_id, + email=email, + expiry_time=expiry_time, + key=final_key_link, + remnawave_link=remnawave_key, + server_id=new_cluster_id, + tariff_id=tariff_id, + alias=alias, + ) diff --git a/handlers/notifications/general_notifications.py b/handlers/notifications/general_notifications.py index b1415928..862e67fe 100644 --- a/handlers/notifications/general_notifications.py +++ b/handlers/notifications/general_notifications.py @@ -33,7 +33,7 @@ from database import ( update_key_expiry, update_key_tariff, ) -from handlers.keys.key_utils import delete_key_from_cluster, renew_key_in_cluster +from handlers.keys.operations import delete_key_from_cluster, renew_key_in_cluster from handlers.notifications.notify_kb import ( build_notification_expired_kb, build_notification_kb, diff --git a/handlers/notifications/special_notifications.py b/handlers/notifications/special_notifications.py index 23c50963..93145868 100644 --- a/handlers/notifications/special_notifications.py +++ b/handlers/notifications/special_notifications.py @@ -22,7 +22,7 @@ from database import ( ) from database.tariffs import get_tariffs from handlers.buttons import CONNECT_DEVICE, CONNECT_PHONE, MAIN_MENU, PC_BUTTON, TV_BUTTON -from handlers.keys.key_utils import get_user_traffic +from handlers.keys.operations import get_user_traffic from handlers.notifications.notify_utils import send_messages_with_limit from handlers.texts import ( TRIAL_INACTIVE_BONUS_MSG, diff --git a/handlers/payments/kassai.py b/handlers/payments/kassai.py index a5ba96dc..fdfc6b5e 100644 --- a/handlers/payments/kassai.py +++ b/handlers/payments/kassai.py @@ -2,7 +2,6 @@ import aiohttp import hashlib import hmac import time -import json from aiogram import F, Router, types from aiogram.fsm.context import FSMContext from aiogram.fsm.state import State, StatesGroup diff --git a/handlers/payments/wata.py b/handlers/payments/wata.py index 68c16d67..47c527f0 100644 --- a/handlers/payments/wata.py +++ b/handlers/payments/wata.py @@ -156,54 +156,61 @@ async def process_custom_amount_button(callback_query: types.CallbackQuery, stat await state.set_state(ReplenishBalanceWataState.entering_custom_amount) -@router.message(ReplenishBalanceWataState.entering_custom_amount) -async def handle_custom_amount_input(message: types.Message, state: FSMContext): - data = await state.get_data() - cassa_name = data.get("wata_cassa") - cassa = next((c for c in WATA_CASSA_CONFIG if c["name"] == cassa_name), None) - if not cassa or not cassa["enable"]: - await edit_or_send_message( - target_message=message, - text="Ошибка: выбранная касса недоступна.", - reply_markup=InlineKeyboardMarkup(inline_keyboard=[]), - force_text=True, - ) - return +async def handle_custom_amount_input( + callback_query: types.CallbackQuery, + state: FSMContext, +): + tg_id = callback_query.from_user.id + target_message = callback_query.message + try: - amount = int(message.text.strip()) - if amount <= 0: - raise ValueError - if cassa_name == "sbp" and amount < 50: + data = await state.get_data() + required_amount = data.get("required_amount", 0) + cassa_name = data.get("wata_cassa", "sbp") + + if not required_amount or required_amount <= 0: await edit_or_send_message( - target_message=message, - text="Минимальная сумма для оплаты через СБП — 50 рублей.", + target_message=target_message, + text="❌ Недостаточная сумма для оплаты.", reply_markup=InlineKeyboardMarkup(inline_keyboard=[]), - force_text=True, ) return - except Exception: - await edit_or_send_message( - target_message=message, - text="Некорректная сумма. Введите целое число больше 0.", - reply_markup=InlineKeyboardMarkup(inline_keyboard=[]), - force_text=True, + + cassa = next((c for c in WATA_CASSA_CONFIG if c["name"] == cassa_name and c["enable"]), None) + if not cassa: + await edit_or_send_message( + target_message=target_message, + text="❌ Выбранная касса WATA недоступна.", + reply_markup=InlineKeyboardMarkup(inline_keyboard=[]), + ) + return + + payment_url = await generate_wata_payment_link(required_amount, tg_id, cassa) + + keyboard = InlineKeyboardMarkup( + inline_keyboard=[ + [InlineKeyboardButton(text=PAY_2, url=payment_url)], + [InlineKeyboardButton(text=BACK, callback_data="balance")], + ] + ) + + message_text = f"💰 Вы выбрали пополнение на {required_amount}₽. Перейдите по ссылке для оплаты:" + await edit_or_send_message( + target_message=target_message, + text=message_text, + reply_markup=keyboard, + disable_web_page_preview=True, + ) + + await state.clear() + + except Exception as e: + logger.error(f"[WATA] Ошибка при создании ссылки для оплаты: {e}", exc_info=True) + await edit_or_send_message( + target_message=target_message, + text="⚠️ Произошла ошибка при создании ссылки на оплату. Попробуйте позже.", + reply_markup=InlineKeyboardMarkup(inline_keyboard=[]), ) - return - await state.update_data(amount=amount) - payment_url = await generate_wata_payment_link(amount, message.chat.id, cassa) - confirm_keyboard = InlineKeyboardMarkup( - inline_keyboard=[ - [InlineKeyboardButton(text=PAY_2, url=payment_url)], - [InlineKeyboardButton(text=BACK, callback_data="balance")], - ] - ) - await edit_or_send_message( - target_message=message, - text=WATA_PAYMENT_MESSAGE.format(amount=amount), - reply_markup=confirm_keyboard, - force_text=True, - ) - await state.set_state(ReplenishBalanceWataState.waiting_for_payment_confirmation) @router.callback_query(F.data.startswith("wata_amount|")) diff --git a/handlers/profile.py b/handlers/profile.py index 37ce344b..1cc9660d 100644 --- a/handlers/profile.py +++ b/handlers/profile.py @@ -1,9 +1,8 @@ -import html import os from aiogram import F, Router from aiogram.fsm.context import FSMContext -from aiogram.types import CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup, Message +from aiogram.types import CallbackQuery, InlineKeyboardButton, Message from aiogram.utils.keyboard import InlineKeyboardBuilder from config import ( @@ -29,12 +28,12 @@ from handlers.buttons import ( TRIAL_SUB, ) from handlers.texts import ADD_SUBSCRIPTION_HINT +from hooks.hook_buttons import insert_hook_buttons from hooks.hooks import run_hooks -from logger import logger from .admin.panel.keyboard import AdminPanelCallback from .texts import profile_message_send -from .utils import edit_or_send_message +from .utils import edit_or_send_message, get_username router = Router() @@ -50,39 +49,25 @@ async def process_callback_view_profile( ): if isinstance(callback_query_or_message, CallbackQuery): chat = callback_query_or_message.message.chat - from_user = callback_query_or_message.from_user - chat_id = chat.id - target_message = callback_query_or_message.message + user = callback_query_or_message.from_user + message = callback_query_or_message.message else: chat = callback_query_or_message.chat - from_user = callback_query_or_message.from_user - chat_id = chat.id - target_message = callback_query_or_message + user = callback_query_or_message.from_user + message = callback_query_or_message - user = chat if chat.type == "private" else from_user - - if getattr(user, "full_name", None): - username = html.escape(user.full_name) - elif getattr(user, "first_name", None): - username = html.escape(user.first_name) - elif getattr(user, "username", None): - username = "@" + html.escape(user.username) - else: - username = "Пользователь" - - image_path = os.path.join("img", "profile.jpg") + chat_id = chat.id + username = get_username(user or chat) key_count = await get_key_count(session, chat_id) balance = await get_balance(session, chat_id) or 0 trial_status = await get_trial(session, chat_id) profile_message = profile_message_send(username, chat_id, int(balance), key_count) - if key_count == 0: - profile_message += ADD_SUBSCRIPTION_HINT - else: - profile_message += f"\n
{NEWS_MESSAGE}
" + profile_message += ADD_SUBSCRIPTION_HINT if key_count == 0 else f"\n
{NEWS_MESSAGE}
" builder = InlineKeyboardBuilder() + if key_count > 0: builder.row(InlineKeyboardButton(text=RENEW_KEY, callback_data="renew_menu")) builder.row(InlineKeyboardButton(text=MY_SUBS, callback_data="view_keys")) @@ -94,68 +79,35 @@ async def process_callback_view_profile( if BALANCE_BUTTON: builder.row(InlineKeyboardButton(text=BALANCE, callback_data="balance")) - row_buttons = [] + extra_buttons = [] if REFERRAL_BUTTON: - row_buttons.append(InlineKeyboardButton(text=INVITE, callback_data="invite")) + extra_buttons.append(InlineKeyboardButton(text=INVITE, callback_data="invite")) if GIFT_BUTTON: - row_buttons.append(InlineKeyboardButton(text=GIFTS, callback_data="gifts")) - if row_buttons: - builder.row(*row_buttons) + extra_buttons.append(InlineKeyboardButton(text=GIFTS, callback_data="gifts")) + if extra_buttons: + builder.row(*extra_buttons) module_buttons = await run_hooks("profile_menu", chat_id=chat_id, admin=admin, session=session) - - for module_btn in module_buttons: - if isinstance(module_btn, dict) and "after" in module_btn: - after_callback = module_btn["after"] - insert_pos = -1 - current_markup = builder.as_markup() - - for i, row in enumerate(current_markup.inline_keyboard): - for btn in row: - if btn.callback_data == after_callback: - insert_pos = i + 1 - break - if insert_pos > 0: - break - - if insert_pos > 0: - new_buttons = [] - for i, row in enumerate(current_markup.inline_keyboard): - if i == insert_pos: - new_buttons.append([module_btn["button"]]) - new_buttons.append(row) - - if insert_pos >= len(current_markup.inline_keyboard): - new_buttons.append([module_btn["button"]]) - - builder = InlineKeyboardBuilder.from_markup(InlineKeyboardMarkup(inline_keyboard=new_buttons)) - else: - builder.row(module_btn["button"]) - else: - if isinstance(module_btn, dict): - builder.row(module_btn["button"]) - else: - builder.row(module_btn) + builder = insert_hook_buttons(builder, module_buttons) if INSTRUCTIONS_BUTTON: builder.row(InlineKeyboardButton(text=INSTRUCTIONS, callback_data="instructions")) + if admin: builder.row( - InlineKeyboardButton( - text="📊 Администратор", - callback_data=AdminPanelCallback(action="admin").pack(), - ) + InlineKeyboardButton(text="📊 Администратор", callback_data=AdminPanelCallback(action="admin").pack()) ) + if SHOW_START_MENU_ONCE: builder.row(InlineKeyboardButton(text=ABOUT_VPN, callback_data="about_vpn")) else: builder.row(InlineKeyboardButton(text=BACK, callback_data="start")) await edit_or_send_message( - target_message=target_message, + target_message=message, text=profile_message, reply_markup=builder.as_markup(), - media_path=image_path, + media_path=os.path.join("img", "profile.jpg"), disable_web_page_preview=False, force_text=True, ) diff --git a/handlers/start.py b/handlers/start.py index 3166d77b..9746aff0 100644 --- a/handlers/start.py +++ b/handlers/start.py @@ -60,94 +60,41 @@ from .utils import edit_or_send_message router = Router() - processing_gifts = set() -@router.callback_query(F.data == "start") -async def handle_start_callback_query( - callback_query: CallbackQuery, - state: FSMContext, - session: Any, - admin: bool, - captcha: bool = False, -): - await start_command(callback_query.message, state, session, admin, captcha) - - @router.message(Command("start")) -async def start_command(message: Message, state: FSMContext, session: Any, admin: bool, captcha: bool = True): - logger.info(f"Вызвана функция start_command для пользователя {message.chat.id}") - +@router.callback_query(F.data == "start") +async def start_entry( + event: Message | CallbackQuery, state: FSMContext, session: Any, admin: bool, captcha: bool = True +): + message = event.message if isinstance(event, CallbackQuery) else event if CAPTCHA_ENABLE and captcha: - user_exists = await check_user_exists(session, message.chat.id) - if not user_exists: + if not await check_user_exists(session, message.chat.id): captcha_data = await generate_captcha(message, state) - await edit_or_send_message( - target_message=message, - text=captcha_data["text"], - reply_markup=captcha_data["markup"], - ) + await edit_or_send_message(message, captcha_data["text"], reply_markup=captcha_data["markup"]) return - - state_data = await state.get_data() - text_to_process = state_data.get("original_text", message.text) - await process_start_logic(message, state, session, admin, text_to_process) + text = getattr(event, "data", None) or message.text + await process_start_logic(message, state, session, admin, text) @router.callback_query(F.data == "check_subscription") -async def check_subscription_callback(callback_query: CallbackQuery, state: FSMContext, session: Any, admin: bool): - user_id = callback_query.from_user.id - logger.info(f"[CALLBACK] Получен callback 'check_subscription' от пользователя {user_id}") - +async def check_subscription_callback(callback: CallbackQuery, state: FSMContext, session: Any, admin: bool): + user_id = callback.from_user.id try: member = await bot.get_chat_member(CHANNEL_ID, user_id) - logger.info(f"[CALLBACK] Статус подписки пользователя {user_id}: {member.status}") - if member.status not in ["member", "administrator", "creator"]: - await callback_query.answer(NOT_SUBSCRIBED_YET_MSG, show_alert=True) - builder = InlineKeyboardBuilder() - builder.row(InlineKeyboardButton(text=SUB_CHANELL, url=CHANNEL_URL)) - builder.row(InlineKeyboardButton(text=SUB_CHANELL_DONE, callback_data="check_subscription")) - await callback_query.message.edit_text( - SUBSCRIPTION_REQUIRED_MSG, - reply_markup=builder.as_markup(), - ) + await prompt_subscription(callback) return - await callback_query.answer(SUBSCRIPTION_CONFIRMED_MSG) + await callback.answer(SUBSCRIPTION_CONFIRMED_MSG) data = await state.get_data() - original_text = data.get("original_text") or callback_query.message.text - user_data = data.get("user_data") - - if not user_data: - user = callback_query.from_user - user_data = { - "tg_id": user.id, - "username": user.username, - "first_name": user.first_name, - "last_name": user.last_name, - "language_code": user.language_code, - "is_bot": user.is_bot, - } - await state.update_data(user_data=user_data) - - if user_data.get("is_bot"): - logger.warning(f"[CALLBACK] Попытка регистрации бота: {user_data}") - return - - await process_start_logic( - message=callback_query.message, - state=state, - session=session, - admin=admin, - text_to_process=original_text, - user_data=user_data, - ) - logger.info(f"[CALLBACK] Завершен вызов process_start_logic для пользователя {user_id}") - + original_text = data.get("original_text") or callback.message.text + user_data = data.get("user_data") or extract_user_data(callback.from_user) + await state.update_data(user_data=user_data) + await process_start_logic(callback.message, state, session, admin, original_text, user_data) except Exception as e: - logger.error(f"[CALLBACK] Ошибка проверки подписки для пользователя {user_id}: {e}", exc_info=True) - await callback_query.answer(SUBSCRIPTION_CHECK_ERROR_MSG, show_alert=True) + logger.error(f"[CALLBACK] Ошибка подписки: {e}", exc_info=True) + await callback.answer(SUBSCRIPTION_CHECK_ERROR_MSG, show_alert=True) async def process_start_logic( @@ -158,222 +105,166 @@ async def process_start_logic( text_to_process: str = None, user_data: dict | None = None, ): - user_data = user_data or { - "tg_id": (message.from_user or message.chat).id, - "username": getattr(message.from_user, "username", None), - "first_name": getattr(message.from_user, "first_name", None), - "last_name": getattr(message.from_user, "last_name", None), - "language_code": getattr(message.from_user, "language_code", None), - "is_bot": getattr(message.from_user, "is_bot", False), - } - + user_data = user_data or extract_user_data(message.from_user or message.chat) text = text_to_process or message.text or message.caption - if not text: - logger.info(f"[StartLogic] Текста нет — вызываю стартовое меню для {user_data['tg_id']}") await show_start_menu(message, admin, session) return if text.startswith("/start "): - parts = text.split(maxsplit=1) - if len(parts) > 1: - text = parts[1] + text = text.split(maxsplit=1)[1] - try: - gift_detected = False - text_parts = text.split("-") + gift_detected = False + for part in text.split("-"): + await run_hooks("start_link", message=message, state=state, session=session, user_data=user_data, part=part) - for part in text_parts: - await run_hooks("start_link", message=message, state=state, session=session, user_data=user_data, part=part) + if "coupons" in part: + await handle_coupon_link(part, message, state, session, admin, user_data) + continue + if "gift" in part: + gift_detected = await handle_gift(part, message, state, session, user_data) + break + if "referral" in part: + await handle_referral_link_safe(part, message, state, session, user_data) + continue + if "utm" in part: + await handle_utm_link(part, message, state, session, user_data) - for part in text_parts: - if "coupons" in part: - logger.info(f"Обнаружена ссылка на купон: {part}") - coupon_code = part.split("coupons")[1].strip("_") - coupon = await get_coupon_by_code(session, coupon_code) - if not coupon: - continue + await state.clear() + if gift_detected: + return - await activate_coupon( - message, - state, - session, - coupon_code, - admin=admin, - user_data=user_data, - ) + if not await check_user_exists(session, user_data["tg_id"]): + await add_user(session=session, **user_data) - if coupon.days: - return - continue + trial_status = await get_trial(session, user_data["tg_id"]) + key_count = await get_key_count(session, user_data["tg_id"]) - if "gift" in part: - gift_raw = part.split("gift")[1].strip("_") - - if not gift_raw: - await message.answer("❌ Неверный формат ссылки на подарок.") - return await process_callback_view_profile(message, state, admin, session) - - gift_id = gift_raw - - if gift_id in processing_gifts: - await message.answer("⏳ Подарок уже обрабатывается, подождите...") - return await process_callback_view_profile(message, state, admin, session) - - processing_gifts.add(gift_id) - - try: - logger.info(f"[GIFT] Обнаружен подарок {gift_id}") - await handle_gift_link(gift_id, message, state, session, user_data=user_data) - gift_detected = True - finally: - processing_gifts.discard(gift_id) - - break - - if "referral" in part: - referrer_tg_id = part.split("referral")[1].strip("_") - try: - referrer_tg_id = int(referrer_tg_id) - await handle_referral_link(referrer_tg_id, message, state, session, user_data=user_data) - except (ValueError, IndexError): - pass - continue - - if "utm" in part: - utm_code = part - logger.info(f"[UTM] Обнаружена ссылка на UTM: {utm_code}") - await handle_utm_link(utm_code, message, state, session, user_data=user_data) - continue - - await state.clear() - if gift_detected: - return - - user_exists = await check_user_exists(session, user_data["tg_id"]) - if not user_exists: - await add_user(session=session, **user_data) - - trial_status = await get_trial(session, user_data["tg_id"]) - key_count = await get_key_count(session, user_data["tg_id"]) - - if SHOW_START_MENU_ONCE: - if key_count > 0: - await process_callback_view_profile(message, state, admin, session) - elif trial_status == 0: - await show_start_menu(message, admin, session) - else: - await process_callback_view_profile(message, state, admin, session) + if SHOW_START_MENU_ONCE: + if key_count > 0 or trial_status != 0: + await process_callback_view_profile(message, state, admin, session) else: await show_start_menu(message, admin, session) - - await state.clear() - - except Exception as e: - logger.error(f"Ошибка при обработке текста {text} — {e}", exc_info=True) - await message.answer("❌ Произошла ошибка. Попробуйте позже.") + else: + await show_start_menu(message, admin, session) -async def handle_utm_link( - utm_code: str, - message: Message, - state: FSMContext, - session: AsyncSession, - user_data: dict, -): +async def handle_coupon_link(part, message, state, session, admin, user_data): + code = part.split("coupons")[1].strip("_") + coupon = await get_coupon_by_code(session, code) + if coupon: + await activate_coupon(message, state, session, code, admin=admin, user_data=user_data) + if coupon.days: + return + + +async def handle_gift(part, message, state, session, user_data): + gift_id = part.split("gift")[1].strip("_") + if not gift_id: + await message.answer("❌ Неверный формат ссылки на подарок.") + await process_callback_view_profile(message, state, False, session) + return False + + if gift_id in processing_gifts: + await message.answer("⏳ Подарок уже обрабатывается, подождите...") + await process_callback_view_profile(message, state, False, session) + return False + + processing_gifts.add(gift_id) + try: + await handle_gift_link(gift_id, message, state, session, user_data=user_data) + return True + finally: + processing_gifts.discard(gift_id) + + +async def handle_referral_link_safe(part, message, state, session, user_data): + try: + referrer_id = int(part.split("referral")[1].strip("_")) + await handle_referral_link(referrer_id, message, state, session, user_data) + except Exception: + pass + + +async def prompt_subscription(callback: CallbackQuery): + await callback.answer(NOT_SUBSCRIBED_YET_MSG, show_alert=True) + kb = InlineKeyboardBuilder() + kb.row(InlineKeyboardButton(text=SUB_CHANELL, url=CHANNEL_URL)) + kb.row(InlineKeyboardButton(text=SUB_CHANELL_DONE, callback_data="check_subscription")) + await callback.message.edit_text(SUBSCRIPTION_REQUIRED_MSG, reply_markup=kb.as_markup()) + + +def extract_user_data(user) -> dict: + return { + "tg_id": user.id, + "username": user.username, + "first_name": user.first_name, + "last_name": user.last_name, + "language_code": user.language_code, + "is_bot": user.is_bot, + } + + +async def handle_utm_link(utm_code: str, message: Message, state: FSMContext, session: AsyncSession, user_data: dict): user_id = user_data["tg_id"] - result = await session.execute(select(TrackingSource).where(TrackingSource.code == utm_code)) - utm_exists = result.scalar_one_or_none() - - if not utm_exists: + if not result.scalar_one_or_none(): await message.answer("❌ UTM ссылка не найдена.") return - result = await session.execute(select(User).where(User.tg_id == user_id)) - user = result.scalar_one_or_none() - if user and user.source_code is None: + user = (await session.execute(select(User).where(User.tg_id == user_id))).scalar_one_or_none() + if user and not user.source_code: user.source_code = utm_code await session.commit() - logger.info(f"[UTM] Привязана {utm_code} к пользователю {user_id}") elif not user: await add_user(session=session, source_code=utm_code, **user_data) - logger.info(f"[UTM] Зарегистрирован и привязан {utm_code} к пользователю {user_id}") async def show_start_menu(message: Message, admin: bool, session: AsyncSession): - """Функция для отображения стандартного меню через редактирование сообщения.""" - image_path = os.path.join("img", "pic.jpg") - builder = InlineKeyboardBuilder() + kb = InlineKeyboardBuilder() - trial_status = None - if session is not None: - trial_status = await get_trial(session, message.chat.id) - else: - logger.warning(f"Сессия базы данных отсутствует, пропускаем проверку триала для {message.chat.id}") + trial_status = await get_trial(session, message.chat.id) if session else None + show_trial = trial_status == 0 and not TRIAL_TIME_DISABLE + show_profile = not SHOW_START_MENU_ONCE or trial_status != 0 or TRIAL_TIME_DISABLE - show_trial_button = trial_status == 0 and not TRIAL_TIME_DISABLE - show_profile_button = not SHOW_START_MENU_ONCE or trial_status != 0 or TRIAL_TIME_DISABLE - - if show_trial_button: - builder.row(InlineKeyboardButton(text=TRIAL_SUB, callback_data="create_key")) - if show_profile_button: - builder.row(InlineKeyboardButton(text=MAIN_MENU, callback_data="profile")) + if show_trial: + kb.row(InlineKeyboardButton(text=TRIAL_SUB, callback_data="create_key")) + if show_profile: + kb.row(InlineKeyboardButton(text=MAIN_MENU, callback_data="profile")) if CHANNEL_EXISTS: - builder.row( + kb.row( InlineKeyboardButton(text=SUPPORT, url=SUPPORT_CHAT_URL), InlineKeyboardButton(text=CHANNEL, url=CHANNEL_URL), ) else: - builder.row(InlineKeyboardButton(text=SUPPORT, url=SUPPORT_CHAT_URL)) + kb.row(InlineKeyboardButton(text=SUPPORT, url=SUPPORT_CHAT_URL)) if admin: - builder.row( - InlineKeyboardButton( - text="📊 Администратор", - callback_data=AdminPanelCallback(action="admin").pack(), - ) - ) + kb.row(InlineKeyboardButton(text="📊 Администратор", callback_data=AdminPanelCallback(action="admin").pack())) - builder.row(InlineKeyboardButton(text=ABOUT_VPN, callback_data="about_vpn")) + kb.row(InlineKeyboardButton(text=ABOUT_VPN, callback_data="about_vpn")) - await edit_or_send_message( - target_message=message, - text=WELCOME_TEXT, - reply_markup=builder.as_markup(), - media_path=image_path, - ) + await edit_or_send_message(message, WELCOME_TEXT, reply_markup=kb.as_markup(), media_path=image_path) @router.callback_query(F.data == "about_vpn") -async def handle_about_vpn(callback_query: CallbackQuery, session: AsyncSession): - user_id = callback_query.from_user.id +async def handle_about_vpn(callback: CallbackQuery, session: AsyncSession): + user_id = callback.from_user.id trial = await get_trial(session, user_id) - back_target = "profile" if SHOW_START_MENU_ONCE and trial > 0 else "start" - builder = InlineKeyboardBuilder() + kb = InlineKeyboardBuilder() if DONATIONS_ENABLE: - builder.row(InlineKeyboardButton(text="💰 Поддержать проект", callback_data="donate")) + kb.row(InlineKeyboardButton(text="💰 Поддержать проект", callback_data="donate")) - support_btn = InlineKeyboardButton(text=SUPPORT, url=SUPPORT_CHAT_URL) + kb.row(InlineKeyboardButton(text=SUPPORT, url=SUPPORT_CHAT_URL)) if CHANNEL_EXISTS: - channel_btn = InlineKeyboardButton(text=CHANNEL, url=CHANNEL_URL) - builder.row(support_btn, channel_btn) - else: - builder.row(support_btn) - - builder.row(InlineKeyboardButton(text=BACK, callback_data=back_target)) + kb.row(InlineKeyboardButton(text=CHANNEL, url=CHANNEL_URL)) + kb.row(InlineKeyboardButton(text=BACK, callback_data=back_target)) text = get_about_vpn("3.2.3-minor") - image_path = os.path.join("img", "pic.jpg") - await edit_or_send_message( - target_message=callback_query.message, - text=text, - reply_markup=builder.as_markup(), - media_path=image_path, - force_text=False, + callback.message, text, reply_markup=kb.as_markup(), media_path=os.path.join("img", "pic.jpg"), force_text=False ) diff --git a/handlers/utils.py b/handlers/utils.py index 6060ce11..18d52977 100644 --- a/handlers/utils.py +++ b/handlers/utils.py @@ -1,3 +1,4 @@ +import html import os import re import secrets @@ -292,3 +293,13 @@ def get_russian_month(date: datetime) -> str: """ english_month = date.strftime("%B") return RUSSIAN_MONTHS.get(english_month, english_month) + + +def get_username(user) -> str: + if getattr(user, "full_name", None): + return html.escape(user.full_name) + if getattr(user, "first_name", None): + return html.escape(user.first_name) + if getattr(user, "username", None): + return "@" + html.escape(user.username) + return "Пользователь" diff --git a/hooks/hook_buttons.py b/hooks/hook_buttons.py new file mode 100644 index 00000000..668c4adb --- /dev/null +++ b/hooks/hook_buttons.py @@ -0,0 +1,31 @@ +from aiogram.types import InlineKeyboardMarkup +from aiogram.utils.keyboard import InlineKeyboardBuilder + + +def insert_hook_buttons(builder: InlineKeyboardBuilder, buttons: list) -> InlineKeyboardBuilder: + """ + Вставляет кнопки из хуков в существующий builder (вставка после указанной кнопки через `after`) + """ + markup = builder.as_markup() + new_rows = markup.inline_keyboard.copy() + + for module in buttons: + if isinstance(module, dict) and "after" in module and "button" in module: + after = module["after"] + button = module["button"] + + insert_pos = -1 + for i, row in enumerate(new_rows): + if any(btn.callback_data == after for btn in row): + insert_pos = i + 1 + break + + if 0 <= insert_pos <= len(new_rows): + new_rows.insert(insert_pos, [button]) + else: + new_rows.append([button]) + else: + button = module.get("button") if isinstance(module, dict) else module + new_rows.append([button]) + + return InlineKeyboardBuilder.from_markup(InlineKeyboardMarkup(inline_keyboard=new_rows)) diff --git a/middlewares/__init__.py b/middlewares/__init__.py index f091229e..827e9bb5 100644 --- a/middlewares/__init__.py +++ b/middlewares/__init__.py @@ -24,9 +24,12 @@ def register_middleware( sessionmaker=None, ) -> None: """Регистрирует middleware в диспетчере.""" - direct_start_blocker = DirectStartBlockerMiddleware() - dispatcher.message.outer_middleware(direct_start_blocker) - dispatcher.callback_query.outer_middleware(direct_start_blocker) + + dispatcher.update.outer_middleware(DirectStartBlockerMiddleware()) + + if sessionmaker: + dispatcher.update.outer_middleware(SubscriptionMiddleware()) + dispatcher.update.outer_middleware(BanCheckerMiddleware(sessionmaker)) if middlewares is None: available_middlewares = { @@ -41,10 +44,6 @@ def register_middleware( exclude_set = set(exclude or []) middlewares = [middleware for name, middleware in available_middlewares.items() if name not in exclude_set] - if sessionmaker: - dispatcher.update.outer_middleware(SubscriptionMiddleware()) - dispatcher.update.outer_middleware(BanCheckerMiddleware(sessionmaker)) - handlers = [ dispatcher.message, dispatcher.callback_query, diff --git a/middlewares/direct_start_blocker.py b/middlewares/direct_start_blocker.py index 122bb308..8f539746 100644 --- a/middlewares/direct_start_blocker.py +++ b/middlewares/direct_start_blocker.py @@ -3,9 +3,11 @@ from typing import Any from aiogram import BaseMiddleware from aiogram.types import Message, Update +from sqlalchemy import select from config import DISABLE_DIRECT_START from database import async_session_maker, check_user_exists +from database.models import Coupon, Gift, TrackingSource, User from logger import logger @@ -19,14 +21,13 @@ class DirectStartBlockerMiddleware(BaseMiddleware): event: Update, data: dict[str, Any], ) -> Any: + logger.debug("[DirectStartBlocker] Middleware triggered") + if not DISABLE_DIRECT_START: return await handler(event, data) - if not isinstance(event, Message): - return await handler(event, data) - - message: Message = event - if not message.text: + message: Message | None = getattr(event, "message", None) + if not message or not message.text: return await handler(event, data) tg_id = message.from_user.id @@ -35,23 +36,60 @@ class DirectStartBlockerMiddleware(BaseMiddleware): async with async_session_maker() as session: user_exists = await check_user_exists(session, tg_id) - if user_exists: - logger.debug(f"[DirectStartBlocker] Пользователь {tg_id} уже есть в базе — пропущен") - return await handler(event, data) + if user_exists: + logger.debug(f"[DirectStartBlocker] Пользователь {tg_id} уже есть в базе — пропущен") + return await handler(event, data) - parts = text.split(maxsplit=1) + parts = text.split(maxsplit=1) + if parts[0] == "/start": + if len(parts) == 1: + logger.info(f"[DirectStartBlocker] Прямой старт запрещён для нового пользователя {tg_id}") + return - if parts[0] != "/start": - return await handler(event, data) + start_param = parts[1].strip() + if not start_param or not start_param.startswith(self.allowed_prefixes): + logger.info(f"[DirectStartBlocker] Отклонена неизвестная ссылка от {tg_id}: {start_param!r}") + return - if len(parts) == 1: - logger.info(f"[DirectStartBlocker] Прямой старт запрещён для нового пользователя {tg_id}") + if start_param.startswith("coupons_"): + code = start_param.removeprefix("coupons_") + result = await session.execute(select(Coupon).where(Coupon.code == code)) + if not result.scalar_one_or_none(): + logger.info(f"[DirectStartBlocker] Купон не найден: {code!r}") + return + + elif start_param.startswith("gift_"): + gift_id = start_param.removeprefix("gift_") + result = await session.execute(select(Gift).where(Gift.id == gift_id)) + if not result.scalar_one_or_none(): + logger.info(f"[DirectStartBlocker] Подарок не найден: {gift_id!r}") + return + + elif start_param.startswith("referral_"): + try: + ref_id = int(start_param.removeprefix("referral_")) + result = await session.execute(select(User).where(User.tg_id == ref_id)) + if not result.scalar_one_or_none(): + logger.info(f"[DirectStartBlocker] Реферал не найден: {ref_id!r}") + return + except ValueError: + logger.info(f"[DirectStartBlocker] Неверный формат referral-ссылки: {start_param!r}") + return + + elif start_param.startswith("utm"): + utm_code = start_param + result = await session.execute(select(TrackingSource).where(TrackingSource.code == utm_code)) + if not result.scalar_one_or_none(): + logger.info(f"[DirectStartBlocker] UTM не найден: {utm_code!r}") + return + + logger.info(f"[DirectStartBlocker] Разрешённая и валидная ссылка от {tg_id}: {start_param!r}") + return await handler(event, data) + + if text.startswith("/"): + logger.info( + f"[DirectStartBlocker] Команда '{text}' отклонена для незарегистрированного пользователя {tg_id}" + ) return - start_param = parts[1].strip() - if not start_param or not start_param.startswith(self.allowed_prefixes): - logger.info(f"[DirectStartBlocker] Отклонена неизвестная ссылка от {tg_id}: {start_param!r}") - return - - logger.debug(f"[DirectStartBlocker] Разрешённая ссылка от {tg_id}: {start_param!r}") return await handler(event, data) diff --git a/middlewares/maintenance.py b/middlewares/maintenance.py index 09a41ae3..ed647371 100644 --- a/middlewares/maintenance.py +++ b/middlewares/maintenance.py @@ -1,23 +1,47 @@ +from collections.abc import Awaitable, Callable +from typing import Any + from aiogram import BaseMiddleware -from aiogram.types import CallbackQuery, Message +from aiogram.types import CallbackQuery, Message, Update from config import ADMIN_ID +from database import async_session_maker +from database.models import Admin maintenance_mode = False class MaintenanceModeMiddleware(BaseMiddleware): - async def __call__(self, handler, event, data): - if maintenance_mode: - user_id = None - if isinstance(event, Message): - user_id = event.from_user.id - elif isinstance(event, CallbackQuery): - user_id = event.from_user.id + async def __call__( + self, + handler: Callable[[Update, dict[str, Any]], Awaitable[Any]], + event: Update, + data: dict[str, Any], + ) -> Any: + if not maintenance_mode: + return await handler(event, data) - if user_id and user_id not in ADMIN_ID: - await event.answer("⚙️ Бот временно недоступен. Ведутся технические работы.") - return + user_id = None + if isinstance(event, Message): + user_id = event.from_user.id + elif isinstance(event, CallbackQuery): + user_id = event.from_user.id - return await handler(event, data) + if not user_id: + return + + if user_id in ADMIN_ID: + return await handler(event, data) + + async with async_session_maker() as session: + db_admin = await session.get(Admin, user_id) + if db_admin: + return await handler(event, data) + + if isinstance(event, CallbackQuery): + await event.answer("⚙️ Бот временно недоступен. Ведутся технические работы.", show_alert=True) + elif isinstance(event, Message): + await event.answer("⚙️ Бот временно недоступен. Ведутся технические работы.") + + return diff --git a/middlewares/subscription.py b/middlewares/subscription.py index c878d5f7..aaf1fdec 100644 --- a/middlewares/subscription.py +++ b/middlewares/subscription.py @@ -2,6 +2,7 @@ from collections.abc import Awaitable, Callable from typing import Any from aiogram import BaseMiddleware +from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError from aiogram.fsm.context import FSMContext from aiogram.types import InlineKeyboardButton, Message, Update from aiogram.utils.keyboard import InlineKeyboardBuilder @@ -15,9 +16,6 @@ from logger import logger class SubscriptionMiddleware(BaseMiddleware): - def __init__(self) -> None: - pass - async def __call__( self, handler: Callable[[Update, dict[str, Any]], Awaitable[Any]], @@ -29,13 +27,16 @@ class SubscriptionMiddleware(BaseMiddleware): tg_id = None message = None + from_user = None if event.message: tg_id = event.message.from_user.id message = event.message + from_user = event.message.from_user elif event.callback_query: tg_id = event.callback_query.from_user.id message = event.callback_query.message + from_user = event.callback_query.from_user else: return await handler(event, data) @@ -43,44 +44,39 @@ class SubscriptionMiddleware(BaseMiddleware): member = await bot.get_chat_member(CHANNEL_ID, tg_id) if member.status not in ("member", "administrator", "creator"): logger.info(f"[SubMiddleware] Пользователь {tg_id} не подписан") - - state: FSMContext = data.get("state") - if state: - original_text = message.text or message.caption - - if event.message: - from_user = event.message.from_user - elif event.callback_query: - from_user = event.callback_query.from_user - else: - logger.warning(f"[SubMiddleware] Не удалось определить пользователя из события: {event}") - return await handler(event, data) - - if from_user.is_bot: - logger.warning(f"[SubMiddleware] Пропуск сохранения is_bot=True для {from_user.id}") - else: - state_data = await state.get_data() - if "original_text" not in state_data or "user_data" not in state_data: - user_data = { - "tg_id": from_user.id, - "username": from_user.username, - "first_name": from_user.first_name, - "last_name": from_user.last_name, - "language_code": from_user.language_code, - "is_bot": from_user.is_bot, - } - await state.update_data( - original_text=original_text, - user_data=user_data, - ) - + await self._store_user_state(data, message, from_user) return await self._ask_to_subscribe(message) - except Exception as e: - logger.warning(f"[SubMiddleware] Ошибка при проверке подписки для {tg_id}: {e}") + except (TelegramBadRequest, TelegramForbiddenError) as e: + logger.warning(f"[SubMiddleware] Ошибка при проверке подписки {tg_id}: {e}") + await self._store_user_state(data, message, from_user) return await self._ask_to_subscribe(message) return await handler(event, data) + async def _store_user_state(self, data: dict, message: Message, from_user): + state: FSMContext = data.get("state") + if not state or not from_user or from_user.is_bot: + return + + state_data = await state.get_data() + if "original_text" in state_data and "user_data" in state_data: + return + + original_text = message.text or message.caption + user_data = { + "tg_id": from_user.id, + "username": from_user.username, + "first_name": from_user.first_name, + "last_name": from_user.last_name, + "language_code": from_user.language_code, + "is_bot": from_user.is_bot, + } + + await state.update_data( + original_text=original_text, + user_data=user_data, + ) + async def _ask_to_subscribe(self, message: Message): builder = InlineKeyboardBuilder() builder.row(InlineKeyboardButton(text=SUB_CHANELL, url=CHANNEL_URL)) diff --git a/requirements.txt b/requirements.txt index 640f0fbe..56f1a43a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,17 +4,23 @@ aiogram==3.13.1 aiohappyeyeballs==2.4.3 aiohttp==3.10.10 aiosignal==1.3.1 +alembic==1.16.4 annotated-types==0.7.0 anyio==4.8.0 +APScheduler==3.11.0 async-timeout==4.0.3 asyncpg==0.30.0 attrs==24.2.0 babel==2.17.0 cachetools==5.5.1 certifi==2023.11.17 +cffi==1.17.1 charset-normalizer==3.4.0 -deprecated==1.2.14 +click==8.2.2 +cryptography==45.0.5 +Deprecated==1.2.14 distro==1.9.0 +fastapi==0.116.1 frozenlist==1.4.1 greenlet==3.1.1 h11==0.14.0 @@ -23,35 +29,38 @@ httpx==0.27.2 idna==3.10 loguru==0.7.3 magic-filter==1.0.12 +Mako==1.3.10 +markdown-it-py==3.0.0 +MarkupSafe==3.0.2 +mdurl==0.1.2 multidict==6.1.0 netaddr==1.3.0 +pillow==11.3.0 ping3==4.0.8 propcache==0.2.0 +psutil==7.0.0 +psycopg2-binary==2.9.10 py3xui==0.3.4 +pycparser==2.22 pydantic==2.9.2 -pydantic-core==2.23.4 +pydantic_core==2.23.4 +Pygments==2.19.2 +python-dateutil==2.9.0.post0 pytz==2025.1 +qrcode==8.2 requests==2.32.3 +rich==14.1.0 robokassa==0.3.2 ruff==0.9.5 +six==1.17.0 sniffio==1.3.1 -sqlalchemy==2.0.38 -strenum==0.4.15 -typing-extensions==4.12.2 +SQLAlchemy==2.0.38 +starlette==0.47.2 +StrEnum==0.4.15 +typing_extensions==4.12.2 +tzlocal==5.3.1 urllib3==2.2.3 +uvicorn==0.35.0 wrapt==1.16.0 yarl==1.15.5 yookassa==3.3.0 -httpx -qrcode -pillow -rich -apscheduler -alembic -psycopg2-binary -psutil -uvicorn -fastapi -python-dateutil -aiohttp>=3.8.0 -cryptography>=41.0.0 \ No newline at end of file diff --git a/web/__init__.py b/web/__init__.py index e83685c7..c8148319 100644 --- a/web/__init__.py +++ b/web/__init__.py @@ -1,7 +1,5 @@ from aiohttp.web_urldispatcher import UrlDispatcher -import bot - from config import TBLOCKER_WEBHOOK_PATH from .heleket_payment import heleket_payment_webhook diff --git a/web/kassai_payment.py b/web/kassai_payment.py index 0112664f..974debdb 100644 --- a/web/kassai_payment.py +++ b/web/kassai_payment.py @@ -1,5 +1,4 @@ import hashlib -import json from aiohttp import web