import math import urllib.parse from datetime import UTC, datetime, timedelta from random import choice from remnawave.models.hwid import HwidDeviceDto from bot import texts from config import settings from schemas.subscriptions import InternalSquads, SubscriptionDuration, SubscriptionPlan from services.rw import RWUserInfo QUOTES: list[str] = [ "Верните себе свободный интернет.", "Свобода должна быть доступной.", "Интернет без границ — не привилегия, а норма.", "Ваш трафик. Ваши правила.", "Тишина в сети — это тоже выбор.", "Открытый мир начинается здесь.", "Без цензуры. Без компромиссов.", "Простота — лучшая форма свободы.", "Один клик между вами и миром.", "Свобода в форме привычки.", "Будьте везде. Оставайтесь невидимым.", "Приватность — это право, а не роскошь.", "Интернет таким, каким он задумывался.", "Ваши данные — только ваше дело.", "Закрытых дверей больше не существует.", "Любой сайт. Из любой точки мира.", "Оставьте след в истории, а не в браузере.", "Ключ к открытой сети.", "Фильтры придумали они. Решение выбрали вы.", "Безопасность, которую вы не замечаете.", "Сотрите цифровые границы. Останьтесь собой.", "Скорость без компромиссов. Сеть без преград.", "Включили и забыли. Вы в безопасности.", "Блокировки иллюзорны. Свобода реальна.", "Никакой слежки. Только вы и интернет.", "Доступ к информации — ваше базовое право.", "Интернет снова ваш.", "Контроль над сетью в ваших руках.", "Невидимый щит для вашей свободы.", "Весь мир на экране. Без лишних вопросов.", ] ADM_QUOTES: list[str] = [ "Они строят стены. Мы пишем для них двери.", "За каждым коммитом — чьё-то право на выбор.", "Цензура — это баг. И мы выкатили патч.", "Обычный день в админке. Глоток свободы для них.", "Наш труд невидим. Наше влияние безгранично.", "Мы не пишем код. Мы держим сеть открытой.", "Пульт управления свободой. Добро пожаловать.", "Malenia дышит. Благодаря тебе.", "Пока они запрещают, мы масштабируем.", "Тихая работа в бэкенде. Громкие результаты в сети.", "Миллионы терабайт трафика. Одна общая цель.", "Инструменты создают люди. Свободу создаешь ты.", "Никакой магии. Только наш код и упорство.", "Будущее интернета компилируется прямо здесь.", "Для кого-то — интерфейс. Для нас — линия фронта.", ] def fetch_quote() -> str: return choice(QUOTES) def adm_fetch_quote() -> str: return choice(ADM_QUOTES) def format_bytes(num_bytes: float) -> str: """Конвертировать байты в читаемый формат: ГБ / МБ / КБ / Б.""" gb_capacity = 1_073_741_824 mb_capacity = 1_048_576 kb_capacity = 1_024 if num_bytes >= gb_capacity: # >= 1 ГБ return f"{num_bytes / gb_capacity:.1f} ГБ" if num_bytes >= mb_capacity: # >= 1 МБ return f"{num_bytes / mb_capacity:.1f} МБ" if num_bytes >= kb_capacity: # >= 1 КБ return f"{num_bytes / kb_capacity:.1f} КБ" return f"{num_bytes:.0f} Б" def format_traffic(used: float, limit: int) -> str: """ Форматировать трафик. limit == 0 → без ограничений: "X ГБ" limit > 0 → с лимитом: "X ГБ / Y ГБ" """ used_str = format_bytes(used) if limit == 0: return texts.TRAFFIC_NO_LIMIT.format(used=used_str) limit_str = format_bytes(float(limit)) return texts.TRAFFIC_WITH_LIMIT.format(used=used_str, limit=limit_str) def format_hwid_limit(limit: int | None) -> str: """Форматировать лимит HWID: число или «без ограничений».""" if limit is None or limit == 0: return texts.UNLIMITED_HWID return str(limit) def format_squads(squads: list[dict]) -> str: """Форматировать список Internal Squads или «нет».""" if not squads: return texts.NO_SQUADS return ", ".join(s["name"] for s in squads) def format_days_left(expire_at: datetime) -> str: """ Вернуть строку с количеством оставшихся дней: - «30 дн.» — если осталось > 0 дней - «истекает сегодня» — если осталось 0 дней - «истекла N дн. назад» — если уже истекла """ now = datetime.now(tz=UTC) # Убедимся что expire_at timezone-aware if expire_at.tzinfo is None: expire_at = expire_at.replace(tzinfo=UTC) delta = expire_at - now days = delta.days # целое, может быть отрицательным if days > 0: return texts.DAYS_LEFT_POSITIVE.format(days=days) if days == 0: return texts.DAYS_LEFT_TODAY return texts.DAYS_LEFT_EXPIRED.format(days=abs(days)) def format_status(status: str) -> tuple[str, str]: """ Вернуть (иконка, текст) для строкового статуса подписки. Статус приводится к верхнему регистру перед сравнением. """ mapping: dict[str, tuple[str, str]] = { "ACTIVE": (texts.STATUS_ICON_ACTIVE, texts.STATUS_ACTIVE), "DISABLED": (texts.STATUS_ICON_DISABLED, texts.STATUS_DISABLED), "EXPIRED": (texts.STATUS_ICON_EXPIRED, texts.STATUS_EXPIRED), "LIMITED": (texts.STATUS_ICON_LIMITED, texts.STATUS_LIMITED), "ON_HOLD": (texts.STATUS_ICON_UNKNOWN, texts.STATUS_UNKNOWN), } return mapping.get(status.upper(), (texts.STATUS_ICON_UNKNOWN, texts.STATUS_UNKNOWN)) def format_username(username: str | None) -> str: """Форматировать username: «@user» или «нет username».""" if username: return "@" + username return texts.NO_USERNAME def format_referal_id(referal_id: int | None) -> str: if referal_id: return str(referal_id) return texts.NOTHING_PLACEHOLDER def convert_duration_to_int(duration: SubscriptionDuration) -> int: mapping: dict[str, int] = { SubscriptionDuration.MONTH.value: 1, SubscriptionDuration.THREE_MONTHS.value: 3, SubscriptionDuration.SIX_MONTHS.value: 6, SubscriptionDuration.YEAR.value: 12, } return mapping.get(duration.value, 1) def convert_int_to_duration(value: int) -> SubscriptionDuration: mapping: dict[int, SubscriptionDuration] = { 1: SubscriptionDuration.MONTH, 3: SubscriptionDuration.THREE_MONTHS, 6: SubscriptionDuration.SIX_MONTHS, 12: SubscriptionDuration.YEAR, } return mapping.get(value, SubscriptionDuration.MONTH) def convert_duration_to_timedelta(duration: SubscriptionDuration) -> timedelta: month_delta = timedelta(days=30) mapping: dict[str, timedelta] = { SubscriptionDuration.MONTH.value: month_delta, SubscriptionDuration.THREE_MONTHS.value: month_delta * 3, SubscriptionDuration.SIX_MONTHS.value: month_delta * 6, SubscriptionDuration.YEAR.value: month_delta * 12, } return mapping.get(duration.value, month_delta) def calculate_price(subscription: SubscriptionPlan): return math.ceil( ( (subscription.devices * settings.per_device_cost) + ( int(subscription.whitelists) * settings.whitelist_cost if subscription.devices < settings.whitelist_device_threshold else 0 ) ) * convert_duration_to_int(subscription.duration) * subscription.discount ) def get_discount(months) -> float: if months <= 0: return 0 calculated = 5 * (math.log2(months / 3) + 1) return max(0, calculated) def format_subscription_link(link: str | None = None): return f"{link or texts.NOTHING_PLACEHOLDER}" def format_share_deep_link(link: str): return f"https://t.me/share/url/?url={urllib.parse.quote_plus(link)}" def format_support_subscription( subscription: SubscriptionPlan, full_name: str, user_id: int, balance: int, username: str | None = None, referal_id: int | None = None, ) -> str: username = format_username(username) whitelist_description = "ВЫКЛ" if subscription.devices >= settings.whitelist_device_threshold: whitelist_description = "ВКЛ (Бесплатно)" elif subscription.whitelists: whitelist_description = "ВКЛ (Платно)" return texts.SUPPORT_SUBSCRIPTION.format( full_name=full_name, telegram_id=user_id, username_display=username, duration=convert_duration_to_int(subscription.duration), devices=subscription.devices, whitelist_description=whitelist_description, price=calculate_price(subscription), balance=balance, referal_id=referal_id, ) def get_subscription_link(short_uuid: str): return settings.subscription_url + "/" + short_uuid def build_main_menu_text(link: str, balance: int) -> str: return texts.MAIN_MENU.format(quote=fetch_quote(), link=link, balance=balance) def build_adm_main_menu_text(user_count: int, user_id: int): return texts.ADM_MAIN_MENU.format( quote=adm_fetch_quote(), user_count=user_count, user_id=user_id ) def is_same_plan(rw_user: RWUserInfo, new_plan: SubscriptionPlan) -> bool: if rw_user.hwid_device_limit != new_plan.devices: return False active_squad_uuids = {squad["uuid"] for squad in rw_user.active_squads} current_has_whitelist = InternalSquads.WHITELISTS in active_squad_uuids return current_has_whitelist == new_plan.whitelists def _format_os_emoji(os: str) -> str: os = os.lower() if os == "windows": return texts.WINDOWS if os == "linux": return texts.LINUX if os == "ios": return texts.APPLE if os == "android": return texts.ANDROID return texts.UNKNOWN_OS def format_hwid_device(device: HwidDeviceDto) -> str: os = device.platform model = device.device_model client = device.user_agent.split("/")[0] emoji = _format_os_emoji(os) return f"{emoji} {os} | {model} | {client}"