107 lines
4.5 KiB
Python
107 lines
4.5 KiB
Python
"""
|
||
Утилиты локализации и зависимости переводов для FastAPI.
|
||
|
||
Содержимое модуля:
|
||
- `get_translations`: зависимость FastAPI, считывающая локаль из query/cookie/header.
|
||
- `get_translations_ws`: аналог для WebSocket.
|
||
- `_load_translations`: низкоуровневая загрузка словаря переводов с fallback.
|
||
- `_()`: обёртка для gettext с дефолтом ru.
|
||
"""
|
||
from fastapi import Request, WebSocket
|
||
from babel.support import Translations
|
||
import logging
|
||
from pathlib import Path
|
||
import os
|
||
from typing import Iterable, Optional
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
def _candidate_locale_dirs() -> Iterable[Path]:
|
||
"""Возвращает возможные каталоги с переводами."""
|
||
configured = os.getenv("GONETS_LOCALES_PATH")
|
||
if configured:
|
||
configured_path = Path(configured).expanduser()
|
||
if configured_path.exists():
|
||
yield configured_path
|
||
srv_root = Path(__file__).resolve().parent.parent
|
||
for candidate in (
|
||
srv_root / "locales",
|
||
srv_root.parent / "locales", # на случай, если каталоги вынесены на уровень выше
|
||
Path("/usr/share/gonets/locales"),
|
||
Path("/usr/local/share/gonets/locales"),
|
||
):
|
||
if candidate.exists():
|
||
yield candidate
|
||
|
||
LOCALE_DIRS = list(dict.fromkeys(_candidate_locale_dirs()))
|
||
if not LOCALE_DIRS:
|
||
logger.warning("Locales directory not found; translations will use NullTranslations")
|
||
|
||
def _load_translations(locale: str) -> Translations:
|
||
"""Подгружает словарь переводов с fallback на ru."""
|
||
last_error: Optional[str] = None
|
||
for directory in LOCALE_DIRS or [None]:
|
||
try:
|
||
if directory is None:
|
||
raise FileNotFoundError("Locales directory list is empty")
|
||
translations = Translations.load(str(directory), [locale], "messages")
|
||
if translations._catalog:
|
||
return translations
|
||
logger.warning(
|
||
"Translations catalog empty for locale %s in %s; trying fallback ru",
|
||
locale,
|
||
directory,
|
||
)
|
||
ru_translations = Translations.load(str(directory), ["ru"], "messages")
|
||
if ru_translations._catalog:
|
||
return ru_translations
|
||
except Exception as exc: # pragma: no cover - логируем и возвращаем ru
|
||
last_error = str(exc)
|
||
continue
|
||
logger.error(
|
||
"Failed to load translations for locale %s from dirs %s: %s. Using NullTranslations",
|
||
locale,
|
||
LOCALE_DIRS,
|
||
last_error,
|
||
)
|
||
return Translations()
|
||
|
||
def _resolve_locale_from_request(request: Request) -> str:
|
||
locale = request.query_params.get("locale")
|
||
if not locale:
|
||
locale = request.cookies.get("locale")
|
||
if not locale:
|
||
locale = request.headers.get("Accept-Language", "ru").split(",")[0].split("-")[0]
|
||
return locale or "ru"
|
||
|
||
def get_translations(request: Request) -> Translations:
|
||
"""Зависимость FastAPI для HTTP-эндпоинтов."""
|
||
locale = _resolve_locale_from_request(request)
|
||
logger.debug(f"Determined locale (http): {locale}")
|
||
return _load_translations(locale)
|
||
|
||
def get_translations_ws(websocket: WebSocket) -> Translations:
|
||
"""Зависимость FastAPI для WebSocket-эндпоинтов."""
|
||
# WebSocket не имеет cookies/headers в виде HTTPRequest, но есть headers
|
||
header_locale = websocket.headers.get("accept-language", "ru").split(",")[0].split("-")[0] if websocket.headers else "ru"
|
||
logger.debug(f"Determined locale (ws): {header_locale}")
|
||
return _load_translations(header_locale or "ru")
|
||
|
||
def _(msgid: str, request: Request = None) -> str:
|
||
"""
|
||
Переводит строку msgid для текущей локали.
|
||
|
||
Args:
|
||
msgid (str): Строка для перевода.
|
||
request (Request, optional): Объект запроса FastAPI для получения локали.
|
||
|
||
Returns:
|
||
str: Переведенная строка.
|
||
"""
|
||
if request is None:
|
||
translations = _load_translations("ru")
|
||
logger.debug(f"Using default ru translations for msgid: {msgid}")
|
||
else:
|
||
translations = get_translations(request)
|
||
logger.debug(f"Using translations for locale {request.cookies.get('locale', 'ru')} for msgid: {msgid}")
|
||
return translations.gettext(msgid)
|