diff --git a/core/bootstrap.py b/core/bootstrap.py index e68d98b0..bf1af583 100644 --- a/core/bootstrap.py +++ b/core/bootstrap.py @@ -1,4 +1,6 @@ from database import async_session_maker +from database.db import warm_pool +from database.tariffs import initialize_all_tariff_weights from .settings.buttons_config import BUTTONS_CONFIG, load_buttons_config, update_buttons_config from .settings.management_config import MANAGEMENT_CONFIG, load_management_config, update_management_config @@ -11,7 +13,9 @@ from .settings.tariffs_config import TARIFFS_CONFIG, load_tariffs_config, update async def bootstrap() -> None: + await warm_pool() async with async_session_maker() as session: + await initialize_all_tariff_weights(session) await load_buttons_config(session) await load_notifications_config(session) await load_modes_config(session) diff --git a/database/coupons.py b/database/coupons.py index 681a78b3..322de6e5 100644 --- a/database/coupons.py +++ b/database/coupons.py @@ -113,6 +113,7 @@ async def create_coupon_usage(session: AsyncSession, coupon_id: int, user_id: in except SQLAlchemyError as e: logger.error(f"❌ Ошибка при сохранении использования купона: {e}") await session.rollback() + raise async def check_coupon_usage(session: AsyncSession, coupon_id: int, user_id: int) -> bool: @@ -136,6 +137,7 @@ async def update_coupon_usage_count(session: AsyncSession, coupon_id: int): except SQLAlchemyError as e: logger.error(f"❌ Ошибка при обновлении купона {coupon_id}: {e}") await session.rollback() + raise def apply_percent_coupon(price_rub: int, coupon: Coupon) -> tuple[int, int]: diff --git a/database/db.py b/database/db.py index 11caa0b3..6fa61ff1 100644 --- a/database/db.py +++ b/database/db.py @@ -1,15 +1,24 @@ +import asyncio + +from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.orm import declarative_base -from config import DATABASE_URL +from config import DATABASE_URL, DB_MAX_OVERFLOW, DB_POOL_SIZE + + +CONCURRENT_UPDATES_LIMIT = DB_POOL_SIZE + DB_MAX_OVERFLOW +MAX_UPDATE_AGE_SEC = 28 engine = create_async_engine( DATABASE_URL, echo=False, future=True, - pool_size=100, - max_overflow=200, + pool_size=DB_POOL_SIZE, + max_overflow=DB_MAX_OVERFLOW, pool_timeout=60, + pool_pre_ping=True, + pool_recycle=300, ) async_session_maker = async_sessionmaker( @@ -19,3 +28,20 @@ async_session_maker = async_sessionmaker( ) Base = declarative_base() + +WARM_POOL_COUNT = 10 + + +async def warm_pool() -> None: + """ + Прогревает пул соединений при старте. + """ + + async def _one() -> None: + async with async_session_maker() as session: + await session.execute(text("SELECT 1")) + + count = min(WARM_POOL_COUNT, DB_POOL_SIZE) + if count <= 0: + return + await asyncio.gather(*[asyncio.create_task(_one()) for _ in range(count)]) diff --git a/database/gifts.py b/database/gifts.py index da6be385..e7754b86 100644 --- a/database/gifts.py +++ b/database/gifts.py @@ -50,4 +50,4 @@ async def store_gift_link( except SQLAlchemyError as e: logger.error(f"❌ Ошибка при сохранении подарка {gift_id}: {e}") await session.rollback() - return False + raise diff --git a/database/importer.py b/database/importer.py index 4dbfa542..674ff7ed 100644 --- a/database/importer.py +++ b/database/importer.py @@ -92,8 +92,9 @@ async def import_keys_from_3xui_db(db_path: str, session: AsyncSession) -> tuple updated_at=datetime.utcnow(), ) ) - except SQLAlchemyError: - continue + except SQLAlchemyError as e: + await session.rollback() + raise RuntimeError(f"Ошибка при импорте пользователя tg_id={tg_id}") from e key_exists = await session.execute(select(Key).where(Key.client_id == client_id)) if key_exists.scalar(): @@ -119,8 +120,9 @@ async def import_keys_from_3xui_db(db_path: str, session: AsyncSession) -> tuple ) ) imported += 1 - except SQLAlchemyError: - continue + except SQLAlchemyError as e: + await session.rollback() + raise RuntimeError(f"Ошибка при импорте ключа client_id={client_id}") from e await session.commit() return imported, skipped diff --git a/database/init_db.py b/database/init_db.py index 6a71654c..c278d755 100644 --- a/database/init_db.py +++ b/database/init_db.py @@ -5,7 +5,6 @@ from sqlalchemy import select from config import ADMIN_ID from database.db import async_session_maker, engine from database.models import Admin, Base, User -from database.tariffs import initialize_all_tariff_weights async def init_db(): @@ -35,5 +34,3 @@ async def init_db(): ) ) await session.commit() - - await initialize_all_tariff_weights(session) diff --git a/database/keys.py b/database/keys.py index 2e276de4..a50d8a35 100644 --- a/database/keys.py +++ b/database/keys.py @@ -85,6 +85,7 @@ async def store_key( except SQLAlchemyError as e: logger.error(f"❌ Ошибка при сохранении ключа: {e}") await session.rollback() + raise async def get_keys(session: AsyncSession, tg_id: int): @@ -155,10 +156,11 @@ async def get_key_count(session: AsyncSession, tg_id: int) -> int: return result.scalar() or 0 -async def delete_key(session: AsyncSession, identifier: int | str): +async def delete_key(session: AsyncSession, identifier: int | str, commit: bool = True): stmt = delete(Key).where(Key.tg_id == identifier if str(identifier).isdigit() else Key.client_id == identifier) await session.execute(stmt) - await session.commit() + if commit: + await session.commit() logger.info(f"Ключ с идентификатором {identifier} удалён") diff --git a/database/notifications.py b/database/notifications.py index f00e2077..de23fd05 100644 --- a/database/notifications.py +++ b/database/notifications.py @@ -31,6 +31,7 @@ async def add_notification(session: AsyncSession, tg_id: int, notification_type: except SQLAlchemyError as e: logger.error(f"❌ Ошибка при добавлении уведомления: {e}") await session.rollback() + raise async def delete_notification(session: AsyncSession, tg_id: int, notification_type: str): @@ -101,6 +102,7 @@ async def check_hot_lead_discount(session: AsyncSession, tg_id: int) -> dict: except Exception as e: logger.error(f"❌ Ошибка при проверке скидки горячего лида для {tg_id}: {e}") + await session.rollback() return {"available": False} @@ -173,4 +175,5 @@ async def check_notifications_bulk( except Exception as e: logger.error(f"Ошибка при массовой проверке уведомлений типа {notification_type}: {e}") + await session.rollback() return [] diff --git a/database/payments.py b/database/payments.py index ceb380e3..a849f432 100644 --- a/database/payments.py +++ b/database/payments.py @@ -106,6 +106,7 @@ async def get_payment_by_id(session: AsyncSession, internal_id: int) -> dict | N } except SQLAlchemyError as e: logger.error(f"Ошибка при поиске платежа id={internal_id}: {e}") + await session.rollback() return None @@ -161,6 +162,7 @@ async def get_payment_by_payment_id(session: AsyncSession, pid: str) -> dict | N } except SQLAlchemyError as e: logger.error(f"Ошибка при поиске платежа payment_id={pid}: {e}") + await session.rollback() return None diff --git a/database/referrals.py b/database/referrals.py index 07e8ab69..730e85ef 100644 --- a/database/referrals.py +++ b/database/referrals.py @@ -224,6 +224,7 @@ async def get_referral_stats(session: AsyncSession, referrer_tg_id: int): except Exception as e: logger.error(f"[ReferralStats] Ошибка при получении статистики для пользователя {referrer_tg_id}: {e}") + await session.rollback() raise diff --git a/database/servers.py b/database/servers.py index bb5090b7..c885e3e4 100644 --- a/database/servers.py +++ b/database/servers.py @@ -100,6 +100,7 @@ async def get_servers(session: AsyncSession, include_enabled: bool = False) -> d return grouped except SQLAlchemyError as e: logger.error(f"Ошибка при получении серверов: {e}") + await session.rollback() return {} @@ -124,6 +125,7 @@ async def check_server_name_by_cluster(session: AsyncSession, server_name: str) return {"cluster_name": row[0]} if row else None except SQLAlchemyError as e: logger.error(f"Ошибка при поиске кластера для сервера {server_name}: {e}") + await session.rollback() return None @@ -161,6 +163,7 @@ async def get_server_by_name(session: AsyncSession, server_name: str) -> dict | return None except SQLAlchemyError as e: logger.error(f"Ошибка при получении сервера {server_name}: {e}") + await session.rollback() return None @@ -209,6 +212,7 @@ async def get_available_clusters(session: AsyncSession) -> list[str]: return [row[0] for row in result.all()] except SQLAlchemyError as e: logger.error(f"Ошибка при получении списка кластеров: {e}") + await session.rollback() return [] diff --git a/database/tariffs.py b/database/tariffs.py index 114a9ed3..0346cde5 100644 --- a/database/tariffs.py +++ b/database/tariffs.py @@ -73,6 +73,7 @@ async def get_tariffs( return tariffs except SQLAlchemyError as e: logger.error(f"[TARIFF] Ошибка при получении тарифов: {e}") + await session.rollback() return [] @@ -83,6 +84,7 @@ async def get_tariff_by_id(session: AsyncSession, tariff_id: int): return dict(tariff.__dict__) if tariff else None except SQLAlchemyError as e: logger.error(f"[TARIFF] Ошибка при получении тарифа по ID {tariff_id}: {e}") + await session.rollback() return None @@ -179,6 +181,7 @@ async def check_tariff_exists(session: AsyncSession, tariff_id: int): return False except SQLAlchemyError as e: logger.error(f"[TARIFF] Ошибка при проверке тарифа {tariff_id}: {e}") + await session.rollback() return False @@ -195,6 +198,7 @@ async def get_tariff_sort_order(session: AsyncSession, tariff_id: int) -> int: return sort_order except SQLAlchemyError as e: logger.error(f"[TARIFF] Ошибка при получении sort_order для тарифа {tariff_id}: {e}") + await session.rollback() return None diff --git a/database/temporary_data.py b/database/temporary_data.py index 55f34ed8..db40d8ea 100644 --- a/database/temporary_data.py +++ b/database/temporary_data.py @@ -25,6 +25,7 @@ async def create_temporary_data(session: AsyncSession, tg_id: int, state: str, d except SQLAlchemyError as e: logger.error(f"❌ Ошибка при сохранении временных данных для {tg_id}: {e}") await session.rollback() + raise async def get_temporary_data(session: AsyncSession, tg_id: int) -> dict | None: diff --git a/database/tracking_sources.py b/database/tracking_sources.py index 37b53bad..17f9da9d 100644 --- a/database/tracking_sources.py +++ b/database/tracking_sources.py @@ -21,6 +21,7 @@ async def create_tracking_source(session: AsyncSession, name: str, code: str, ty except SQLAlchemyError as e: logger.error(f"❌ Ошибка при создании источника {code}: {e}") await session.rollback() + raise async def get_all_tracking_sources(session: AsyncSession) -> list[dict]: diff --git a/database/users.py b/database/users.py index 31a49dfe..e51e1e3a 100644 --- a/database/users.py +++ b/database/users.py @@ -100,6 +100,7 @@ async def set_user_balance(session: AsyncSession, tg_id: int, balance: float) -> except SQLAlchemyError as e: logger.error(f"Ошибка при установке баланса для пользователя {tg_id}: {e}") await session.rollback() + raise async def update_trial(session: AsyncSession, tg_id: int, status: int): @@ -110,6 +111,7 @@ async def update_trial(session: AsyncSession, tg_id: int, status: int): except SQLAlchemyError as e: logger.error(f"[DB] Ошибка при обновлении триала пользователя {tg_id}: {e}") await session.rollback() + raise async def get_trial(session: AsyncSession, tg_id: int) -> int: @@ -205,7 +207,7 @@ async def delete_user_data(session: AsyncSession, tg_id: int): delete(Referral).where(or_(Referral.referrer_tg_id == tg_id, Referral.referred_tg_id == tg_id)) ) await session.execute(delete(CouponUsage).where(CouponUsage.user_id == tg_id)) - await delete_key(session, tg_id) + await delete_key(session, tg_id, commit=False) await session.execute(delete(TemporaryData).where(TemporaryData.tg_id == tg_id)) await session.execute(delete(BlockedUser).where(BlockedUser.tg_id == tg_id)) await session.execute(delete(User).where(User.tg_id == tg_id)) diff --git a/filters/admin.py b/filters/admin.py index 5daba74d..23b1a5f2 100644 --- a/filters/admin.py +++ b/filters/admin.py @@ -1,20 +1,48 @@ +import time + from aiogram.filters import BaseFilter from aiogram.types import CallbackQuery, Message -from sqlalchemy import select, exists +from sqlalchemy import select +from config import ADMIN_ID from database.db import async_session_maker from database.models import Admin +_ADMIN_CACHE: dict[int, tuple[float, bool, bool]] = {} +_ADMIN_CACHE_TTL = 60 + + +def _get_cached_admin(user_id: int) -> tuple[bool, bool] | None: + now = time.time() + entry = _ADMIN_CACHE.get(user_id) + if entry and entry[0] > now: + return entry[1], entry[2] + return None + + +def _set_cached_admin(user_id: int, is_admin: bool, is_superadmin: bool) -> None: + _ADMIN_CACHE[user_id] = (time.time() + _ADMIN_CACHE_TTL, is_admin, is_superadmin) + + class IsAdminFilter(BaseFilter): async def __call__(self, event: Message | CallbackQuery) -> bool: if not event.from_user: return False + user_id = event.from_user.id + cached = _get_cached_admin(user_id) + if cached is not None: + return cached[0] + try: async with async_session_maker() as session: - result = await session.execute(select(exists().where(Admin.tg_id == event.from_user.id))) - return result.scalar() + admin = (await session.execute(select(Admin).where(Admin.tg_id == user_id))).scalar_one_or_none() + admin_ids = (ADMIN_ID,) if isinstance(ADMIN_ID, int) else ADMIN_ID + is_admin = admin is not None or user_id in admin_ids + is_super = admin.role != "moderator" if admin else (user_id in admin_ids) + _set_cached_admin(user_id, is_admin, is_super) + return is_admin except (Exception,): return False @@ -24,13 +52,19 @@ class IsSuperAdminFilter(BaseFilter): if not event.from_user: return False + user_id = event.from_user.id + cached = _get_cached_admin(user_id) + if cached is not None: + return cached[1] + try: async with async_session_maker() as session: - admin = ( - await session.execute(select(Admin).where(Admin.tg_id == event.from_user.id)) - ).scalar_one_or_none() + admin = (await session.execute(select(Admin).where(Admin.tg_id == user_id))).scalar_one_or_none() if not admin: + _set_cached_admin(user_id, False, False) return False - return admin.role != "moderator" + is_super = admin.role != "moderator" + _set_cached_admin(user_id, True, is_super) + return is_super except (Exception,): return False diff --git a/handlers/fallback_router.py b/handlers/fallback_router.py index 9bc7c7af..bb50c8d4 100644 --- a/handlers/fallback_router.py +++ b/handlers/fallback_router.py @@ -1,9 +1,10 @@ +from typing import Any + from aiogram import F, Router from aiogram.types import InlineKeyboardButton, Message from aiogram.utils.keyboard import InlineKeyboardBuilder from config import SUPPORT_CHAT_URL -from database import async_session_maker from handlers.buttons import MAIN_MENU, SUPPORT from handlers.texts import FALLBACK_MESSAGE from hooks.hooks import run_hooks @@ -13,23 +14,21 @@ fallback_router = Router() @fallback_router.message(F.text) -async def handle_unhandled_messages(message: Message): - async with async_session_maker() as session: - await run_hooks( - "user_message", - user_id=message.from_user.id, - message_text=message.text, - username=message.from_user.username, - first_name=message.from_user.first_name, - last_name=message.from_user.last_name, - session=session, - message=message, - ) +async def handle_unhandled_messages(message: Message, session: Any): + await run_hooks( + "user_message", + user_id=message.from_user.id, + message_text=message.text, + username=message.from_user.username, + first_name=message.from_user.first_name, + last_name=message.from_user.last_name, + session=session, + message=message, + ) keyboard = InlineKeyboardBuilder() keyboard.row(InlineKeyboardButton(text=SUPPORT, url=SUPPORT_CHAT_URL)) keyboard.row(InlineKeyboardButton(text=MAIN_MENU, callback_data="profile")) - await message.answer( FALLBACK_MESSAGE, reply_markup=keyboard.as_markup(), diff --git a/handlers/keys/key_create.py b/handlers/keys/key_create.py index 34065db6..3599ec8b 100644 --- a/handlers/keys/key_create.py +++ b/handlers/keys/key_create.py @@ -306,6 +306,7 @@ async def handle_key_creation( tg_id=tg_id, cluster_name=cluster_name, group_code=group_code, + tariff_subgroup_hash=None, ) await state.set_state(Form.waiting_for_server_selection) @@ -329,6 +330,8 @@ async def show_tariffs_in_subgroup_user(callback: CallbackQuery, state: FSMConte ) return + await state.update_data(tariff_subgroup_hash=subgroup_hash) + tariffs_for_cluster = await get_tariffs_for_cluster(session, cluster_name) filtered: list[dict[str, Any]] = [] @@ -383,6 +386,59 @@ async def back_to_tariff_group_list(callback: CallbackQuery, state: FSMContext, ) +@router.callback_query(F.data == "back_to_subgroup_tariffs") +async def back_to_subgroup_tariffs(callback: CallbackQuery, state: FSMContext, session: AsyncSession): + """Возврат к списку тарифов текущей подгруппы (из конфигуратора).""" + data = await state.get_data() + subgroup_hash = data.get("tariff_subgroup_hash") + if not subgroup_hash: + await back_to_tariff_group_list(callback, state, session) + return + cluster_name = data.get("cluster_name") + group_code = data.get("group_code") + + subgroup = await find_subgroup_by_hash(session, subgroup_hash, group_code) + if not subgroup: + await back_to_tariff_group_list(callback, state, session) + return + + tariffs_for_cluster = await get_tariffs_for_cluster(session, cluster_name) + filtered: list[dict[str, Any]] = [] + if tariffs_for_cluster: + gc = tariffs_for_cluster[0].get("group_code") + if gc: + tariffs = await get_tariffs(session, group_code=gc) + filtered = [ + t for t in tariffs if t.get("subgroup_title") == subgroup and t.get("is_active") + ] + + if not filtered: + await back_to_tariff_group_list(callback, state, session) + return + + tg_id = callback.from_user.id + language_code = getattr(callback.from_user, "language_code", None) + builder = InlineKeyboardBuilder() + for tariff in filtered: + await add_tariff_button_generic( + builder=builder, + tariff=tariff, + session=session, + tg_id=tg_id, + language_code=language_code, + callback_prefix="select_tariff_plan", + ) + builder.row(InlineKeyboardButton(text=BACK, callback_data="back_to_tariff_group_list")) + builder.row(InlineKeyboardButton(text=MAIN_MENU, callback_data="profile")) + + await edit_or_send_message( + target_message=callback.message, + text=f"{subgroup}\n\nВыберите тариф:", + reply_markup=builder.as_markup(), + ) + await callback.answer() + + async def create_key( tg_id: int, expiry_time: datetime, diff --git a/handlers/payments/heleket/handlers.py b/handlers/payments/heleket/handlers.py index 238a4817..8676f29e 100644 --- a/handlers/payments/heleket/handlers.py +++ b/handlers/payments/heleket/handlers.py @@ -83,7 +83,7 @@ async def handle_custom_amount_input_heleket( method = enabled_methods[0] try: - payment_url = await generate_heleket_payment_link(amount, tg_id, method) + payment_url = await generate_heleket_payment_link(amount, tg_id, method, session) if not payment_url or payment_url == "https://heleket.com/": await edit_or_send_message( diff --git a/handlers/payments/heleket/service.py b/handlers/payments/heleket/service.py index 6a3ab162..56d2eb24 100644 --- a/handlers/payments/heleket/service.py +++ b/handlers/payments/heleket/service.py @@ -243,7 +243,7 @@ async def handle_custom_amount_input(message: types.Message, state: FSMContext, amount_rub = int(await to_rub(user_amount, "USD", session=session_http)) await state.update_data(amount=amount_rub) - payment_url = await generate_heleket_payment_link(amount_rub, message.chat.id, method) + payment_url = await generate_heleket_payment_link(amount_rub, message.chat.id, method, session) if not payment_url or payment_url == "https://heleket.com/": await edit_or_send_message( @@ -298,7 +298,7 @@ async def process_amount_selection(callback_query: types.CallbackQuery, state: F return await state.update_data(amount=amount) - payment_url = await generate_heleket_payment_link(amount, callback_query.message.chat.id, method) + payment_url = await generate_heleket_payment_link(amount, callback_query.message.chat.id, method, session) if not payment_url or payment_url == "https://heleket.com/": await edit_or_send_message( @@ -323,10 +323,13 @@ async def process_amount_selection(callback_query: types.CallbackQuery, state: F await state.set_state(ReplenishBalanceHeleket.waiting_for_payment_confirmation) -async def generate_heleket_payment_link(amount: int, tg_id: int, method: dict) -> str: +async def generate_heleket_payment_link( + amount: int, tg_id: int, method: dict, session: AsyncSession | None = None +) -> str: """ Создание платежа в Heleket и получение ссылки на оплату. amount — сумма в RUB, method['currency'] — валюта провайдера (обычно USD). + session — сессия из хендлера; если не передана, создаётся своя (лишняя нагрузка на пул). """ url = "https://api.heleket.com/v1/payment" unique_order_id = f"{int(time.time())}_{tg_id}" @@ -371,9 +374,9 @@ async def generate_heleket_payment_link(amount: int, tg_id: int, method: dict) - if resp_json.get("state") == 0: payment_url = resp_json.get("result", {}).get("url") if payment_url: - async with async_session_maker() as dbs: + if session is not None: await add_payment( - session=dbs, + session=session, tg_id=tg_id, amount=float(amount), payment_system="HELEKET", @@ -381,6 +384,17 @@ async def generate_heleket_payment_link(amount: int, tg_id: int, method: dict) - currency="RUB", payment_id=unique_order_id, ) + else: + async with async_session_maker() as dbs: + await add_payment( + session=dbs, + tg_id=tg_id, + amount=float(amount), + payment_system="HELEKET", + status="pending", + currency="RUB", + payment_id=unique_order_id, + ) logger.info(f"Heleket payment URL created for user {tg_id}") return payment_url else: diff --git a/handlers/payments/kassai/handlers.py b/handlers/payments/kassai/handlers.py index dd43ca90..704f440f 100644 --- a/handlers/payments/kassai/handlers.py +++ b/handlers/payments/kassai/handlers.py @@ -107,7 +107,7 @@ async def _handle_custom_amount_input_kassai( return try: - payment_url = await generate_kassai_payment_link(amount, tg_id, method) + payment_url = await generate_kassai_payment_link(amount, tg_id, method, session) if not payment_url or payment_url == "https://fk.life/": await edit_or_send_message( diff --git a/handlers/payments/kassai/service.py b/handlers/payments/kassai/service.py index 9382ed16..7ea57e82 100644 --- a/handlers/payments/kassai/service.py +++ b/handlers/payments/kassai/service.py @@ -266,7 +266,7 @@ async def handle_custom_amount_input(message: types.Message, state: FSMContext, amount_rub = int(await to_rub(user_amount, "USD", session=session_http)) await state.update_data(amount=amount_rub) - payment_url = await generate_kassai_payment_link(amount_rub, message.chat.id, method) + payment_url = await generate_kassai_payment_link(amount_rub, message.chat.id, method, session) if not payment_url or payment_url == "https://fk.life/": await edit_or_send_message( @@ -328,7 +328,7 @@ async def process_amount_selection(callback_query: types.CallbackQuery, state: F return await state.update_data(amount=amount) - payment_url = await generate_kassai_payment_link(amount, callback_query.message.chat.id, method) + payment_url = await generate_kassai_payment_link(amount, callback_query.message.chat.id, method, session) if not payment_url or payment_url == "https://fk.life/": await edit_or_send_message( @@ -353,9 +353,12 @@ async def process_amount_selection(callback_query: types.CallbackQuery, state: F await state.set_state(ReplenishBalanceKassaiState.waiting_for_payment_confirmation) -async def generate_kassai_payment_link(amount: int, tg_id: int, method: dict) -> str: +async def generate_kassai_payment_link( + amount: int, tg_id: int, method: dict, session: AsyncSession | None = None +) -> str: """ - Создание заказа в KassaAI и получение ссылки на оплату + Создание заказа в KassaAI и получение ссылки на оплату. + session — сессия из хендлера; если не передана, создаётся своя (лишняя нагрузка на пул). """ nonce = int(time.time()) unique_payment_id = f"{nonce}_{tg_id}" @@ -393,9 +396,9 @@ async def generate_kassai_payment_link(amount: int, tg_id: int, method: dict) -> if resp_json.get("type") == "success": payment_url = resp_json.get("location") if payment_url: - async with async_session_maker() as dbs: + if session is not None: await add_payment( - session=dbs, + session=session, tg_id=tg_id, amount=float(amount), payment_system="KASSAI", @@ -403,6 +406,17 @@ async def generate_kassai_payment_link(amount: int, tg_id: int, method: dict) -> currency="RUB", payment_id=unique_payment_id, ) + else: + async with async_session_maker() as dbs: + await add_payment( + session=dbs, + tg_id=tg_id, + amount=float(amount), + payment_system="KASSAI", + status="pending", + currency="RUB", + payment_id=unique_payment_id, + ) logger.info(f"KassaAI payment URL created for user {tg_id}") return payment_url logger.error(f"KassaAI: No location in response: {resp_json}") diff --git a/handlers/payments/yookassa/service.cpython-312-x86_64-linux-gnu.so b/handlers/payments/yookassa/service.cpython-312-x86_64-linux-gnu.so index 95da092c..c3cf9a82 100644 Binary files a/handlers/payments/yookassa/service.cpython-312-x86_64-linux-gnu.so and b/handlers/payments/yookassa/service.cpython-312-x86_64-linux-gnu.so differ diff --git a/handlers/profile.py b/handlers/profile.py index 3b15fa7e..c9c17e28 100644 --- a/handlers/profile.py +++ b/handlers/profile.py @@ -1,4 +1,3 @@ -import asyncio import os from aiogram import F, Router @@ -63,44 +62,35 @@ async def process_callback_view_profile( chat_id = chat.id username = get_username(user or chat) - key_count, balance_rub, trial_status = await asyncio.gather( - get_key_count(session, chat_id), - get_balance(session, chat_id), - get_trial(session, chat_id), - ) + key_count = await get_key_count(session, chat_id) + balance_rub = await get_balance(session, chat_id) + trial_status = await get_trial(session, chat_id) balance_rub = balance_rub or 0 - balance_text_task = asyncio.create_task( - format_for_user( - session, - chat_id, - balance_rub, - getattr(user, "language_code", None), - ) + balance_text = await format_for_user( + session, + chat_id, + balance_rub, + getattr(user, "language_code", None), ) - profile_menu_buttons_task = asyncio.create_task( - run_hooks("profile_menu", chat_id=chat_id, admin=admin, session=session) + profile_menu_buttons = await run_hooks( + "profile_menu", chat_id=chat_id, admin=admin, session=session ) - profile_text_hooks_task = asyncio.create_task( - run_hooks( - "profile_text", - username=username, - chat_id=chat_id, - balance=int(balance_rub), - key_count=key_count, - session=session, - ) + text_hooks = await run_hooks( + "profile_text", + username=username, + chat_id=chat_id, + balance=int(balance_rub), + key_count=key_count, + session=session, ) - balance_text = await balance_text_task - profile_message = profile_message_send(username, chat_id, balance_text, key_count) if key_count == 0: profile_message += ADD_SUBSCRIPTION_HINT else: profile_message += f"\n
{NEWS_MESSAGE}" - text_hooks = await profile_text_hooks_task if text_hooks: profile_message = text_hooks[0] @@ -127,7 +117,6 @@ async def process_callback_view_profile( if extra_buttons: builder.row(*extra_buttons) - profile_menu_buttons = await profile_menu_buttons_task builder = insert_hook_buttons(builder, profile_menu_buttons) if BUTTONS_CONFIG.get("INSTRUCTIONS_BUTTON_ENABLE", INSTRUCTIONS_BUTTON): diff --git a/handlers/tariffs/addons/key_addons_pack.py b/handlers/tariffs/addons/key_addons_pack.py index 9532504c..91cd5dd7 100644 --- a/handlers/tariffs/addons/key_addons_pack.py +++ b/handlers/tariffs/addons/key_addons_pack.py @@ -16,7 +16,6 @@ from database import ( get_key_details, get_tariff_by_id, save_key_config_with_mode, - update_balance, ) from database.models import User from handlers.buttons import BACK, CONFIRM_ADDON_BUTTON_TEXT, PAYMENT diff --git a/handlers/tariffs/buy/key_tariffs.py b/handlers/tariffs/buy/key_tariffs.py index b78d8419..9b1b3473 100644 --- a/handlers/tariffs/buy/key_tariffs.py +++ b/handlers/tariffs/buy/key_tariffs.py @@ -12,7 +12,7 @@ from config import USE_NEW_PAYMENT_FLOW from core.settings.tariffs_config import normalize_tariff_config from database import get_balance, get_tariff_by_id from database.notifications import check_hot_lead_discount -from handlers.buttons import CONFIG_PAY_BUTTON_TEXT, MAIN_MENU, PAYMENT +from handlers.buttons import BACK, CONFIG_PAY_BUTTON_TEXT, MAIN_MENU, PAYMENT from handlers.payments.currency_rates import format_for_user from handlers.payments.fast_payment_flow import try_fast_payment_flow from handlers.tariffs.tariff_display import GB @@ -237,6 +237,7 @@ async def proceed_purchase_with_values( builder = InlineKeyboardBuilder() builder.row(InlineKeyboardButton(text=PAYMENT, callback_data="pay")) + builder.row(InlineKeyboardButton(text=BACK, callback_data="back_to_tariff_group_list")) builder.row(InlineKeyboardButton(text=MAIN_MENU, callback_data="profile")) await edit_or_send_message( target_message=callback_query.message, @@ -507,12 +508,18 @@ async def render_user_config_screen( is_renew_mode = data.get("renew_mode") == "renew" confirm_prefix = "cfg_renew_confirm" if is_renew_mode else "cfg_user_confirm" + back_callback = ( + "back_to_subgroup_tariffs" + if data.get("tariff_subgroup_hash") + else "back_to_tariff_group_list" + ) builder.row( InlineKeyboardButton( text=CONFIG_PAY_BUTTON_TEXT.format(amount=price_text), callback_data=f"{confirm_prefix}|{tariff_id}", ) ) + builder.row(InlineKeyboardButton(text=BACK, callback_data=back_callback)) builder.row(InlineKeyboardButton(text=MAIN_MENU, callback_data="profile")) await state.update_data( @@ -590,31 +597,17 @@ async def start_user_tariff_configurator( await render_user_config_screen(callback_query, state, session) -async def show_price_and_confirm(callback_query: CallbackQuery, state: FSMContext, session: Any | None): +async def show_price_and_confirm(callback_query: CallbackQuery, state: FSMContext, session: Any): """Обновляет экран конфигурации и показывает актуальную цену.""" - if session is None: - from database import async_session_maker - - async with async_session_maker() as new_session: - await show_price_and_confirm(callback_query, state, new_session) - return - await render_user_config_screen(callback_query, state, session) -async def finalize_config_and_purchase(callback_query: CallbackQuery, state: FSMContext, session: Any | None): +async def finalize_config_and_purchase(callback_query: CallbackQuery, state: FSMContext, session: Any): """Фиксирует выбор пользователя и проводит оплату тарифа.""" data = await state.get_data() tariff_id = data.get("config_tariff_id") cfg = data.get("tariff_config") or {} - if session is None: - from database import async_session_maker - - async with async_session_maker() as new_session: - await finalize_config_and_purchase(callback_query, state, new_session) - return - tariff = await get_tariff_by_id(session, int(tariff_id)) if not tariff: await edit_or_send_message( diff --git a/middlewares/__init__.py b/middlewares/__init__.py index 3f323477..18664bb3 100644 --- a/middlewares/__init__.py +++ b/middlewares/__init__.py @@ -8,6 +8,7 @@ from middlewares.subscription import SubscriptionMiddleware from .admin import AdminMiddleware from .answer import CallbackAnswerMiddleware +from .concurrency import ConcurrencyLimiterMiddleware from .direct_start_blocker import DirectStartBlockerMiddleware from .loggings import LoggingMiddleware from .maintenance import MaintenanceModeMiddleware @@ -34,6 +35,7 @@ def register_middleware( dispatcher.update.outer_middleware(StreamProbeMiddleware("global")) if sessionmaker: + dispatcher.update.outer_middleware(wrap(ConcurrencyLimiterMiddleware(), "concurrency")) dispatcher.update.outer_middleware(wrap(SessionMiddleware(sessionmaker), "session")) if DISABLE_DIRECT_START: diff --git a/middlewares/answer.py b/middlewares/answer.py index 5da28936..94400486 100644 --- a/middlewares/answer.py +++ b/middlewares/answer.py @@ -15,8 +15,14 @@ class CallbackAnswerMiddleware(BaseMiddleware): data: dict[str, Any], ) -> Any: if isinstance(event, CallbackQuery): - await event.answer() + try: + await event.answer() + except Exception: + pass if isinstance(event.message, InaccessibleMessage): - new_message = await bot.send_message(event.message.chat.id, "⏳") - object.__setattr__(event, "message", new_message) + try: + new_message = await bot.send_message(event.message.chat.id, "⏳") + object.__setattr__(event, "message", new_message) + except Exception: + pass return await handler(event, data) diff --git a/middlewares/concurrency.py b/middlewares/concurrency.py new file mode 100644 index 00000000..f583f00c --- /dev/null +++ b/middlewares/concurrency.py @@ -0,0 +1,59 @@ +import asyncio +import time +from collections.abc import Awaitable, Callable +from typing import Any + +from aiogram import BaseMiddleware, Bot +from aiogram.types import CallbackQuery, Message, TelegramObject + +from database.db import CONCURRENT_UPDATES_LIMIT, MAX_UPDATE_AGE_SEC + + +class ConcurrencyLimiterMiddleware(BaseMiddleware): + """ + Регистрируется до SessionMiddleware. Ограничивает число апдейтов, одновременно + получающих сессию, и отсекает апдейты, ждавшие слишком долго. + """ + + def __init__(self) -> None: + self._semaphore = asyncio.Semaphore(CONCURRENT_UPDATES_LIMIT) + + async def __call__( + self, + handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]], + event: TelegramObject, + data: dict[str, Any], + ) -> Any: + data["request_time"] = time.monotonic() + await self._semaphore.acquire() + try: + age = time.monotonic() - data["request_time"] + if age > MAX_UPDATE_AGE_SEC: + await self._reject_stale(event, data) + return None + return await handler(event, data) + finally: + self._semaphore.release() + + async def _reject_stale(self, event: TelegramObject, data: dict[str, Any]) -> None: + if isinstance(event, CallbackQuery): + bot: Bot = data.get("bot") + if bot: + try: + await bot.answer_callback_query( + event.id, + text="Время ожидания истекло. Нажмите ещё раз.", + show_alert=False, + ) + except Exception: + pass + elif isinstance(event, Message) and event.text and event.chat: + bot: Bot = data.get("bot") + if bot: + try: + await bot.send_message( + event.chat.id, + "Сейчас высокая нагрузка. Отправьте команду ещё раз через пару секунд.", + ) + except Exception: + pass diff --git a/middlewares/session.py b/middlewares/session.py index 741aecf3..35803280 100644 --- a/middlewares/session.py +++ b/middlewares/session.py @@ -9,10 +9,20 @@ class SessionMiddleware(BaseMiddleware): if data.get("session"): return await handler(event, data) - async with self.sessionmaker() as session: - data["session"] = session + session = self.sessionmaker() + data["session"] = session + try: + result = await handler(event, data) + await session.commit() + return result + except Exception: try: - return await handler(event, data) - except Exception: await session.rollback() - raise + except Exception: + pass + raise + finally: + try: + await session.close() + except Exception: + pass diff --git a/utils/custom_emojis.py b/utils/custom_emojis.py index 108d9bb0..5685bad4 100644 --- a/utils/custom_emojis.py +++ b/utils/custom_emojis.py @@ -71,8 +71,7 @@ async def _fetch_placeholder(emoji_id: str) -> str: async def _replace_markers(text: str) -> tuple[str, list[MessageEntity]]: - """Replace markers with placeholders and build custom emoji entities. - """ + """Replace markers with placeholders and build custom emoji entities.""" if not text: return text, [] diff --git a/utils/errors.py b/utils/errors.py index 41cf7e8b..82e398e8 100644 --- a/utils/errors.py +++ b/utils/errors.py @@ -1,3 +1,5 @@ +import html +import re import traceback from aiogram import Bot, Dispatcher @@ -11,6 +13,15 @@ from database import async_session_maker from logger import logger +_OBFUSCATED_MIN_SEQ = 15 +_PLACEHOLDER = "