diff --git a/bot/handlers/admin/sync_admin.py b/bot/handlers/admin/sync_admin.py index ed26d0e..4beca0f 100644 --- a/bot/handlers/admin/sync_admin.py +++ b/bot/handlers/admin/sync_admin.py @@ -16,11 +16,14 @@ router = Router(name="admin_sync_router") @router.message(Command("sync")) -async def sync_command_handler(message_event: Union[types.Message, - types.CallbackQuery], - bot: Bot, settings: Settings, i18n_data: dict, - panel_service: PanelApiService, - session: AsyncSession): +async def sync_command_handler( + message_event: Union[types.Message, types.CallbackQuery], + bot: Bot, + settings: Settings, + i18n_data: dict, + panel_service: PanelApiService, + session: AsyncSession, +): current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE) i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") if not i18n: @@ -33,14 +36,15 @@ async def sync_command_handler(message_event: Union[types.Message, return _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) - target_chat_id = message_event.chat.id if isinstance( - message_event, types.Message) else ( - message_event.message.chat.id if message_event.message else None) + target_chat_id = ( + message_event.chat.id + if isinstance(message_event, types.Message) + else (message_event.message.chat.id if message_event.message else None) + ) if not target_chat_id: logging.error("Sync handler: could not determine target_chat_id.") if isinstance(message_event, types.CallbackQuery): - await message_event.answer("Error initiating sync.", - show_alert=True) + await message_event.answer("Error initiating sync.", show_alert=True) return if isinstance(message_event, types.Message): @@ -59,21 +63,21 @@ async def sync_command_handler(message_event: Union[types.Message, if panel_users_data is None: error_msg = "Failed to fetch users from panel or panel API issue." sync_errors.append(error_msg) - await panel_sync_dal.update_panel_sync_status( - session, "failed", error_msg) + await panel_sync_dal.update_panel_sync_status(session, "failed", error_msg) await session.commit() - await bot.send_message(target_chat_id, - _("sync_failed", details=error_msg)) + await bot.send_message(target_chat_id, _("sync_failed", details=error_msg)) return if not panel_users_data: status_msg = "No users found in the panel to sync." await panel_sync_dal.update_panel_sync_status( - session, "success", status_msg, 0, 0) + session, "success", status_msg, 0, 0 + ) await session.commit() await bot.send_message( target_chat_id, - _("sync_completed", status="Success", details=status_msg)) + _("sync_completed", status="Success", details=status_msg), + ) return total_panel_users = len(panel_users_data) @@ -81,9 +85,9 @@ async def sync_command_handler(message_event: Union[types.Message, for panel_user_dict in panel_users_data: users_processed_count += 1 - panel_uuid = panel_user_dict.get('uuid') - telegram_id_from_panel_str = panel_user_dict.get('telegramId') - panel_username = panel_user_dict.get('username') + panel_uuid = panel_user_dict.get("uuid") + telegram_id_from_panel_str = panel_user_dict.get("telegramId") + panel_username = panel_user_dict.get("username") if not panel_uuid: logging.warning( @@ -111,26 +115,22 @@ async def sync_command_handler(message_event: Union[types.Message, continue - bot_user = await user_dal.get_user_by_id(session, - telegram_id_from_panel) + bot_user = await user_dal.get_user_by_id(session, telegram_id_from_panel) if not bot_user: user_data_to_create = { - "user_id": - telegram_id_from_panel, - "username": - panel_username, - "panel_user_uuid": - panel_uuid, - "language_code": - settings.DEFAULT_LANGUAGE, - "registration_date": - datetime.fromisoformat( - panel_user_dict['createdAt'].replace("Z", "+00:00")) - if panel_user_dict.get('createdAt') else datetime.now( - timezone.utc) + "user_id": telegram_id_from_panel, + "username": panel_username, + "panel_user_uuid": panel_uuid, + "language_code": settings.DEFAULT_LANGUAGE, + "registration_date": ( + datetime.fromisoformat( + panel_user_dict["createdAt"].replace("Z", "+00:00") + ) + if panel_user_dict.get("createdAt") + else datetime.now(timezone.utc) + ), } - bot_user = await user_dal.create_user(session, - user_data_to_create) + bot_user = await user_dal.create_user(session, user_data_to_create) logging.info( f"Sync: Created new local user {telegram_id_from_panel} from panel data {panel_uuid}." ) @@ -142,8 +142,12 @@ async def sync_command_handler(message_event: Union[types.Message, ) conflicting_user = await user_dal.get_user_by_panel_uuid( - session, panel_uuid) - if conflicting_user and conflicting_user.user_id != telegram_id_from_panel: + session, panel_uuid + ) + if ( + conflicting_user + and conflicting_user.user_id != telegram_id_from_panel + ): sync_errors.append( f"Panel UUID {panel_uuid} for TG {telegram_id_from_panel} already linked to another TG user {conflicting_user.user_id}." ) @@ -151,60 +155,64 @@ async def sync_command_handler(message_event: Union[types.Message, continue await user_dal.update_user( - session, telegram_id_from_panel, { - "panel_user_uuid": panel_uuid, - "username": panel_username - }) + session, + telegram_id_from_panel, + {"panel_user_uuid": panel_uuid, "username": panel_username}, + ) logging.info( f"Sync: Updated panel_uuid for local user {telegram_id_from_panel} to {panel_uuid}." ) panel_sub_link_id = panel_user_dict.get( - 'subscriptionUuid') or panel_user_dict.get('shortUuid') + "subscriptionUuid" + ) or panel_user_dict.get("shortUuid") if panel_sub_link_id: - end_date_str = panel_user_dict.get('expireAt') - start_date_str = panel_user_dict.get('createdAt') + end_date_str = panel_user_dict.get("expireAt") + start_date_str = panel_user_dict.get("createdAt") if end_date_str: try: end_date_obj = datetime.fromisoformat( - end_date_str.replace("Z", "+00:00")) - start_date_obj = datetime.fromisoformat( - start_date_str.replace( - "Z", - "+00:00")) if start_date_str else datetime.now( - timezone.utc) + end_date_str.replace("Z", "+00:00") + ) + start_date_obj = ( + datetime.fromisoformat( + start_date_str.replace("Z", "+00:00") + ) + if start_date_str + else datetime.now(timezone.utc) + ) status_from_panel = panel_user_dict.get( - 'status', 'UNKNOWN').upper() - is_active_flag = 1 if status_from_panel == 'ACTIVE' and end_date_obj > datetime.now( - timezone.utc) else 0 + "status", "UNKNOWN" + ).upper() + is_active_flag = ( + 1 + if status_from_panel == "ACTIVE" + and end_date_obj > datetime.now(timezone.utc) + else 0 + ) sub_payload = { - "user_id": - telegram_id_from_panel, - "panel_user_uuid": - panel_uuid, - "panel_subscription_uuid": - panel_sub_link_id, - "start_date": - start_date_obj, - "end_date": - end_date_obj, - "is_active": - is_active_flag, - "status_from_panel": - status_from_panel, - "traffic_limit_bytes": - panel_user_dict.get('trafficLimitBytes'), - "traffic_used_bytes": - panel_user_dict.get('usedTrafficBytes') + "user_id": telegram_id_from_panel, + "panel_user_uuid": panel_uuid, + "panel_subscription_uuid": panel_sub_link_id, + "start_date": start_date_obj, + "end_date": end_date_obj, + "is_active": is_active_flag, + "status_from_panel": status_from_panel, + "traffic_limit_bytes": panel_user_dict.get( + "trafficLimitBytes" + ), + "traffic_used_bytes": panel_user_dict.get( + "usedTrafficBytes" + ), } await subscription_dal.deactivate_other_active_subscriptions( - session, panel_uuid, panel_sub_link_id) - await subscription_dal.upsert_subscription( - session, sub_payload) + session, panel_uuid, panel_sub_link_id + ) + await subscription_dal.upsert_subscription(session, sub_payload) subscriptions_synced_count += 1 users_synced_successfully += 1 except ValueError as e_date: @@ -217,7 +225,8 @@ async def sync_command_handler(message_event: Union[types.Message, except Exception as e_sub_sync: logging.error( f"Sync: Error syncing subscription for panel user {panel_uuid} (TG ID: {telegram_id_from_panel}): {e_sub_sync}", - exc_info=True) + exc_info=True, + ) sync_errors.append( f"Sub sync error for panel user {panel_uuid} (TG ID: {telegram_id_from_panel})." ) @@ -228,7 +237,8 @@ async def sync_command_handler(message_event: Union[types.Message, else: await subscription_dal.deactivate_other_active_subscriptions( - session, panel_uuid, None) + session, panel_uuid, None + ) logging.info( f"Sync: Panel user {panel_uuid} (TG ID: {telegram_id_from_panel}) has no subscription link on panel. Deactivated local subs if any." ) @@ -239,6 +249,17 @@ async def sync_command_handler(message_event: Union[types.Message, f"Sync progress: {users_processed_count}/{total_panel_users} users processed from panel." ) + panel_uuid_set = {u.get("uuid") for u in panel_users_data if u.get("uuid")} + local_users_with_uuid = await user_dal.get_all_users_with_panel_uuid(session) + for local_user in local_users_with_uuid: + if local_user.panel_user_uuid not in panel_uuid_set: + await subscription_dal.deactivate_other_active_subscriptions( + session, local_user.panel_user_uuid, None + ) + logging.info( + f"Sync: Local user {local_user.user_id} with panel UUID {local_user.panel_user_uuid} not found on panel. Deactivated local subs." + ) + status_msg_key = "sync_completed_details" final_status_type = "success" @@ -251,36 +272,51 @@ async def sync_command_handler(message_event: Union[types.Message, details_for_db = f"Successfully processed {users_processed_count} users. Synced {subscriptions_synced_count} subscriptions." await panel_sync_dal.update_panel_sync_status( - session, final_status_type, details_for_db, users_processed_count, - subscriptions_synced_count) + session, + final_status_type, + details_for_db, + users_processed_count, + subscriptions_synced_count, + ) await session.commit() - final_user_message = _(status_msg_key, - total_checked=total_panel_users, - users_synced=users_synced_successfully, - subs_synced=subscriptions_synced_count, - errors_count=len(sync_errors), - error_details_preview=error_preview - if sync_errors else _("no_errors_placeholder")) + final_user_message = _( + status_msg_key, + total_checked=total_panel_users, + users_synced=users_synced_successfully, + subs_synced=subscriptions_synced_count, + errors_count=len(sync_errors), + error_details_preview=( + error_preview if sync_errors else _("no_errors_placeholder") + ), + ) await bot.send_message(target_chat_id, final_user_message) except Exception as e_sync_global: await session.rollback() - logging.error(f"Global error during /sync command: {e_sync_global}", - exc_info=True) - error_detail_for_db = f"An unexpected error occurred during sync: {str(e_sync_global)[:200]}" + logging.error( + f"Global error during /sync command: {e_sync_global}", exc_info=True + ) + error_detail_for_db = ( + f"An unexpected error occurred during sync: {str(e_sync_global)[:200]}" + ) await panel_sync_dal.update_panel_sync_status( - session, "failed", error_detail_for_db, users_processed_count, - subscriptions_synced_count) + session, + "failed", + error_detail_for_db, + users_processed_count, + subscriptions_synced_count, + ) - await bot.send_message(target_chat_id, - _("sync_failed", details=error_detail_for_db)) + await bot.send_message( + target_chat_id, _("sync_failed", details=error_detail_for_db) + ) @router.message(Command("syncstatus")) -async def sync_status_command_handler(message: types.Message, i18n_data: dict, - settings: Settings, - session: AsyncSession): +async def sync_status_command_handler( + message: types.Message, i18n_data: dict, settings: Settings, session: AsyncSession +): current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE) i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") if not i18n: @@ -292,13 +328,16 @@ async def sync_status_command_handler(message: types.Message, i18n_data: dict, response_text = "" if status_record_model: last_time_val = status_record_model.last_sync_time - last_time_str = last_time_val.strftime( - '%Y-%m-%d %H:%M:%S UTC') if last_time_val else "N/A" + last_time_str = ( + last_time_val.strftime("%Y-%m-%d %H:%M:%S UTC") if last_time_val else "N/A" + ) details_val = status_record_model.details - details_str = (details_val[:200] + - "...") if details_val and len(details_val) > 200 else ( - details_val or "N/A") + details_str = ( + (details_val[:200] + "...") + if details_val and len(details_val) > 200 + else (details_val or "N/A") + ) response_text = ( f"{_('admin_stats_last_sync_header')}\n" @@ -306,7 +345,8 @@ async def sync_status_command_handler(message: types.Message, i18n_data: dict, f" {_('admin_stats_sync_status')}: {status_record_model.status}\n" f" {_('admin_stats_sync_users_processed')}: {status_record_model.users_processed_from_panel}\n" f" {_('admin_stats_sync_subs_synced')}: {status_record_model.subscriptions_synced}\n" - f" {_('admin_stats_sync_details_label')}: {details_str}") + f" {_('admin_stats_sync_details_label')}: {details_str}" + ) else: response_text = _("admin_sync_status_never_run") diff --git a/bot/handlers/user/trial_handler.py b/bot/handlers/user/trial_handler.py index 158fa9b..530c2c0 100644 --- a/bot/handlers/user/trial_handler.py +++ b/bot/handlers/user/trial_handler.py @@ -7,7 +7,10 @@ from datetime import datetime from config.settings import Settings from bot.services.subscription_service import SubscriptionService from bot.services.panel_api_service import PanelApiService -from bot.keyboards.inline.user_keyboards import get_trial_confirmation_keyboard, get_main_menu_inline_keyboard +from bot.keyboards.inline.user_keyboards import ( + get_trial_confirmation_keyboard, + get_main_menu_inline_keyboard, +) from bot.middlewares.i18n import JsonI18n from .start import send_main_menu @@ -15,13 +18,16 @@ router = Router(name="user_trial_router") async def request_trial_confirmation_handler( - callback: types.CallbackQuery, settings: Settings, i18n_data: dict, - subscription_service: SubscriptionService, session: AsyncSession): + callback: types.CallbackQuery, + settings: Settings, + i18n_data: dict, + subscription_service: SubscriptionService, + session: AsyncSession, +): user_id = callback.from_user.id current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE) i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") - _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs - ) if i18n else key + _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) if i18n else key if not i18n or not callback.message: await callback.answer(_("error_occurred_try_again"), show_alert=True) @@ -29,50 +35,60 @@ async def request_trial_confirmation_handler( show_trial_btn_in_menu_if_fail = False if settings.TRIAL_ENABLED: - if not await subscription_service.has_had_any_subscription( - session, user_id): + if not await subscription_service.has_had_any_subscription(session, user_id): show_trial_btn_in_menu_if_fail = True if not settings.TRIAL_ENABLED: await callback.message.edit_text( _("trial_feature_disabled"), - reply_markup=get_main_menu_inline_keyboard(current_lang, i18n, - settings, False)) + reply_markup=get_main_menu_inline_keyboard( + current_lang, i18n, settings, False + ), + ) await callback.answer() return if await subscription_service.has_had_any_subscription(session, user_id): await callback.message.edit_text( _("trial_already_had_subscription_or_trial"), - reply_markup=get_main_menu_inline_keyboard(current_lang, i18n, - settings, False)) + reply_markup=get_main_menu_inline_keyboard( + current_lang, i18n, settings, False + ), + ) await callback.answer() return - traffic_gb_display = str( - settings.TRIAL_TRAFFIC_LIMIT_GB - ) if settings.TRIAL_TRAFFIC_LIMIT_GB and settings.TRIAL_TRAFFIC_LIMIT_GB > 0 else _( - "traffic_unlimited") + traffic_gb_display = ( + str(settings.TRIAL_TRAFFIC_LIMIT_GB) + if settings.TRIAL_TRAFFIC_LIMIT_GB and settings.TRIAL_TRAFFIC_LIMIT_GB > 0 + else _("traffic_unlimited") + ) await callback.message.edit_text( - text=_("trial_confirm_prompt", - days=settings.TRIAL_DURATION_DAYS, - traffic_gb=traffic_gb_display), - reply_markup=get_trial_confirmation_keyboard(current_lang, i18n)) + text=_( + "trial_confirm_prompt", + days=settings.TRIAL_DURATION_DAYS, + traffic_gb=traffic_gb_display, + ), + reply_markup=get_trial_confirmation_keyboard(current_lang, i18n), + ) await callback.answer() @router.callback_query(F.data == "trial_action:confirm_activate") async def confirm_activate_trial_handler( - callback: types.CallbackQuery, settings: Settings, i18n_data: dict, - subscription_service: SubscriptionService, - panel_service: PanelApiService, session: AsyncSession): + callback: types.CallbackQuery, + settings: Settings, + i18n_data: dict, + subscription_service: SubscriptionService, + panel_service: PanelApiService, + session: AsyncSession, +): user_id = callback.from_user.id current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE) i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance") - _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs - ) if i18n else key + _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs) if i18n else key if not i18n or not callback.message: await callback.answer(_("error_occurred_try_again"), show_alert=True) @@ -81,26 +97,22 @@ async def confirm_activate_trial_handler( if not settings.TRIAL_ENABLED: await callback.answer(_("trial_feature_disabled"), show_alert=True) - await send_main_menu(callback, - settings, - i18n_data, - subscription_service, - session, - is_edit=True) + await send_main_menu( + callback, settings, i18n_data, subscription_service, session, is_edit=True + ) return if await subscription_service.has_had_any_subscription(session, user_id): - await callback.answer(_("trial_already_had_subscription_or_trial"), - show_alert=True) - await send_main_menu(callback, - settings, - i18n_data, - subscription_service, - session, - is_edit=True) + await callback.answer( + _("trial_already_had_subscription_or_trial"), show_alert=True + ) + await send_main_menu( + callback, settings, i18n_data, subscription_service, session, is_edit=True + ) return activation_result = await subscription_service.activate_trial_subscription( - session, user_id) + session, user_id + ) final_message_text_in_chat = "" show_trial_button_after_action = False @@ -110,35 +122,43 @@ async def confirm_activate_trial_handler( end_date_obj = activation_result.get("end_date") config_link_for_trial = activation_result.get("subscription_url") or _( - "config_link_not_available") + "config_link_not_available" + ) - if config_link_for_trial == _( - "config_link_not_available") and activation_result.get( - "panel_short_uuid"): - generated_link = await panel_service.get_subscription_link( - activation_result["panel_short_uuid"]) - if generated_link: config_link_for_trial = generated_link - - traffic_gb_val = activation_result.get("traffic_gb", - settings.TRIAL_TRAFFIC_LIMIT_GB) - traffic_display = f"{traffic_gb_val} GB" if traffic_gb_val and traffic_gb_val > 0 else _( - "traffic_unlimited") + traffic_gb_val = activation_result.get( + "traffic_gb", settings.TRIAL_TRAFFIC_LIMIT_GB + ) + traffic_display = ( + f"{traffic_gb_val} GB" + if traffic_gb_val and traffic_gb_val > 0 + else _("traffic_unlimited") + ) final_message_text_in_chat = _( "trial_activated_details_message", days=activation_result.get("days", settings.TRIAL_DURATION_DAYS), - end_date=end_date_obj.strftime('%Y-%m-%d') if isinstance( - end_date_obj, datetime) else "N/A", + end_date=( + end_date_obj.strftime("%Y-%m-%d") + if isinstance(end_date_obj, datetime) + else "N/A" + ), config_link=config_link_for_trial, - traffic_gb=traffic_display) + traffic_gb=traffic_display, + ) else: - message_key_from_service = activation_result.get( - "message_key", "trial_activation_failed" - ) if activation_result else "trial_activation_failed" + message_key_from_service = ( + activation_result.get("message_key", "trial_activation_failed") + if activation_result + else "trial_activation_failed" + ) final_message_text_in_chat = _(message_key_from_service) await callback.answer(final_message_text_in_chat, show_alert=True) - if settings.TRIAL_ENABLED and not await subscription_service.has_had_any_subscription( - session, user_id): + if ( + settings.TRIAL_ENABLED + and not await subscription_service.has_had_any_subscription( + session, user_id + ) + ): show_trial_button_after_action = True try: @@ -146,31 +166,38 @@ async def confirm_activate_trial_handler( final_message_text_in_chat, parse_mode="HTML", reply_markup=get_main_menu_inline_keyboard( - current_lang, i18n, settings, show_trial_button_after_action), - disable_web_page_preview=True) + current_lang, i18n, settings, show_trial_button_after_action + ), + disable_web_page_preview=True, + ) except Exception as e_edit: logging.warning( - f"Could not edit trial result message: {e_edit}. Sending new one.") + f"Could not edit trial result message: {e_edit}. Sending new one." + ) - if callback.message and hasattr(callback.message, - 'chat') and callback.message.chat: + if ( + callback.message + and hasattr(callback.message, "chat") + and callback.message.chat + ): await callback.message.chat.send_message( final_message_text_in_chat, parse_mode="HTML", reply_markup=get_main_menu_inline_keyboard( - current_lang, i18n, settings, - show_trial_button_after_action), - disable_web_page_preview=True) + current_lang, i18n, settings, show_trial_button_after_action + ), + disable_web_page_preview=True, + ) @router.callback_query(F.data == "main_action:cancel_trial") -async def cancel_trial_activation(callback: types.CallbackQuery, - settings: Settings, i18n_data: dict, - subscription_service: SubscriptionService, - session: AsyncSession): - await send_main_menu(callback, - settings, - i18n_data, - subscription_service, - session, - is_edit=True) +async def cancel_trial_activation( + callback: types.CallbackQuery, + settings: Settings, + i18n_data: dict, + subscription_service: SubscriptionService, + session: AsyncSession, +): + await send_main_menu( + callback, settings, i18n_data, subscription_service, session, is_edit=True + ) diff --git a/bot/main_bot.py b/bot/main_bot.py index a934f3f..789d932 100644 --- a/bot/main_bot.py +++ b/bot/main_bot.py @@ -43,14 +43,17 @@ class DBSessionMiddleware(BaseMiddleware): super().__init__() self.async_session_factory = async_session_factory - async def __call__(self, handler: Callable[[Update, Dict[str, Any]], - Awaitable[Any]], event: Update, - data: Dict[str, Any]) -> Any: + async def __call__( + self, + handler: Callable[[Update, Dict[str, Any]], Awaitable[Any]], + event: Update, + data: Dict[str, Any], + ) -> Any: if self.async_session_factory is None: - logging.critical( - "DBSessionMiddleware: async_session_factory is None!") + logging.critical("DBSessionMiddleware: async_session_factory is None!") raise RuntimeError( - "async_session_factory not provided to DBSessionMiddleware") + "async_session_factory not provided to DBSessionMiddleware" + ) async with self.async_session_factory() as session: data["session"] = session @@ -62,8 +65,8 @@ class DBSessionMiddleware(BaseMiddleware): except Exception: await session.rollback() logging.error( - "DBSessionMiddleware: Exception caused rollback.", - exc_info=True) + "DBSessionMiddleware: Exception caused rollback.", exc_info=True + ) raise @@ -95,25 +98,29 @@ async def on_startup_configured(dispatcher: Dispatcher): existing_scheduler: Optional[AsyncIOScheduler] = dispatcher.get("scheduler") if existing_scheduler and existing_scheduler.running: - logging.warning( - "STARTUP: Scheduler already running, skipping initialization.") + logging.warning("STARTUP: Scheduler already running, skipping initialization.") else: scheduler = AsyncIOScheduler(timezone="UTC") try: await schedule_subscription_notifications( - bot, settings, i18n_instance, scheduler, panel_service, - async_session_factory) + bot, + settings, + i18n_instance, + scheduler, + panel_service, + async_session_factory, + ) scheduler.start() dispatcher["scheduler"] = scheduler logging.info("STARTUP: APScheduler started.") except Exception as e: - logging.error( - f"STARTUP: Failed to start APScheduler: {e}", exc_info=True) + logging.error(f"STARTUP: Failed to start APScheduler: {e}", exc_info=True) - telegram_webhook_url_to_set = getattr(settings, - 'TELEGRAM_WEBHOOK_BASE_URL', None) + telegram_webhook_url_to_set = getattr(settings, "TELEGRAM_WEBHOOK_BASE_URL", None) if telegram_webhook_url_to_set: - full_telegram_webhook_url = f"{str(telegram_webhook_url_to_set).rstrip('/')}/{settings.BOT_TOKEN}" + full_telegram_webhook_url = ( + f"{str(telegram_webhook_url_to_set).rstrip('/')}/{settings.BOT_TOKEN}" + ) logging.info( f"STARTUP: Attempting to set Telegram webhook to: {full_telegram_webhook_url if full_telegram_webhook_url != 'ERROR_URL_TOKEN_DETECTED' else 'HIDDEN DUE TO TOKEN'}" @@ -129,7 +136,8 @@ async def on_startup_configured(dispatcher: Dispatcher): set_success = await bot.set_webhook( url=full_telegram_webhook_url, drop_pending_updates=True, - allowed_updates=dispatcher.resolve_used_update_types()) + allowed_updates=dispatcher.resolve_used_update_types(), + ) if set_success: logging.info( f"STARTUP: bot.set_webhook to {full_telegram_webhook_url} returned SUCCESS (True)." @@ -151,7 +159,8 @@ async def on_startup_configured(dispatcher: Dispatcher): except Exception as e_setwebhook: logging.error( f"STARTUP: EXCEPTION during set/get Telegram webhook: {e_setwebhook}", - exc_info=True) + exc_info=True, + ) else: logging.error( "STARTUP: Skipped setting Telegram webhook due to security or configuration error." @@ -174,11 +183,12 @@ async def on_shutdown_configured(dispatcher: Dispatcher): scheduler.shutdown(wait=False) logging.info("SHUTDOWN: APScheduler shut down.") except Exception as e: - logging.error(f"SHUTDOWN: Error shutting down APScheduler: {e}", - exc_info=True) + logging.error( + f"SHUTDOWN: Error shutting down APScheduler: {e}", exc_info=True + ) panel_service: Optional[PanelApiService] = dispatcher.get("panel_service") - if panel_service and hasattr(panel_service, 'close_session'): + if panel_service and hasattr(panel_service, "close_session"): await panel_service.close_session() logging.info("Panel API service session closed on shutdown.") @@ -191,6 +201,7 @@ async def on_shutdown_configured(dispatcher: Dispatcher): logging.warning(f"SHUTDOWN: Failed to close bot session: {e}") from db.database_setup import async_engine as global_async_engine + if global_async_engine: logging.info("SHUTDOWN: Disposing SQLAlchemy engine...") await global_async_engine.dispose() @@ -223,28 +234,40 @@ async def run_bot(settings_param: Settings): f"Failed to get bot info (e.g., for YooKassa default URL): {e}. Using fallback: {actual_bot_username}" ) - i18n_instance = get_i18n_instance(path="locales", - default=settings_param.DEFAULT_LANGUAGE) + i18n_instance = get_i18n_instance( + path="locales", default=settings_param.DEFAULT_LANGUAGE + ) yookassa_service = YooKassaService( shop_id=settings_param.YOOKASSA_SHOP_ID, secret_key=settings_param.YOOKASSA_SECRET_KEY, configured_return_url=settings_param.YOOKASSA_RETURN_URL, bot_username_for_default_return=actual_bot_username, - settings_obj=settings_param) + settings_obj=settings_param, + ) panel_service = PanelApiService(settings_param) - subscription_service = SubscriptionService(settings_param, panel_service) - referral_service = ReferralService(settings_param, subscription_service, - bot, i18n_instance) - promo_code_service = PromoCodeService(settings_param, subscription_service, - bot, i18n_instance) - stars_service = StarsService(bot, settings_param, i18n_instance, - subscription_service, referral_service) - tribute_service = TributeService(bot, settings_param, i18n_instance, - local_async_session_factory, - panel_service, subscription_service, - referral_service) + subscription_service = SubscriptionService( + settings_param, panel_service, bot, i18n_instance + ) + referral_service = ReferralService( + settings_param, subscription_service, bot, i18n_instance + ) + promo_code_service = PromoCodeService( + settings_param, subscription_service, bot, i18n_instance + ) + stars_service = StarsService( + bot, settings_param, i18n_instance, subscription_service, referral_service + ) + tribute_service = TributeService( + bot, + settings_param, + i18n_instance, + local_async_session_factory, + panel_service, + subscription_service, + referral_service, + ) dp["i18n_instance"] = i18n_instance dp["yookassa_service"] = yookassa_service @@ -256,13 +279,13 @@ async def run_bot(settings_param: Settings): dp["tribute_service"] = tribute_service dp["async_session_factory"] = local_async_session_factory + dp.update.outer_middleware(DBSessionMiddleware(local_async_session_factory)) dp.update.outer_middleware( - DBSessionMiddleware(local_async_session_factory)) + I18nMiddleware(i18n=i18n_instance, settings=settings_param) + ) dp.update.outer_middleware( - I18nMiddleware(i18n=i18n_instance, settings=settings_param)) - dp.update.outer_middleware( - BanCheckMiddleware(settings=settings_param, - i18n_instance=i18n_instance)) + BanCheckMiddleware(settings=settings_param, i18n_instance=i18n_instance) + ) dp.update.outer_middleware(ActionLoggerMiddleware(settings=settings_param)) dp.startup.register(on_startup_configured) @@ -270,13 +293,12 @@ async def run_bot(settings_param: Settings): await register_all_routers(dp, settings_param) - tg_webhook_base = getattr(settings_param, 'TELEGRAM_WEBHOOK_BASE_URL', - None) - yk_webhook_base = getattr(settings_param, 'YOOKASSA_WEBHOOK_BASE_URL', - None) + tg_webhook_base = getattr(settings_param, "TELEGRAM_WEBHOOK_BASE_URL", None) + yk_webhook_base = getattr(settings_param, "YOOKASSA_WEBHOOK_BASE_URL", None) should_run_aiohttp_server = bool(tg_webhook_base) or ( - bool(yk_webhook_base) and bool(settings_param.yookassa_webhook_path)) + bool(yk_webhook_base) and bool(settings_param.yookassa_webhook_path) + ) telegram_uses_webhook_mode = bool(tg_webhook_base) run_telegram_polling = not telegram_uses_webhook_mode @@ -297,27 +319,28 @@ async def run_bot(settings_param: Settings): if should_run_aiohttp_server: app = web.Application() - app['bot'] = bot - app['dp'] = dp - app['settings'] = settings_param - app['i18n'] = i18n_instance - app['async_session_factory'] = local_async_session_factory + app["bot"] = bot + app["dp"] = dp + app["settings"] = settings_param + app["i18n"] = i18n_instance + app["async_session_factory"] = local_async_session_factory - app['yookassa_service'] = yookassa_service - app['subscription_service'] = subscription_service - app['referral_service'] = referral_service - app['panel_service'] = panel_service - app['stars_service'] = stars_service - app['tribute_service'] = tribute_service + app["yookassa_service"] = yookassa_service + app["subscription_service"] = subscription_service + app["referral_service"] = referral_service + app["panel_service"] = panel_service + app["stars_service"] = stars_service + app["tribute_service"] = tribute_service setup_application(app, dp, bot=bot) if telegram_uses_webhook_mode: telegram_webhook_path = f"/{settings_param.BOT_TOKEN}" - if not telegram_webhook_path.startswith('/'): - telegram_webhook_path = '/' + telegram_webhook_path - app.router.add_post(telegram_webhook_path, - SimpleRequestHandler(dispatcher=dp, bot=bot)) + if not telegram_webhook_path.startswith("/"): + telegram_webhook_path = "/" + telegram_webhook_path + app.router.add_post( + telegram_webhook_path, SimpleRequestHandler(dispatcher=dp, bot=bot) + ) logging.info( f"Telegram webhook route configured at: [POST] {telegram_webhook_path} (relative to base URL)" ) @@ -328,48 +351,52 @@ async def run_bot(settings_param: Settings): logging.error( f"YooKassa webhook path is invalid or not configured in settings: {yk_path}. Skipping YooKassa webhook setup." ) - elif not yk_path.startswith('/'): + elif not yk_path.startswith("/"): logging.error( f"CRITICAL: YooKassa webhook path '{yk_path}' from settings does not start with '/'. Correct settings.py or .env. Skipping YooKassa webhook." ) else: app.router.add_post( - yk_path, - user_payment_webhook_module.yookassa_webhook_route) - logging.info( - f"YooKassa webhook route configured at: [POST] {yk_path}") + yk_path, user_payment_webhook_module.yookassa_webhook_route + ) + logging.info(f"YooKassa webhook route configured at: [POST] {yk_path}") tribute_path = settings_param.tribute_webhook_path - if tribute_path.startswith('/'): - app.router.add_post( - tribute_path, - tribute_webhook_route) - logging.info( - f"Tribute webhook route configured at: [POST] {tribute_path}") + if tribute_path.startswith("/"): + app.router.add_post(tribute_path, tribute_webhook_route) + logging.info(f"Tribute webhook route configured at: [POST] {tribute_path}") web_app_runner = web.AppRunner(app) await web_app_runner.setup() - site = web.TCPSite(web_app_runner, - host=settings_param.WEB_SERVER_HOST, - port=settings_param.WEB_SERVER_PORT) + site = web.TCPSite( + web_app_runner, + host=settings_param.WEB_SERVER_HOST, + port=settings_param.WEB_SERVER_PORT, + ) async def web_server_task(): await site.start() logging.info( f"AIOHTTP server started on http://{settings_param.WEB_SERVER_HOST}:{settings_param.WEB_SERVER_PORT}" ) - await asyncio.Event().wait( - ) if not run_telegram_polling else await asyncio.sleep(31536000) + ( + await asyncio.Event().wait() + if not run_telegram_polling + else await asyncio.sleep(31536000) + ) main_tasks.append( - asyncio.create_task(web_server_task(), name="AIOHTTPServerTask")) + asyncio.create_task(web_server_task(), name="AIOHTTPServerTask") + ) if run_telegram_polling: logging.info("Starting bot in Telegram Polling mode...") main_tasks.append( - asyncio.create_task(dp.start_polling( - bot, allowed_updates=dp.resolve_used_update_types()), - name="TelegramPollingTask")) + asyncio.create_task( + dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types()), + name="TelegramPollingTask", + ) + ) if not main_tasks: logging.error( @@ -385,8 +412,7 @@ async def run_bot(settings_param: Settings): try: await asyncio.gather(*main_tasks) except (KeyboardInterrupt, SystemExit, asyncio.CancelledError) as e: - logging.info( - f"Main bot loop interrupted/cancelled: {type(e).__name__} - {e}") + logging.info(f"Main bot loop interrupted/cancelled: {type(e).__name__} - {e}") finally: logging.info("Initiating final bot shutdown sequence...") for task in main_tasks: @@ -401,7 +427,8 @@ async def run_bot(settings_param: Settings): except Exception as e_task_cancel: logging.error( f"Error during cancellation of task '{task.get_name()}': {e_task_cancel}", - exc_info=True) + exc_info=True, + ) if web_app_runner: await web_app_runner.cleanup() diff --git a/bot/services/subscription_service.py b/bot/services/subscription_service.py index d9fb834..9e6be1b 100644 --- a/bot/services/subscription_service.py +++ b/bot/services/subscription_service.py @@ -2,6 +2,8 @@ import logging from sqlalchemy.ext.asyncio import AsyncSession from datetime import datetime, timedelta, timezone from typing import Optional, Dict, Any, List, Tuple +from aiogram import Bot +from bot.middlewares.i18n import JsonI18n from db.dal import user_dal, subscription_dal, promo_code_dal, payment_dal from db.models import User, Subscription @@ -12,26 +14,48 @@ from .panel_api_service import PanelApiService class SubscriptionService: - def __init__(self, settings: Settings, panel_service: PanelApiService): + def __init__( + self, + settings: Settings, + panel_service: PanelApiService, + bot: Optional[Bot] = None, + i18n: Optional[JsonI18n] = None, + ): self.settings = settings self.panel_service = panel_service + self.bot = bot + self.i18n = i18n - async def get_user_language(self, session: AsyncSession, - user_id: int) -> str: + async def get_user_language(self, session: AsyncSession, user_id: int) -> str: user_record = await user_dal.get_user_by_id(session, user_id) - return user_record.language_code if user_record and user_record.language_code else self.settings.DEFAULT_LANGUAGE + return ( + user_record.language_code + if user_record and user_record.language_code + else self.settings.DEFAULT_LANGUAGE + ) - async def has_had_any_subscription(self, session: AsyncSession, - user_id: int) -> bool: + async def has_had_any_subscription( + self, session: AsyncSession, user_id: int + ) -> bool: - return await subscription_dal.has_any_subscription_for_user( - session, user_id) + return await subscription_dal.has_any_subscription_for_user(session, user_id) + + async def _notify_admin_panel_user_creation_failed(self, user_id: int): + if not self.bot or not self.i18n or not self.settings.ADMIN_IDS: + return + admin_lang = self.settings.DEFAULT_LANGUAGE + _adm = lambda k, **kw: self.i18n.gettext(admin_lang, k, **kw) + msg = _adm("admin_panel_user_creation_failed", user_id=user_id) + for admin_id in self.settings.ADMIN_IDS: + try: + await self.bot.send_message(admin_id, msg) + except Exception as e: + logging.error( + f"Failed to notify admin {admin_id} about panel user creation failure: {e}" + ) async def _get_or_create_panel_user_link_details( - self, - session: AsyncSession, - user_id: int, - db_user: Optional[User] = None + self, session: AsyncSession, user_id: int, db_user: Optional[User] = None ) -> Tuple[Optional[str], Optional[str], Optional[str], bool]: if not db_user: db_user = await user_dal.get_user_by_id(session, user_id) @@ -49,7 +73,8 @@ class SubscriptionService: panel_user_created_or_linked_now = False panel_users_by_tg_id_list = await self.panel_service.get_users_by_filter( - telegram_id=user_id) + telegram_id=user_id + ) if panel_users_by_tg_id_list and len(panel_users_by_tg_id_list) == 1: panel_user_obj_from_api = panel_users_by_tg_id_list[0] logging.info( @@ -68,7 +93,8 @@ class SubscriptionService: f"User {user_id} (local panel_uuid: {current_local_panel_uuid}) not found on panel by TG ID. Fetching by panel_uuid." ) panel_user_obj_from_api = await self.panel_service.get_user_by_uuid( - current_local_panel_uuid) + current_local_panel_uuid + ) if not panel_user_obj_from_api: logging.warning( f"Local panel_uuid {current_local_panel_uuid} for TG user {user_id} also not found on panel. User might be deleted from panel or UUID desynced." @@ -83,26 +109,31 @@ class SubscriptionService: username_on_panel=panel_username_on_panel_standard, telegram_id=user_id, ) - if creation_response and not creation_response.get( - "error") and creation_response.get("response"): + if ( + creation_response + and not creation_response.get("error") + and creation_response.get("response") + ): panel_user_obj_from_api = creation_response.get("response") panel_user_created_or_linked_now = True - elif creation_response and creation_response.get( - "errorCode") == "A019": + elif creation_response and creation_response.get("errorCode") == "A019": logging.warning( f"Panel user '{panel_username_on_panel_standard}' already exists (errorCode A019). Fetching by username." ) - fetched_by_username_list = await self.panel_service.get_users_by_filter( - username=panel_username_on_panel_standard) - if fetched_by_username_list and len( - fetched_by_username_list) == 1: + fetched_by_username_list = ( + await self.panel_service.get_users_by_filter( + username=panel_username_on_panel_standard + ) + ) + if fetched_by_username_list and len(fetched_by_username_list) == 1: panel_user_obj_from_api = fetched_by_username_list[0] if not panel_user_obj_from_api: logging.error( f"Failed to create or link panel user for TG_ID {user_id} with panel username '{panel_username_on_panel_standard}'. Response: {creation_response if 'creation_response' in locals() else 'N/A'}" ) + await self._notify_admin_panel_user_creation_failed(user_id) return None, None, None, False if not panel_user_obj_from_api: @@ -110,33 +141,47 @@ class SubscriptionService: f"Could not obtain panel user object for TG user {user_id} after all checks." ) - return current_local_panel_uuid if current_local_panel_uuid else None, None, None, panel_user_created_or_linked_now + return ( + current_local_panel_uuid if current_local_panel_uuid else None, + None, + None, + panel_user_created_or_linked_now, + ) - actual_panel_uuid_from_api = panel_user_obj_from_api.get('uuid') - actual_panel_username_from_api = panel_user_obj_from_api.get( - 'username') - panel_telegram_id_from_api = panel_user_obj_from_api.get('telegramId') + actual_panel_uuid_from_api = panel_user_obj_from_api.get("uuid") + actual_panel_username_from_api = panel_user_obj_from_api.get("username") + panel_telegram_id_from_api = panel_user_obj_from_api.get("telegramId") if not actual_panel_uuid_from_api: logging.error( f"Panel user object for TG user {user_id} does not contain 'uuid'. Data: {panel_user_obj_from_api}" ) - return current_local_panel_uuid, None, None, panel_user_created_or_linked_now + return ( + current_local_panel_uuid, + None, + None, + panel_user_created_or_linked_now, + ) needs_local_panel_uuid_update = False if current_local_panel_uuid is None and actual_panel_uuid_from_api: needs_local_panel_uuid_update = True - elif current_local_panel_uuid is not None and current_local_panel_uuid != actual_panel_uuid_from_api: + elif ( + current_local_panel_uuid is not None + and current_local_panel_uuid != actual_panel_uuid_from_api + ): logging.warning( f"Local panel_uuid for user {user_id} ('{current_local_panel_uuid}') " f"differs from panel's UUID ('{actual_panel_uuid_from_api}') for their telegramId. " - f"Will attempt to update local to panel's version.") + f"Will attempt to update local to panel's version." + ) needs_local_panel_uuid_update = True if needs_local_panel_uuid_update: conflicting_user_record = await user_dal.get_user_by_panel_uuid( - session, actual_panel_uuid_from_api) + session, actual_panel_uuid_from_api + ) if conflicting_user_record and conflicting_user_record.user_id != user_id: logging.error( f"CRITICAL CONFLICT: Panel UUID {actual_panel_uuid_from_api} (from panel for TG ID {user_id}) " @@ -151,14 +196,20 @@ class SubscriptionService: "panel_user_uuid": actual_panel_uuid_from_api } - if actual_panel_username_from_api and actual_panel_username_from_api != panel_username_on_panel_standard and ( + if ( + actual_panel_username_from_api + and actual_panel_username_from_api + != panel_username_on_panel_standard + and ( db_user.username is None - or db_user.username != actual_panel_username_from_api): - update_data_for_local_user[ - "username"] = actual_panel_username_from_api + or db_user.username != actual_panel_username_from_api + ) + ): + update_data_for_local_user["username"] = ( + actual_panel_username_from_api + ) - await user_dal.update_user(session, user_id, - update_data_for_local_user) + await user_dal.update_user(session, user_id, update_data_for_local_user) db_user.panel_user_uuid = actual_panel_uuid_from_api if "username" in update_data_for_local_user: db_user.username = update_data_for_local_user["username"] @@ -175,69 +226,79 @@ class SubscriptionService: except ValueError: pass - if panel_user_obj_from_api and current_local_panel_uuid and panel_telegram_id_int != user_id: + if ( + panel_user_obj_from_api + and current_local_panel_uuid + and panel_telegram_id_int != user_id + ): logging.info( f"Panel user {current_local_panel_uuid} has telegramId '{panel_telegram_id_from_api}'. Updating on panel to '{user_id}'." ) await self.panel_service.update_user_details_on_panel( - current_local_panel_uuid, {"telegramId": user_id}) + current_local_panel_uuid, {"telegramId": user_id} + ) panel_sub_link_id = panel_user_obj_from_api.get( - 'subscriptionUuid') or panel_user_obj_from_api.get('shortUuid') - panel_short_uuid = panel_user_obj_from_api.get('shortUuid') + "subscriptionUuid" + ) or panel_user_obj_from_api.get("shortUuid") + panel_short_uuid = panel_user_obj_from_api.get("shortUuid") if not panel_sub_link_id and current_local_panel_uuid: logging.warning( f"No subscriptionUuid or shortUuid found on panel for panel_user_uuid {current_local_panel_uuid} (TG ID: {user_id})." ) - return current_local_panel_uuid, panel_sub_link_id, panel_short_uuid, panel_user_created_or_linked_now + return ( + current_local_panel_uuid, + panel_sub_link_id, + panel_short_uuid, + panel_user_created_or_linked_now, + ) async def activate_trial_subscription( - self, session: AsyncSession, - user_id: int) -> Optional[Dict[str, Any]]: + self, session: AsyncSession, user_id: int + ) -> Optional[Dict[str, Any]]: if not self.settings.TRIAL_ENABLED or self.settings.TRIAL_DURATION_DAYS <= 0: return { "eligible": False, "activated": False, - "message_key": "trial_feature_disabled" + "message_key": "trial_feature_disabled", } db_user = await user_dal.get_user_by_id(session, user_id) if not db_user: - logging.error( - f"User {user_id} not found in DB, cannot activate trial.") + logging.error(f"User {user_id} not found in DB, cannot activate trial.") return { "eligible": False, "activated": False, - "message_key": "user_not_found_for_trial" + "message_key": "user_not_found_for_trial", } if await self.has_had_any_subscription(session, user_id): return { "eligible": False, "activated": False, - "message_key": "trial_already_had_subscription_or_trial" + "message_key": "trial_already_had_subscription_or_trial", } - panel_user_uuid, panel_sub_link_id, panel_short_uuid, panel_user_created_now = await self._get_or_create_panel_user_link_details( - session, user_id, db_user) + panel_user_uuid, panel_sub_link_id, panel_short_uuid, panel_user_created_now = ( + await self._get_or_create_panel_user_link_details(session, user_id, db_user) + ) if not panel_user_uuid or not panel_sub_link_id: - logging.error( - f"Failed to get panel link details for trial user {user_id}.") + logging.error(f"Failed to get panel link details for trial user {user_id}.") return { "eligible": True, "activated": False, - "message_key": "trial_activation_failed_panel_link" + "message_key": "trial_activation_failed_panel_link", } start_date = datetime.now(timezone.utc) - end_date = start_date + timedelta( - days=self.settings.TRIAL_DURATION_DAYS) + end_date = start_date + timedelta(days=self.settings.TRIAL_DURATION_DAYS) await subscription_dal.deactivate_other_active_subscriptions( - session, panel_user_uuid, panel_sub_link_id) + session, panel_user_uuid, panel_sub_link_id + ) trial_sub_data = { "user_id": user_id, @@ -255,34 +316,34 @@ class SubscriptionService: except Exception as e_upsert: logging.error( f"Failed to upsert trial subscription for user {user_id}: {e_upsert}", - exc_info=True) + exc_info=True, + ) await session.rollback() return { "eligible": True, "activated": False, - "message_key": "trial_activation_failed_db" + "message_key": "trial_activation_failed_db", } panel_update_payload: Dict[str, Any] = { - "uuid": - panel_user_uuid, - "expireAt": - end_date.isoformat(timespec='milliseconds').replace('+00:00', 'Z'), - "status": - "ACTIVE", - "trafficLimitBytes": - self.settings.trial_traffic_limit_bytes, - "trafficLimitStrategy": - self.settings.PANEL_USER_DEFAULT_TRAFFIC_STRATEGY, + "uuid": panel_user_uuid, + "expireAt": end_date.isoformat(timespec="milliseconds").replace( + "+00:00", "Z" + ), + "status": "ACTIVE", + "trafficLimitBytes": self.settings.trial_traffic_limit_bytes, + "trafficLimitStrategy": self.settings.PANEL_USER_DEFAULT_TRAFFIC_STRATEGY, } if self.settings.parsed_default_panel_user_inbound_uuids: - panel_update_payload[ - "activeUserInbounds"] = self.settings.parsed_default_panel_user_inbound_uuids + panel_update_payload["activeUserInbounds"] = ( + self.settings.parsed_default_panel_user_inbound_uuids + ) elif panel_user_created_now: panel_update_payload["activateAllInbounds"] = True updated_panel_user = await self.panel_service.update_user_details_on_panel( - panel_user_uuid, panel_update_payload) + panel_user_uuid, panel_update_payload + ) if not updated_panel_user or updated_panel_user.get("error"): logging.warning( f"Panel user details update FAILED for trial user {panel_user_uuid}. Response: {updated_panel_user}" @@ -291,14 +352,13 @@ class SubscriptionService: return { "eligible": True, "activated": False, - "message_key": "trial_activation_failed_panel_update" + "message_key": "trial_activation_failed_panel_update", } await session.commit() - final_subscription_url = updated_panel_user.get('subscriptionUrl') - final_panel_short_uuid = updated_panel_user.get( - 'shortUuid', panel_short_uuid) + final_subscription_url = updated_panel_user.get("subscriptionUrl") + final_panel_short_uuid = updated_panel_user.get("shortUuid", panel_short_uuid) return { "eligible": True, @@ -308,7 +368,7 @@ class SubscriptionService: "traffic_gb": self.settings.TRIAL_TRAFFIC_LIMIT_GB, "panel_user_uuid": panel_user_uuid, "panel_short_uuid": final_panel_short_uuid, - "subscription_url": final_subscription_url + "subscription_url": final_subscription_url, } async def activate_subscription( @@ -319,7 +379,7 @@ class SubscriptionService: payment_amount: float, payment_db_id: int, promo_code_id_from_payment: Optional[int] = None, - provider: str = "yookassa" + provider: str = "yookassa", ) -> Optional[Dict[str, Any]]: db_user = await user_dal.get_user_by_id(session, user_id) @@ -329,8 +389,9 @@ class SubscriptionService: ) return None - panel_user_uuid, panel_sub_link_id, panel_short_uuid, panel_user_created_now = await self._get_or_create_panel_user_link_details( - session, user_id, db_user) + panel_user_uuid, panel_sub_link_id, panel_short_uuid, panel_user_created_now = ( + await self._get_or_create_panel_user_link_details(session, user_id, db_user) + ) if not panel_user_uuid or not panel_sub_link_id: logging.error( @@ -339,9 +400,14 @@ class SubscriptionService: return None current_active_sub = await subscription_dal.get_active_subscription_by_user_id( - session, user_id, panel_user_uuid) + session, user_id, panel_user_uuid + ) start_date = datetime.now(timezone.utc) - if current_active_sub and current_active_sub.end_date and current_active_sub.end_date > start_date: + if ( + current_active_sub + and current_active_sub.end_date + and current_active_sub.end_date > start_date + ): start_date = current_active_sub.end_date duration_days_total = months * 30 @@ -349,8 +415,13 @@ class SubscriptionService: if promo_code_id_from_payment: promo_model = await promo_code_dal.get_promo_code_by_id( - session, promo_code_id_from_payment) - if promo_model and promo_model.is_active and promo_model.current_activations < promo_model.max_activations: + session, promo_code_id_from_payment + ) + if ( + promo_model + and promo_model.is_active + and promo_model.current_activations < promo_model.max_activations + ): applied_promo_bonus_days = promo_model.bonus_days duration_days_total += applied_promo_bonus_days @@ -358,10 +429,12 @@ class SubscriptionService: session, promo_code_id_from_payment, user_id, - payment_id=payment_db_id) + payment_id=payment_db_id, + ) if activation: await promo_code_dal.increment_promo_code_usage( - session, promo_code_id_from_payment) + session, promo_code_id_from_payment + ) else: logging.warning( f"Promo code {promo_code_id_from_payment} was already activated by user {user_id}, but bonus applied via payment {payment_db_id}." @@ -374,7 +447,8 @@ class SubscriptionService: final_end_date = start_date + timedelta(days=duration_days_total) await subscription_dal.deactivate_other_active_subscriptions( - session, panel_user_uuid, panel_sub_link_id) + session, panel_user_uuid, panel_sub_link_id + ) sub_payload = { "user_id": user_id, @@ -385,50 +459,48 @@ class SubscriptionService: "duration_months": months, "is_active": True, "status_from_panel": "ACTIVE", - "traffic_limit_bytes": - self.settings.PANEL_USER_DEFAULT_TRAFFIC_BYTES, + "traffic_limit_bytes": self.settings.PANEL_USER_DEFAULT_TRAFFIC_BYTES, "provider": provider, "skip_notifications": provider == "tribute", } try: new_or_updated_sub = await subscription_dal.upsert_subscription( - session, sub_payload) + session, sub_payload + ) except Exception as e_upsert_sub: logging.error( f"Failed to upsert paid subscription for user {user_id}: {e_upsert_sub}", - exc_info=True) + exc_info=True, + ) return None panel_update_payload = { - "uuid": - panel_user_uuid, - "expireAt": - final_end_date.isoformat(timespec='milliseconds').replace( - '+00:00', 'Z'), - "status": - "ACTIVE", - "trafficLimitBytes": - self.settings.PANEL_USER_DEFAULT_TRAFFIC_BYTES, - "trafficLimitStrategy": - self.settings.PANEL_USER_DEFAULT_TRAFFIC_STRATEGY + "uuid": panel_user_uuid, + "expireAt": final_end_date.isoformat(timespec="milliseconds").replace( + "+00:00", "Z" + ), + "status": "ACTIVE", + "trafficLimitBytes": self.settings.PANEL_USER_DEFAULT_TRAFFIC_BYTES, + "trafficLimitStrategy": self.settings.PANEL_USER_DEFAULT_TRAFFIC_STRATEGY, } if self.settings.parsed_default_panel_user_inbound_uuids: - panel_update_payload[ - "activeUserInbounds"] = self.settings.parsed_default_panel_user_inbound_uuids + panel_update_payload["activeUserInbounds"] = ( + self.settings.parsed_default_panel_user_inbound_uuids + ) elif panel_user_created_now: panel_update_payload["activateAllInbounds"] = True updated_panel_user = await self.panel_service.update_user_details_on_panel( - panel_user_uuid, panel_update_payload) + panel_user_uuid, panel_update_payload + ) if not updated_panel_user or updated_panel_user.get("error"): logging.warning( f"Panel user details update FAILED for paid sub user {panel_user_uuid}. Response: {updated_panel_user}" ) return None - final_subscription_url = updated_panel_user.get('subscriptionUrl') - final_panel_short_uuid = updated_panel_user.get( - 'shortUuid', panel_short_uuid) + final_subscription_url = updated_panel_user.get("subscriptionUrl") + final_panel_short_uuid = updated_panel_user.get("shortUuid", panel_short_uuid) return { "subscription_id": new_or_updated_sub.subscription_id, @@ -437,15 +509,16 @@ class SubscriptionService: "panel_user_uuid": panel_user_uuid, "panel_short_uuid": final_panel_short_uuid, "subscription_url": final_subscription_url, - "applied_promo_bonus_days": applied_promo_bonus_days + "applied_promo_bonus_days": applied_promo_bonus_days, } async def extend_active_subscription_days( - self, - session: AsyncSession, - user_id: int, - bonus_days: int, - reason: str = "bonus") -> Optional[datetime]: + self, + session: AsyncSession, + user_id: int, + bonus_days: int, + reason: str = "bonus", + ) -> Optional[datetime]: user = await user_dal.get_user_by_id(session, user_id) if not user or not user.panel_user_uuid: logging.warning( @@ -454,7 +527,8 @@ class SubscriptionService: return None active_sub = await subscription_dal.get_active_subscription_by_user_id( - session, user_id, user.panel_user_uuid) + session, user_id, user.panel_user_uuid + ) if not active_sub or not active_sub.end_date: logging.info( f"No active extendable subscription found for user {user_id} (panel: {user.panel_user_uuid}) for reason: {reason}." @@ -463,19 +537,26 @@ class SubscriptionService: current_end_date = active_sub.end_date now_utc = datetime.now(timezone.utc) - start_point_for_bonus = current_end_date if current_end_date > now_utc else now_utc + start_point_for_bonus = ( + current_end_date if current_end_date > now_utc else now_utc + ) new_end_date_obj = start_point_for_bonus + timedelta(days=bonus_days) updated_sub_model = await subscription_dal.update_subscription_end_date( - session, active_sub.subscription_id, new_end_date_obj) + session, active_sub.subscription_id, new_end_date_obj + ) if updated_sub_model: - panel_update_success = await self.panel_service.update_user_details_on_panel( - user.panel_user_uuid, { - "expireAt": - new_end_date_obj.isoformat( - timespec='milliseconds').replace('+00:00', 'Z') - }) + panel_update_success = ( + await self.panel_service.update_user_details_on_panel( + user.panel_user_uuid, + { + "expireAt": new_end_date_obj.isoformat( + timespec="milliseconds" + ).replace("+00:00", "Z") + }, + ) + ) if not panel_update_success: logging.warning( f"Panel expiry update failed for {user.panel_user_uuid} after {reason} bonus. Local DB was updated to {new_end_date_obj}." @@ -492,8 +573,8 @@ class SubscriptionService: return None async def get_active_subscription_details( - self, session: AsyncSession, - user_id: int) -> Optional[Dict[str, Any]]: + self, session: AsyncSession, user_id: int + ) -> Optional[Dict[str, Any]]: db_user = await user_dal.get_user_by_id(session, user_id) if not db_user or not db_user.panel_user_uuid: logging.info( @@ -503,136 +584,144 @@ class SubscriptionService: panel_user_uuid = db_user.panel_user_uuid local_active_sub = await subscription_dal.get_active_subscription_by_user_id( - session, user_id, panel_user_uuid) - panel_user_data = await self.panel_service.get_user_by_uuid( - panel_user_uuid) + session, user_id, panel_user_uuid + ) + panel_user_data = await self.panel_service.get_user_by_uuid(panel_user_uuid) if not panel_user_data: logging.warning( f"Panel user {panel_user_uuid} not found on panel for user {user_id}. Using local data if available." ) - if local_active_sub and local_active_sub.end_date and local_active_sub.end_date > datetime.now( - timezone.utc): + if ( + local_active_sub + and local_active_sub.end_date + and local_active_sub.end_date > datetime.now(timezone.utc) + ): return { - "end_date": - local_active_sub.end_date, - "status_from_panel": - (local_active_sub.status_from_panel or "UNKNOWN") + - " (Panel Unreachable)", - "config_link": - await self.panel_service.get_subscription_link( - local_active_sub.panel_subscription_uuid) - if local_active_sub.panel_subscription_uuid else "N/A", - "traffic_limit_bytes": - local_active_sub.traffic_limit_bytes, - "traffic_used_bytes": - local_active_sub.traffic_used_bytes, - "user_bot_username": - db_user.username, - "is_panel_data": - False + "end_date": local_active_sub.end_date, + "status_from_panel": local_active_sub.status_from_panel + or "UNKNOWN", + "config_link": None, + "traffic_limit_bytes": local_active_sub.traffic_limit_bytes, + "traffic_used_bytes": local_active_sub.traffic_used_bytes, + "user_bot_username": db_user.username, + "is_panel_data": False, } return None if local_active_sub: update_payload_local = {} - panel_status = panel_user_data.get('status', 'UNKNOWN').upper() - panel_expire_at_str = panel_user_data.get('expireAt') - panel_traffic_used = panel_user_data.get('usedTrafficBytes') - panel_traffic_limit = panel_user_data.get('trafficLimitBytes') + panel_status = panel_user_data.get("status", "UNKNOWN").upper() + panel_expire_at_str = panel_user_data.get("expireAt") + panel_traffic_used = panel_user_data.get("usedTrafficBytes") + panel_traffic_limit = panel_user_data.get("trafficLimitBytes") panel_sub_uuid_from_panel = panel_user_data.get( - 'subscriptionUuid') or panel_user_data.get('shortUuid') + "subscriptionUuid" + ) or panel_user_data.get("shortUuid") if local_active_sub.status_from_panel != panel_status: - update_payload_local['status_from_panel'] = panel_status + update_payload_local["status_from_panel"] = panel_status if panel_expire_at_str: panel_expire_dt = datetime.fromisoformat( - panel_expire_at_str.replace("Z", "+00:00")) + panel_expire_at_str.replace("Z", "+00:00") + ) if local_active_sub.end_date.replace( - microsecond=0) != panel_expire_dt.replace( - microsecond=0): - update_payload_local['end_date'] = panel_expire_dt - update_payload_local['last_notification_sent'] = None - if panel_traffic_used is not None and local_active_sub.traffic_used_bytes != panel_traffic_used: - update_payload_local['traffic_used_bytes'] = panel_traffic_used - if panel_traffic_limit is not None and local_active_sub.traffic_limit_bytes != panel_traffic_limit: - update_payload_local[ - 'traffic_limit_bytes'] = panel_traffic_limit - if panel_sub_uuid_from_panel and local_active_sub.panel_subscription_uuid != panel_sub_uuid_from_panel: - update_payload_local[ - 'panel_subscription_uuid'] = panel_sub_uuid_from_panel + microsecond=0 + ) != panel_expire_dt.replace(microsecond=0): + update_payload_local["end_date"] = panel_expire_dt + update_payload_local["last_notification_sent"] = None + if ( + panel_traffic_used is not None + and local_active_sub.traffic_used_bytes != panel_traffic_used + ): + update_payload_local["traffic_used_bytes"] = panel_traffic_used + if ( + panel_traffic_limit is not None + and local_active_sub.traffic_limit_bytes != panel_traffic_limit + ): + update_payload_local["traffic_limit_bytes"] = panel_traffic_limit + if ( + panel_sub_uuid_from_panel + and local_active_sub.panel_subscription_uuid + != panel_sub_uuid_from_panel + ): + update_payload_local["panel_subscription_uuid"] = ( + panel_sub_uuid_from_panel + ) is_active_based_on_panel = panel_status == "ACTIVE" and ( panel_expire_dt > datetime.now(timezone.utc) - if panel_expire_dt else False) + if panel_expire_dt + else False + ) if local_active_sub.is_active != is_active_based_on_panel: - update_payload_local['is_active'] = is_active_based_on_panel + update_payload_local["is_active"] = is_active_based_on_panel if update_payload_local: await subscription_dal.update_subscription( - session, local_active_sub.subscription_id, - update_payload_local) + session, local_active_sub.subscription_id, update_payload_local + ) - panel_end_date = datetime.fromisoformat( - panel_user_data['expireAt'].replace( - "Z", "+00:00")) if panel_user_data.get('expireAt') else None + panel_end_date = ( + datetime.fromisoformat(panel_user_data["expireAt"].replace("Z", "+00:00")) + if panel_user_data.get("expireAt") + else None + ) return { - "end_date": - panel_end_date, - "status_from_panel": - panel_user_data.get('status', 'UNKNOWN').upper(), - "config_link": - panel_user_data.get('subscriptionUrl') - or (await self.panel_service.get_subscription_link( - panel_user_data['shortUuid']) - if panel_user_data.get('shortUuid') else "N/A"), - "traffic_limit_bytes": - panel_user_data.get('trafficLimitBytes'), - "traffic_used_bytes": - panel_user_data.get('usedTrafficBytes'), - "user_bot_username": - db_user.username, - "is_panel_data": - True + "end_date": panel_end_date, + "status_from_panel": panel_user_data.get("status", "UNKNOWN").upper(), + "config_link": panel_user_data.get("subscriptionUrl"), + "traffic_limit_bytes": panel_user_data.get("trafficLimitBytes"), + "traffic_used_bytes": panel_user_data.get("usedTrafficBytes"), + "user_bot_username": db_user.username, + "is_panel_data": True, } async def get_subscriptions_ending_soon( - self, session: AsyncSession, - days_threshold: int) -> List[Dict[str, Any]]: - subs_models_with_users = await subscription_dal.get_subscriptions_near_expiration( - session, days_threshold) + self, session: AsyncSession, days_threshold: int + ) -> List[Dict[str, Any]]: + subs_models_with_users = ( + await subscription_dal.get_subscriptions_near_expiration( + session, days_threshold + ) + ) results = [] for sub_model in subs_models_with_users: - if sub_model.user and sub_model.end_date and not sub_model.skip_notifications: - days_left = (sub_model.end_date - datetime.now( - timezone.utc)).total_seconds() / (24 * 3600) - results.append({ - 'user_id': - sub_model.user_id, - 'first_name': - sub_model.user.first_name or f"User {sub_model.user_id}", - 'language_code': - sub_model.user.language_code - or self.settings.DEFAULT_LANGUAGE, - 'end_date_str': - sub_model.end_date.strftime('%Y-%m-%d'), - 'days_left': - max(0, int(round(days_left))), - 'subscription_end_date_iso_for_update': - sub_model.end_date - }) + if ( + sub_model.user + and sub_model.end_date + and not sub_model.skip_notifications + ): + days_left = ( + sub_model.end_date - datetime.now(timezone.utc) + ).total_seconds() / (24 * 3600) + results.append( + { + "user_id": sub_model.user_id, + "first_name": sub_model.user.first_name + or f"User {sub_model.user_id}", + "language_code": sub_model.user.language_code + or self.settings.DEFAULT_LANGUAGE, + "end_date_str": sub_model.end_date.strftime("%Y-%m-%d"), + "days_left": max(0, int(round(days_left))), + "subscription_end_date_iso_for_update": sub_model.end_date, + } + ) return results - async def update_last_notification_sent(self, session: AsyncSession, - user_id: int, - subscription_end_date: datetime): - sub_to_update = await subscription_dal.find_subscription_for_notification_update( - session, user_id, subscription_end_date) + async def update_last_notification_sent( + self, session: AsyncSession, user_id: int, subscription_end_date: datetime + ): + sub_to_update = ( + await subscription_dal.find_subscription_for_notification_update( + session, user_id, subscription_end_date + ) + ) if sub_to_update: await subscription_dal.update_subscription_notification_time( - session, sub_to_update.subscription_id, - datetime.now(timezone.utc)) + session, sub_to_update.subscription_id, datetime.now(timezone.utc) + ) logging.info( f"Updated last_notification_sent for user {user_id}, sub_id {sub_to_update.subscription_id}" ) diff --git a/db/dal/user_dal.py b/db/dal/user_dal.py index 526d4ef..e6cb4be 100644 --- a/db/dal/user_dal.py +++ b/db/dal/user_dal.py @@ -9,33 +9,31 @@ from datetime import datetime from ..models import User, Subscription -async def get_user_by_id(session: AsyncSession, - user_id: int) -> Optional[User]: +async def get_user_by_id(session: AsyncSession, user_id: int) -> Optional[User]: stmt = select(User).where(User.user_id == user_id) result = await session.execute(stmt) return result.scalar_one_or_none() -async def get_user_by_username(session: AsyncSession, - username: str) -> Optional[User]: - clean_username = username.lstrip('@').lower() +async def get_user_by_username(session: AsyncSession, username: str) -> Optional[User]: + clean_username = username.lstrip("@").lower() stmt = select(User).where(func.lower(User.username) == clean_username) result = await session.execute(stmt) return result.scalar_one_or_none() -async def get_user_by_panel_uuid(session: AsyncSession, - panel_uuid: str) -> Optional[User]: +async def get_user_by_panel_uuid( + session: AsyncSession, panel_uuid: str +) -> Optional[User]: stmt = select(User).where(User.panel_user_uuid == panel_uuid) result = await session.execute(stmt) return result.scalar_one_or_none() -async def create_user(session: AsyncSession, user_data: Dict[str, - Any]) -> User: +async def create_user(session: AsyncSession, user_data: Dict[str, Any]) -> User: - if 'registration_date' not in user_data: - user_data['registration_date'] = datetime.now() + if "registration_date" not in user_data: + user_data["registration_date"] = datetime.now() new_user = User(**user_data) session.add(new_user) @@ -47,8 +45,9 @@ async def create_user(session: AsyncSession, user_data: Dict[str, return new_user -async def update_user(session: AsyncSession, user_id: int, - update_data: Dict[str, Any]) -> Optional[User]: +async def update_user( + session: AsyncSession, user_id: int, update_data: Dict[str, Any] +) -> Optional[User]: user = await get_user_by_id(session, user_id) if user: for key, value in update_data.items(): @@ -58,16 +57,17 @@ async def update_user(session: AsyncSession, user_id: int, return user -async def update_user_language(session: AsyncSession, user_id: int, - lang_code: str) -> bool: - stmt = update(User).where(User.user_id == user_id).values( - language_code=lang_code) +async def update_user_language( + session: AsyncSession, user_id: int, lang_code: str +) -> bool: + stmt = update(User).where(User.user_id == user_id).values(language_code=lang_code) result = await session.execute(stmt) return result.rowcount > 0 -async def set_user_ban_status(session: AsyncSession, user_id: int, - is_banned: bool) -> bool: +async def set_user_ban_status( + session: AsyncSession, user_id: int, is_banned: bool +) -> bool: user = await get_user_by_id(session, user_id) if user: user.is_banned = is_banned @@ -77,23 +77,27 @@ async def set_user_ban_status(session: AsyncSession, user_id: int, return False -async def get_banned_users_paginated(session: AsyncSession, limit: int, - offset: int) -> Tuple[List[User], int]: - stmt_users = select(User).where(User.is_banned == True).order_by( - User.registration_date.desc()).limit(limit).offset(offset) +async def get_banned_users_paginated( + session: AsyncSession, limit: int, offset: int +) -> Tuple[List[User], int]: + stmt_users = ( + select(User) + .where(User.is_banned == True) + .order_by(User.registration_date.desc()) + .limit(limit) + .offset(offset) + ) result_users = await session.execute(stmt_users) users_list = result_users.scalars().all() - stmt_count = select( - func.count()).select_from(User).where(User.is_banned == True) + stmt_count = select(func.count()).select_from(User).where(User.is_banned == True) result_count = await session.execute(stmt_count) total_banned = result_count.scalar_one() return users_list, total_banned -async def get_all_active_user_ids_for_broadcast( - session: AsyncSession) -> List[int]: +async def get_all_active_user_ids_for_broadcast(session: AsyncSession) -> List[int]: stmt = select(User.user_id).where(User.is_banned == False) result = await session.execute(stmt) return result.scalars().all() @@ -101,24 +105,31 @@ async def get_all_active_user_ids_for_broadcast( async def get_user_count_stats_dal(session: AsyncSession) -> Dict[str, int]: total_users_stmt = select(func.count(User.user_id)).select_from(User) - banned_users_stmt = select(func.count( - User.user_id)).select_from(User).where(User.is_banned == True) + banned_users_stmt = ( + select(func.count(User.user_id)).select_from(User).where(User.is_banned == True) + ) - active_subs_stmt = (select(func.count( - func.distinct(Subscription.user_id))).join( - User, Subscription.user_id == User.user_id).where( - Subscription.is_active == True).where( - Subscription.end_date > datetime.now())) + active_subs_stmt = ( + select(func.count(func.distinct(Subscription.user_id))) + .join(User, Subscription.user_id == User.user_id) + .where(Subscription.is_active == True) + .where(Subscription.end_date > datetime.now()) + ) - total_users = (await - session.execute(total_users_stmt)).scalar_one_or_none() or 0 - banned_users = ( - await session.execute(banned_users_stmt)).scalar_one_or_none() or 0 + total_users = (await session.execute(total_users_stmt)).scalar_one_or_none() or 0 + banned_users = (await session.execute(banned_users_stmt)).scalar_one_or_none() or 0 active_subs_users = ( - await session.execute(active_subs_stmt)).scalar_one_or_none() or 0 + await session.execute(active_subs_stmt) + ).scalar_one_or_none() or 0 return { "total_users": total_users, "banned_users": banned_users, "users_with_active_subscriptions": active_subs_users, } + + +async def get_all_users_with_panel_uuid(session: AsyncSession) -> List[User]: + stmt = select(User).where(User.panel_user_uuid.is_not(None)) + result = await session.execute(stmt) + return result.scalars().all() diff --git a/locales/en.json b/locales/en.json index f20c452..359394f 100644 --- a/locales/en.json +++ b/locales/en.json @@ -198,6 +198,7 @@ "sync_completed_with_errors_details": "Checked: {total_checked} entries.\nUsers synced/updated: {users_synced}.\nSubscriptions synced/updated: {subs_synced}.\nErrors: {errors_count}.\n\nFirst errors:\n{error_details_preview}", "no_errors_placeholder": "none", "admin_sync_initiated_from_panel": "Sync initiated...", + "admin_panel_user_creation_failed": "❌ Failed to create panel user for TG ID {user_id}. Panel unreachable?", "error_displaying_logs_too_long": "Error: logs too long to display in one message. Try viewing logs for a specific user.", "error_displaying_statistics": "Error displaying statistics.", "stub_page_display": "Page", diff --git a/locales/ru.json b/locales/ru.json index 5e852f6..1250214 100644 --- a/locales/ru.json +++ b/locales/ru.json @@ -198,6 +198,7 @@ "sync_completed_with_errors_details": "Проверено: {total_checked} записей.\nПользователей синхронизировано/обновлено: {users_synced}.\nПодписок синхронизировано/обновлено: {subs_synced}.\nОшибок: {errors_count}.\n\nПервые ошибки:\n{error_details_preview}", "no_errors_placeholder": "нет", "admin_sync_initiated_from_panel": "Синхронизация запущена...", + "admin_panel_user_creation_failed": "❌ Не удалось создать пользователя на панели для TG ID {user_id}. Панель недоступна?", "error_displaying_logs_too_long": "Ошибка: логи слишком длинные для отображения одним сообщением. Попробуйте найти логи по конкретному пользователю.", "error_displaying_statistics": "Ошибка отображения статистики.", "stub_page_display": "Страница",