syncClaster/picGifts/mailingCluster/renewButtin and more

client.py

@@ -41,8 +41,15 @@ async def add_client(
         return response if response else {"status": "failed"}
 
     except Exception as e:
-        logger.error(f"Ошибка при добавлении клиента {email}: {e}")
-        return {"status": "failed", "error": str(e)}
+        error_message = str(e)
+
+        if "Duplicate email" in error_message:
+            logger.warning(f"Дублированный email: {email}. Пропуск. Сообщение: {error_message}")
+            return {"status": "duplicate", "email": email}
+
+        logger.error(f"Ошибка при добавлении клиента {email}: {error_message}")
+        return {"status": "failed", "error": error_message}
+
 
 
 async def extend_client_key(

handlers/admin/admin_panel.py

@@ -14,9 +14,9 @@ from aiogram.utils.keyboard import InlineKeyboardBuilder
 from backup import backup_database
 from bot import bot
 from config import DATABASE_URL
+from database import delete_user_data
 from filters.admin import IsAdminFilter
 from logger import logger
-from database import delete_user_data
 
 router = Router()
 
@@ -37,7 +37,7 @@ async def handle_admin_callback_query(callback_query: CallbackQuery, state: FSMC
 async def handle_admin_message(message: types.Message, state: FSMContext):
     await state.clear()
 
-    BOT_VERSION = "4.0.0-preAlpha"  # Укажите текущую версию бота
+    BOT_VERSION = "4.0.0-preAlpha(9)" 
 
     builder = InlineKeyboardBuilder()
     builder.row(
@@ -274,6 +274,7 @@ async def handle_send_to_all(callback_query: CallbackQuery, state: FSMContext):
     builder.row(InlineKeyboardButton(text="📢 Отправить всем", callback_data="send_to_all"))
     builder.row(InlineKeyboardButton(text="📢 Отправить с подпиской", callback_data="send_to_subscribed"))
     builder.row(InlineKeyboardButton(text="📢 Отправить без подписки", callback_data="send_to_unsubscribed"))
+    builder.row(InlineKeyboardButton(text="📢 Рассылка по кластеру", callback_data="send_to_cluster"))
     builder.row(InlineKeyboardButton(text="🔙 Назад", callback_data="admin"))
     await callback_query.message.answer(
         "✍️ Выберите группу пользователей и введите текст сообщения для рассылки:",
@@ -304,6 +305,35 @@ async def handle_send_to_unsubscribed(callback_query: CallbackQuery, state: FSMC
     )
     await state.set_state(UserEditorState.waiting_for_message)
 
+@router.callback_query(F.data == "send_to_cluster", IsAdminFilter())
+async def handle_send_to_cluster(callback_query: CallbackQuery, state: FSMContext, session: Any):
+    clusters = await session.fetch("SELECT DISTINCT cluster_name FROM servers")
+
+    builder = InlineKeyboardBuilder()
+    for cluster in clusters:
+        builder.row(
+            InlineKeyboardButton(
+                text=f"🌐 {cluster['cluster_name']}",
+                callback_data=f"send_cluster|{cluster['cluster_name']}"
+            )
+        )
+
+    builder.row(InlineKeyboardButton(text="🔙 Назад", callback_data="send_to"))
+    await callback_query.message.answer(
+        "✍️ Выберите кластер для рассылки сообщений:",
+        reply_markup=builder.as_markup(),
+    )
+
+
+@router.callback_query(F.data.startswith("send_cluster|"), IsAdminFilter())
+async def handle_send_cluster(callback_query: CallbackQuery, state: FSMContext):
+    cluster_name = callback_query.data.split("|")[1]
+    await state.update_data(send_to="cluster", cluster_name=cluster_name)
+    await callback_query.message.answer(
+        f"✍️ Введите текст сообщения для рассылки пользователям кластера <b>{cluster_name}</b>:"
+    )
+    await state.set_state(UserEditorState.waiting_for_message)
+
 
 @router.message(UserEditorState.waiting_for_message, IsAdminFilter())
 async def process_message_to_all(
@@ -332,6 +362,15 @@ async def process_message_to_all(
                 GROUP BY c.tg_id
                 HAVING COUNT(k.tg_id) = 0 OR MAX(k.expiry_time) <= $1
             """, int(datetime.utcnow().timestamp() * 1000))
+        elif send_to == 'cluster':
+            cluster_name = state_data.get('cluster_name')
+            tg_ids = await session.fetch("""
+                SELECT DISTINCT c.tg_id
+                FROM connections c
+                JOIN keys k ON c.tg_id = k.tg_id
+                JOIN servers s ON k.server_id = s.cluster_name
+                WHERE s.cluster_name = $1
+            """, cluster_name)
 
         total_users = len(tg_ids)
         success_count = 0

handlers/admin/admin_servers.py

@@ -1,3 +1,5 @@
+import asyncio
+
 import asyncpg
 from aiogram import F, Router, types
 from aiogram.fsm.context import FSMContext
@@ -10,6 +12,8 @@ from backup import create_backup_and_send_to_admins
 from config import ADMIN_PASSWORD, ADMIN_USERNAME, DATABASE_URL
 from database import check_unique_server_name, get_servers_from_db
 from filters.admin import IsAdminFilter
+from handlers.keys.key_utils import create_key_on_cluster
+from logger import logger
 
 router = Router()
 
@@ -328,6 +332,13 @@ async def handle_manage_cluster(callback_query: types.CallbackQuery, state: FSMC
         )
     )
 
+    builder.row(
+        InlineKeyboardButton(
+            text="🔄 Синхронизировать",
+            callback_data=f"sync_cluster|{cluster_name}",
+        )
+    )
+
     builder.row(
         InlineKeyboardButton(
             text="🔙 Назад в управление кластерами", callback_data="servers_editor"
@@ -340,6 +351,64 @@ async def handle_manage_cluster(callback_query: types.CallbackQuery, state: FSMC
     )
 
 
+@router.callback_query(F.data.startswith("sync_cluster|"), IsAdminFilter())
+async def sync_cluster_handler(callback_query: types.CallbackQuery):
+    """Обработчик для синхронизации ключей на всех серверах выбранного кластера."""
+    cluster_name = callback_query.data.split("|")[1]
+
+    conn = await asyncpg.connect(DATABASE_URL)
+    try:
+        query_keys = """
+            SELECT tg_id, client_id, email, expiry_time
+            FROM keys
+            WHERE server_id = $1
+        """
+        keys_to_sync = await conn.fetch(query_keys, cluster_name)
+
+        if not keys_to_sync:
+            await callback_query.message.answer(
+                f"❌ Нет ключей для синхронизации в кластере {cluster_name}.",
+                reply_markup=InlineKeyboardBuilder()
+                .row(InlineKeyboardButton(text="🔙 Назад", callback_data="servers_editor"))
+                .as_markup(),
+            )
+            return
+
+        tasks = []
+
+        for key in keys_to_sync:
+            tasks.append(
+                asyncio.create_task(
+                    create_key_on_cluster(
+                        cluster_name,
+                        key["tg_id"],
+                        key["client_id"],
+                        key["email"],
+                        key["expiry_time"],
+                    )
+                )
+            )
+
+        await asyncio.gather(*tasks)
+        await callback_query.message.answer(
+            f"✅ Ключи успешно синхронизированы для кластера {cluster_name}.",
+            reply_markup=InlineKeyboardBuilder()
+            .row(InlineKeyboardButton(text="🔙 Назад", callback_data="servers_editor"))
+            .as_markup(),
+        )
+    except Exception as e:
+        logger.error(f"Ошибка синхронизации ключей в кластере {cluster_name}: {e}")
+        await callback_query.message.answer(
+            f"❌ Произошла ошибка при синхронизации: {e}",
+            reply_markup=InlineKeyboardBuilder()
+            .row(InlineKeyboardButton(text="🔙 Назад", callback_data="servers_editor"))
+            .as_markup(),
+        )
+    finally:
+        await conn.close()
+
+
+
 @router.callback_query(F.data.startswith("server_availability|"), IsAdminFilter())
 async def handle_check_server_availability(callback_query: types.CallbackQuery):
     cluster_name = callback_query.data.split("|")[1]

handlers/keys/key_utils.py

@@ -1,15 +1,24 @@
 import asyncio
 
-import asyncpg
 from py3xui import AsyncApi
 
 from client import add_client, delete_client, extend_client_key
-from config import ADMIN_PASSWORD, ADMIN_USERNAME, DATABASE_URL, LIMIT_IP, TOTAL_GB
+from config import ADMIN_PASSWORD, ADMIN_USERNAME, LIMIT_IP, TOTAL_GB
 from database import get_servers_from_db
 from logger import logger
 
 
 async def create_key_on_cluster(cluster_id, tg_id, client_id, email, expiry_timestamp):
+    """
+    Создает ключ на всех серверах указанного кластера.
+    
+    :param cluster_id: ID кластера.
+    :param tg_id: Telegram ID пользователя.
+    :param client_id: Уникальный идентификатор клиента.
+    :param email: Email клиента.
+    :param expiry_timestamp: Время истечения ключа (timestamp в миллисекундах).
+    :param allow_existing: Игнорируется, ключи всегда продолжают выполнение.
+    """
     try:
         tasks = []
         servers = await get_servers_from_db()
@@ -32,14 +41,6 @@ async def create_key_on_cluster(cluster_id, tg_id, client_id, email, expiry_time
                 )
                 continue
 
-            conn = await asyncpg.connect(DATABASE_URL)
-            existing_key = await conn.fetchrow(
-                "SELECT 1 FROM keys WHERE email = $1", email
-            )
-
-            if existing_key:
-                raise ValueError(f"Email {email} уже существует в базе данных.")
-
             tasks.append(
                 add_client(
                     xui,
@@ -54,7 +55,6 @@ async def create_key_on_cluster(cluster_id, tg_id, client_id, email, expiry_time
                     inbound_id=int(inbound_id),
                 )
             )
-            await conn.close()
 
         await asyncio.gather(*tasks)
 

handlers/keys/keys.py

@@ -13,6 +13,7 @@ from config import (
     CONNECT_IOS,
     DOWNLOAD_ANDROID,
     DOWNLOAD_IOS,
+    ENABLE_UPDATE_SUBSCRIPTION_BUTTON,
     PUBLIC_LINK,
     RENEWAL_PLANS,
     TOTAL_GB,
@@ -166,13 +167,18 @@ async def process_callback_view_key(callback_query: types.CallbackQuery, session
 
             if time_left.total_seconds() <= 0:
                 days_left_message = (
-                    "<b>🕒 Статус подписки:</b>\n🔴 Истекла\nОсталось часов: 0"
+                    "<b>🕒 Статус подписки:</b>\n🔴 Истекла\nОсталось часов: 0\nОсталось минут: 0"
                 )
-            elif time_left.days > 0:
-                days_left_message = f"Осталось дней: <b>{time_left.days}</b>"
             else:
-                hours_left = time_left.seconds // 3600
-                days_left_message = f"Осталось часов: <b>{hours_left}</b>"
+                total_seconds = int(time_left.total_seconds())
+                days = total_seconds // 86400
+                hours = (total_seconds % 86400) // 3600
+                minutes = (total_seconds % 3600) // 60
+
+                days_left_message = (
+                    f"<b>🕒 Статус подписки:</b>\n"
+                    f"Осталось: <b>{days}</b> дней, <b>{hours}</b> часов, <b>{minutes}</b> минут"
+                )
 
             formatted_expiry_date = expiry_date.strftime("%d %B %Y года")
             response_message = key_message(
@@ -181,10 +187,11 @@ async def process_callback_view_key(callback_query: types.CallbackQuery, session
 
             builder = InlineKeyboardBuilder()
 
-            builder.row(
+            if not key.startswith(PUBLIC_LINK) or ENABLE_UPDATE_SUBSCRIPTION_BUTTON:
+                builder.row(
                     InlineKeyboardButton(
                         text="🔄 Обновить подписку",
-                        callback_data=f"update_subscription|{key_name}",
+                        callback_data=f"update_subscription|{key_name}"
                     )
                 )