Subscription.py - master - github_better/GenAI

Subscription.py
422 строки15 KB
github_better
New
22 дек 2025, 08:52
22 дек 2025, 08:52b68d8c4
import logging
import os

from aiogram import types

import Database.Database as Database

from Lib.Debugger import debug_print

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def get_tarrif_plan(currency: str) -> dict[str, int]:
    debug_print(logger)
    """
    Возвращает словарь тарифов по валюте на первой позиции название на второй цена в нужной валюте

    :param currency: валюта юзера
    :return: словарь тарифов
    """

    if not currency:
        raise ValueError("currency must be non-empty string")

    # Нормализуем валюту к верхнему регистру, т.к. в файлах ты используешь RUB
    currency = currency.upper()

    # Строим путь до файла тарифов относительно корня проекта
    base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    # Например: Text/Tarif_plan_RUB.txt
    tariffs_filename = f"Tarif_plan_{currency}.txt"
    tariffs_path = os.path.join(base_dir, "Text", tariffs_filename)

    if not os.path.exists(tariffs_path):
        logger.error(f"Tariff file not found for currency {currency}: {tariffs_path}")
        raise FileNotFoundError(f"Tariff file not found: {tariffs_path}")

    tariffs: dict[str, int] = {}

    with open(tariffs_path, "r", encoding="utf-8") as f:
        for raw_line in f:
            line = raw_line.strip()
            if not line:
                continue
            # Ожидаемый формат: NAME:149RUB
            if ":" not in line:
                logger.warning(f"Invalid tariff line (no colon) in {tariffs_path!r}: {line!r}")
                continue

            name_part, price_part = line.split(":", 1)
            name = name_part.strip()
            price_part = price_part.strip()

            if not price_part.upper().endswith(currency):
                logger.warning(
                    f"Tariff line currency mismatch in {tariffs_path!r}: {line!r}, "
                    f"expected currency suffix {currency!r}"
                )
                continue

            numeric_part = price_part[: -len(currency)]
            numeric_part = numeric_part.strip()

            try:
                price_value = int(numeric_part)
            except ValueError:
                logger.warning(
                    f"Cannot parse price as int in {tariffs_path!r}: {line!r}, "
                    f"numeric_part={numeric_part!r}"
                )
                continue

            tariffs[name] = price_value

    return tariffs

def get_tarrif_plan_STR(currency: str) -> dict[str, int]:
    debug_print(logger)
    """
    Читает файл с тарифами в "звёздах" для указанной валюты и возвращает словарь:
        {
            'MINIMUM': 1,
            'USUAL': 2,
            ...
        }

    Формат строк в файле (пример):
        MINIMUM:1⭐
        USUAL:2⭐
        ...

    Из правой части берутся ведущие цифры до первого нецифрового символа.

    :param currency: валюта юзера
    :return: тарифы в звездах
    """

    if not currency:
        raise ValueError("currency must be non-empty string")

    # Нормализуем валюту
    currency = currency.upper()

    # Путь к файлу вида Text/Tarif_plan_RUB_STARS.txt
    base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    tariffs_filename = f"Tarif_plan_{currency}_STARS.txt"
    tariffs_path = os.path.join(base_dir, "Text", tariffs_filename)

    if not os.path.exists(tariffs_path):
        logger.error(f"Tariff STARS file not found for currency {currency}: {tariffs_path}")
        raise FileNotFoundError(f"Tariff STARS file not found: {tariffs_path}")

    tariffs: dict[str, int] = {}

    with open(tariffs_path, "r", encoding="utf-8") as f:
        for raw_line in f:
            line = raw_line.strip()
            if not line:
                continue

            # Ожидаемый формат: NAME:что‑то
            if ":" not in line:
                logger.warning(f"Invalid tariff STARS line (no colon) in {tariffs_path!r}: {line!r}")
                continue

            name_part, value_part = line.split(":", 1)
            name = name_part.strip()
            value_part = value_part.strip()

            # Берём ведущие цифры (подойдёт как для '1', так и для '1STAR', '1⭐' и т.п.)
            digits = ""
            for ch in value_part:
                if ch.isdigit():
                    digits += ch
                else:
                    break

            if not digits:
                logger.warning(
                    f"Cannot parse stars value as int in {tariffs_path!r}: {line!r}, "
                    f"value_part={value_part!r}"
                )
                continue

            stars_value = int(digits)
            tariffs[name] = stars_value

    return tariffs

def get_single_tarrif_plan(currency: str) -> dict[str, float]:
    debug_print(logger)
    """
    Читает файл одноразовых тарифов для указанной валюты и возвращает словарь
    СТАВОК за один запрос / одно изображение, например:
        {
            'SINGLE_TEXT': 0.8,
            'SINGLE_PIC': 9.5,
            ...
        }

    Формат строк в файле Text/Tarif_single_{CURRENCY}.txt:
        SINGLE_TEXT:0.8RUB
        SINGLE_PIC:9.5RUB
        ...

    Числа могут быть с дробной частью.

    :param currency: валюта
    :return: одноразовый тарифный план
    """

    if not currency:
        raise ValueError("currency must be non-empty string")

    currency = currency.upper()

    base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    tariffs_filename = f"Tarif_single_{currency}.txt"
    tariffs_path = os.path.join(base_dir, "Text", tariffs_filename)

    if not os.path.exists(tariffs_path):
        logger.error(f"Single tariff file not found for currency {currency}: {tariffs_path}")
        raise FileNotFoundError(f"Single tariff file not found: {tariffs_path}")

    tariffs: dict[str, float] = {}

    with open(tariffs_path, "r", encoding="utf-8") as f:
        for raw_line in f:
            line = raw_line.strip()
            if not line:
                continue

            # Ожидаемый формат: NAME:0.8RUB
            if ":" not in line:
                logger.warning(f"Invalid single tariff line (no colon) in {tariffs_path!r}: {line!r}")
                continue

            name_part, price_part = line.split(":", 1)
            name = name_part.strip()
            price_part = price_part.strip()

            if not price_part.upper().endswith(currency):
                logger.warning(
                    f"Single tariff line currency mismatch in {tariffs_path!r}: {line!r}, "
                    f"expected currency suffix {currency!r}"
                )
                continue

            numeric_part = price_part[: -len(currency)].strip()

            try:
                # Поддерживаем и 0.8, и 0,8
                price_value = float(numeric_part.replace(",", "."))
            except ValueError:
                logger.warning(
                    f"Cannot parse single tariff price as float in {tariffs_path!r}: {line!r}, "
                    f"numeric_part={numeric_part!r}"
                )
                continue

            tariffs[name] = price_value

    return tariffs


def get_single_tarrif_plan_STR(currency: str) -> dict[str, float]:
    debug_print(logger)
    """
    Читает файл одноразовых тарифов в Telegram Stars и возвращает словарь
    СТАВОК в звёздах за один запрос / одно изображение, например:
        {
            'SINGLE_TEXT': 1,
            'SINGLE_PIC': 10,
            ...
        }

    Формат строк в файле Text/Tarif_single_{CURRENCY}_STARS.txt:
        SINGLE_TEXT:1STR
        SINGLE_PIC:10STR
        ...

    Из правой части берутся ведущие цифры до первого нецифрового символа.
    
    :param currency: валюта юзера 
    :return: тариф в тг звездах
    """

    if not currency:
        raise ValueError("currency must be non-empty string")

    currency = currency.upper()

    base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    tariffs_filename = f"Tarif_single_{currency}_STARS.txt"
    tariffs_path = os.path.join(base_dir, "Text", tariffs_filename)

    if not os.path.exists(tariffs_path):
        logger.error(f"Single tariff STARS file not found for currency {currency}: {tariffs_path}")
        raise FileNotFoundError(f"Single tariff STARS file not found: {tariffs_path}")

    tariffs: dict[str, float] = {}

    with open(tariffs_path, "r", encoding="utf-8") as f:
        for raw_line in f:
            line = raw_line.strip()
            if not line:
                continue

            if ":" not in line:
                logger.warning(
                    f"Invalid single tariff STARS line (no colon) in {tariffs_path!r}: {line!r}"
                )
                continue

            name_part, value_part = line.split(":", 1)
            name = name_part.strip()
            value_part = value_part.strip()

            digits = ""
            for ch in value_part:
                if ch.isdigit():
                    digits += ch
                else:
                    break

            if not digits:
                logger.warning(
                    f"Cannot parse single stars value as int in {tariffs_path!r}: {line!r}, "
                    f"value_part={value_part!r}"
                )
                continue

            stars_value = int(digits)
            tariffs[name] = stars_value

    return tariffs


def beautiful_tariffs_send(currency: str) -> str:
    debug_print(logger)
    """
    Возвращает путь до изображения с тарифами для указанной валюты.

    Например:
        currency = 'RUB' -> 'Image_in_bot/tarrifs_pic_RUB.png'
        
    :param currency: валюта юзера
    :return: путь до картинки с тарифами 
    """

    if not currency:
        raise ValueError("currency must be non-empty string")

    currency = currency.upper()

    base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    image_filename = f"tarrifs_pic_{currency}.png"
    image_path = os.path.join(base_dir, "Image_in_bot", image_filename)

    if not os.path.exists(image_path):
        logger.error(f"Tariff image not found for currency {currency}: {image_path}")
        raise FileNotFoundError(f"Tariff image not found: {image_path}")

    return image_path

def get_name_of_current_tariff(user_id):
    debug_print(logger)
    """
    Возвращает код текущего тарифа пользователя по его user_id (UUID из mainDB.user_id).
    Если пользователь не найден или план не задан, вернёт 'Free'.
    
    :param user_id: айди юзера
    :return: текущий тариф юзера 
    """

    if not user_id:
        raise ValueError("user_id must be non-empty")

    plan = Database.take_value(
        "mainDB",
        {"user_id": user_id},
        "current_plan"
    )

    # В схеме по умолчанию current_plan = 'Free', но подстрахуемся
    return plan if plan else "Free"


def get_text_and_im_credits(user_id) -> [int, int]:
    debug_print(logger)
    """
    Возвращает список [text_credits, image_credits] для пользователя по его user_id.
    Если пользователь не найден или значения NULL, вернёт [0, 0].
    
    :param user_id: айди юзера
    :return: список текущих текстовых кредитов и кредитов на изображение
    """

    if not user_id:
        raise ValueError("user_id must be non-empty")

    text_raw = Database.take_value(
        "mainDB",
        {"user_id": user_id},
        "text_credits"
    )
    image_raw = Database.take_value(
        "mainDB",
        {"user_id": user_id},
        "image_credits"
    )

    text_credits = int(text_raw) if text_raw is not None else 0
    image_credits = int(image_raw) if image_raw is not None else 0

    return [text_credits, image_credits]


async def create_stars_invoice(
        bot,
        chat_id: int,
        title: str,
        description: str,
        payload: str,
        amount_stars: int
) -> bool:
    debug_print(logger)
    """
    Создает инвойс для оплаты через Telegram Stars.

    :param bot: объект бота
    :param chat_id: ID чата пользователя
    :param title: название товара
    :param description: описание товара
    :param payload: полезная нагрузка для идентификации платежа (например, "text_50" или "image_20")
    :param amount_stars: стоимость в Telegram Stars
    :return: True если инвойс отправлен успешно, False иначе
    """

    try:
        # Формируем prices для Telegram Stars
        # В API Telegram Stars цена передается в XTR (Telegram Stars)
        prices = [types.LabeledPrice(label="XTR", amount=amount_stars)]

        # Отправляем инвойс
        await bot.send_invoice(
            chat_id=chat_id,
            title=title,
            description=description,
            payload=payload,
            provider_token="",  # Для Stars токен не нужен
            currency="XTR",  # Валюта Telegram Stars
            prices=prices,
            start_parameter=f"stars_payment_{chat_id}"
        )

        logger.info(f"✅ Invoice sent to {chat_id}: {amount_stars} stars for {payload}")
        return True

    except Exception as e:
        logger.exception(f"❌ Error sending invoice to {chat_id}: {e}")
        return False