Move DTO definitions from dto/ to schemas/ and update imports. Add buy handler with FSM state management for device selection, whitelist toggling, and duration choice. Include pricing logic, inline keyboards, and new configuration fields.
172 lines
7.1 KiB
Python
172 lines
7.1 KiB
Python
from typing import Optional
|
||
from random import choice
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
from config import settings
|
||
from schemas.subscriptions import SubscriptionDuration, SubscriptionPlan
|
||
import texts
|
||
|
||
QUOTES: list[str] = [
|
||
"Верните себе свободный интернет.",
|
||
"Свобода должна быть доступной.",
|
||
"Интернет без границ — не привилегия, а норма.",
|
||
"Ваш трафик. Ваши правила.",
|
||
"Тишина в сети — это тоже выбор.",
|
||
"Открытый мир начинается здесь.",
|
||
"Без цензуры. Без компромиссов.",
|
||
"Простота — лучшая форма свободы.",
|
||
"Один клик между вами и миром.",
|
||
"Свобода в форме привычки.",
|
||
"Будьте везде. Оставайтесь невидимым.",
|
||
"Приватность — это право, а не роскошь.",
|
||
"Интернет таким, каким он задумывался.",
|
||
"Ваши данные — только ваше дело.",
|
||
"Закрытых дверей больше не существует.",
|
||
"Любой сайт. Из любой точки мира.",
|
||
"Оставьте след в истории, а не в браузере.",
|
||
"Ключ к открытой сети.",
|
||
"Фильтры придумали они. Решение выбрали вы.",
|
||
"Безопасность, которую вы не замечаете.",
|
||
"Сотрите цифровые границы. Останьтесь собой.",
|
||
"Скорость без компромиссов. Сеть без преград.",
|
||
"Включили и забыли. Вы в безопасности.",
|
||
"Блокировки иллюзорны. Свобода реальна.",
|
||
"Никакой слежки. Только вы и интернет.",
|
||
"Доступ к информации — ваше базовое право.",
|
||
"Интернет снова ваш.",
|
||
"Контроль над сетью в ваших руках.",
|
||
"Невидимый щит для вашей свободы.",
|
||
"Весь мир на экране. Без лишних вопросов.",
|
||
]
|
||
|
||
|
||
def fetch_quote() -> str:
|
||
return choice(QUOTES)
|
||
|
||
|
||
def format_bytes(num_bytes: float) -> str:
|
||
"""Конвертировать байты в читаемый формат: ГБ / МБ / КБ / Б."""
|
||
if num_bytes >= 1_073_741_824: # >= 1 ГБ
|
||
return f"{num_bytes / 1_073_741_824:.1f} ГБ"
|
||
if num_bytes >= 1_048_576: # >= 1 МБ
|
||
return f"{num_bytes / 1_048_576:.1f} МБ"
|
||
if num_bytes >= 1_024: # >= 1 КБ
|
||
return f"{num_bytes / 1_024:.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: Optional[int]) -> 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=timezone.utc)
|
||
|
||
# Убедимся что expire_at timezone-aware
|
||
if expire_at.tzinfo is None:
|
||
expire_at = expire_at.replace(tzinfo=timezone.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: Optional[str]) -> str:
|
||
"""Форматировать username: «@user» или «нет username»."""
|
||
if username:
|
||
return "@" + username
|
||
return texts.NO_USERNAME
|
||
|
||
|
||
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 (
|
||
(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)
|