From 73a1ea8e1bef21ceda1760d3d89f7fb5c830d914 Mon Sep 17 00:00:00 2001 From: Capybara-z Date: Sun, 21 Dec 2025 16:36:18 +0300 Subject: [PATCH] refact sync --- handlers/admin/clusters/cluster_sync.py | 465 ++++++++++++++++-------- handlers/keys/operations/creation.py | 10 + 2 files changed, 330 insertions(+), 145 deletions(-) diff --git a/handlers/admin/clusters/cluster_sync.py b/handlers/admin/clusters/cluster_sync.py index 650307b5..6cd8c6fc 100644 --- a/handlers/admin/clusters/cluster_sync.py +++ b/handlers/admin/clusters/cluster_sync.py @@ -221,6 +221,10 @@ async def handle_sync_server( Key.expiry_time, Key.tariff_id, Key.remnawave_link, + Key.selected_device_limit, + Key.selected_traffic_limit, + Key.current_device_limit, + Key.current_traffic_limit, ) .join(Key, Server.server_name == Key.server_id) .where(Server.server_name == server_name) @@ -238,6 +242,10 @@ async def handle_sync_server( Key.expiry_time, Key.tariff_id, Key.remnawave_link, + Key.selected_device_limit, + Key.selected_traffic_limit, + Key.current_device_limit, + Key.current_traffic_limit, ) .join(Key, Server.cluster_name == Key.server_id) .where(Server.server_name == server_name) @@ -257,33 +265,37 @@ async def handle_sync_server( text=f"🔄 Синхронизация сервера {server_name}\n\n🔑 Количество ключей: {len(keys_to_sync)}" ) + all_servers = await get_servers(session) + cluster_servers = all_servers.get(cluster_name, []) + + server_info = None + for s in cluster_servers: + if s.get("server_name") == server_name: + server_info = s + break + + tariff_ids = {key["tariff_id"] for key in keys_to_sync if key["tariff_id"]} + tariffs_cache = {} + if tariff_ids: + tariffs_result = await session.execute(select(Tariff).where(Tariff.id.in_(tariff_ids))) + tariffs_list = tariffs_result.scalars().all() + tariffs_cache = {t.id: dict(t.__dict__) for t in tariffs_list} + semaphore = asyncio.Semaphore(2) for key in keys_to_sync: try: if key["panel_type"] == "remnawave": - tariff = None - if key["tariff_id"]: - tariff = await session.get(Tariff, key["tariff_id"]) - if tariff: - servers = await get_servers(session) - server_info = None - for cluster_servers in servers.values(): - for s in cluster_servers: - if s.get("server_name") == server_name: - server_info = s - break - if server_info: - break + tariff = tariffs_cache.get(key["tariff_id"]) if key["tariff_id"] else None + + if tariff and server_info: + if tariff.get("subgroup_title") and tariff.get("subgroup_title") not in server_info.get( + "tariff_subgroups", [] + ): + continue - if server_info: - if tariff.subgroup_title and tariff.subgroup_title not in server_info.get( - "tariff_subgroups", [] - ): - continue - - if tariff.group_code and tariff.group_code.lower() in ALLOWED_GROUP_CODES: - if tariff.group_code.lower() not in server_info.get("special_groups", []): - continue + if tariff.get("group_code") and tariff.get("group_code").lower() in ALLOWED_GROUP_CODES: + if tariff.get("group_code").lower() not in server_info.get("special_groups", []): + continue expire_iso = ( datetime.utcfromtimestamp(key["expiry_time"] / 1000).replace(tzinfo=timezone.utc).isoformat() @@ -296,10 +308,26 @@ async def handle_sync_server( traffic_limit_bytes = 0 hwid_limit = 0 + + current_device_limit_from_key = key.get("current_device_limit") + current_traffic_limit_gb_from_key = key.get("current_traffic_limit") + selected_device_limit_from_key = key.get("selected_device_limit") + selected_traffic_limit_gb_from_key = key.get("selected_traffic_limit") + if tariff: - if tariff.traffic_limit is not None: - traffic_limit_bytes = int(tariff.traffic_limit * 1024**3) - hwid_limit = tariff.device_limit + if current_traffic_limit_gb_from_key is not None: + traffic_limit_bytes = int(current_traffic_limit_gb_from_key * 1024**3) + elif selected_traffic_limit_gb_from_key is not None: + traffic_limit_bytes = int(selected_traffic_limit_gb_from_key * 1024**3) + elif tariff.get("traffic_limit") is not None: + traffic_limit_bytes = int(tariff.get("traffic_limit") * 1024**3) + + if current_device_limit_from_key is not None: + hwid_limit = int(current_device_limit_from_key) + elif selected_device_limit_from_key is not None: + hwid_limit = int(selected_device_limit_from_key) + else: + hwid_limit = tariff.get("device_limit") success = await remna.update_user( uuid=key["client_id"], @@ -318,14 +346,6 @@ async def handle_sync_server( new_remnawave_link = sub.get("subscriptionUrl") if new_remnawave_link: - server_result = await session.execute( - select(Server.cluster_name).where(Server.server_name == server_name) - ) - cluster_name = server_result.scalar() - - servers = await get_servers(session) - cluster_servers = servers.get(cluster_name, []) - key_value = await make_aggregated_link( session=session, cluster_all=cluster_servers, @@ -334,7 +354,7 @@ async def handle_sync_server( client_id=key["client_id"], tg_id=key["tg_id"], remna_link_override=None, - plan=key["tariff_id"], + plan=tariff, ) await session.execute( @@ -361,6 +381,13 @@ async def handle_sync_server( plan=key["tariff_id"], session=session, remnawave_link=key["remnawave_link"], + hwid_limit=hwid_limit, + traffic_limit_bytes=traffic_limit_bytes, + selected_device_limit=key.get("selected_device_limit"), + selected_traffic_limit_gb=key.get("selected_traffic_limit"), + current_device_limit=key.get("current_device_limit"), + current_traffic_limit_gb=key.get("current_traffic_limit"), + selected_price_rub=key.get("selected_price_rub"), ) else: await create_client_on_server( @@ -377,7 +404,7 @@ async def handle_sync_server( plan=key["tariff_id"], session=session, ) - await asyncio.sleep(0.6) + await asyncio.sleep(0.6) except Exception as e: logger.error(f"Ошибка при синхронизации ключа {key['client_id']} в сервер {server_name}: {e}") @@ -424,6 +451,10 @@ async def handle_sync_cluster( Key.remnawave_link, Key.tariff_id, Key.server_id, + Key.selected_device_limit, + Key.selected_traffic_limit, + Key.current_device_limit, + Key.current_traffic_limit, ).where(Key.server_id.in_(server_names), Key.is_frozen.is_(False)) ) else: @@ -436,6 +467,10 @@ async def handle_sync_cluster( Key.remnawave_link, Key.tariff_id, Key.server_id, + Key.selected_device_limit, + Key.selected_traffic_limit, + Key.current_device_limit, + Key.current_traffic_limit, ).where(Key.server_id == cluster_name, Key.is_frozen.is_(False)) ) @@ -453,131 +488,201 @@ async def handle_sync_cluster( text=f"🔄 Синхронизация кластера {cluster_name}\n\n🔑 Количество ключей: {len(keys_to_sync)}" ) - for key in keys_to_sync: - try: - if only_remnawave: - expire_iso = ( - datetime.utcfromtimestamp(key["expiry_time"] / 1000).replace(tzinfo=timezone.utc).isoformat() - ) + tariff_ids = {key["tariff_id"] for key in keys_to_sync if key["tariff_id"]} + tariffs_cache = {} + if tariff_ids: + tariffs_result = await session.execute(select(Tariff).where(Tariff.id.in_(tariff_ids))) + tariffs_list = tariffs_result.scalars().all() + tariffs_cache = {t.id: dict(t.__dict__) for t in tariffs_list} - traffic_limit_bytes = 0 - hwid_limit = 0 - subgroup_title = None - tariff = None - if key["tariff_id"]: - tariff = await session.get(Tariff, key["tariff_id"]) + if only_remnawave: + batch_size = 50 + total_keys = len(keys_to_sync) + processed_count = 0 + + for batch_start in range(0, total_keys, batch_size): + batch = keys_to_sync[batch_start:batch_start + batch_size] + batch_end = batch_start + len(batch) + logger.info(f"[Sync] Обработка батча {batch_start}-{batch_end} из {total_keys}") + + async def update_remnawave_api(key): + try: + traffic_limit_bytes = 0 + hwid_limit = 0 + subgroup_title = None + tariff = tariffs_cache.get(key["tariff_id"]) if key["tariff_id"] else None + + current_device_limit_from_key = key.get("current_device_limit") + current_traffic_limit_gb_from_key = key.get("current_traffic_limit") + selected_device_limit_from_key = key.get("selected_device_limit") + selected_traffic_limit_gb_from_key = key.get("selected_traffic_limit") + if tariff: - if tariff.traffic_limit is not None: - traffic_limit_bytes = int(tariff.traffic_limit * 1024**3) + if current_traffic_limit_gb_from_key is not None: + traffic_limit_bytes = int(current_traffic_limit_gb_from_key * 1024**3) + elif selected_traffic_limit_gb_from_key is not None: + traffic_limit_bytes = int(selected_traffic_limit_gb_from_key * 1024**3) + elif tariff.get("traffic_limit") is not None: + traffic_limit_bytes = int(tariff.get("traffic_limit") * 1024**3) else: traffic_limit_bytes = 0 - hwid_limit = tariff.device_limit - subgroup_title = tariff.subgroup_title + + if current_device_limit_from_key is not None: + hwid_limit = int(current_device_limit_from_key) + elif selected_device_limit_from_key is not None: + hwid_limit = int(selected_device_limit_from_key) + else: + hwid_limit = tariff.get("device_limit") + + subgroup_title = tariff.get("subgroup_title") + + expire_iso = ( + datetime.utcfromtimestamp(key["expiry_time"] / 1000).replace(tzinfo=timezone.utc).isoformat() + ) + + if use_country_selection: + user_server = None + for s in cluster_servers: + if s.get("server_name") == key["server_id"]: + user_server = s + break + + if not user_server: + return {"key": key, "success": False, "error": "Server not found"} + + remna = RemnawaveAPI(user_server["api_url"]) + inbound_ids = [user_server["inbound_id"]] if user_server.get("inbound_id") else [] else: - logger.warning( - f"[Sync] Ключ {key['client_id']} с несуществующим тарифом ID={key['tariff_id']} — " - f"обновим без лимитов" - ) + remna = RemnawaveAPI(cluster_servers[0]["api_url"]) - if use_country_selection: - user_server = None - for s in cluster_servers: - if s.get("server_name") == key["server_id"]: - user_server = s - break - - if not user_server: - logger.warning( - f"[Sync] Сервер {key['server_id']} не найден в кластере {cluster_name}, пропускаем ключ" - ) - continue - - remna = RemnawaveAPI(user_server["api_url"]) - inbound_ids = [user_server["inbound_id"]] if user_server.get("inbound_id") else [] - else: - remna = RemnawaveAPI(cluster_servers[0]["api_url"]) - - filtered_servers = cluster_servers - if subgroup_title: - filtered_servers = [ - s for s in cluster_servers if subgroup_title in s.get("tariff_subgroups", []) - ] - if not filtered_servers: - logger.warning( - f"[Sync] В кластере {cluster_name} не найдено серверов для подгруппы " - f"'{subgroup_title}'. Использую весь кластер." - ) - filtered_servers = cluster_servers - - if tariff and tariff.group_code: - group_code = tariff.group_code.lower() - if group_code in ALLOWED_GROUP_CODES: - special_filtered = [ - s for s in filtered_servers if group_code in (s.get("special_groups") or []) + filtered_servers = cluster_servers + if subgroup_title: + filtered_servers = [ + s for s in cluster_servers if subgroup_title in s.get("tariff_subgroups", []) ] - if special_filtered: - filtered_servers = special_filtered - else: - logger.warning( - f"[Sync] В кластере {cluster_name} нет серверов со спецгруппой " - f"'{group_code}'. Использую весь кластер." - ) + if not filtered_servers: + filtered_servers = cluster_servers - inbound_ids = [s["inbound_id"] for s in filtered_servers if s.get("inbound_id")] + if tariff and tariff.get("group_code"): + group_code = tariff.get("group_code").lower() + if group_code in ALLOWED_GROUP_CODES: + special_filtered = [ + s for s in filtered_servers if group_code in (s.get("special_groups") or []) + ] + if special_filtered: + filtered_servers = special_filtered - if not await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD): - raise Exception("Не удалось авторизоваться в Remnawave") + inbound_ids = [s["inbound_id"] for s in filtered_servers if s.get("inbound_id")] - success = await remna.update_user( - uuid=key["client_id"], - expire_at=expire_iso, - telegram_id=key["tg_id"], - email=f"{key['email']}@fake.local", - active_user_inbounds=inbound_ids, - traffic_limit_bytes=traffic_limit_bytes, - hwid_device_limit=hwid_limit, - ) + if not await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD): + return {"key": key, "success": False, "error": "Login failed"} - if success: - try: + success = await remna.update_user( + uuid=key["client_id"], + expire_at=expire_iso, + telegram_id=key["tg_id"], + email=f"{key['email']}@fake.local", + active_user_inbounds=inbound_ids, + traffic_limit_bytes=traffic_limit_bytes, + hwid_device_limit=hwid_limit, + ) + + if success: sub = await remna.get_subscription_by_username(key["email"]) - if sub: - new_remnawave_link = sub.get("subscriptionUrl") + new_link = sub.get("subscriptionUrl") if sub else None + return { + "key": key, + "success": True, + "new_link": new_link, + "tariff": tariff, + "traffic_limit_bytes": traffic_limit_bytes, + "hwid_limit": hwid_limit, + } + else: + return {"key": key, "success": False, "needs_recreate": True, "tariff": tariff, "traffic_limit_bytes": traffic_limit_bytes, "hwid_limit": hwid_limit} - if new_remnawave_link: - servers = await get_servers(session) - cluster_servers = servers.get(cluster_name, []) + except Exception as e: + logger.error(f"[Sync] Ошибка API для {key.get('email')}: {e}") + return {"key": key, "success": False, "error": str(e)} - key_value = await make_aggregated_link( - session=session, - cluster_all=cluster_servers, - cluster_id=cluster_name, - email=key["email"], - client_id=key["client_id"], - tg_id=key["tg_id"], - remna_link_override=None, - plan=key["tariff_id"], - ) + tasks = [update_remnawave_api(key) for key in batch] + results = await asyncio.gather(*tasks, return_exceptions=True) - await session.execute( - update(Key) - .where(Key.tg_id == key["tg_id"], Key.client_id == key["client_id"]) - .values(remnawave_link=new_remnawave_link, key=key_value) - ) - await session.commit() - logger.info(f"[Sync] Обновлена ссылка для {key['email']}: {new_remnawave_link}") - except Exception as e: - logger.warning(f"[Sync] Не удалось получить ссылку для {key['email']}: {e}") + bulk_updates = [] + recreate_tasks = [] + + for result in results: + if isinstance(result, Exception): + logger.error(f"[Sync] Exception в батче: {result}") + continue + + if not isinstance(result, dict): + continue + + key = result.get("key") + if not key: + continue + + try: + if result.get("success") and result.get("new_link"): + new_link = result["new_link"] + tariff = result.get("tariff") + + key_value = await make_aggregated_link( + session=session, + cluster_all=cluster_servers, + cluster_id=cluster_name, + email=key["email"], + client_id=key["client_id"], + tg_id=key["tg_id"], + remna_link_override=None, + plan=tariff, + ) - if not success: - logger.warning("[Sync] ошибка обновления, пробуем пересоздать") + bulk_updates.append({ + "client_id": key["client_id"], + "remnawave_link": new_link, + "key": key_value, + }) + + elif result.get("needs_recreate"): + recreate_tasks.append((key, result)) + + except Exception as e: + logger.error(f"[Sync] Ошибка подготовки для {key.get('email')}: {e}") + if bulk_updates: + try: + await session.run_sync( + lambda sync_session: sync_session.bulk_update_mappings(Key, bulk_updates) + ) + await session.commit() + logger.info(f"[Sync] Bulk: обновлено {len(bulk_updates)} ключей") + except Exception as bulk_error: + logger.warning(f"[Sync] Bulk упал, fallback: {bulk_error}") + await session.rollback() + + for upd in bulk_updates: + try: + await session.execute( + update(Key) + .where(Key.client_id == upd["client_id"]) + .values(remnawave_link=upd["remnawave_link"], key=upd["key"]) + ) + await session.commit() + except Exception as e: + logger.error(f"[Sync] Fallback ошибка {upd['client_id']}: {e}") + await session.rollback() + + for key, result in recreate_tasks: + try: + logger.warning(f"[Sync] Пересоздание {key['email']}") await delete_key_from_cluster(cluster_name, key["email"], key["client_id"], session) - await session.execute( delete(Key).where(Key.tg_id == key["tg_id"], Key.client_id == key["client_id"]) ) - + await session.commit() + cluster_id_for_recreate = key["server_id"] if use_country_selection else cluster_name await create_key_on_cluster( cluster_id_for_recreate, @@ -588,11 +693,70 @@ async def handle_sync_cluster( plan=key["tariff_id"], session=session, remnawave_link=key["remnawave_link"], + hwid_limit=result.get("hwid_limit"), + traffic_limit_bytes=result.get("traffic_limit_bytes"), + selected_device_limit=key.get("selected_device_limit"), + selected_traffic_limit_gb=key.get("selected_traffic_limit"), + current_device_limit=key.get("current_device_limit"), + current_traffic_limit_gb=key.get("current_traffic_limit"), + selected_price_rub=key.get("selected_price_rub"), + ) + except Exception as e: + logger.error(f"[Sync] Пересоздание ошибка {key.get('email')}: {e}") + + processed_count = batch_end + progress_percent = int((processed_count / total_keys) * 100) + progress_bar = "█" * (progress_percent // 5) + "░" * (20 - progress_percent // 5) + + try: + await callback_query.message.edit_text( + text=( + f"🔄 Синхронизация кластера {cluster_name}\n\n" + f"🔑 Количество ключей: {total_keys}\n\n" + f"Обработано: {processed_count}/{total_keys}\n" + f"{progress_bar}" + ) + ) + except Exception: + pass + + else: + for key in keys_to_sync: + try: + traffic_limit_bytes = 0 + hwid_limit = 0 + subgroup_title = None + tariff = tariffs_cache.get(key["tariff_id"]) if key["tariff_id"] else None + + current_device_limit_from_key = key.get("current_device_limit") + current_traffic_limit_gb_from_key = key.get("current_traffic_limit") + selected_device_limit_from_key = key.get("selected_device_limit") + selected_traffic_limit_gb_from_key = key.get("selected_traffic_limit") + + if tariff: + if current_traffic_limit_gb_from_key is not None: + traffic_limit_bytes = int(current_traffic_limit_gb_from_key * 1024**3) + elif selected_traffic_limit_gb_from_key is not None: + traffic_limit_bytes = int(selected_traffic_limit_gb_from_key * 1024**3) + elif tariff.get("traffic_limit") is not None: + traffic_limit_bytes = int(tariff.get("traffic_limit") * 1024**3) + else: + traffic_limit_bytes = 0 + + if current_device_limit_from_key is not None: + hwid_limit = int(current_device_limit_from_key) + elif selected_device_limit_from_key is not None: + hwid_limit = int(selected_device_limit_from_key) + else: + hwid_limit = tariff.get("device_limit") + + subgroup_title = tariff.get("subgroup_title") + elif key["tariff_id"]: + logger.warning( + f"[Sync] Ключ {key['client_id']} с несуществующим тарифом ID={key['tariff_id']} — " + f"обновим без лимитов" ) - await asyncio.sleep(0.1) - - else: await delete_key_from_cluster(cluster_name, key["email"], key["client_id"], session) await session.execute( @@ -609,15 +773,26 @@ async def handle_sync_cluster( plan=key["tariff_id"], session=session, remnawave_link=key["remnawave_link"], + hwid_limit=hwid_limit, + traffic_limit_bytes=traffic_limit_bytes, + selected_device_limit=key.get("selected_device_limit"), + selected_traffic_limit_gb=key.get("selected_traffic_limit"), + current_device_limit=key.get("current_device_limit"), + current_traffic_limit_gb=key.get("current_traffic_limit"), + selected_price_rub=key.get("selected_price_rub"), ) await asyncio.sleep(0.5) - except Exception as e: - logger.error(f"[Sync] Ошибка при обработке ключа {key['client_id']} в {cluster_name}: {e}") + except Exception as e: + logger.error(f"[Sync] Ошибка при обработке ключа {key['client_id']} в {cluster_name}: {e}") await callback_query.message.edit_text( - text=f"✅ Ключи успешно синхронизированы для кластера {cluster_name}", + text=( + f"✅ Синхронизация завершена\n\n" + f"📊 Кластер: {cluster_name}\n" + f"🔑 Обработано ключей: {len(keys_to_sync)}" + ), reply_markup=build_admin_back_kb("clusters"), ) diff --git a/handlers/keys/operations/creation.py b/handlers/keys/operations/creation.py index fdc0e921..7afe83f5 100644 --- a/handlers/keys/operations/creation.py +++ b/handlers/keys/operations/creation.py @@ -33,6 +33,11 @@ async def create_key_on_cluster( hwid_limit: int = None, traffic_limit_bytes: int = None, is_trial: bool = False, + selected_device_limit: int = None, + selected_traffic_limit_gb: int = None, + current_device_limit: int = None, + current_traffic_limit_gb: int = None, + selected_price_rub: int = None, ): try: servers = await get_servers(session) @@ -271,6 +276,11 @@ async def create_key_on_cluster( server_id=server_id_to_store, remnawave_link=remnawave_link_value if remnawave_created else None, tariff_id=plan, + selected_device_limit=selected_device_limit, + selected_traffic_limit=selected_traffic_limit_gb, + current_device_limit=current_device_limit, + current_traffic_limit=current_traffic_limit_gb, + selected_price_rub=selected_price_rub, ) await session.execute(update(User).where(User.tg_id == tg_id, User.trial.in_([0, -1])).values(trial=1)) await session.commit()