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()