mirror of
https://github.com/BEDOLAGA-DEV/remnawave-bedolaga-telegram-bot.git
synced 2026-01-20 11:50:27 +00:00
282 lines
8.9 KiB
Python
282 lines
8.9 KiB
Python
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
from typing import Any, Dict
|
||
|
||
from app.config import settings
|
||
from app.localization.loader import (
|
||
DEFAULT_LANGUAGE,
|
||
clear_locale_cache,
|
||
load_locale,
|
||
)
|
||
|
||
_logger = logging.getLogger(__name__)
|
||
|
||
_cached_rules: Dict[str, str] = {}
|
||
|
||
|
||
_LANGUAGE_ALIASES = {
|
||
"uk": "ua",
|
||
}
|
||
|
||
|
||
_DYNAMIC_LANGUAGE_CONFIGS = {
|
||
"ru": {
|
||
"traffic_pattern": "📊 {size} ГБ - {price}",
|
||
"unlimited_pattern": "📊 Безлимит - {price}",
|
||
"support_info": (
|
||
"\n🛟 <b>Поддержка</b>\n\n"
|
||
"Это центр тикетов: создавайте обращения, просматривайте ответы и историю.\n\n"
|
||
"• 🎫 Создать тикет — опишите проблему или вопрос\n"
|
||
"• 📋 Мои тикеты — статус и переписка\n"
|
||
"• 💬 Связаться — написать напрямую (если нужно)\n\n"
|
||
"Старайтесь использовать тикеты — так мы быстрее поможем и ничего не потеряется.\n"
|
||
),
|
||
},
|
||
"en": {
|
||
"traffic_pattern": "📊 {size} GB - {price}",
|
||
"unlimited_pattern": "📊 Unlimited - {price}",
|
||
"support_info": (
|
||
"\n🛟 <b>RemnaWave Support</b>\n\n"
|
||
"This is the ticket center: create requests, view replies and history.\n\n"
|
||
"• 🎫 Create ticket — describe your issue or question\n"
|
||
"• 📋 My tickets — status and conversation\n"
|
||
"• 💬 Contact — message directly if needed\n\n"
|
||
"Prefer tickets — it helps us respond faster and keep context.\n"
|
||
),
|
||
},
|
||
"ua": {
|
||
"traffic_pattern": "📊 {size} ГБ - {price}",
|
||
"unlimited_pattern": "📊 Безліміт - {price}",
|
||
"support_info": (
|
||
"\n🛠️ <b>Технічна підтримка</b>\n\n"
|
||
"З усіх питань звертайтеся до нашої підтримки:\n\n"
|
||
"👤 {support_username}\n\n"
|
||
"Ми допоможемо з:\n"
|
||
"• Налаштуванням підключення\n"
|
||
"• Вирішенням технічних проблем\n"
|
||
"• Питаннями щодо оплати\n"
|
||
"• Іншими питаннями\n\n"
|
||
"⏰ Час відповіді: зазвичай протягом 1-2 годин\n"
|
||
),
|
||
},
|
||
"zh": {
|
||
"traffic_pattern": "📊{size}GB-{price}",
|
||
"unlimited_pattern": "📊无限-{price}",
|
||
"support_info": (
|
||
"\n🛠️ <b>技术支持</b>\n\n"
|
||
"如有任何问题，请联系我们的支持团队：\n\n"
|
||
"👤 {support_username}\n\n"
|
||
"我们将帮助您：\n"
|
||
"• 设置连接\n"
|
||
"• 解决技术问题\n"
|
||
"• 付款问题\n"
|
||
"• 其他问题\n\n"
|
||
"⏰ 响应时间：通常在 1-2 小时内\n"
|
||
),
|
||
},
|
||
}
|
||
|
||
|
||
_TRAFFIC_TIERS = (
|
||
("TRAFFIC_5GB", "5", "PRICE_TRAFFIC_5GB"),
|
||
("TRAFFIC_10GB", "10", "PRICE_TRAFFIC_10GB"),
|
||
("TRAFFIC_25GB", "25", "PRICE_TRAFFIC_25GB"),
|
||
("TRAFFIC_50GB", "50", "PRICE_TRAFFIC_50GB"),
|
||
("TRAFFIC_100GB", "100", "PRICE_TRAFFIC_100GB"),
|
||
("TRAFFIC_250GB", "250", "PRICE_TRAFFIC_250GB"),
|
||
)
|
||
|
||
|
||
def _get_cached_rules_value(language: str) -> str:
|
||
if language in _cached_rules:
|
||
return _cached_rules[language]
|
||
|
||
default = _get_default_rules(language)
|
||
_cached_rules[language] = default
|
||
return default
|
||
|
||
|
||
def _build_dynamic_values(language: str) -> Dict[str, Any]:
|
||
language_code = (language or DEFAULT_LANGUAGE).split("-")[0].lower()
|
||
|
||
language_code = _LANGUAGE_ALIASES.get(language_code, language_code)
|
||
config = _DYNAMIC_LANGUAGE_CONFIGS.get(language_code)
|
||
|
||
if not config:
|
||
return {}
|
||
|
||
values: Dict[str, Any] = {}
|
||
traffic_pattern = config["traffic_pattern"]
|
||
for key, size, price_attr in _TRAFFIC_TIERS:
|
||
price_value = getattr(settings, price_attr)
|
||
values[key] = traffic_pattern.format(
|
||
size=size,
|
||
price=settings.format_price(price_value),
|
||
)
|
||
|
||
values["TRAFFIC_UNLIMITED"] = config["unlimited_pattern"].format(
|
||
price=settings.format_price(settings.PRICE_TRAFFIC_UNLIMITED)
|
||
)
|
||
|
||
support_template = config.get("support_info")
|
||
if support_template:
|
||
values["SUPPORT_INFO"] = support_template.format(
|
||
support_username=settings.SUPPORT_USERNAME
|
||
)
|
||
|
||
return values
|
||
|
||
|
||
class Texts:
|
||
def __init__(self, language: str = DEFAULT_LANGUAGE):
|
||
self.language = language or DEFAULT_LANGUAGE
|
||
raw_data = load_locale(self.language)
|
||
self._values = {key: value for key, value in raw_data.items()}
|
||
|
||
if self.language != DEFAULT_LANGUAGE:
|
||
fallback_data = load_locale(DEFAULT_LANGUAGE)
|
||
else:
|
||
fallback_data = self._values
|
||
|
||
self._fallback_values = {
|
||
key: value for key, value in fallback_data.items() if key not in self._values
|
||
}
|
||
|
||
self._values.update(_build_dynamic_values(self.language))
|
||
|
||
def __getattr__(self, item: str) -> Any:
|
||
if item == "language":
|
||
return super().__getattribute__(item)
|
||
try:
|
||
return self._get_value(item)
|
||
except KeyError as error:
|
||
raise AttributeError(item) from error
|
||
|
||
def __getitem__(self, item: str) -> Any:
|
||
return self._get_value(item)
|
||
|
||
def get(self, item: str, default: Any = None) -> Any:
|
||
try:
|
||
return self._get_value(item)
|
||
except KeyError:
|
||
return default
|
||
|
||
def t(self, key: str, default: Any = None) -> Any:
|
||
try:
|
||
return self._get_value(key)
|
||
except KeyError:
|
||
if default is not None:
|
||
return default
|
||
raise
|
||
|
||
def _get_value(self, item: str) -> Any:
|
||
if item == "RULES_TEXT":
|
||
return _get_cached_rules_value(self.language)
|
||
|
||
if item in self._values:
|
||
return self._values[item]
|
||
|
||
if item in self._fallback_values:
|
||
return self._fallback_values[item]
|
||
|
||
_logger.warning(
|
||
"Missing localization key '%s' for language '%s'",
|
||
item,
|
||
self.language,
|
||
)
|
||
raise KeyError(item)
|
||
|
||
@staticmethod
|
||
def format_price(kopeks: int) -> str:
|
||
return settings.format_price(kopeks)
|
||
|
||
@staticmethod
|
||
def format_traffic(gb: float) -> str:
|
||
if gb == 0:
|
||
return "∞ (безлимит)"
|
||
if gb >= 1024:
|
||
return f"{gb / 1024:.1f} ТБ"
|
||
return f"{gb:.0f} ГБ"
|
||
|
||
|
||
def get_texts(language: str = DEFAULT_LANGUAGE) -> Texts:
|
||
return Texts(language)
|
||
|
||
|
||
async def get_rules_from_db(language: str = DEFAULT_LANGUAGE) -> str:
|
||
try:
|
||
from app.database.database import get_db
|
||
from app.database.crud.rules import get_current_rules_content
|
||
|
||
async for db in get_db():
|
||
rules = await get_current_rules_content(db, language)
|
||
if rules:
|
||
_cached_rules[language] = rules
|
||
return rules
|
||
break
|
||
|
||
except Exception as error: # pragma: no cover - defensive logging
|
||
_logger.warning("Failed to load rules from DB for %s: %s", language, error)
|
||
|
||
default = _get_default_rules(language)
|
||
_cached_rules[language] = default
|
||
return default
|
||
|
||
|
||
def _get_default_rules(language: str = DEFAULT_LANGUAGE) -> str:
|
||
default_key = "RULES_TEXT_DEFAULT"
|
||
locale = load_locale(language)
|
||
if default_key in locale:
|
||
return locale[default_key]
|
||
fallback = load_locale(DEFAULT_LANGUAGE)
|
||
return fallback.get(default_key, "")
|
||
|
||
|
||
def _get_default_privacy_policy(language: str = DEFAULT_LANGUAGE) -> str:
|
||
default_key = "PRIVACY_POLICY_TEXT_DEFAULT"
|
||
locale = load_locale(language)
|
||
if default_key in locale:
|
||
return locale[default_key]
|
||
fallback = load_locale(DEFAULT_LANGUAGE)
|
||
return fallback.get(default_key, "")
|
||
|
||
|
||
def get_privacy_policy(language: str = DEFAULT_LANGUAGE) -> str:
|
||
return _get_default_privacy_policy(language)
|
||
|
||
|
||
def get_rules_sync(language: str = DEFAULT_LANGUAGE) -> str:
|
||
if language in _cached_rules:
|
||
return _cached_rules[language]
|
||
|
||
try:
|
||
loop = asyncio.get_running_loop()
|
||
except RuntimeError:
|
||
return asyncio.run(get_rules(language))
|
||
|
||
loop.create_task(get_rules(language))
|
||
return _get_cached_rules_value(language)
|
||
|
||
|
||
async def get_rules(language: str = DEFAULT_LANGUAGE) -> str:
|
||
if language in _cached_rules:
|
||
return _cached_rules[language]
|
||
|
||
return await get_rules_from_db(language)
|
||
|
||
|
||
async def refresh_rules_cache(language: str = DEFAULT_LANGUAGE) -> None:
|
||
if language in _cached_rules:
|
||
del _cached_rules[language]
|
||
await get_rules_from_db(language)
|
||
|
||
|
||
def clear_rules_cache() -> None:
|
||
_cached_rules.clear()
|
||
|
||
|
||
def reload_locales() -> None:
|
||
clear_locale_cache()
|