houseassassin/Solo_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refact sync

Browse Source
This commit is contained in:
Capybara-z
2025-12-21 16:36:18 +03:00
parent 39f91d0d19
commit 73a1ea8e1b
2 changed files with 330 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files
handlers/admin/clusters/cluster_sync.py
+320 -145
View File
@@ -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"<b>🔄 Синхронизация сервера {server_name}</b>\n\n🔑 Количество ключей: <b>{len(keys_to_sync)}</b>"
)
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"<b>🔄 Синхронизация кластера {cluster_name}</b>\n\n🔑 Количество ключей: <b>{len(keys_to_sync)}</b>"
)
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"<b>🔄 Синхронизация кластера {cluster_name}</b>\n\n"
f"🔑 Количество ключей: <b>{total_keys}</b>\n\n"
f"Обработано: <b>{processed_count}/{total_keys}</b>\n"
f"<code>{progress_bar}</code>"
)
)
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"✅ <b>Синхронизация завершена</b>\n\n"
f"📊 Кластер: <b>{cluster_name}</b>\n"
f"🔑 Обработано ключей: <b>{len(keys_to_sync)}</b>"
),
reply_markup=build_admin_back_kb("clusters"),
)
handlers/keys/operations/creation.py
+10
View File
@@ -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()