[add] i18n support zh,en

This commit is contained in:
Mark
2026-03-11 10:45:07 +08:00
parent f1207dc8b9
commit 4f5ee24bc5
44 changed files with 5730 additions and 75 deletions

61
api/app/i18n/README.md Normal file
View File

@@ -0,0 +1,61 @@
# Internationalization (i18n) Module
This module provides internationalization support for the MemoryBear API.
## Components
- `service.py` - Translation service and core translation logic
- `middleware.py` - Language detection middleware
- `dependencies.py` - FastAPI dependency injection functions
- `exceptions.py` - Internationalized exception classes
## Usage
### Basic Translation
```python
from app.i18n import t
# Simple translation
message = t("common.success.created")
# Parameterized translation
message = t("common.validation.required", field="Name")
```
### Enum Translation
```python
from app.i18n import t_enum
# Translate enum value
role_display = t_enum("workspace_role", "manager")
```
### In FastAPI Endpoints
```python
from fastapi import Depends
from app.i18n.dependencies import get_translator
@router.post("/workspaces")
async def create_workspace(
data: WorkspaceCreate,
t: Callable = Depends(get_translator)
):
workspace = await workspace_service.create(data)
return {
"success": True,
"message": t("workspace.created_successfully"),
"data": workspace
}
```
## Configuration
See `app/core/config.py` for i18n configuration options:
- `I18N_DEFAULT_LANGUAGE` - Default language (default: "zh")
- `I18N_SUPPORTED_LANGUAGES` - Supported languages (default: "zh,en")
- `I18N_ENABLE_TRANSLATION_CACHE` - Enable caching (default: true)
- `I18N_LOG_MISSING_TRANSLATIONS` - Log missing translations (default: true)

113
api/app/i18n/__init__.py Normal file
View File

@@ -0,0 +1,113 @@
"""
Internationalization (i18n) module for MemoryBear Enterprise.
This module provides complete i18n support for the backend API including:
- Translation loading from multiple directories (community + enterprise)
- Translation service with caching and fallback
- Language detection middleware
- Dependency injection for FastAPI
- Convenience functions for easy usage
Usage:
from app.i18n import t, t_enum
# Simple translation
message = t("common.success.created")
# Parameterized translation
error = t("common.validation.required", field="名称")
# Enum translation
role_display = t_enum("workspace_role", "manager")
"""
from app.i18n.dependencies import (
get_current_language,
get_enum_translator,
get_translator,
)
from app.i18n.exceptions import (
BadRequestError,
ConflictError,
FileNotFoundError,
FileTooLargeError,
ForbiddenError,
I18nException,
InternalServerError,
InvalidCredentialsError,
InvalidFileTypeError,
NotFoundError,
QuotaExceededError,
RateLimitExceededError,
ServiceUnavailableError,
TenantNotFoundError,
TenantSuspendedError,
TokenExpiredError,
TokenInvalidError,
UnauthorizedError,
UserAlreadyExistsError,
UserNotFoundError,
ValidationError,
WorkspaceNotFoundError,
WorkspacePermissionDeniedError,
get_current_locale,
set_current_locale,
)
from app.i18n.loader import TranslationLoader
from app.i18n.logger import (
TranslationLogger,
get_translation_logger,
log_missing_translation,
log_translation_error,
)
from app.i18n.middleware import LanguageMiddleware
from app.i18n.service import (
TranslationService,
get_translation_service,
t,
t_enum,
)
__all__ = [
"TranslationLoader",
"LanguageMiddleware",
"TranslationService",
"get_translation_service",
"t",
"t_enum",
"get_current_language",
"get_translator",
"get_enum_translator",
# Context management
"get_current_locale",
"set_current_locale",
# Logging
"TranslationLogger",
"get_translation_logger",
"log_missing_translation",
"log_translation_error",
# Exception classes
"I18nException",
"BadRequestError",
"UnauthorizedError",
"ForbiddenError",
"NotFoundError",
"ConflictError",
"ValidationError",
"InternalServerError",
"ServiceUnavailableError",
"WorkspaceNotFoundError",
"WorkspacePermissionDeniedError",
"UserNotFoundError",
"UserAlreadyExistsError",
"TenantNotFoundError",
"TenantSuspendedError",
"InvalidCredentialsError",
"TokenExpiredError",
"TokenInvalidError",
"FileNotFoundError",
"FileTooLargeError",
"InvalidFileTypeError",
"RateLimitExceededError",
"QuotaExceededError",
]

291
api/app/i18n/cache.py Normal file
View File

@@ -0,0 +1,291 @@
"""
Advanced caching system for i18n translations.
This module provides:
- LRU cache for hot translations
- Lazy loading mechanism
- Memory optimization
- Cache statistics
"""
import logging
from functools import lru_cache
from typing import Any, Dict, Optional
from collections import OrderedDict
import time
logger = logging.getLogger(__name__)
class TranslationCache:
"""
Advanced translation cache with LRU eviction and lazy loading.
Features:
- LRU cache for frequently accessed translations
- Lazy loading to reduce startup time
- Memory-efficient storage
- Cache hit/miss statistics
"""
def __init__(self, max_lru_size: int = 1000, enable_lazy_load: bool = True):
"""
Initialize the translation cache.
Args:
max_lru_size: Maximum size of LRU cache for hot translations
enable_lazy_load: Enable lazy loading of locales
"""
self.max_lru_size = max_lru_size
self.enable_lazy_load = enable_lazy_load
# Main cache: {locale: {namespace: {key: value}}}
self._main_cache: Dict[str, Dict[str, Any]] = {}
# LRU cache for hot translations
self._lru_cache: OrderedDict = OrderedDict()
# Loaded locales tracker
self._loaded_locales: set = set()
# Statistics
self._stats = {
"hits": 0,
"misses": 0,
"lru_hits": 0,
"lru_misses": 0,
"lazy_loads": 0
}
logger.info(
f"TranslationCache initialized with LRU size: {max_lru_size}, "
f"lazy loading: {enable_lazy_load}"
)
def set_locale_data(self, locale: str, data: Dict[str, Any]):
"""
Set translation data for a locale.
Args:
locale: Locale code
data: Translation data dictionary
"""
self._main_cache[locale] = data
self._loaded_locales.add(locale)
logger.debug(f"Loaded locale '{locale}' into cache")
def get_translation(
self,
locale: str,
namespace: str,
key_path: list
) -> Optional[str]:
"""
Get translation from cache with LRU optimization.
Args:
locale: Locale code
namespace: Translation namespace
key_path: List of nested keys
Returns:
Translation string or None if not found
"""
# Build cache key for LRU
cache_key = f"{locale}:{namespace}:{'.'.join(key_path)}"
# Check LRU cache first (hot translations)
if cache_key in self._lru_cache:
self._stats["lru_hits"] += 1
self._stats["hits"] += 1
# Move to end (most recently used)
self._lru_cache.move_to_end(cache_key)
return self._lru_cache[cache_key]
self._stats["lru_misses"] += 1
# Check main cache
if locale not in self._main_cache:
self._stats["misses"] += 1
return None
if namespace not in self._main_cache[locale]:
self._stats["misses"] += 1
return None
# Navigate through nested keys
current = self._main_cache[locale][namespace]
for key in key_path:
if isinstance(current, dict) and key in current:
current = current[key]
else:
self._stats["misses"] += 1
return None
# Return only if it's a string value
if not isinstance(current, str):
self._stats["misses"] += 1
return None
self._stats["hits"] += 1
# Add to LRU cache
self._add_to_lru(cache_key, current)
return current
def _add_to_lru(self, key: str, value: str):
"""
Add translation to LRU cache.
Args:
key: Cache key
value: Translation value
"""
# Remove oldest if cache is full
if len(self._lru_cache) >= self.max_lru_size:
self._lru_cache.popitem(last=False)
self._lru_cache[key] = value
def is_locale_loaded(self, locale: str) -> bool:
"""
Check if a locale is loaded.
Args:
locale: Locale code
Returns:
True if locale is loaded
"""
return locale in self._loaded_locales
def get_loaded_locales(self) -> list:
"""
Get list of loaded locales.
Returns:
List of locale codes
"""
return list(self._loaded_locales)
def clear_lru(self):
"""Clear the LRU cache."""
self._lru_cache.clear()
logger.info("LRU cache cleared")
def clear_locale(self, locale: str):
"""
Clear cache for a specific locale.
Args:
locale: Locale code
"""
if locale in self._main_cache:
del self._main_cache[locale]
self._loaded_locales.discard(locale)
# Clear related LRU entries
keys_to_remove = [k for k in self._lru_cache if k.startswith(f"{locale}:")]
for key in keys_to_remove:
del self._lru_cache[key]
logger.info(f"Cleared cache for locale '{locale}'")
def clear_all(self):
"""Clear all caches."""
self._main_cache.clear()
self._lru_cache.clear()
self._loaded_locales.clear()
logger.info("All caches cleared")
def get_stats(self) -> Dict[str, Any]:
"""
Get cache statistics.
Returns:
Dictionary with cache statistics
"""
total_requests = self._stats["hits"] + self._stats["misses"]
hit_rate = (
self._stats["hits"] / total_requests * 100
if total_requests > 0
else 0
)
lru_total = self._stats["lru_hits"] + self._stats["lru_misses"]
lru_hit_rate = (
self._stats["lru_hits"] / lru_total * 100
if lru_total > 0
else 0
)
return {
"total_requests": total_requests,
"hits": self._stats["hits"],
"misses": self._stats["misses"],
"hit_rate": round(hit_rate, 2),
"lru_hits": self._stats["lru_hits"],
"lru_misses": self._stats["lru_misses"],
"lru_hit_rate": round(lru_hit_rate, 2),
"lru_size": len(self._lru_cache),
"lru_max_size": self.max_lru_size,
"loaded_locales": len(self._loaded_locales),
"lazy_loads": self._stats["lazy_loads"]
}
def reset_stats(self):
"""Reset cache statistics."""
self._stats = {
"hits": 0,
"misses": 0,
"lru_hits": 0,
"lru_misses": 0,
"lazy_loads": 0
}
logger.info("Cache statistics reset")
def get_memory_usage(self) -> Dict[str, Any]:
"""
Estimate memory usage of the cache.
Returns:
Dictionary with memory usage information
"""
import sys
main_cache_size = sys.getsizeof(self._main_cache)
lru_cache_size = sys.getsizeof(self._lru_cache)
# Rough estimate of nested data
for locale_data in self._main_cache.values():
main_cache_size += sys.getsizeof(locale_data)
for namespace_data in locale_data.values():
main_cache_size += sys.getsizeof(namespace_data)
return {
"main_cache_bytes": main_cache_size,
"lru_cache_bytes": lru_cache_size,
"total_bytes": main_cache_size + lru_cache_size,
"main_cache_mb": round(main_cache_size / 1024 / 1024, 2),
"lru_cache_mb": round(lru_cache_size / 1024 / 1024, 2),
"total_mb": round((main_cache_size + lru_cache_size) / 1024 / 1024, 2)
}
@lru_cache(maxsize=128)
def get_cached_translation_key(locale: str, namespace: str, key: str) -> str:
"""
LRU cached function for building translation cache keys.
This reduces string concatenation overhead for frequently accessed keys.
Args:
locale: Locale code
namespace: Translation namespace
key: Translation key
Returns:
Cache key string
"""
return f"{locale}:{namespace}:{key}"

View File

@@ -0,0 +1,158 @@
"""
FastAPI dependency injection functions for i18n.
This module provides dependency injection functions that can be used
in FastAPI route handlers to access the current language and translator.
"""
import logging
from typing import Callable
from fastapi import Request
from app.i18n.service import get_translation_service
logger = logging.getLogger(__name__)
async def get_current_language(request: Request) -> str:
"""
Get the current language from the request context.
This dependency extracts the language that was determined by the
LanguageMiddleware and stored in request.state.
Args:
request: FastAPI request object
Returns:
Language code (e.g., "zh", "en")
Usage:
@router.get("/example")
async def example(language: str = Depends(get_current_language)):
return {"language": language}
"""
# Get language from request state (set by LanguageMiddleware)
language = getattr(request.state, "language", None)
if language is None:
# Fallback to default language if not set
from app.core.config import settings
language = settings.I18N_DEFAULT_LANGUAGE
logger.warning(
"Language not found in request.state, using default: "
f"{language}"
)
return language
async def get_translator(request: Request) -> Callable:
"""
Get a translator function bound to the current request's language.
This dependency returns a translation function that automatically
uses the current request's language, making it easy to translate
strings in route handlers.
Args:
request: FastAPI request object
Returns:
Translation function with signature: t(key: str, **params) -> str
Usage:
@router.post("/workspaces")
async def create_workspace(
data: WorkspaceCreate,
t: Callable = Depends(get_translator)
):
workspace = await workspace_service.create(data)
return {
"success": True,
"message": t("workspace.created_successfully"),
"data": workspace
}
# With parameters
@router.get("/items")
async def get_items(t: Callable = Depends(get_translator)):
count = 5
return {
"message": t("items.found", count=count)
}
"""
# Get current language
language = await get_current_language(request)
# Get translation service
service = get_translation_service()
# Return a bound translation function
def translate(key: str, **params) -> str:
"""
Translate a key using the current request's language.
Args:
key: Translation key (e.g., "common.success.created")
**params: Parameters for parameterized messages
Returns:
Translated string
"""
return service.translate(key, language, **params)
return translate
async def get_enum_translator(request: Request) -> Callable:
"""
Get an enum translator function bound to the current request's language.
This dependency returns a function for translating enum values
that automatically uses the current request's language.
Args:
request: FastAPI request object
Returns:
Enum translation function with signature:
t_enum(enum_type: str, value: str) -> str
Usage:
@router.get("/workspace/{id}")
async def get_workspace(
id: str,
t_enum: Callable = Depends(get_enum_translator)
):
workspace = await workspace_service.get(id)
return {
"id": workspace.id,
"role": workspace.role,
"role_display": t_enum("workspace_role", workspace.role),
"status": workspace.status,
"status_display": t_enum("workspace_status", workspace.status)
}
"""
# Get current language
language = await get_current_language(request)
# Get translation service
service = get_translation_service()
# Return a bound enum translation function
def translate_enum(enum_type: str, value: str) -> str:
"""
Translate an enum value using the current request's language.
Args:
enum_type: Enum type name (e.g., "workspace_role")
value: Enum value (e.g., "manager")
Returns:
Translated enum display name
"""
return service.translate_enum(enum_type, value, language)
return translate_enum

495
api/app/i18n/exceptions.py Normal file
View File

@@ -0,0 +1,495 @@
"""
Internationalized exception classes for i18n system.
This module provides exception classes that automatically translate
error messages based on the current request's language.
"""
import logging
from contextvars import ContextVar
from typing import Any, Dict, Optional
from fastapi import HTTPException, Request
from app.i18n.service import get_translation_service
logger = logging.getLogger(__name__)
# Context variable to store current locale
_current_locale: ContextVar[Optional[str]] = ContextVar("current_locale", default=None)
def set_current_locale(locale: str) -> None:
"""
Set the current locale in the context variable.
This should be called by the LanguageMiddleware.
Args:
locale: Locale code (e.g., "zh", "en")
"""
_current_locale.set(locale)
def get_current_locale() -> Optional[str]:
"""
Get the current locale from the context variable.
Returns:
Locale code or None if not set
"""
return _current_locale.get()
class I18nException(HTTPException):
"""
Base exception class with automatic i18n support.
This exception automatically translates error messages based on:
1. The current request's language (from request.state.language)
2. The fallback language if request language is not available
3. The error key itself if no translation is found
Features:
- Automatic error message translation
- Parameterized error messages support
- Consistent error response format
- Language-aware error handling
Usage:
# Simple error
raise I18nException(
error_key="errors.workspace.not_found",
status_code=404
)
# Error with parameters
raise I18nException(
error_key="errors.validation.missing_field",
status_code=400,
field="name"
)
# Custom error code
raise I18nException(
error_key="errors.workspace.not_found",
error_code="WORKSPACE_NOT_FOUND",
status_code=404,
workspace_id="123"
)
"""
def __init__(
self,
error_key: str,
status_code: int = 400,
error_code: Optional[str] = None,
locale: Optional[str] = None,
headers: Optional[Dict[str, str]] = None,
**params
):
"""
Initialize the i18n exception.
Args:
error_key: Translation key for the error message
(e.g., "errors.workspace.not_found")
status_code: HTTP status code (default: 400)
error_code: Custom error code for API clients
(default: derived from error_key)
locale: Target locale for translation (optional)
If not provided, uses current request's language
headers: Additional HTTP headers
**params: Parameters for parameterized error messages
"""
self.error_key = error_key
self.error_code = error_code or self._generate_error_code(error_key)
self.params = params
# Get locale from request context if not provided
if locale is None:
locale = self._get_current_locale()
# Translate error message
translation_service = get_translation_service()
message = translation_service.translate(
error_key,
locale,
**params
)
# Build error detail
detail = {
"error_code": self.error_code,
"message": message,
}
# Add parameters to detail if provided
if params:
detail["params"] = params
# Initialize HTTPException
super().__init__(
status_code=status_code,
detail=detail,
headers=headers
)
logger.debug(
f"I18nException raised: {self.error_code} "
f"(key: {error_key}, locale: {locale})"
)
def _get_current_locale(self) -> str:
"""
Get the current locale from request context.
Returns:
Locale code (e.g., "zh", "en")
"""
try:
# Try to get locale from context variable
locale = _current_locale.get()
if locale:
return locale
except Exception as e:
logger.debug(f"Could not get locale from context: {e}")
# Fallback to default locale
from app.core.config import settings
return settings.I18N_DEFAULT_LANGUAGE
def _generate_error_code(self, error_key: str) -> str:
"""
Generate error code from error key.
Converts "errors.workspace.not_found" to "WORKSPACE_NOT_FOUND"
Args:
error_key: Translation key
Returns:
Error code in UPPER_SNAKE_CASE
"""
# Remove "errors." prefix if present
if error_key.startswith("errors."):
error_key = error_key[7:]
# Convert to UPPER_SNAKE_CASE
parts = error_key.split(".")
return "_".join(parts).upper()
# Specific exception classes for common errors
class BadRequestError(I18nException):
"""Bad request error (400)."""
def __init__(
self,
error_key: str = "errors.common.bad_request",
error_code: Optional[str] = None,
**params
):
super().__init__(
error_key=error_key,
status_code=400,
error_code=error_code,
**params
)
class UnauthorizedError(I18nException):
"""Unauthorized error (401)."""
def __init__(
self,
error_key: str = "errors.auth.unauthorized",
error_code: Optional[str] = None,
**params
):
super().__init__(
error_key=error_key,
status_code=401,
error_code=error_code,
**params
)
class ForbiddenError(I18nException):
"""Forbidden error (403)."""
def __init__(
self,
error_key: str = "errors.auth.forbidden",
error_code: Optional[str] = None,
**params
):
super().__init__(
error_key=error_key,
status_code=403,
error_code=error_code,
**params
)
class NotFoundError(I18nException):
"""Not found error (404)."""
def __init__(
self,
error_key: str = "errors.common.not_found",
error_code: Optional[str] = None,
**params
):
super().__init__(
error_key=error_key,
status_code=404,
error_code=error_code,
**params
)
class ConflictError(I18nException):
"""Conflict error (409)."""
def __init__(
self,
error_key: str = "errors.common.conflict",
error_code: Optional[str] = None,
**params
):
super().__init__(
error_key=error_key,
status_code=409,
error_code=error_code,
**params
)
class ValidationError(I18nException):
"""Validation error (422)."""
def __init__(
self,
error_key: str = "errors.common.validation_failed",
error_code: Optional[str] = None,
**params
):
super().__init__(
error_key=error_key,
status_code=422,
error_code=error_code,
**params
)
class InternalServerError(I18nException):
"""Internal server error (500)."""
def __init__(
self,
error_key: str = "errors.common.internal_error",
error_code: Optional[str] = None,
**params
):
super().__init__(
error_key=error_key,
status_code=500,
error_code=error_code,
**params
)
class ServiceUnavailableError(I18nException):
"""Service unavailable error (503)."""
def __init__(
self,
error_key: str = "errors.common.service_unavailable",
error_code: Optional[str] = None,
**params
):
super().__init__(
error_key=error_key,
status_code=503,
error_code=error_code,
**params
)
# Domain-specific exception classes
class WorkspaceNotFoundError(NotFoundError):
"""Workspace not found error."""
def __init__(self, workspace_id: Optional[str] = None, **params):
if workspace_id:
params["workspace_id"] = workspace_id
super().__init__(
error_key="errors.workspace.not_found",
error_code="WORKSPACE_NOT_FOUND",
**params
)
class WorkspacePermissionDeniedError(ForbiddenError):
"""Workspace permission denied error."""
def __init__(self, workspace_id: Optional[str] = None, **params):
if workspace_id:
params["workspace_id"] = workspace_id
super().__init__(
error_key="errors.workspace.permission_denied",
error_code="WORKSPACE_PERMISSION_DENIED",
**params
)
class UserNotFoundError(NotFoundError):
"""User not found error."""
def __init__(self, user_id: Optional[str] = None, **params):
if user_id:
params["user_id"] = user_id
super().__init__(
error_key="errors.user.not_found",
error_code="USER_NOT_FOUND",
**params
)
class UserAlreadyExistsError(ConflictError):
"""User already exists error."""
def __init__(self, identifier: Optional[str] = None, **params):
if identifier:
params["identifier"] = identifier
super().__init__(
error_key="errors.user.already_exists",
error_code="USER_ALREADY_EXISTS",
**params
)
class TenantNotFoundError(NotFoundError):
"""Tenant not found error."""
def __init__(self, tenant_id: Optional[str] = None, **params):
if tenant_id:
params["tenant_id"] = tenant_id
super().__init__(
error_key="errors.tenant.not_found",
error_code="TENANT_NOT_FOUND",
**params
)
class TenantSuspendedError(ForbiddenError):
"""Tenant suspended error."""
def __init__(self, tenant_id: Optional[str] = None, **params):
if tenant_id:
params["tenant_id"] = tenant_id
super().__init__(
error_key="errors.tenant.suspended",
error_code="TENANT_SUSPENDED",
**params
)
class InvalidCredentialsError(UnauthorizedError):
"""Invalid credentials error."""
def __init__(self, **params):
super().__init__(
error_key="errors.auth.invalid_credentials",
error_code="INVALID_CREDENTIALS",
**params
)
class TokenExpiredError(UnauthorizedError):
"""Token expired error."""
def __init__(self, **params):
super().__init__(
error_key="errors.auth.token_expired",
error_code="TOKEN_EXPIRED",
**params
)
class TokenInvalidError(UnauthorizedError):
"""Token invalid error."""
def __init__(self, **params):
super().__init__(
error_key="errors.auth.token_invalid",
error_code="TOKEN_INVALID",
**params
)
class FileNotFoundError(NotFoundError):
"""File not found error."""
def __init__(self, file_id: Optional[str] = None, **params):
if file_id:
params["file_id"] = file_id
super().__init__(
error_key="errors.file.not_found",
error_code="FILE_NOT_FOUND",
**params
)
class FileTooLargeError(BadRequestError):
"""File too large error."""
def __init__(self, max_size: Optional[str] = None, **params):
if max_size:
params["max_size"] = max_size
super().__init__(
error_key="errors.file.too_large",
error_code="FILE_TOO_LARGE",
**params
)
class InvalidFileTypeError(BadRequestError):
"""Invalid file type error."""
def __init__(self, file_type: Optional[str] = None, **params):
if file_type:
params["file_type"] = file_type
super().__init__(
error_key="errors.file.invalid_type",
error_code="INVALID_FILE_TYPE",
**params
)
class RateLimitExceededError(I18nException):
"""Rate limit exceeded error (429)."""
def __init__(self, **params):
super().__init__(
error_key="errors.api.rate_limit_exceeded",
status_code=429,
error_code="RATE_LIMIT_EXCEEDED",
**params
)
class QuotaExceededError(ForbiddenError):
"""Quota exceeded error."""
def __init__(self, resource: Optional[str] = None, **params):
if resource:
params["resource"] = resource
super().__init__(
error_key="errors.api.quota_exceeded",
error_code="QUOTA_EXCEEDED",
**params
)

199
api/app/i18n/loader.py Normal file
View File

@@ -0,0 +1,199 @@
"""
Translation file loader for i18n system.
This module handles loading translation files from multiple directories
(community edition + enterprise edition) and provides hot reload support.
"""
import json
import logging
from pathlib import Path
from typing import Any, Dict, List, Optional
logger = logging.getLogger(__name__)
class TranslationLoader:
"""
Translation file loader that supports:
- Loading from multiple directories (community + enterprise)
- Hot reload of translation files
- Automatic locale detection
"""
def __init__(self, locales_dirs: Optional[List[str]] = None):
"""
Initialize the translation loader.
Args:
locales_dirs: List of directories containing translation files.
If None, will auto-detect from settings.
"""
if locales_dirs is None:
locales_dirs = self._detect_locales_dirs()
self.locales_dirs = [Path(d) for d in locales_dirs]
logger.info(f"TranslationLoader initialized with directories: {self.locales_dirs}")
def _detect_locales_dirs(self) -> List[str]:
"""
Auto-detect translation directories from settings.
Returns:
List of translation directory paths
"""
from app.core.config import settings
dirs = []
# 1. Core locales directory (community edition, required)
core_dir = Path(settings.I18N_CORE_LOCALES_DIR)
if core_dir.exists():
dirs.append(str(core_dir))
logger.debug(f"Found core locales directory: {core_dir}")
else:
logger.warning(f"Core locales directory not found: {core_dir}")
# 2. Premium locales directory (enterprise edition, optional)
if settings.I18N_PREMIUM_LOCALES_DIR:
premium_dir = Path(settings.I18N_PREMIUM_LOCALES_DIR)
if premium_dir.exists():
dirs.append(str(premium_dir))
logger.debug(f"Found premium locales directory: {premium_dir}")
else:
# Auto-detect premium directory
premium_dir = Path("premium/locales")
if premium_dir.exists():
dirs.append(str(premium_dir))
logger.debug(f"Auto-detected premium locales directory: {premium_dir}")
if not dirs:
logger.error("No translation directories found!")
return dirs
def get_available_locales(self) -> List[str]:
"""
Get list of all available locales across all directories.
Returns:
List of locale codes (e.g., ['zh', 'en'])
"""
locales = set()
for locales_dir in self.locales_dirs:
if not locales_dir.exists():
continue
for locale_dir in locales_dir.iterdir():
if locale_dir.is_dir() and not locale_dir.name.startswith('.'):
locales.add(locale_dir.name)
return sorted(list(locales))
def load_locale(self, locale: str) -> Dict[str, Any]:
"""
Load all translation files for a specific locale from all directories.
Translation files are merged with priority:
- Later directories override earlier directories
- Enterprise translations override community translations
Args:
locale: Locale code (e.g., 'zh', 'en')
Returns:
Dictionary of translations organized by namespace
Format: {namespace: {key: value, ...}, ...}
"""
translations = {}
# Load from each directory in order (later directories override earlier)
for locales_dir in self.locales_dirs:
locale_dir = locales_dir / locale
if not locale_dir.exists():
logger.debug(f"Locale directory not found: {locale_dir}")
continue
# Load all JSON files in this locale directory
for json_file in locale_dir.glob("*.json"):
namespace = json_file.stem
try:
with open(json_file, "r", encoding="utf-8") as f:
new_translations = json.load(f)
# Merge translations (deep merge)
if namespace in translations:
translations[namespace] = self._deep_merge(
translations[namespace],
new_translations
)
logger.debug(
f"Merged translations: {locale}/{namespace} from {json_file}"
)
else:
translations[namespace] = new_translations
logger.debug(
f"Loaded translations: {locale}/{namespace} from {json_file}"
)
except json.JSONDecodeError as e:
logger.error(
f"Failed to parse JSON file {json_file}: {e}"
)
except Exception as e:
logger.error(
f"Failed to load translation file {json_file}: {e}"
)
if not translations:
logger.warning(f"No translations found for locale: {locale}")
return translations
def reload(self, locale: Optional[str] = None) -> Dict[str, Dict[str, Any]]:
"""
Reload translation files.
Args:
locale: Specific locale to reload. If None, reloads all locales.
Returns:
Dictionary of reloaded translations
Format: {locale: {namespace: {key: value}}}
"""
if locale:
logger.info(f"Reloading translations for locale: {locale}")
return {locale: self.load_locale(locale)}
else:
logger.info("Reloading all translations")
all_translations = {}
for loc in self.get_available_locales():
all_translations[loc] = self.load_locale(loc)
return all_translations
def _deep_merge(self, base: Dict, override: Dict) -> Dict:
"""
Deep merge two dictionaries.
Args:
base: Base dictionary
override: Dictionary with values to override
Returns:
Merged dictionary
"""
result = base.copy()
for key, value in override.items():
if (
key in result
and isinstance(result[key], dict)
and isinstance(value, dict)
):
result[key] = self._deep_merge(result[key], value)
else:
result[key] = value
return result

382
api/app/i18n/logger.py Normal file
View File

@@ -0,0 +1,382 @@
"""
Translation logging for i18n system.
This module provides:
- TranslationLogger for recording missing translations
- Missing translation report generation
- Integration with existing logging system
- Structured logging for translation events
"""
import logging
from typing import Dict, List, Optional, Set
from datetime import datetime
from collections import defaultdict
from pathlib import Path
import json
from app.core.logging_config import get_logger
logger = get_logger(__name__)
class TranslationLogger:
"""
Logger for translation events and missing translations.
Features:
- Records missing translations with context
- Generates missing translation reports
- Integrates with existing logging system
- Provides structured logging for analysis
"""
def __init__(self, log_file: Optional[str] = None):
"""
Initialize translation logger.
Args:
log_file: Optional custom log file path for missing translations
"""
self.log_file = log_file or "logs/i18n/missing_translations.log"
self._missing_translations: Dict[str, Set[str]] = defaultdict(set)
self._missing_with_context: List[Dict] = []
self._max_context_entries = 10000 # Keep last 10k entries
# Ensure log directory exists
log_path = Path(self.log_file)
log_path.parent.mkdir(parents=True, exist_ok=True)
# Create dedicated file handler for missing translations
self._file_handler = logging.FileHandler(
self.log_file,
encoding='utf-8'
)
self._file_handler.setLevel(logging.WARNING)
# Create formatter
formatter = logging.Formatter(
fmt='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
self._file_handler.setFormatter(formatter)
# Create dedicated logger for missing translations
self._logger = logging.getLogger("i18n.missing_translations")
self._logger.setLevel(logging.WARNING)
self._logger.addHandler(self._file_handler)
self._logger.propagate = False # Don't propagate to root logger
logger.info(f"TranslationLogger initialized with log file: {self.log_file}")
def log_missing_translation(
self,
key: str,
locale: str,
context: Optional[Dict] = None
):
"""
Log a missing translation.
Args:
key: Translation key that was not found
locale: Locale code
context: Optional context information (e.g., request path, user info)
"""
# Add to missing set
self._missing_translations[locale].add(key)
# Create context entry
entry = {
"timestamp": datetime.now().isoformat(),
"key": key,
"locale": locale,
"context": context or {}
}
# Keep only recent entries to avoid memory bloat
if len(self._missing_with_context) >= self._max_context_entries:
self._missing_with_context.pop(0)
self._missing_with_context.append(entry)
# Log to file
context_str = f" (context: {context})" if context else ""
self._logger.warning(
f"Missing translation: key='{key}', locale='{locale}'{context_str}"
)
def log_translation_error(
self,
error_type: str,
message: str,
key: Optional[str] = None,
locale: Optional[str] = None,
context: Optional[Dict] = None
):
"""
Log a translation error.
Args:
error_type: Type of error (e.g., "format_error", "parameter_missing")
message: Error message
key: Translation key (optional)
locale: Locale code (optional)
context: Optional context information
"""
error_data = {
"error_type": error_type,
"message": message,
"key": key,
"locale": locale,
"context": context or {},
"timestamp": datetime.now().isoformat()
}
self._logger.error(
f"Translation error: {error_type} - {message} "
f"(key: {key}, locale: {locale})"
)
def log_translation_success(
self,
key: str,
locale: str,
duration_ms: Optional[float] = None
):
"""
Log a successful translation (debug level).
Args:
key: Translation key
locale: Locale code
duration_ms: Optional duration in milliseconds
"""
duration_str = f" ({duration_ms:.3f}ms)" if duration_ms else ""
logger.debug(
f"Translation success: key='{key}', locale='{locale}'{duration_str}"
)
def get_missing_translations(
self,
locale: Optional[str] = None
) -> Dict[str, List[str]]:
"""
Get missing translations.
Args:
locale: Specific locale (optional, returns all if None)
Returns:
Dictionary of missing translations by locale
"""
if locale:
return {locale: sorted(list(self._missing_translations.get(locale, set())))}
return {
loc: sorted(list(keys))
for loc, keys in self._missing_translations.items()
}
def get_missing_with_context(
self,
locale: Optional[str] = None,
limit: Optional[int] = None
) -> List[Dict]:
"""
Get missing translations with context.
Args:
locale: Filter by locale (optional)
limit: Maximum number of entries to return (optional)
Returns:
List of missing translation entries with context
"""
entries = self._missing_with_context
# Filter by locale if specified
if locale:
entries = [e for e in entries if e["locale"] == locale]
# Apply limit if specified
if limit:
entries = entries[-limit:]
return entries
def generate_report(
self,
locale: Optional[str] = None,
output_file: Optional[str] = None
) -> Dict:
"""
Generate a missing translation report.
Args:
locale: Specific locale (optional, generates for all if None)
output_file: Optional file path to save report as JSON
Returns:
Report dictionary
"""
missing = self.get_missing_translations(locale)
report = {
"generated_at": datetime.now().isoformat(),
"total_missing": sum(len(keys) for keys in missing.values()),
"missing_by_locale": {
loc: {
"count": len(keys),
"keys": keys
}
for loc, keys in missing.items()
},
"recent_context": self.get_missing_with_context(locale, limit=100)
}
# Save to file if specified
if output_file:
output_path = Path(output_file)
output_path.parent.mkdir(parents=True, exist_ok=True)
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(report, f, indent=2, ensure_ascii=False)
logger.info(f"Missing translation report saved to: {output_file}")
return report
def get_statistics(self) -> Dict:
"""
Get statistics about missing translations.
Returns:
Dictionary with statistics
"""
total_missing = sum(len(keys) for keys in self._missing_translations.values())
# Count by namespace
namespace_counts = defaultdict(int)
for locale, keys in self._missing_translations.items():
for key in keys:
namespace = key.split('.')[0] if '.' in key else 'unknown'
namespace_counts[namespace] += 1
return {
"total_missing": total_missing,
"locales_affected": len(self._missing_translations),
"missing_by_locale": {
loc: len(keys)
for loc, keys in self._missing_translations.items()
},
"missing_by_namespace": dict(namespace_counts),
"total_context_entries": len(self._missing_with_context)
}
def clear(self, locale: Optional[str] = None):
"""
Clear missing translation records.
Args:
locale: Specific locale to clear (optional, clears all if None)
"""
if locale:
self._missing_translations.pop(locale, None)
self._missing_with_context = [
e for e in self._missing_with_context
if e["locale"] != locale
]
logger.info(f"Cleared missing translations for locale: {locale}")
else:
self._missing_translations.clear()
self._missing_with_context.clear()
logger.info("Cleared all missing translations")
def export_to_json(self, output_file: str):
"""
Export all missing translations to JSON file.
Args:
output_file: Output file path
"""
data = {
"exported_at": datetime.now().isoformat(),
"missing_translations": self.get_missing_translations(),
"statistics": self.get_statistics(),
"recent_context": self.get_missing_with_context(limit=1000)
}
output_path = Path(output_file)
output_path.parent.mkdir(parents=True, exist_ok=True)
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2, ensure_ascii=False)
logger.info(f"Missing translations exported to: {output_file}")
def __del__(self):
"""Cleanup file handler on deletion."""
try:
if hasattr(self, '_file_handler'):
self._file_handler.close()
self._logger.removeHandler(self._file_handler)
except Exception:
pass
# Global translation logger instance
_translation_logger: Optional[TranslationLogger] = None
def get_translation_logger() -> TranslationLogger:
"""
Get the global translation logger instance.
Returns:
TranslationLogger singleton
"""
global _translation_logger
if _translation_logger is None:
_translation_logger = TranslationLogger()
return _translation_logger
def log_missing_translation(
key: str,
locale: str,
context: Optional[Dict] = None
):
"""
Log a missing translation (convenience function).
Args:
key: Translation key
locale: Locale code
context: Optional context information
"""
translation_logger = get_translation_logger()
translation_logger.log_missing_translation(key, locale, context)
def log_translation_error(
error_type: str,
message: str,
key: Optional[str] = None,
locale: Optional[str] = None,
context: Optional[Dict] = None
):
"""
Log a translation error (convenience function).
Args:
error_type: Type of error
message: Error message
key: Translation key (optional)
locale: Locale code (optional)
context: Optional context information
"""
translation_logger = get_translation_logger()
translation_logger.log_translation_error(
error_type, message, key, locale, context
)

337
api/app/i18n/metrics.py Normal file
View File

@@ -0,0 +1,337 @@
"""
Performance monitoring and metrics for i18n system.
This module provides:
- Translation request counters
- Translation timing metrics
- Missing translation tracking
- Performance monitoring decorators
- Prometheus-compatible metrics
"""
import logging
import time
from functools import wraps
from typing import Any, Callable, Dict, Optional
from collections import defaultdict
from datetime import datetime
logger = logging.getLogger(__name__)
class TranslationMetrics:
"""
Metrics collector for translation operations.
Tracks:
- Translation request counts
- Translation timing (latency)
- Missing translations
- Cache performance
- Locale usage
"""
def __init__(self):
"""Initialize metrics collector."""
# Request counters by locale
self._request_counts: Dict[str, int] = defaultdict(int)
# Missing translation tracker
self._missing_translations: Dict[str, set] = defaultdict(set)
# Timing metrics (in milliseconds)
self._timing_data: list = []
self._max_timing_samples = 10000 # Keep last 10k samples
# Locale usage
self._locale_usage: Dict[str, int] = defaultdict(int)
# Namespace usage
self._namespace_usage: Dict[str, int] = defaultdict(int)
# Error counts
self._error_counts: Dict[str, int] = defaultdict(int)
# Start time
self._start_time = datetime.now()
logger.info("TranslationMetrics initialized")
def record_request(self, locale: str, namespace: str = None):
"""
Record a translation request.
Args:
locale: Locale code
namespace: Translation namespace (optional)
"""
self._request_counts[locale] += 1
self._locale_usage[locale] += 1
if namespace:
self._namespace_usage[namespace] += 1
def record_missing(self, key: str, locale: str):
"""
Record a missing translation.
Args:
key: Translation key
locale: Locale code
"""
self._missing_translations[locale].add(key)
logger.debug(f"Missing translation recorded: {key} (locale: {locale})")
def record_timing(self, duration_ms: float, locale: str, operation: str = "translate"):
"""
Record translation operation timing.
Args:
duration_ms: Duration in milliseconds
locale: Locale code
operation: Operation type
"""
# Keep only recent samples to avoid memory bloat
if len(self._timing_data) >= self._max_timing_samples:
self._timing_data.pop(0)
self._timing_data.append({
"duration_ms": duration_ms,
"locale": locale,
"operation": operation,
"timestamp": time.time()
})
def record_error(self, error_type: str):
"""
Record an error.
Args:
error_type: Type of error
"""
self._error_counts[error_type] += 1
def get_summary(self) -> Dict[str, Any]:
"""
Get metrics summary.
Returns:
Dictionary with metrics summary
"""
total_requests = sum(self._request_counts.values())
total_missing = sum(len(keys) for keys in self._missing_translations.values())
# Calculate timing statistics
timing_stats = self._calculate_timing_stats()
# Calculate uptime
uptime_seconds = (datetime.now() - self._start_time).total_seconds()
return {
"uptime_seconds": round(uptime_seconds, 2),
"total_requests": total_requests,
"requests_per_locale": dict(self._request_counts),
"total_missing_translations": total_missing,
"missing_by_locale": {
locale: len(keys)
for locale, keys in self._missing_translations.items()
},
"timing": timing_stats,
"locale_usage": dict(self._locale_usage),
"namespace_usage": dict(self._namespace_usage),
"error_counts": dict(self._error_counts)
}
def _calculate_timing_stats(self) -> Dict[str, Any]:
"""
Calculate timing statistics.
Returns:
Dictionary with timing statistics
"""
if not self._timing_data:
return {
"count": 0,
"avg_ms": 0,
"min_ms": 0,
"max_ms": 0,
"p50_ms": 0,
"p95_ms": 0,
"p99_ms": 0
}
durations = [d["duration_ms"] for d in self._timing_data]
durations.sort()
count = len(durations)
avg = sum(durations) / count
# Calculate percentiles
p50_idx = int(count * 0.50)
p95_idx = int(count * 0.95)
p99_idx = int(count * 0.99)
return {
"count": count,
"avg_ms": round(avg, 3),
"min_ms": round(durations[0], 3),
"max_ms": round(durations[-1], 3),
"p50_ms": round(durations[p50_idx], 3),
"p95_ms": round(durations[p95_idx], 3),
"p99_ms": round(durations[p99_idx], 3)
}
def get_missing_translations(self, locale: Optional[str] = None) -> Dict[str, list]:
"""
Get missing translations.
Args:
locale: Specific locale (optional, returns all if None)
Returns:
Dictionary of missing translations by locale
"""
if locale:
return {locale: list(self._missing_translations.get(locale, set()))}
return {
locale: list(keys)
for locale, keys in self._missing_translations.items()
}
def reset(self):
"""Reset all metrics."""
self._request_counts.clear()
self._missing_translations.clear()
self._timing_data.clear()
self._locale_usage.clear()
self._namespace_usage.clear()
self._error_counts.clear()
self._start_time = datetime.now()
logger.info("Metrics reset")
def export_prometheus(self) -> str:
"""
Export metrics in Prometheus format.
Returns:
Prometheus-formatted metrics string
"""
lines = []
# Translation requests counter
lines.append("# HELP i18n_translation_requests_total Total number of translation requests")
lines.append("# TYPE i18n_translation_requests_total counter")
for locale, count in self._request_counts.items():
lines.append(f'i18n_translation_requests_total{{locale="{locale}"}} {count}')
# Missing translations counter
lines.append("# HELP i18n_missing_translations_total Total number of missing translations")
lines.append("# TYPE i18n_missing_translations_total counter")
for locale, keys in self._missing_translations.items():
lines.append(f'i18n_missing_translations_total{{locale="{locale}"}} {len(keys)}')
# Timing metrics
timing_stats = self._calculate_timing_stats()
lines.append("# HELP i18n_translation_duration_ms Translation operation duration in milliseconds")
lines.append("# TYPE i18n_translation_duration_ms summary")
lines.append(f'i18n_translation_duration_ms{{quantile="0.5"}} {timing_stats["p50_ms"]}')
lines.append(f'i18n_translation_duration_ms{{quantile="0.95"}} {timing_stats["p95_ms"]}')
lines.append(f'i18n_translation_duration_ms{{quantile="0.99"}} {timing_stats["p99_ms"]}')
lines.append(f'i18n_translation_duration_ms_sum {sum(d["duration_ms"] for d in self._timing_data)}')
lines.append(f'i18n_translation_duration_ms_count {timing_stats["count"]}')
# Error counter
lines.append("# HELP i18n_errors_total Total number of i18n errors")
lines.append("# TYPE i18n_errors_total counter")
for error_type, count in self._error_counts.items():
lines.append(f'i18n_errors_total{{type="{error_type}"}} {count}')
return "\n".join(lines)
# Global metrics instance
_metrics: Optional[TranslationMetrics] = None
def get_metrics() -> TranslationMetrics:
"""
Get the global metrics instance.
Returns:
TranslationMetrics singleton
"""
global _metrics
if _metrics is None:
_metrics = TranslationMetrics()
return _metrics
def monitor_performance(operation: str = "translate"):
"""
Decorator to monitor translation operation performance.
Args:
operation: Operation name for metrics
Returns:
Decorated function
Example:
@monitor_performance("translate")
def translate(key: str, locale: str) -> str:
...
"""
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.perf_counter()
try:
result = func(*args, **kwargs)
# Record timing
duration_ms = (time.perf_counter() - start_time) * 1000
# Try to extract locale from args/kwargs
locale = kwargs.get("locale", "unknown")
if not locale and len(args) > 1:
locale = args[1] if isinstance(args[1], str) else "unknown"
metrics = get_metrics()
metrics.record_timing(duration_ms, locale, operation)
return result
except Exception as e:
# Record error
metrics = get_metrics()
metrics.record_error(type(e).__name__)
raise
return wrapper
return decorator
def track_missing_translation(key: str, locale: str):
"""
Track a missing translation.
Args:
key: Translation key
locale: Locale code
"""
metrics = get_metrics()
metrics.record_missing(key, locale)
def track_translation_request(locale: str, namespace: str = None):
"""
Track a translation request.
Args:
locale: Locale code
namespace: Translation namespace (optional)
"""
metrics = get_metrics()
metrics.record_request(locale, namespace)

202
api/app/i18n/middleware.py Normal file
View File

@@ -0,0 +1,202 @@
"""
Language detection middleware for i18n system.
This middleware determines the language to use for each request based on:
1. Query parameter (?lang=en)
2. Accept-Language HTTP header
3. User language preference (from database)
4. Tenant default language
5. System default language
The detected language is injected into request.state.language and
added to the response Content-Language header.
"""
import logging
import re
from typing import Optional
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
logger = logging.getLogger(__name__)
class LanguageMiddleware(BaseHTTPMiddleware):
"""
Language detection middleware.
Determines the language for each request based on multiple sources
with a clear priority order, validates the language is supported,
and injects it into the request context.
"""
async def dispatch(self, request: Request, call_next):
"""
Process the request and determine the language.
Args:
request: The incoming request
call_next: The next middleware/handler in the chain
Returns:
Response with Content-Language header added
"""
# Determine the language for this request
language = await self._determine_language(request)
# Validate language is supported
from app.core.config import settings
if language not in settings.I18N_SUPPORTED_LANGUAGES:
logger.warning(
f"Unsupported language '{language}' requested, "
f"falling back to default: {settings.I18N_DEFAULT_LANGUAGE}"
)
language = settings.I18N_DEFAULT_LANGUAGE
# Inject language into request state
request.state.language = language
# Also set in context variable for exception handling
from app.i18n.exceptions import set_current_locale
set_current_locale(language)
logger.debug(f"Request language set to: {language}")
# Process the request
response = await call_next(request)
# Add Content-Language header to response
response.headers["Content-Language"] = language
return response
async def _determine_language(self, request: Request) -> str:
"""
Determine the language to use based on priority order.
Priority:
1. Query parameter (?lang=en)
2. Accept-Language HTTP header
3. User language preference (from database)
4. Tenant default language
5. System default language
Args:
request: The incoming request
Returns:
Language code (e.g., "zh", "en")
"""
from app.core.config import settings
# 1. Check query parameter (?lang=en)
if "lang" in request.query_params:
lang = request.query_params["lang"].strip().lower()
if lang:
logger.debug(f"Language from query parameter: {lang}")
return lang
# 2. Check Accept-Language HTTP header
if "Accept-Language" in request.headers:
lang = self._parse_accept_language(
request.headers["Accept-Language"]
)
if lang:
logger.debug(f"Language from Accept-Language header: {lang}")
return lang
# 3. Check user language preference (requires authentication)
# Note: This assumes user is already loaded into request.state by auth middleware
if hasattr(request.state, "user") and request.state.user:
user = request.state.user
if hasattr(user, "preferred_language") and user.preferred_language:
logger.debug(
f"Language from user preference: {user.preferred_language}"
)
return user.preferred_language
# 4. Check tenant default language
# Note: This assumes tenant is already loaded into request.state
if hasattr(request.state, "tenant") and request.state.tenant:
tenant = request.state.tenant
if hasattr(tenant, "default_language") and tenant.default_language:
logger.debug(
f"Language from tenant default: {tenant.default_language}"
)
return tenant.default_language
# 5. Fall back to system default language
logger.debug(
f"Using system default language: {settings.I18N_DEFAULT_LANGUAGE}"
)
return settings.I18N_DEFAULT_LANGUAGE
def _parse_accept_language(self, header: str) -> Optional[str]:
"""
Parse the Accept-Language HTTP header.
The Accept-Language header format:
Accept-Language: zh-CN,zh;q=0.9,en;q=0.8,en-US;q=0.7
This method:
1. Parses all language codes and their quality values
2. Extracts the base language code (zh-CN -> zh)
3. Sorts by quality value (higher first)
4. Returns the first supported language
Args:
header: Accept-Language header value
Returns:
Language code if found and supported, None otherwise
Examples:
_parse_accept_language("zh-CN,zh;q=0.9,en;q=0.8")
# => "zh" (if zh is supported)
_parse_accept_language("en-US,en;q=0.9")
# => "en" (if en is supported)
"""
from app.core.config import settings
if not header:
return None
# Parse language preferences with quality values
languages = []
for item in header.split(","):
item = item.strip()
if not item:
continue
# Split language code and quality value
parts = item.split(";")
lang_code = parts[0].strip()
# Extract base language code (zh-CN -> zh, en-US -> en)
base_lang = lang_code.split("-")[0].lower()
# Extract quality value (default: 1.0)
quality = 1.0
if len(parts) > 1:
# Look for q=0.9 pattern
q_match = re.search(r"q=([\d.]+)", parts[1])
if q_match:
try:
quality = float(q_match.group(1))
except ValueError:
quality = 1.0
languages.append((base_lang, quality))
# Sort by quality value (descending)
languages.sort(key=lambda x: x[1], reverse=True)
# Return the first supported language
for lang_code, _ in languages:
if lang_code in settings.I18N_SUPPORTED_LANGUAGES:
return lang_code
return None

219
api/app/i18n/serializers.py Normal file
View File

@@ -0,0 +1,219 @@
"""
国际化响应序列化器
提供基础的 I18nResponseMixin 类,用于为 API 响应添加国际化字段。
"""
from typing import Any, Dict, List, Union
from pydantic import BaseModel
class I18nResponseMixin:
"""国际化响应混入类
为响应数据添加国际化字段,特别是为枚举值添加 _display 后缀的翻译字段。
使用方法:
1. 继承此类
2. 实现 _get_enum_fields() 方法定义需要翻译的枚举字段
3. 调用 serialize_with_i18n() 方法序列化数据
示例:
class WorkspaceSerializer(I18nResponseMixin):
def _get_enum_fields(self) -> Dict[str, str]:
return {
"role": "workspace_role",
"status": "workspace_status"
}
def serialize(self, workspace: Workspace, locale: str = "zh") -> Dict:
data = {
"id": str(workspace.id),
"name": workspace.name,
"role": workspace.role,
"status": workspace.status
}
return self.serialize_with_i18n(data, locale)
"""
def serialize_with_i18n(
self,
data: Any,
locale: str = "zh"
) -> Union[Dict, List[Dict], Any]:
"""序列化数据并添加国际化字段
Args:
data: 要序列化的数据(字典、列表或 Pydantic 模型)
locale: 语言代码
Returns:
序列化后的数据,包含国际化字段
"""
# 如果是 Pydantic 模型,转换为字典
if isinstance(data, BaseModel):
data = data.model_dump()
# 处理不同类型的数据
if isinstance(data, dict):
return self._serialize_dict(data, locale)
elif isinstance(data, list):
return [self._serialize_dict(item, locale) if isinstance(item, dict) else item for item in data]
else:
return data
def _serialize_dict(self, data: Dict, locale: str) -> Dict:
"""序列化字典并添加 _display 字段
Args:
data: 字典数据
locale: 语言代码
Returns:
添加了 _display 字段的字典
"""
from app.i18n.service import translation_service
result = data.copy()
# 获取需要翻译的枚举字段
enum_fields = self._get_enum_fields()
# 为每个枚举字段添加 _display 字段
for field, enum_type in enum_fields.items():
if field in result and result[field] is not None:
value = result[field]
# 翻译枚举值
display_value = translation_service.translate_enum(
enum_type=enum_type,
value=str(value),
locale=locale
)
# 添加 _display 字段
result[f"{field}_display"] = display_value
return result
def _get_enum_fields(self) -> Dict[str, str]:
"""获取需要翻译的枚举字段
子类必须实现此方法,返回字段名到枚举类型的映射。
Returns:
字段名到枚举类型的映射
例如: {"role": "workspace_role", "status": "workspace_status"}
"""
return {}
class WorkspaceSerializer(I18nResponseMixin):
"""工作空间序列化器
为工作空间响应添加国际化字段。
"""
def _get_enum_fields(self) -> Dict[str, str]:
"""定义工作空间的枚举字段"""
return {
"role": "workspace_role",
"status": "workspace_status"
}
def serialize(self, workspace_data: Union[Dict, BaseModel], locale: str = "zh") -> Dict:
"""序列化工作空间数据
Args:
workspace_data: 工作空间数据(字典或 Pydantic 模型)
locale: 语言代码
Returns:
序列化后的工作空间数据,包含国际化字段
"""
return self.serialize_with_i18n(workspace_data, locale)
def serialize_list(self, workspaces: List[Union[Dict, BaseModel]], locale: str = "zh") -> List[Dict]:
"""序列化工作空间列表
Args:
workspaces: 工作空间列表
locale: 语言代码
Returns:
序列化后的工作空间列表
"""
return [self.serialize(ws, locale) for ws in workspaces]
class WorkspaceMemberSerializer(I18nResponseMixin):
"""工作空间成员序列化器
为工作空间成员响应添加国际化字段。
"""
def _get_enum_fields(self) -> Dict[str, str]:
"""定义工作空间成员的枚举字段"""
return {
"role": "workspace_role"
}
def serialize(self, member_data: Union[Dict, BaseModel], locale: str = "zh") -> Dict:
"""序列化工作空间成员数据
Args:
member_data: 成员数据(字典或 Pydantic 模型)
locale: 语言代码
Returns:
序列化后的成员数据,包含国际化字段
"""
return self.serialize_with_i18n(member_data, locale)
def serialize_list(self, members: List[Union[Dict, BaseModel]], locale: str = "zh") -> List[Dict]:
"""序列化工作空间成员列表
Args:
members: 成员列表
locale: 语言代码
Returns:
序列化后的成员列表
"""
return [self.serialize(member, locale) for member in members]
class WorkspaceInviteSerializer(I18nResponseMixin):
"""工作空间邀请序列化器
为工作空间邀请响应添加国际化字段。
"""
def _get_enum_fields(self) -> Dict[str, str]:
"""定义工作空间邀请的枚举字段"""
return {
"status": "invite_status",
"role": "workspace_role"
}
def serialize(self, invite_data: Union[Dict, BaseModel], locale: str = "zh") -> Dict:
"""序列化工作空间邀请数据
Args:
invite_data: 邀请数据(字典或 Pydantic 模型)
locale: 语言代码
Returns:
序列化后的邀请数据,包含国际化字段
"""
return self.serialize_with_i18n(invite_data, locale)
def serialize_list(self, invites: List[Union[Dict, BaseModel]], locale: str = "zh") -> List[Dict]:
"""序列化工作空间邀请列表
Args:
invites: 邀请列表
locale: 语言代码
Returns:
序列化后的邀请列表
"""
return [self.serialize(invite, locale) for invite in invites]

370
api/app/i18n/service.py Normal file
View File

@@ -0,0 +1,370 @@
"""
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)