hexdev
287f8eebda
- Implemented SubscriptionRepository with methods to get, create, and update subscriptions. - Added BalanceTXRepository for creating and retrieving balance transactions. - Introduced a test script for simulating Pally webhook for local testing.
236 lines
8.9 KiB
Python
236 lines
8.9 KiB
Python
import math
|
||
import urllib.parse
|
||
from datetime import UTC, datetime, timedelta
|
||
from random import choice
|
||
|
||
from bot import texts
|
||
from config import settings
|
||
from schemas.subscriptions import SubscriptionDuration, SubscriptionPlan
|
||
|
||
QUOTES: list[str] = [
|
||
"Верните себе свободный интернет.",
|
||
"Свобода должна быть доступной.",
|
||
"Интернет без границ — не привилегия, а норма.",
|
||
"Ваш трафик. Ваши правила.",
|
||
"Тишина в сети — это тоже выбор.",
|
||
"Открытый мир начинается здесь.",
|
||
"Без цензуры. Без компромиссов.",
|
||
"Простота — лучшая форма свободы.",
|
||
"Один клик между вами и миром.",
|
||
"Свобода в форме привычки.",
|
||
"Будьте везде. Оставайтесь невидимым.",
|
||
"Приватность — это право, а не роскошь.",
|
||
"Интернет таким, каким он задумывался.",
|
||
"Ваши данные — только ваше дело.",
|
||
"Закрытых дверей больше не существует.",
|
||
"Любой сайт. Из любой точки мира.",
|
||
"Оставьте след в истории, а не в браузере.",
|
||
"Ключ к открытой сети.",
|
||
"Фильтры придумали они. Решение выбрали вы.",
|
||
"Безопасность, которую вы не замечаете.",
|
||
"Сотрите цифровые границы. Останьтесь собой.",
|
||
"Скорость без компромиссов. Сеть без преград.",
|
||
"Включили и забыли. Вы в безопасности.",
|
||
"Блокировки иллюзорны. Свобода реальна.",
|
||
"Никакой слежки. Только вы и интернет.",
|
||
"Доступ к информации — ваше базовое право.",
|
||
"Интернет снова ваш.",
|
||
"Контроль над сетью в ваших руках.",
|
||
"Невидимый щит для вашей свободы.",
|
||
"Весь мир на экране. Без лишних вопросов.",
|
||
]
|
||
|
||
|
||
def fetch_quote() -> str:
|
||
return choice(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"<code>{link or texts.NOTHING_PLACEHOLDER}</code>"
|
||
|
||
|
||
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)
|