Пиздец очередной

app/services/db_manager.py

@@ -0,0 +1,211 @@
+from instance.model import User, Subscription, Transaction, Administrators
+from sqlalchemy.future import select
+from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy import desc
+from dateutil.relativedelta import relativedelta
+from datetime import datetime
+from .xui_rep import PanelInteraction
+from .mongo_rep import MongoDBRepository
+import random
+import string
+import logging
+import asyncio
+
+class DatabaseManager:
+    def __init__(self, session_generator):
+        """
+        Инициализация с асинхронным генератором сессий (например, get_postgres_session).
+        """
+        self.session_generator = session_generator
+        self.logger = logging.getLogger(__name__)
+        self.mongo_repo = MongoDBRepository()
+
+    async def create_user(self, telegram_id: int):
+        """
+        Создаёт нового пользователя, если его нет.
+        """
+        async for session in self.session_generator():
+            try:
+                username = self.generate_string(6)
+                result = await session.execute(select(User).where(User.telegram_id == int(telegram_id)))
+                user = result.scalars().first()
+                if not user:
+                    new_user = User(telegram_id=int(telegram_id), username=username)
+                    session.add(new_user)
+                    await session.commit()
+                    return new_user
+                return user
+            except SQLAlchemyError as e:
+                self.logger.error(f"Ошибка при создании пользователя {telegram_id}: {e}")
+                await session.rollback()
+                return "ERROR"
+
+    async def get_user_by_telegram_id(self, telegram_id: int):
+        """
+        Возвращает пользователя по Telegram ID.
+        """
+        async for session in self.session_generator():
+            try:
+                result = await session.execute(select(User).where(User.telegram_id == telegram_id))
+                return result.scalars().first()
+            except SQLAlchemyError as e:
+                self.logger.error(f"Ошибка при получении пользователя {telegram_id}: {e}")
+                return None
+
+    async def add_transaction(self, user_id: int, amount: float):
+        """
+        Добавляет транзакцию для пользователя.
+        """
+        async for session in self.session_generator():
+            try:
+                transaction = Transaction(user_id=user_id, amount=amount)
+                session.add(transaction)
+                await session.commit()
+            except SQLAlchemyError as e:
+                self.logger.error(f"Ошибка добавления транзакции для пользователя {user_id}: {e}")
+                await session.rollback()
+
+    async def update_balance(self, telegram_id: int, amount: float):
+        """
+        Обновляет баланс пользователя и добавляет транзакцию.
+        """
+        async for session in self.session_generator():
+            try:
+                result = await session.execute(select(User).where(User.telegram_id == telegram_id))
+                user = result.scalars().first()
+                if user:
+                    user.balance += int(amount)
+                    await self.add_transaction(user.id, amount)
+                    await session.commit()
+                else:
+                    self.logger.warning(f"Пользователь с Telegram ID {telegram_id} не найден.")
+                    return "ERROR"
+            except SQLAlchemyError as e:
+                self.logger.error(f"Ошибка при обновлении баланса: {e}")
+                await session.rollback()
+                return "ERROR"
+
+    async def last_subscription(self, user_id: int):
+        """
+        Возвращает список подписок пользователя.
+        """
+        async for session in self.session_generator():
+            try:
+                result = await session.execute(
+                    select(Subscription)
+                    .where(Subscription.user_id == user_id)
+                    .order_by(desc(Subscription.created_at))
+                )
+                return result.scalars().all()
+            except SQLAlchemyError as e:
+                self.logger.error(f"Ошибка при получении последней подписки пользователя {user_id}: {e}")
+                return "ERROR"
+
+    async def last_transaction(self, user_id: int):
+        """
+        Возвращает список транзакций пользователя.
+        """
+        async for session in self.session_generator():
+            try:
+                result = await session.execute(
+                    select(Transaction)
+                    .where(Transaction.user_id == user_id)
+                    .order_by(desc(Transaction.created_at))
+                )
+                transactions = result.scalars().all()
+                return transactions
+            except SQLAlchemyError as e:
+                self.logger.error(f"Ошибка при получении транзакций пользователя {user_id}: {e}")
+                return "ERROR"
+
+    async def buy_sub(self, telegram_id: str, plan_id: str):
+        async for session in self.session_generator():
+            try:
+                result = await self.create_user(telegram_id)
+                if not result:
+                    self.logger.error(f"Пользователь с Telegram ID {telegram_id} не найден.")
+                    return "ERROR"
+
+                # Получение тарифного плана из MongoDB
+                plan = await self.mongo_repo.get_subscription_plan(plan_id)
+                if not plan:
+                    self.logger.error(f"Тарифный план {plan_id} не найден.")
+                    return "ERROR"
+
+                # Проверка достаточности средств
+                cost = int(plan["price"])
+                if result.balance < cost:
+                    self.logger.error(f"Недостаточно средств у пользователя {telegram_id} для покупки плана {plan_id}.")
+                    return "INSUFFICIENT_FUNDS"
+
+                # Списываем средства
+                result.balance -= cost
+
+                # Создаем подписку
+                expiry_date = datetime.utcnow() + relativedelta(months=plan["duration_months"])
+                server = await self.mongo_repo.get_server_with_least_clients()
+                self.logger.info(f"Выбран сервер для подписки: {server}")
+                new_subscription = Subscription(
+                    user_id=result.id,
+                    vpn_server_id=str(server['server']["name"]),
+                    plan=plan_id,
+                    expiry_date=expiry_date
+                )
+                session.add(new_subscription)
+
+                # Попытка добавить пользователя на сервер
+                # Получаем информацию о пользователе
+                user = result  # так как result уже содержит пользователя
+                if not user:
+                    self.logger.error(f"Не удалось найти пользователя для добавления на сервер.")
+                    await session.rollback()
+                    return "ERROR"
+
+                # Получаем сервер из MongoDB
+                server_data = await self.mongo_repo.get_server(new_subscription.vpn_server_id)
+                if not server_data:
+                    self.logger.error(f"Не удалось найти сервер с ID {new_subscription.vpn_server_id}.")
+                    await session.rollback()
+                    return "ERROR"
+
+                server_info = server_data['server']
+                url_base = f"https://{server_info['ip']}:{server_info['port']}/{server_info['secretKey']}"
+                login_data = {
+                    'username': server_info['login'],
+                    'password': server_info['password'],
+                }
+
+                panel = PanelInteraction(url_base, login_data, self.logger,server_info['certificate']['data'])
+                expiry_date_iso = new_subscription.expiry_date.isoformat()
+
+                # Добавляем на сервер
+                response = await panel.add_client(user.id, expiry_date_iso, user.username)
+
+                if response != "OK":
+                    self.logger.error(f"Ошибка при добавлении клиента {telegram_id} на сервер: {response}")
+                    # Если не получилось добавить на сервер, откатываем транзакцию
+                    await session.rollback()
+                    return "ERROR"
+
+                # Если мы здесь - значит и подписка, и добавление на сервер успешны
+                await session.commit()
+                self.logger.info(f"Подписка успешно оформлена для пользователя {telegram_id} на план {plan_id} и клиент добавлен на сервер.")
+                return "OK"
+
+            except SQLAlchemyError as e:
+                self.logger.error(f"Ошибка при покупке подписки {plan_id} для пользователя {telegram_id}: {e}")
+                await session.rollback()
+                return "ERROR"
+            except Exception as e:
+                self.logger.error(f"Непредвиденная ошибка: {e}")
+                await session.rollback()
+                return "ERROR"
+
+
+    @staticmethod
+    def generate_string(length):
+        """
+        Генерирует случайную строку заданной длины.
+        """
+        characters = string.ascii_lowercase + string.digits
+        return ''.join(random.choices(characters, k=length))

app/services/mongo_rep.py

@@ -0,0 +1,158 @@
+import os
+from motor.motor_asyncio import AsyncIOMotorClient
+from pymongo.errors import DuplicateKeyError, NetworkTimeout
+import logging
+
+
+class MongoDBRepository:
+    def __init__(self):
+        # Настройки MongoDB из переменных окружения
+        mongo_uri = os.getenv("MONGO_URL")
+        database_name = os.getenv("DB_NAME")
+        server_collection = os.getenv("SERVER_COLLECTION", "servers")
+        plan_collection = os.getenv("PLAN_COLLECTION", "plans")
+
+        # Подключение к базе данных и коллекциям
+        self.client = AsyncIOMotorClient(mongo_uri)
+        self.db = self.client[database_name]
+        self.collection = self.db[server_collection]  # Коллекция серверов
+        self.plans_collection = self.db[plan_collection]  # Коллекция планов
+        self.logger = logging.getLogger(__name__)
+
+    async def add_subscription_plan(self, plan_data):
+        """Добавляет новый тарифный план в коллекцию."""
+        try:
+            result = await self.plans_collection.insert_one(plan_data)
+            self.logger.debug(f"Тарифный план добавлен с ID: {result.inserted_id}")
+            return result.inserted_id
+        except DuplicateKeyError:
+            self.logger.error("Дублирующий ключ.")
+        except NetworkTimeout:
+            self.logger.error("Сетевой таймаут.")
+
+    async def get_subscription_plan(self, plan_name):
+        """Получает тарифный план по его имени."""
+        try:
+            plan = await self.plans_collection.find_one({"name": plan_name})
+            if plan:
+                self.logger.debug(f"Найден тарифный план: {plan}")
+            else:
+                self.logger.error(f"Тарифный план {plan_name} не найден.")
+            return plan
+        except DuplicateKeyError:
+            self.logger.error("Дублирующий ключ.")
+        except NetworkTimeout:
+            self.logger.error("Сетевой таймаут.")
+
+    async def add_server(self, server_data):
+        """Добавляет новый VPN сервер в коллекцию."""
+        try:
+            result = await self.collection.insert_one(server_data)
+            self.logger.debug(f"VPN сервер добавлен с ID: {result.inserted_id}")
+            return result.inserted_id
+        except DuplicateKeyError:
+            self.logger.error("Дублирующий ключ.")
+        except NetworkTimeout:
+            self.logger.error("Сетевой таймаут.")
+
+    async def get_server(self, server_name: str):
+        """Получает сервер VPN по его ID."""
+        try:
+            server = await self.collection.find_one({"server.name": server_name})
+            if server:
+                self.logger.debug(f"Найден VPN сервер: {server}")
+            else:
+                self.logger.debug(f"VPN сервер с ID {server_name} не найден.")
+            return server
+        except DuplicateKeyError:
+            self.logger.error("Дублирующий ключ.")
+        except NetworkTimeout:
+            self.logger.error("Сетевой таймаут.")
+            
+    async def get_server_with_least_clients(self):
+        """Возвращает сервер с наименьшим количеством подключенных клиентов."""
+        try:
+            pipeline = [
+                {
+                    "$addFields": {
+                        "current_clients": {"$size": {"$ifNull": ["$clients", []]}}
+                    }
+                },
+                {
+                    "$sort": {"current_clients": 1}
+                },
+                {
+                    "$limit": 1
+                }
+            ]
+
+            result = await self.collection.aggregate(pipeline).to_list(length=1)
+            if result:
+                server = result[0]
+                self.logger.debug(f"Найден сервер с наименьшим количеством клиентов: {server}")
+                return server
+            else:
+                self.logger.debug("Не найдено серверов.")
+                return None
+        except DuplicateKeyError:
+            self.logger.error("Дублирующий ключ.")
+        except NetworkTimeout:
+            self.logger.error("Сетевой таймаут.")
+
+    async def update_server(self, server_name, update_data):
+        """Обновляет данные VPN сервера."""
+        try:
+            result = await self.collection.update_one({"server_name": server_name}, {"$set": update_data})
+            if result.matched_count > 0:
+                self.logger.debug(f"VPN сервер с ID {server_name} обновлен.")
+            else:
+                self.logger.debug(f"VPN сервер с ID {server_name} не найден.")
+            return result.matched_count > 0
+        except DuplicateKeyError:
+            self.logger.error("Дублирующий ключ.")
+        except NetworkTimeout:
+            self.logger.error("Сетевой таймаут.")
+
+    async def delete_server(self, server_name):
+        """Удаляет VPN сервер по его ID."""
+        try:
+            result = await self.collection.delete_one({"name": server_name})
+            if result.deleted_count > 0:
+                self.logger.debug(f"VPN сервер с ID {server_name} удален.")
+            else:
+                self.logger.debug(f"VPN сервер с ID {server_name} не найден.")
+            return result.deleted_count > 0
+        except DuplicateKeyError:
+            self.logger.error("Дублирующий ключ.")
+        except NetworkTimeout:
+            self.logger.error("Сетевой таймаут.")
+
+    async def list_servers(self):
+        """Возвращает список всех VPN серверов."""
+        try:
+            servers = await self.collection.find().to_list(length=1000)  # Получить до 1000 серверов (можно настроить)
+            self.logger.debug(f"Найдено {len(servers)} VPN серверов.")
+            return servers
+        except DuplicateKeyError:
+            self.logger.error("Дублирующий ключ.")
+        except NetworkTimeout:
+            self.logger.error("Сетевой таймаут.")
+
+
+    async def __aenter__(self):
+        """
+        Метод вызывается при входе в блок with.
+        """
+        self.logger.debug("Контекстный менеджер: подключение открыто.")
+        return self  
+
+    async def __aexit__(self, exc_type, exc_value, traceback):
+        """
+        Метод вызывается при выходе из блока with.
+        """
+        await self.close_connection() 
+        if exc_type:
+            self.logger.error(f"Контекстный менеджер завершён с ошибкой: {exc_value}")
+        else:
+            self.logger.debug("Контекстный менеджер: подключение закрыто.")
+

app/services/postgres_rep.py

@@ -0,0 +1,102 @@
+from sqlalchemy.future import select
+from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy import desc
+from instance.model import User, Subscription, Transaction
+
+
+class PostgresRepository:
+    def __init__(self, session_generator, logger):
+        self.session_generator = session_generator
+        self.logger = logger
+
+    async def create_user(self, telegram_id: int, username: str):
+        """
+        Создаёт нового пользователя в PostgreSQL.
+        """
+        async for session in self.session_generator():
+            try:
+                new_user = User(telegram_id=telegram_id, username=username)
+                session.add(new_user)
+                await session.commit()
+                return new_user
+            except SQLAlchemyError as e:
+                self.logger.error(f"Ошибка при создании пользователя {telegram_id}: {e}")
+                await session.rollback()
+                return None
+
+    async def get_user_by_telegram_id(self, telegram_id: int):
+        """
+        Возвращает пользователя по Telegram ID.
+        """
+        async for session in self.session_generator():
+            try:
+                result = await session.execute(select(User).where(User.telegram_id == telegram_id))
+                return result.scalars().first()
+            except SQLAlchemyError as e:
+                self.logger.error(f"Ошибка при получении пользователя {telegram_id}: {e}")
+                return None
+
+    async def add_transaction(self, user_id: int, amount: float):
+        """
+        Добавляет транзакцию для пользователя.
+        """
+        async for session in self.session_generator():
+            try:
+                transaction = Transaction(user_id=user_id, amount=amount)
+                session.add(transaction)
+                await session.commit()
+            except SQLAlchemyError as e:
+                self.logger.error(f"Ошибка добавления транзакции для пользователя {user_id}: {e}")
+                await session.rollback()
+
+    async def update_balance(self, telegram_id: int, amount: float):
+        """
+        Обновляет баланс пользователя.
+        """
+        async for session in self.session_generator():
+            try:
+                result = await session.execute(select(User).where(User.telegram_id == telegram_id))
+                user = result.scalars().first()
+                if user:
+                    user.balance += amount
+                    await session.commit()
+                    return user
+                else:
+                    self.logger.warning(f"Пользователь с Telegram ID {telegram_id} не найден.")
+                    return None
+            except SQLAlchemyError as e:
+                self.logger.error(f"Ошибка при обновлении баланса: {e}")
+                await session.rollback()
+                return None
+
+    async def last_subscription(self, user_id: int):
+        """
+        Возвращает последние подписки пользователя.
+        """
+        async for session in self.session_generator():
+            try:
+                result = await session.execute(
+                    select(Subscription)
+                    .where(Subscription.user_id == user_id)
+                    .order_by(desc(Subscription.created_at))
+                )
+                return result.scalars().all()
+            except SQLAlchemyError as e:
+                self.logger.error(f"Ошибка при получении последней подписки пользователя {user_id}: {e}")
+                return None
+
+    async def last_transaction(self, user_id: int):
+        """
+        Возвращает последние транзакции пользователя.
+        """
+        async for session in self.session_generator():
+            try:
+                result = await session.execute(
+                    select(Transaction)
+                    .where(Transaction.user_id == user_id)
+                    .order_by(desc(Transaction.created_at))
+                )
+                return result.scalars().all()
+            except SQLAlchemyError as e:
+                self.logger.error(f"Ошибка при получении транзакций пользователя {user_id}: {e}")
+                return None

app/services/xui_rep.py

@@ -0,0 +1,232 @@
+import aiohttp
+import uuid
+import base64
+import ssl
+
+def generate_uuid():
+    return str(uuid.uuid4())
+
+
+class PanelInteraction:
+    def __init__(self, base_url, login_data, logger, certificate=None, is_encoded=True):
+        """
+        Initialize the PanelInteraction class.
+
+        :param base_url: Base URL for the panel.
+        :param login_data: Login data (username/password or token).
+        :param logger: Logger for debugging.
+        :param certificate: Certificate content (Base64-encoded or raw string).
+        :param is_encoded: Indicates whether the certificate is Base64-encoded.
+        """
+        self.base_url = base_url
+        self.login_data = login_data
+        self.logger = logger
+        self.cert_content = self._decode_certificate(certificate, is_encoded)
+        self.session_id = None  # Session ID will be initialized lazily
+        self.headers = None
+
+    def _decode_certificate(self, certificate, is_encoded):
+        """
+        Decode the provided certificate content.
+
+        :param certificate: Certificate content (Base64-encoded or raw string).
+        :param is_encoded: Indicates whether the certificate is Base64-encoded.
+        :return: Decoded certificate content as bytes.
+        """
+        
+        if not certificate:
+            self.logger.error("No certificate provided.")
+            raise ValueError("Certificate is required.")
+        try:
+            # Создаем SSLContext
+            ssl_context = ssl.create_default_context()
+
+            # Декодируем, если нужно
+            if is_encoded:
+                certificate = base64.b64decode(certificate).decode()
+
+            # Загружаем сертификат в SSLContext
+            ssl_context.load_verify_locations(cadata=certificate)
+            return ssl_context
+
+        except Exception as e:
+            self.logger.error(f"Error while decoding certificate: {e}")
+            raise ValueError("Invalid certificate format or content.") from e
+        
+
+    async def _ensure_logged_in(self):
+        """
+        Ensure the session ID is available for authenticated requests.
+        """
+        if not self.session_id:
+            self.session_id = await self.login()
+            if self.session_id:
+                self.headers = {
+                    'Accept': 'application/json',
+                    'Cookie': f'3x-ui={self.session_id}',
+                    'Content-Type': 'application/json'
+                }
+            else:
+                raise ValueError("Unable to log in and retrieve session ID.")
+
+    async def login(self):
+        """
+        Perform login to the panel.
+
+        :return: Session ID or None.
+        """
+        login_url = f"{self.base_url}/login"
+        self.logger.info(f"Attempting to login at: {login_url}")
+
+        async with aiohttp.ClientSession() as session:
+            try:
+                async with session.post(
+                    login_url, data=self.login_data, ssl=self.cert_content, timeout=10
+                ) as response:
+                    if response.status == 200:
+                        session_id = response.cookies.get("3x-ui")
+                        if session_id:
+                            return session_id.value
+                        else:
+                            self.logger.error("Login failed: No session ID received.")
+                            return None
+                    else:
+                        self.logger.error(f"Login failed: {response.status}")
+                        return None
+            except aiohttp.ClientError as e:
+                self.logger.error(f"Login request failed: {e}")
+                return None
+
+    async def get_inbound_info(self, inbound_id):
+        """
+        Fetch inbound information by ID.
+
+        :param inbound_id: ID of the inbound.
+        :return: JSON response or None.
+        """
+        await self._ensure_logged_in()
+        url = f"{self.base_url}/panel/api/inbounds/get/{inbound_id}"
+        async with aiohttp.ClientSession() as session:
+            try:
+                async with session.get(
+                    url, headers=self.headers, ssl=self.cert_content, timeout=10
+                ) as response:
+                    if response.status == 200:
+                        return await response.json()
+                    else:
+                        self.logger.error(f"Failed to get inbound info: {response.status}")
+                        return None
+            except aiohttp.ClientError as e:
+                self.logger.error(f"Get inbound info request failed: {e}")
+                return None
+
+    async def get_client_traffic(self, email):
+        """
+        Fetch traffic information for a specific client.
+
+        :param email: Client's email.
+        :return: JSON response or None.
+        """
+        await self._ensure_logged_in()
+        url = f"{self.base_url}/panel/api/inbounds/getClientTraffics/{email}"
+        async with aiohttp.ClientSession() as session:
+            try:
+                async with session.get(
+                    url, headers=self.headers, ssl=self.cert_content, timeout=10
+                ) as response:
+                    if response.status == 200:
+                        return await response.json()
+                    else:
+                        self.logger.error(f"Failed to get client traffic: {response.status}")
+                        return None
+            except aiohttp.ClientError as e:
+                self.logger.error(f"Get client traffic request failed: {e}")
+                return None
+
+    async def update_client_expiry(self, client_uuid, new_expiry_time, client_email):
+        """
+        Update the expiry date of a specific client.
+
+        :param client_uuid: UUID of the client.
+        :param new_expiry_time: New expiry date in ISO format.
+        :param client_email: Client's email.
+        :return: None.
+        """
+        await self._ensure_logged_in()
+        url = f"{self.base_url}/panel/api/inbounds/updateClient"
+        update_data = {
+            "id": 1,
+            "settings": {
+                "clients": [
+                    {
+                        "id": client_uuid,
+                        "alterId": 0,
+                        "email": client_email,
+                        "limitIp": 2,
+                        "totalGB": 0,
+                        "expiryTime": new_expiry_time,
+                        "enable": True,
+                        "tgId": "",
+                        "subId": ""
+                    }
+                ]
+            }
+        }
+
+        async with aiohttp.ClientSession() as session:
+            try:
+                async with session.post(
+                    url, headers=self.headers, json=update_data, ssl=self.cert_content
+                ) as response:
+                    if response.status == 200:
+                        self.logger.info("Client expiry updated successfully.")
+                    else:
+                        self.logger.error(f"Failed to update client expiry: {response.status}")
+            except aiohttp.ClientError as e:
+                self.logger.error(f"Update client expiry request failed: {e}")
+
+    async def add_client(self, inbound_id, expiry_date, email):
+        """
+        Add a new client to an inbound.
+
+        :param inbound_id: ID of the inbound.
+        :param expiry_date: Expiry date in ISO format.
+        :param email: Client's email.
+        :return: JSON response or None.
+        """
+        await self._ensure_logged_in()
+        url = f"{self.base_url}/panel/api/inbounds/addClient"
+        client_info = {
+            "clients": [
+                {
+                    "id": generate_uuid(),
+                    "alterId": 0,
+                    "email": email,
+                    "limitIp": 2,
+                    "totalGB": 0,
+                    "flow": "xtls-rprx-vision",
+                    "expiryTime": expiry_date,
+                    "enable": True,
+                    "tgId": "",
+                    "subId": ""
+                }
+            ]
+        }
+        payload = {
+            "id": inbound_id,
+            "settings": client_info
+        }
+
+        async with aiohttp.ClientSession() as session:
+            try:
+                async with session.post(
+                    url, headers=self.headers, json=payload, ssl=self.cert_content
+                ) as response:
+                    if response.status == 200:
+                        return await response.status
+                    else:
+                        self.logger.error(f"Failed to add client: {response.status}")
+                        return None
+            except aiohttp.ClientError as e:
+                self.logger.error(f"Add client request failed: {e}")
+                return None