From 1f75413abd360530f2a97dd2bfa578cdf842a06f Mon Sep 17 00:00:00 2001 From: Egor Date: Fri, 28 Nov 2025 02:48:49 +0300 Subject: [PATCH] Handle RemnaWave API status and expire constraints --- app/handlers/admin/remnawave.py | 54 ++++++++ app/handlers/admin/users.py | 26 +++- app/keyboards/admin.py | 6 + app/localization/locales/ru.json | 1 + app/localization/locales/ua.json | 1 + app/services/monitoring_service.py | 22 +++- app/services/remnawave_service.py | 183 +++++++++++++++++---------- app/services/subscription_service.py | 68 ++++++++-- 8 files changed, 274 insertions(+), 87 deletions(-) diff --git a/app/handlers/admin/remnawave.py b/app/handlers/admin/remnawave.py index e63c1c80..968a56f2 100644 --- a/app/handlers/admin/remnawave.py +++ b/app/handlers/admin/remnawave.py @@ -2285,6 +2285,9 @@ async def show_sync_options( "• При полной синхронизации подписки пользователей, отсутствующих в панели, будут деактивированы\n" "• Рекомендуется делать полную синхронизацию ежедневно\n" "• Баланс пользователей НЕ удаляется\n\n" + "⬆️ Обратная синхронизация:\n" + "• Отправляет активных пользователей из бота в панель\n" + "• Используйте при сбоях панели или для восстановления данных\n\n" + "\n".join(status_lines) ) @@ -2295,6 +2298,12 @@ async def show_sync_options( callback_data="sync_all_users", ) ], + [ + types.InlineKeyboardButton( + text="⬆️ Синхронизация в панель", + callback_data="sync_to_panel", + ) + ], [ types.InlineKeyboardButton( text="⚙️ Настройки автосинхронизации", @@ -2654,6 +2663,50 @@ async def sync_all_users( ) await callback.answer() + +@admin_required +@error_handler +async def sync_users_to_panel( + callback: types.CallbackQuery, + db_user: User, + db: AsyncSession, +): + await callback.message.edit_text( + "⬆️ Выполняется синхронизация данных бота в панель Remnawave...\n\n" + "Это может занять несколько минут.", + reply_markup=None, + ) + + remnawave_service = RemnaWaveService() + stats = await remnawave_service.sync_users_to_panel(db) + + if stats["errors"] == 0: + status_emoji = "✅" + status_text = "успешно завершена" + else: + status_emoji = "⚠️" if (stats["created"] + stats["updated"]) > 0 else "❌" + status_text = "завершена с предупреждениями" if status_emoji == "⚠️" else "завершена с ошибками" + + text = ( + f"{status_emoji} Синхронизация в панель {status_text}\n\n" + "📊 Результаты:\n" + f"• 🆕 Создано: {stats['created']}\n" + f"• 🔄 Обновлено: {stats['updated']}\n" + f"• ❌ Ошибок: {stats['errors']}" + ) + + keyboard = [ + [types.InlineKeyboardButton(text="🔄 Повторить", callback_data="sync_to_panel")], + [types.InlineKeyboardButton(text="🔄 Полная синхронизация", callback_data="sync_all_users")], + [types.InlineKeyboardButton(text="⬅️ К синхронизации", callback_data="admin_rw_sync")], + ] + + await callback.message.edit_text( + text, + reply_markup=types.InlineKeyboardMarkup(inline_keyboard=keyboard), + ) + await callback.answer() + @admin_required @error_handler async def show_sync_recommendations( @@ -3126,6 +3179,7 @@ def register_handlers(dp: Dispatcher): dp.callback_query.register(cancel_auto_sync_schedule, F.data == "remnawave_auto_sync_cancel") dp.callback_query.register(run_auto_sync_now, F.data == "remnawave_auto_sync_run") dp.callback_query.register(sync_all_users, F.data == "sync_all_users") + dp.callback_query.register(sync_users_to_panel, F.data == "sync_to_panel") dp.callback_query.register(show_squad_migration_menu, F.data == "admin_rw_migration") dp.callback_query.register(paginate_migration_source, F.data.startswith("admin_migration_source_page_")) dp.callback_query.register(handle_migration_source_selection, F.data.startswith("admin_migration_source_")) diff --git a/app/handlers/admin/users.py b/app/handlers/admin/users.py index d25499e6..9c9c9cb9 100644 --- a/app/handlers/admin/users.py +++ b/app/handlers/admin/users.py @@ -4595,10 +4595,19 @@ async def admin_buy_subscription_execute( if target_user.remnawave_uuid: async with remnawave_service.get_api_client() as api: + expire_at = remnawave_service.ensure_future_expire_at( + subscription.end_date + ) + status = ( + UserStatus.ACTIVE + if subscription.is_active + else UserStatus.DISABLED + ) + update_kwargs = dict( uuid=target_user.remnawave_uuid, - status=UserStatus.ACTIVE if subscription.is_active else UserStatus.EXPIRED, - expire_at=subscription.end_date, + status=status, + expire_at=expire_at, traffic_limit_bytes=subscription.traffic_limit_gb * (1024**3) if subscription.traffic_limit_gb > 0 else 0, traffic_limit_strategy=TrafficLimitStrategy.MONTH, description=settings.format_remnawave_user_description( @@ -4620,10 +4629,19 @@ async def admin_buy_subscription_execute( telegram_id=target_user.telegram_id, ) async with remnawave_service.get_api_client() as api: + expire_at = remnawave_service.ensure_future_expire_at( + subscription.end_date + ) + status = ( + UserStatus.ACTIVE + if subscription.is_active + else UserStatus.DISABLED + ) + create_kwargs = dict( username=username, - expire_at=subscription.end_date, - status=UserStatus.ACTIVE if subscription.is_active else UserStatus.EXPIRED, + expire_at=expire_at, + status=status, traffic_limit_bytes=subscription.traffic_limit_gb * (1024**3) if subscription.traffic_limit_gb > 0 else 0, traffic_limit_strategy=TrafficLimitStrategy.MONTH, telegram_id=target_user.telegram_id, diff --git a/app/keyboards/admin.py b/app/keyboards/admin.py index 6272a531..964984fb 100644 --- a/app/keyboards/admin.py +++ b/app/keyboards/admin.py @@ -1118,6 +1118,12 @@ def get_sync_options_keyboard(language: str = "ru") -> InlineKeyboardMarkup: callback_data="sync_all_users" ) ], + [ + InlineKeyboardButton( + text=_t(texts, "ADMIN_SYNC_TO_PANEL", "⬆️ Синхронизация в панель"), + callback_data="sync_to_panel" + ) + ], [ InlineKeyboardButton( text=_t(texts, "ADMIN_SYNC_ONLY_NEW", "🆕 Только новые"), diff --git a/app/localization/locales/ru.json b/app/localization/locales/ru.json index 3519200a..3d3aef2c 100644 --- a/app/localization/locales/ru.json +++ b/app/localization/locales/ru.json @@ -688,6 +688,7 @@ "ADMIN_SUPPORT_SUBMENU_TITLE": "🛟 **Поддержка**\n\n", "ADMIN_SUPPORT_TICKETS": "🎫 Тикеты поддержки", "ADMIN_SYNC_BACK": "⬅️ К синхронизации", + "ADMIN_SYNC_TO_PANEL": "⬆️ Синхронизация в панель", "ADMIN_SYNC_CLEANUP": "🧹 Очистка", "ADMIN_SYNC_CONFIRM": "✅ Подтвердить", "ADMIN_SYNC_FULL": "🔄 Полная синхронизация", diff --git a/app/localization/locales/ua.json b/app/localization/locales/ua.json index f0b24625..d291f56a 100644 --- a/app/localization/locales/ua.json +++ b/app/localization/locales/ua.json @@ -687,6 +687,7 @@ "ADMIN_SUPPORT_SUBMENU_TITLE": "🛟 **Підтримка**\n\n", "ADMIN_SUPPORT_TICKETS": "🎫 Тікети підтримки", "ADMIN_SYNC_BACK": "⬅️ До синхронізації", + "ADMIN_SYNC_TO_PANEL": "⬆️ Синхронізація в панель", "ADMIN_SYNC_CLEANUP": "🧹 Очищення", "ADMIN_SYNC_CONFIRM": "✅ Підтвердити", "ADMIN_SYNC_FULL": "🔄 Повна синхронізація", diff --git a/app/services/monitoring_service.py b/app/services/monitoring_service.py index 9a0f6f95..1a0c300c 100644 --- a/app/services/monitoring_service.py +++ b/app/services/monitoring_service.py @@ -281,11 +281,15 @@ class MonitoringService: return None current_time = datetime.utcnow() - is_active = (subscription.status == SubscriptionStatus.ACTIVE.value and - subscription.end_date > current_time) + is_active = ( + subscription.status == SubscriptionStatus.ACTIVE.value + and subscription.end_date > current_time + ) - if (subscription.status == SubscriptionStatus.ACTIVE.value and - subscription.end_date <= current_time): + if ( + subscription.status == SubscriptionStatus.ACTIVE.value + and subscription.end_date <= current_time + ): subscription.status = SubscriptionStatus.EXPIRED.value await db.commit() is_active = False @@ -301,10 +305,16 @@ class MonitoringService: async with self.subscription_service.get_api_client() as api: hwid_limit = resolve_hwid_device_limit_for_payload(subscription) + expire_at = self.subscription_service.ensure_future_expire_at( + subscription.end_date + ) + + status = UserStatus.ACTIVE if is_active else UserStatus.DISABLED + update_kwargs = dict( uuid=user.remnawave_uuid, - status=UserStatus.ACTIVE if is_active else UserStatus.EXPIRED, - expire_at=subscription.end_date, + status=status, + expire_at=expire_at, traffic_limit_bytes=self._gb_to_bytes(subscription.traffic_limit_gb), traffic_limit_strategy=TrafficLimitStrategy.MONTH, description=settings.format_remnawave_user_description( diff --git a/app/services/remnawave_service.py b/app/services/remnawave_service.py index e2cb7db5..22f9e183 100644 --- a/app/services/remnawave_service.py +++ b/app/services/remnawave_service.py @@ -220,6 +220,27 @@ class RemnaWaveService: """Возвращает текущее время в UTC без привязки к часовому поясу.""" return datetime.now(self._utc_timezone).replace(tzinfo=None) + def ensure_future_expire_at(self, expire_at: Optional[datetime]) -> datetime: + """Приводит дату окончания подписки к будущему значению для API панели.""" + + safe_now = self._now_utc() + + if expire_at is None: + adjusted = safe_now + timedelta(days=30) + logger.debug("⚙️ Используем дефолтную дату окончания подписки: %s", adjusted) + return adjusted + + if expire_at <= safe_now: + adjusted = safe_now + timedelta(minutes=1) + logger.debug( + "⚙️ Корректируем просроченную дату подписки %s → %s для RemnaWave API", + expire_at, + adjusted, + ) + return adjusted + + return expire_at + def _parse_remnawave_date(self, date_str: str) -> datetime: if not date_str: return self._now_utc() + timedelta(days=30) @@ -1615,77 +1636,109 @@ class RemnaWaveService: async def sync_users_to_panel(self, db: AsyncSession) -> Dict[str, int]: try: stats = {"created": 0, "updated": 0, "errors": 0} - - users = await get_users_list(db, offset=0, limit=10000) - + + batch_size = 100 + offset = 0 + async with self.get_api_client() as api: - for user in users: - if not user.subscription: - continue + while True: + users = await get_users_list(db, offset=offset, limit=batch_size) + + if not users: + break + + for user in users: + if not user.subscription: + continue + + try: + subscription = user.subscription + hwid_limit = resolve_hwid_device_limit_for_payload(subscription) + + expire_at = self.ensure_future_expire_at( + subscription.end_date + ) + + status = ( + UserStatus.ACTIVE + if subscription.is_active + else UserStatus.DISABLED + ) + + if user.remnawave_uuid: + update_kwargs = dict( + uuid=user.remnawave_uuid, + status=status, + expire_at=expire_at, + traffic_limit_bytes=subscription.traffic_limit_gb * (1024**3) if subscription.traffic_limit_gb > 0 else 0, + traffic_limit_strategy=TrafficLimitStrategy.MONTH, + description=settings.format_remnawave_user_description( + full_name=user.full_name, + username=user.username, + telegram_id=user.telegram_id + ), + active_internal_squads=subscription.connected_squads, + ) + + if hwid_limit is not None: + update_kwargs['hwid_device_limit'] = hwid_limit + + await api.update_user(**update_kwargs) + stats["updated"] += 1 + else: + username = settings.format_remnawave_username( + full_name=user.full_name, + username=user.username, + telegram_id=user.telegram_id, + ) + + create_kwargs = dict( + username=username, + expire_at=expire_at, + status=status, + traffic_limit_bytes=subscription.traffic_limit_gb * (1024**3) if subscription.traffic_limit_gb > 0 else 0, + traffic_limit_strategy=TrafficLimitStrategy.MONTH, + telegram_id=user.telegram_id, + description=settings.format_remnawave_user_description( + full_name=user.full_name, + username=user.username, + telegram_id=user.telegram_id + ), + active_internal_squads=subscription.connected_squads, + ) + + if hwid_limit is not None: + create_kwargs['hwid_device_limit'] = hwid_limit + + new_user = await api.create_user(**create_kwargs) + + user.remnawave_uuid = new_user.uuid + subscription.remnawave_short_uuid = new_user.short_uuid + + stats["created"] += 1 + + except Exception as e: + logger.error(f"Ошибка синхронизации пользователя {user.telegram_id} в панель: {e}") + stats["errors"] += 1 try: - subscription = user.subscription - hwid_limit = resolve_hwid_device_limit_for_payload(subscription) + await db.commit() + except Exception as commit_error: + logger.error( + "Ошибка фиксации транзакции при синхронизации в панель: %s", + commit_error, + ) + await db.rollback() + stats["errors"] += len(users) - if user.remnawave_uuid: - update_kwargs = dict( - uuid=user.remnawave_uuid, - status=UserStatus.ACTIVE if subscription.is_active else UserStatus.EXPIRED, - expire_at=subscription.end_date, - traffic_limit_bytes=subscription.traffic_limit_gb * (1024**3) if subscription.traffic_limit_gb > 0 else 0, - traffic_limit_strategy=TrafficLimitStrategy.MONTH, - description=settings.format_remnawave_user_description( - full_name=user.full_name, - username=user.username, - telegram_id=user.telegram_id - ), - active_internal_squads=subscription.connected_squads, - ) + if len(users) < batch_size: + break - if hwid_limit is not None: - update_kwargs['hwid_device_limit'] = hwid_limit + offset += batch_size - await api.update_user(**update_kwargs) - stats["updated"] += 1 - else: - username = settings.format_remnawave_username( - full_name=user.full_name, - username=user.username, - telegram_id=user.telegram_id, - ) - - create_kwargs = dict( - username=username, - expire_at=subscription.end_date, - status=UserStatus.ACTIVE if subscription.is_active else UserStatus.EXPIRED, - traffic_limit_bytes=subscription.traffic_limit_gb * (1024**3) if subscription.traffic_limit_gb > 0 else 0, - traffic_limit_strategy=TrafficLimitStrategy.MONTH, - telegram_id=user.telegram_id, - description=settings.format_remnawave_user_description( - full_name=user.full_name, - username=user.username, - telegram_id=user.telegram_id - ), - active_internal_squads=subscription.connected_squads, - ) - - if hwid_limit is not None: - create_kwargs['hwid_device_limit'] = hwid_limit - - new_user = await api.create_user(**create_kwargs) - - await update_user(db, user, remnawave_uuid=new_user.uuid) - subscription.remnawave_short_uuid = new_user.short_uuid - # Убираем немедленный коммит для пакетной обработки - # await db.commit() - - stats["created"] += 1 - - except Exception as e: - logger.error(f"Ошибка синхронизации пользователя {user.telegram_id} в панель: {e}") - stats["errors"] += 1 - - logger.info(f"✅ Синхронизация в панель завершена: создано {stats['created']}, обновлено {stats['updated']}, ошибок {stats['errors']}") + logger.info( + f"✅ Синхронизация в панель завершена: создано {stats['created']}, обновлено {stats['updated']}, ошибок {stats['errors']}" + ) return stats except Exception as e: diff --git a/app/services/subscription_service.py b/app/services/subscription_service.py index 623aeaf9..d04284b5 100644 --- a/app/services/subscription_service.py +++ b/app/services/subscription_service.py @@ -153,6 +153,27 @@ class SubscriptionService: assert self.api is not None async with self.api as api: yield api + + def ensure_future_expire_at(self, expire_at: Optional[datetime]) -> datetime: + """Гарантирует, что дата окончания передается в будущее значение для RemnaWave API.""" + + safe_now = datetime.utcnow() + + if expire_at is None: + adjusted = safe_now + timedelta(days=30) + logger.debug("⚙️ Используем дефолтную дату окончания подписки: %s", adjusted) + return adjusted + + if expire_at <= safe_now: + adjusted = safe_now + timedelta(minutes=1) + logger.debug( + "⚙️ Корректируем просроченную дату подписки %s → %s для RemnaWave API", + expire_at, + adjusted, + ) + return adjusted + + return expire_at async def create_remnawave_user( self, @@ -187,10 +208,18 @@ class SubscriptionService: except Exception as hwid_error: logger.warning(f"⚠️ Не удалось сбросить HWID: {hwid_error}") + expire_at = self.ensure_future_expire_at(subscription.end_date) + + status = ( + UserStatus.ACTIVE + if subscription.is_active + else UserStatus.DISABLED + ) + update_kwargs = dict( uuid=remnawave_user.uuid, - status=UserStatus.ACTIVE, - expire_at=subscription.end_date, + status=status, + expire_at=expire_at, traffic_limit_bytes=self._gb_to_bytes(subscription.traffic_limit_gb), traffic_limit_strategy=get_traffic_reset_strategy(), description=settings.format_remnawave_user_description( @@ -221,10 +250,18 @@ class SubscriptionService: username=user.username, telegram_id=user.telegram_id, ) + expire_at = self.ensure_future_expire_at(subscription.end_date) + + status = ( + UserStatus.ACTIVE + if subscription.is_active + else UserStatus.DISABLED + ) + create_kwargs = dict( username=username, - expire_at=subscription.end_date, - status=UserStatus.ACTIVE, + expire_at=expire_at, + status=status, traffic_limit_bytes=self._gb_to_bytes(subscription.traffic_limit_gb), traffic_limit_strategy=get_traffic_reset_strategy(), telegram_id=user.telegram_id, @@ -285,25 +322,32 @@ class SubscriptionService: return None current_time = datetime.utcnow() - is_actually_active = (subscription.status == SubscriptionStatus.ACTIVE.value and - subscription.end_date > current_time) + is_actually_active = ( + subscription.status == SubscriptionStatus.ACTIVE.value + and subscription.end_date > current_time + ) - if (subscription.status == SubscriptionStatus.ACTIVE.value and - subscription.end_date <= current_time): - + if ( + subscription.status == SubscriptionStatus.ACTIVE.value + and subscription.end_date <= current_time + ): subscription.status = SubscriptionStatus.EXPIRED.value subscription.updated_at = current_time await db.commit() is_actually_active = False logger.info(f"🔔 Статус подписки {subscription.id} автоматически изменен на 'expired'") - + async with self.get_api_client() as api: hwid_limit = resolve_hwid_device_limit_for_payload(subscription) + expire_at = self.ensure_future_expire_at(subscription.end_date) + + status = UserStatus.ACTIVE if is_actually_active else UserStatus.DISABLED + update_kwargs = dict( uuid=user.remnawave_uuid, - status=UserStatus.ACTIVE if is_actually_active else UserStatus.EXPIRED, - expire_at=subscription.end_date, + status=status, + expire_at=expire_at, traffic_limit_bytes=self._gb_to_bytes(subscription.traffic_limit_gb), traffic_limit_strategy=get_traffic_reset_strategy(), description=settings.format_remnawave_user_description(