371 lines
11 KiB
Python
371 lines
11 KiB
Python
"""
|
|
Translation service for i18n system.
|
|
|
|
This module provides the core translation functionality including:
|
|
- Translation lookup with fallback mechanism
|
|
- Parameterized message support
|
|
- Enum value translation
|
|
- Memory caching for performance
|
|
- Performance monitoring and metrics
|
|
"""
|
|
|
|
import logging
|
|
from functools import lru_cache
|
|
from typing import Any, Dict, Optional
|
|
|
|
from app.i18n.loader import TranslationLoader
|
|
from app.i18n.cache import TranslationCache
|
|
from app.i18n.metrics import get_metrics, monitor_performance, track_missing_translation, track_translation_request
|
|
from app.i18n.logger import get_translation_logger
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class TranslationService:
|
|
"""
|
|
Translation service that provides:
|
|
- Fast translation lookup with memory cache
|
|
- Parameterized message support ({param} syntax)
|
|
- Fallback mechanism (current locale → default locale → key)
|
|
- Enum value translation
|
|
- Deep merge of multi-directory translations
|
|
"""
|
|
|
|
def __init__(self, locales_dirs: Optional[list] = None):
|
|
"""
|
|
Initialize the translation service.
|
|
|
|
Args:
|
|
locales_dirs: List of directories containing translation files.
|
|
If None, will auto-detect from settings.
|
|
"""
|
|
from app.core.config import settings
|
|
|
|
self.loader = TranslationLoader(locales_dirs)
|
|
self.default_locale = settings.I18N_DEFAULT_LANGUAGE
|
|
self.fallback_locale = settings.I18N_FALLBACK_LANGUAGE
|
|
self.log_missing = settings.I18N_LOG_MISSING_TRANSLATIONS
|
|
self.enable_cache = settings.I18N_ENABLE_TRANSLATION_CACHE
|
|
|
|
# Initialize advanced cache with LRU
|
|
lru_cache_size = getattr(settings, 'I18N_LRU_CACHE_SIZE', 1000)
|
|
self.cache = TranslationCache(
|
|
max_lru_size=lru_cache_size,
|
|
enable_lazy_load=False # Load all at startup for now
|
|
)
|
|
|
|
# Load all translations into cache
|
|
self._load_all_locales()
|
|
|
|
# Initialize metrics
|
|
self.metrics = get_metrics()
|
|
|
|
# Initialize translation logger
|
|
self.translation_logger = get_translation_logger()
|
|
|
|
logger.info(
|
|
f"TranslationService initialized with default locale: {self.default_locale}, "
|
|
f"LRU cache size: {lru_cache_size}"
|
|
)
|
|
|
|
def _load_all_locales(self):
|
|
"""Load all available locales into memory cache."""
|
|
available_locales = self.loader.get_available_locales()
|
|
logger.info(f"Loading translations for locales: {available_locales}")
|
|
|
|
for locale in available_locales:
|
|
locale_data = self.loader.load_locale(locale)
|
|
self.cache.set_locale_data(locale, locale_data)
|
|
|
|
logger.info(f"Loaded {len(available_locales)} locales into cache")
|
|
|
|
@monitor_performance("translate")
|
|
def translate(
|
|
self,
|
|
key: str,
|
|
locale: Optional[str] = None,
|
|
**params
|
|
) -> str:
|
|
"""
|
|
Translate a key to the target locale.
|
|
|
|
Supports:
|
|
- Dot-separated keys (e.g., "common.success.created")
|
|
- Parameterized messages (e.g., "Hello {name}")
|
|
- Fallback mechanism
|
|
|
|
Args:
|
|
key: Translation key (format: "namespace.key.subkey")
|
|
locale: Target locale (defaults to default locale)
|
|
**params: Parameters for parameterized messages
|
|
|
|
Returns:
|
|
Translated string, or the key itself if translation not found
|
|
|
|
Examples:
|
|
translate("common.success.created", "zh")
|
|
# => "创建成功"
|
|
|
|
translate("common.validation.required", "zh", field="名称")
|
|
# => "名称不能为空"
|
|
"""
|
|
if locale is None:
|
|
locale = self.default_locale
|
|
|
|
# Parse key (namespace.key.subkey)
|
|
parts = key.split(".", 1)
|
|
if len(parts) < 2:
|
|
if self.log_missing:
|
|
logger.warning(f"Invalid translation key format: {key}")
|
|
return key
|
|
|
|
namespace = parts[0]
|
|
key_path = parts[1].split(".")
|
|
|
|
# Track request
|
|
track_translation_request(locale, namespace)
|
|
|
|
# Get translation from cache
|
|
translation = self.cache.get_translation(locale, namespace, key_path)
|
|
|
|
# Fallback to default locale if not found
|
|
if translation is None and locale != self.fallback_locale:
|
|
translation = self.cache.get_translation(
|
|
self.fallback_locale, namespace, key_path
|
|
)
|
|
|
|
# If still not found, return the key itself
|
|
if translation is None:
|
|
if self.log_missing:
|
|
logger.warning(
|
|
f"Missing translation: {key} (locale: {locale})"
|
|
)
|
|
track_missing_translation(key, locale)
|
|
|
|
# Log to translation logger with context
|
|
self.translation_logger.log_missing_translation(
|
|
key=key,
|
|
locale=locale,
|
|
context={"namespace": namespace}
|
|
)
|
|
return key
|
|
|
|
# Apply parameters if provided
|
|
if params:
|
|
try:
|
|
translation = translation.format(**params)
|
|
except KeyError as e:
|
|
error_msg = f"Missing parameter in translation '{key}': {e}"
|
|
logger.error(error_msg)
|
|
self.translation_logger.log_translation_error(
|
|
error_type="parameter_missing",
|
|
message=error_msg,
|
|
key=key,
|
|
locale=locale,
|
|
context={"params": list(params.keys())}
|
|
)
|
|
except Exception as e:
|
|
error_msg = f"Error formatting translation '{key}': {e}"
|
|
logger.error(error_msg)
|
|
self.translation_logger.log_translation_error(
|
|
error_type="format_error",
|
|
message=error_msg,
|
|
key=key,
|
|
locale=locale
|
|
)
|
|
|
|
return translation
|
|
|
|
def _get_translation(
|
|
self,
|
|
locale: str,
|
|
namespace: str,
|
|
key_path: list
|
|
) -> Optional[str]:
|
|
"""
|
|
Get translation from cache (deprecated, use cache.get_translation).
|
|
|
|
Args:
|
|
locale: Locale code
|
|
namespace: Translation namespace
|
|
key_path: List of nested keys
|
|
|
|
Returns:
|
|
Translation string or None if not found
|
|
"""
|
|
return self.cache.get_translation(locale, namespace, key_path)
|
|
|
|
@monitor_performance("translate_enum")
|
|
def translate_enum(
|
|
self,
|
|
enum_type: str,
|
|
value: str,
|
|
locale: Optional[str] = None
|
|
) -> str:
|
|
"""
|
|
Translate an enum value.
|
|
|
|
Args:
|
|
enum_type: Enum type name (e.g., "workspace_role")
|
|
value: Enum value (e.g., "manager")
|
|
locale: Target locale
|
|
|
|
Returns:
|
|
Translated enum display name
|
|
|
|
Examples:
|
|
translate_enum("workspace_role", "manager", "zh")
|
|
# => "管理员"
|
|
|
|
translate_enum("invite_status", "pending", "en")
|
|
# => "Pending"
|
|
"""
|
|
key = f"enums.{enum_type}.{value}"
|
|
return self.translate(key, locale)
|
|
|
|
def has_translation(self, key: str, locale: str) -> bool:
|
|
"""
|
|
Check if a translation exists for the given key and locale.
|
|
|
|
Args:
|
|
key: Translation key
|
|
locale: Locale code
|
|
|
|
Returns:
|
|
True if translation exists, False otherwise
|
|
"""
|
|
parts = key.split(".", 1)
|
|
if len(parts) < 2:
|
|
return False
|
|
|
|
namespace = parts[0]
|
|
key_path = parts[1].split(".")
|
|
|
|
translation = self.cache.get_translation(locale, namespace, key_path)
|
|
return translation is not None
|
|
|
|
def reload(self, locale: Optional[str] = None):
|
|
"""
|
|
Reload translation files.
|
|
|
|
Args:
|
|
locale: Specific locale to reload. If None, reloads all locales.
|
|
"""
|
|
logger.info(f"Reloading translations for locale: {locale or 'all'}")
|
|
|
|
if locale:
|
|
locale_data = self.loader.load_locale(locale)
|
|
self.cache.set_locale_data(locale, locale_data)
|
|
# Clear LRU cache for this locale
|
|
self.cache.clear_locale(locale)
|
|
else:
|
|
self._load_all_locales()
|
|
# Clear all LRU cache
|
|
self.cache.clear_lru()
|
|
|
|
logger.info("Translation reload completed")
|
|
|
|
def get_available_locales(self) -> list:
|
|
"""
|
|
Get list of all available locales.
|
|
|
|
Returns:
|
|
List of locale codes
|
|
"""
|
|
return self.cache.get_loaded_locales()
|
|
|
|
def get_cache_stats(self) -> Dict[str, Any]:
|
|
"""
|
|
Get cache statistics.
|
|
|
|
Returns:
|
|
Dictionary with cache statistics
|
|
"""
|
|
return self.cache.get_stats()
|
|
|
|
def get_metrics_summary(self) -> Dict[str, Any]:
|
|
"""
|
|
Get metrics summary.
|
|
|
|
Returns:
|
|
Dictionary with metrics summary
|
|
"""
|
|
return self.metrics.get_summary()
|
|
|
|
def get_memory_usage(self) -> Dict[str, Any]:
|
|
"""
|
|
Get memory usage information.
|
|
|
|
Returns:
|
|
Dictionary with memory usage information
|
|
"""
|
|
return self.cache.get_memory_usage()
|
|
|
|
def get_loaded_dirs(self) -> list:
|
|
"""
|
|
Get list of loaded translation directories.
|
|
|
|
Returns:
|
|
List of directory paths
|
|
"""
|
|
return self.loader.locales_dirs
|
|
|
|
|
|
# Global singleton instance
|
|
_translation_service: Optional[TranslationService] = None
|
|
|
|
|
|
def get_translation_service() -> TranslationService:
|
|
"""
|
|
Get the global translation service instance.
|
|
|
|
Returns:
|
|
TranslationService singleton
|
|
"""
|
|
global _translation_service
|
|
if _translation_service is None:
|
|
_translation_service = TranslationService()
|
|
return _translation_service
|
|
|
|
|
|
# Convenience functions for easy access
|
|
def t(key: str, locale: Optional[str] = None, **params) -> str:
|
|
"""
|
|
Translate a key (convenience function).
|
|
|
|
Args:
|
|
key: Translation key
|
|
locale: Target locale (optional, uses default if not provided)
|
|
**params: Parameters for parameterized messages
|
|
|
|
Returns:
|
|
Translated string
|
|
|
|
Examples:
|
|
t("common.success.created")
|
|
t("common.validation.required", field="名称")
|
|
t("workspace.member_count", count=5)
|
|
"""
|
|
service = get_translation_service()
|
|
return service.translate(key, locale, **params)
|
|
|
|
|
|
def t_enum(enum_type: str, value: str, locale: Optional[str] = None) -> str:
|
|
"""
|
|
Translate an enum value (convenience function).
|
|
|
|
Args:
|
|
enum_type: Enum type name
|
|
value: Enum value
|
|
locale: Target locale
|
|
|
|
Returns:
|
|
Translated enum display name
|
|
|
|
Examples:
|
|
t_enum("workspace_role", "manager")
|
|
t_enum("invite_status", "pending", "en")
|
|
"""
|
|
service = get_translation_service()
|
|
return service.translate_enum(enum_type, value, locale)
|