austnv/remnawave-minishop
1
0
Fork 0
Code Issues 1 Pull Requests Activity

feat: add runtime locale overrides

Browse Source
This commit is contained in:
3252a8
2026-05-25 17:47:45 +03:00
parent 23784d00fe
commit 59fa301344
51 changed files with 4026 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/bot/middlewares/i18n.py
+439 -9
View File
@@ -1,7 +1,10 @@
import json
import logging
import os
from typing import Any, Awaitable, Callable, Dict, Optional
import re
import time
from pathlib import Path
from typing import Any, Awaitable, Callable, Dict, Iterable, List, Optional, Set, Tuple
from aiogram import BaseMiddleware
from aiogram.types import Update, User
@@ -10,14 +13,309 @@ from sqlalchemy.ext.asyncio import AsyncSession
from config.settings import Settings
from db.dal import user_dal
LocaleOverrides = Dict[str, Dict[str, str]]
_LOCALE_LANGUAGE_CODE_RE = re.compile(r"^[a-z][a-z0-9]*(?:-[a-z0-9]+)*$")
LANGUAGE_LABELS: Dict[str, str] = {
"ru": "Русский",
"en": "English",
"de": "Deutsch",
"es": "Español",
"fr": "Français",
"pt-br": "Português (BR)",
"tr": "Türkçe",
"uk": "Українська",
}
LANGUAGE_FLAGS: Dict[str, str] = {
"ru": "🇷🇺",
"en": "🇬🇧",
"de": "🇩🇪",
"es": "🇪🇸",
"fr": "🇫🇷",
"pt-br": "🇧🇷",
"tr": "🇹🇷",
"uk": "🇺🇦",
}
DEFAULT_LANGUAGE_ORDER = ("ru", "en")
LOCALE_KEY_ALIASES: Dict[str, str] = {
"admin_apply": "wa_apply",
"admin_ads_col_status": "admin_status",
"admin_ad_label_source": "admin_ads_col_source",
"admin_back": "wa_back",
"admin_btn_refresh": "admin_refresh",
"admin_btn_save": "admin_save",
"admin_btn_saving": "admin_saving",
"admin_close": "wa_close",
"admin_copied": "wa_copied",
"admin_copy": "wa_copy",
"admin_csv_amount": "admin_amount",
"admin_csv_description": "admin_description",
"admin_csv_payment_id": "admin_id",
"admin_csv_status": "admin_status",
"admin_link_copied": "wa_link_copied",
"admin_next": "wa_next",
"admin_payment_detail_copied": "wa_copied",
"admin_payment_detail_provider": "admin_provider",
"admin_payment_detail_provider_section": "admin_provider",
"admin_payment_detail_user_section": "admin_user",
"admin_payments_col_user_id": "admin_id",
"admin_promo_col_code": "admin_promo_csv_code",
"admin_promo_col_status": "admin_status",
"admin_promo_csv_is_active": "admin_badge_active",
"admin_promo_csv_status": "admin_status",
"admin_promo_label_code": "admin_promo_csv_code",
"admin_promo_unlimited_validity": "admin_promo_unlimited",
"admin_stats_revenue_custom_range_apply": "wa_apply",
"admin_stats_revenue_tooltip_amount": "admin_amount",
"admin_stats_sync_status": "admin_status",
"admin_status_active": "admin_badge_active",
"admin_support_category": "wa_support_category",
"admin_support_category_account": "wa_support_category_account",
"admin_support_category_billing": "wa_support_category_billing",
"admin_support_category_other": "wa_support_category_other",
"admin_support_category_technical": "wa_support_category_technical",
"admin_support_close_ticket": "wa_close",
"admin_support_empty": "wa_support_empty",
"admin_support_filter_active": "wa_support_filter_active",
"admin_support_filter_all": "wa_support_filter_all",
"admin_support_internal_note": "wa_support_internal_note",
"admin_support_no_messages": "wa_support_no_messages",
"admin_support_priority": "wa_support_priority",
"admin_support_priority_high": "wa_support_priority_high",
"admin_support_priority_low": "wa_support_priority_low",
"admin_support_priority_normal": "wa_support_priority_normal",
"admin_support_priority_urgent": "wa_support_priority_urgent",
"admin_support_role_system": "wa_support_role_system",
"admin_support_role_user": "admin_user",
"admin_support_search": "admin_search",
"admin_support_status": "admin_status",
"admin_support_status_awaiting_admin": "wa_support_status_awaiting_admin",
"admin_support_status_awaiting_user": "wa_support_status_awaiting_user",
"admin_support_status_closed": "wa_support_status_closed",
"admin_support_status_open": "wa_support_status_open",
"admin_support_status_resolved": "wa_support_status_resolved",
"admin_support_ticket_number": "wa_support_ticket_number",
"admin_support_user_context": "admin_user",
"admin_tariffs_legacy_traffic_packages": "admin_tariff_traffic_packages",
"admin_tariffs_stat_enabled": "admin_enabled",
"admin_user_btn_cancel": "wa_cancel",
"admin_user_history_until": "wa_until_date",
"admin_user_label_provider": "admin_provider",
"admin_user_short": "admin_user",
"admin_user_stats_total_label": "admin_total",
"back_to_autopay_method_choice_button": "back_to_main_menu_button",
"back_to_payment_methods_button": "back_to_main_menu_button",
"cancel_broadcast_button": "cancel_button",
"csv_no": "no_button",
"csv_yes": "yes_button",
"user_premium_override_status_unlimited": "user_regular_override_status_unlimited",
"user_regular_override_save": "admin_save",
"wa_devices_disconnect_title": "wa_devices_disconnect",
"wa_install_link_copied": "wa_link_copied",
"wa_link_email_modal_title": "wa_settings_link_email_action",
}
def resolve_locale_key(key: object) -> str:
value = str(key or "").strip()
seen: Set[str] = set()
while value in LOCALE_KEY_ALIASES and value not in seen:
seen.add(value)
value = LOCALE_KEY_ALIASES[value]
return value
def is_valid_locale_language_code(value: str) -> bool:
return 2 <= len(value) <= 16 and bool(_LOCALE_LANGUAGE_CODE_RE.fullmatch(value))
def normalize_locale_language_code(
raw: object,
valid_languages: Optional[Set[str]] = None,
*,
prefer_known_base: bool = True,
) -> str:
value = str(raw or "").strip().lower().replace("_", "-")
if not value:
return ""
if prefer_known_base and valid_languages and value not in valid_languages:
base = value.split("-", 1)[0]
if base in valid_languages:
return base
return value
def _normalize_language_code(raw: object, valid_languages: Optional[Set[str]] = None) -> str:
return normalize_locale_language_code(raw, valid_languages)
def locale_language_label(code: object) -> str:
value = normalize_locale_language_code(code, prefer_known_base=False)
return LANGUAGE_LABELS.get(value, value.upper())
def locale_language_flag(code: object) -> str:
value = normalize_locale_language_code(code, prefer_known_base=False)
return LANGUAGE_FLAGS.get(value, "🏳️")
def sort_locale_language_codes(codes: Iterable[object]) -> List[str]:
normalized = {
normalize_locale_language_code(code, prefer_known_base=False)
for code in codes
}
normalized = {code for code in normalized if code and is_valid_locale_language_code(code)}
preferred = [code for code in DEFAULT_LANGUAGE_ORDER if code in normalized]
rest = sorted(code for code in normalized if code not in DEFAULT_LANGUAGE_ORDER)
return [*preferred, *rest]
def locale_language_options(
codes: Iterable[object],
*,
base_languages: Iterable[object] = (),
) -> List[Dict[str, Any]]:
base_set = set(sort_locale_language_codes(base_languages))
return [
{
"code": code,
"label": locale_language_label(code),
"flag": locale_language_flag(code),
"base": code in base_set,
}
for code in sort_locale_language_codes(codes)
]
def _valid_locale_keys_by_language(
locales_data: Dict[str, Dict[str, str]],
) -> Dict[str, Set[str]]:
return {
lang: {str(key) for key in messages.keys()}
for lang, messages in locales_data.items()
if isinstance(messages, dict)
}
def normalize_locale_overrides_payload(
payload: object,
*,
valid_languages: Optional[Iterable[str]] = None,
valid_keys_by_language: Optional[Dict[str, Set[str]]] = None,
allow_extra_languages: bool = False,
key_aliases: Optional[Dict[str, str]] = None,
) -> Tuple[LocaleOverrides, Dict[str, str]]:
"""Normalize a user/admin supplied locale override JSON payload.
The canonical shape is ``{"ru": {"welcome": "..."}, "en": {...}}``.
For convenience, files may also wrap it as ``{"overrides": {...}}`` or
``{"locales": {...}}``.
"""
if not isinstance(payload, dict):
return {}, {"_payload": "invalid_payload"}
raw_payload = payload
for wrapper_key in ("overrides", "locales"):
wrapped = raw_payload.get(wrapper_key)
if isinstance(wrapped, dict):
raw_payload = wrapped
break
valid_lang_set = {str(lang).lower() for lang in valid_languages or []}
aliases = key_aliases or LOCALE_KEY_ALIASES
def resolve_payload_key(raw_key: str) -> str:
value = raw_key
seen: Set[str] = set()
while value in aliases and value not in seen:
seen.add(value)
value = aliases[value]
return value
all_valid_keys: Set[str] = set()
if valid_keys_by_language:
for keys in valid_keys_by_language.values():
all_valid_keys.update(str(key) for key in keys)
overrides: LocaleOverrides = {}
errors: Dict[str, str] = {}
for raw_lang, raw_messages in raw_payload.items():
lang = normalize_locale_language_code(
raw_lang,
valid_lang_set or None,
prefer_known_base=not allow_extra_languages,
)
error_key = str(raw_lang or "_language")
if not lang:
errors[error_key] = "invalid_language"
continue
if valid_lang_set and lang not in valid_lang_set:
if not allow_extra_languages:
errors[error_key] = "unknown_language"
continue
if not is_valid_locale_language_code(lang):
errors[error_key] = "invalid_language"
continue
elif allow_extra_languages and not is_valid_locale_language_code(lang):
errors[error_key] = "invalid_language"
continue
if not isinstance(raw_messages, dict):
errors[lang] = "invalid_language_bucket"
continue
lang_keys = valid_keys_by_language.get(lang, set()) if valid_keys_by_language else set()
bucket: Dict[str, str] = {}
for raw_key, raw_value in raw_messages.items():
raw_key_text = str(raw_key or "").strip()
key = resolve_payload_key(raw_key_text)
item_error_key = f"{lang}.{raw_key_text or '_key'}"
if not raw_key_text or not key:
errors[item_error_key] = "invalid_key"
continue
if all_valid_keys and key not in all_valid_keys and key not in lang_keys:
errors[item_error_key] = "unknown_key"
continue
if raw_value is None:
continue
if not isinstance(raw_value, str):
errors[item_error_key] = "invalid_value"
continue
if len(raw_value) > 20000:
errors[item_error_key] = "value_too_long"
continue
if raw_key_text in aliases and key in bucket:
continue
bucket[key] = raw_value
if bucket:
overrides[lang] = dict(sorted(bucket.items()))
return dict(sorted(overrides.items())), errors
class JsonI18n:
def __init__(self, path: str, default: str = "en", domain: str = "bot"):
def __init__(
self,
path: str,
default: str = "en",
domain: str = "bot",
overrides_path: Optional[str] = None,
):
self.domain = domain
self.path = path
self.default_lang = default
self.base_locales_data: Dict[str, Dict[str, str]] = {}
self.locale_overrides: LocaleOverrides = {}
self.locales_data: Dict[str, Dict[str, str]] = {}
self._overrides_path: Optional[Path] = None
self._overrides_file_mtime_ns: Optional[int] = None
self._overrides_file_next_check = 0.0
self._overrides_file_check_interval_seconds = 1.0
self._load_locales()
if overrides_path:
self.configure_overrides_file(overrides_path)
self.reload_overrides_from_file(force=True)
logging.info(
f"JsonI18n initialized. Loaded languages: {list(self.locales_data.keys())}. Default: {self.default_lang}" # noqa: E501
)
@@ -26,13 +324,26 @@ class JsonI18n:
if not os.path.isdir(self.path):
logging.error(f"Locales path not found or not a directory: {self.path}")
return
loaded: Dict[str, Dict[str, str]] = {}
for item in os.listdir(self.path):
if item.endswith(".json"):
lang_code = item.split(".")[0]
file_path = os.path.join(self.path, item)
try:
with open(file_path, "r", encoding="utf-8") as f:
self.locales_data[lang_code] = json.load(f)
data = json.load(f)
if isinstance(data, dict):
loaded[lang_code] = {
str(key): str(value)
for key, value in data.items()
if isinstance(value, str)
}
else:
logging.error(
"Locale %s from %s is not a JSON object",
lang_code,
file_path,
)
except json.JSONDecodeError as e_json_load:
logging.error(
f"Error loading locale {lang_code} from {file_path} (JSON Decode Error): {e_json_load}" # noqa: E501
@@ -42,24 +353,143 @@ class JsonI18n:
f"Error loading locale {lang_code} from {file_path}: {e_load}",
exc_info=True,
)
self.base_locales_data = loaded
self._rebuild_effective_locales()
def _rebuild_effective_locales(self) -> None:
effective: Dict[str, Dict[str, str]] = {}
for lang, messages in self.base_locales_data.items():
merged = dict(messages)
merged.update(self.locale_overrides.get(lang, {}))
effective[lang] = merged
fallback_base = (
self.base_locales_data.get(self.default_lang)
or self.base_locales_data.get("en")
or next(iter(self.base_locales_data.values()), {})
)
for lang, messages in self.locale_overrides.items():
if lang in effective:
continue
merged = dict(fallback_base)
merged.update(messages)
effective[lang] = merged
self.locales_data = effective
def _valid_keys_by_language(self) -> Dict[str, Set[str]]:
return _valid_locale_keys_by_language(self.base_locales_data)
def language_options(self) -> List[Dict[str, Any]]:
self.reload_overrides_from_file()
return locale_language_options(
self.locales_data.keys(),
base_languages=self.base_locales_data.keys(),
)
def set_locale_overrides(self, overrides: object) -> Dict[str, str]:
normalized, errors = normalize_locale_overrides_payload(
overrides,
valid_languages=set(self.base_locales_data.keys()),
valid_keys_by_language=self._valid_keys_by_language(),
allow_extra_languages=True,
)
if errors:
logging.warning("Some locale overrides were skipped: %s", errors)
self.locale_overrides = normalized
self._rebuild_effective_locales()
return errors
def configure_overrides_file(self, path: str | Path) -> None:
self._overrides_path = Path(path)
try:
self._overrides_file_mtime_ns = self._overrides_path.stat().st_mtime_ns
except FileNotFoundError:
self._overrides_file_mtime_ns = None
except OSError as exc:
logging.warning("Failed to stat locale overrides file %s: %s", path, exc)
self._overrides_file_mtime_ns = None
def reload_overrides_from_file(self, *, force: bool = False) -> bool:
if self._overrides_path is None:
return False
now = time.monotonic()
if not force and now < self._overrides_file_next_check:
return False
self._overrides_file_next_check = now + self._overrides_file_check_interval_seconds
try:
stat = self._overrides_path.stat()
except FileNotFoundError:
if self._overrides_file_mtime_ns is None:
return False
self._overrides_file_mtime_ns = None
logging.info(
"Locale overrides file removed; keeping current in-memory overrides until "
"the DB fallback is reloaded"
)
return False
except OSError as exc:
logging.warning(
"Failed to stat locale overrides file %s: %s",
self._overrides_path,
exc,
)
return False
if not force and stat.st_mtime_ns == self._overrides_file_mtime_ns:
return False
try:
payload = json.loads(self._overrides_path.read_text(encoding="utf-8"))
except json.JSONDecodeError as exc:
logging.warning(
"Failed to parse locale overrides file %s: %s",
self._overrides_path,
exc,
)
self._overrides_file_mtime_ns = stat.st_mtime_ns
return False
except OSError as exc:
logging.warning(
"Failed to read locale overrides file %s: %s",
self._overrides_path,
exc,
)
return False
self._overrides_file_mtime_ns = stat.st_mtime_ns
self.set_locale_overrides(payload)
logging.info("Locale overrides reloaded from %s", self._overrides_path)
return True
def gettext(self, lang_code: Optional[str], key: str, **kwargs) -> str:
self.reload_overrides_from_file()
lookup_key = resolve_locale_key(key)
requested_lang_code = normalize_locale_language_code(
lang_code,
set(self.locales_data.keys()),
prefer_known_base=False,
)
requested_base_lang_code = requested_lang_code.split("-", 1)[0]
# Determine effective language with robust fallback
if lang_code and lang_code in self.locales_data:
effective_lang_code = lang_code
if requested_lang_code and requested_lang_code in self.locales_data:
effective_lang_code = requested_lang_code
elif requested_base_lang_code and requested_base_lang_code in self.locales_data:
effective_lang_code = requested_base_lang_code
elif self.default_lang in self.locales_data:
effective_lang_code = self.default_lang
elif "en" in self.locales_data:
effective_lang_code = "en"
else:
effective_lang_code = lang_code or self.default_lang
effective_lang_code = requested_lang_code or self.default_lang
lang_data = self.locales_data.get(effective_lang_code)
if lang_data is None:
# Try explicit fallback to English if available
fallback_data = self.locales_data.get("en")
if fallback_data is not None:
text = fallback_data.get(key)
text = fallback_data.get(lookup_key)
if text is not None:
try:
return text.format(**kwargs) if kwargs else text
@@ -70,11 +500,11 @@ class JsonI18n:
)
return key.format(**kwargs) if kwargs else key
text = lang_data.get(key)
text = lang_data.get(lookup_key)
if text is None:
if effective_lang_code != self.default_lang:
default_lang_data = self.locales_data.get(self.default_lang, {})
text = default_lang_data.get(key)
text = default_lang_data.get(lookup_key)
if text is None:
logging.warning(