i18n: crowdin integration

src/common/i18n/__init__.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+from collections.abc import Iterator
+from datetime import date, datetime
+from decimal import Decimal
+
+from .loaders import DEFAULT_LOCALE
+
+
+def _get_manager():
+    from .manager import I18nManager
+
+    manager = getattr(_get_manager, "_manager", None)
+    if manager is None:
+        manager = I18nManager()
+        _get_manager._manager = manager
+    return manager
+
+
+def set_locale(locale: str) -> str:
+    return _get_manager().set_locale(locale)
+
+
+def get_locale() -> str:
+    return _get_manager().get_locale()
+
+
+def reload_translations(locale: str | None = None) -> None:
+    _get_manager().reload(locale)
+
+
+def t(key: str, locale: str | None = None, **kwargs: object) -> str:
+    return _get_manager().t(key, locale=locale, **kwargs)
+
+
+def tn(key: str, count: int | float, locale: str | None = None, **kwargs: object) -> str:
+    return _get_manager().tn(key, count=count, locale=locale, **kwargs)
+
+
+def use_locale(locale: str) -> Iterator[None]:
+    return _get_manager().use_locale(locale)
+
+
+def format_datetime_localized(value: datetime | date, locale: str | None = None, format: str = "medium") -> str:
+    from .formatting import format_datetime_localized as _format_datetime_localized
+
+    return _format_datetime_localized(value, locale=locale or get_locale(), format=format)
+
+
+def format_number_localized(value: int | float | Decimal, locale: str | None = None) -> str:
+    from .formatting import format_number_localized as _format_number_localized
+
+    return _format_number_localized(value, locale=locale or get_locale())
+
+
+def format_decimal_localized(value: int | float | Decimal, locale: str | None = None) -> str:
+    from .formatting import format_decimal_localized as _format_decimal_localized
+
+    return _format_decimal_localized(value, locale=locale or get_locale())
+
+
+__all__ = [
+    "DEFAULT_LOCALE",
+    "format_datetime_localized",
+    "format_decimal_localized",
+    "format_number_localized",
+    "get_locale",
+    "reload_translations",
+    "set_locale",
+    "t",
+    "tn",
+    "use_locale",
+]

src/common/i18n/exceptions.py

@@ -0,0 +1,18 @@
+class I18nError(Exception):
+    """国际化基础异常。"""
+
+
+class InvalidLocaleError(I18nError):
+    """Locale 格式非法。"""
+
+
+class LocaleNotFoundError(I18nError):
+    """未找到指定 locale 的翻译目录。"""
+
+
+class InvalidTranslationFileError(I18nError):
+    """翻译文件结构非法。"""
+
+
+class DuplicateTranslationKeyError(I18nError):
+    """同一 locale 下存在重复的翻译 key。"""

src/common/i18n/formatting.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+from datetime import date, datetime, time
+from decimal import Decimal
+from string import Formatter
+
+from babel import Locale
+from babel.dates import format_datetime as babel_format_datetime
+from babel.numbers import format_decimal as babel_format_decimal
+
+from .loaders import DEFAULT_LOCALE, to_babel_locale
+
+FORMATTER = Formatter()
+
+
+def extract_placeholders(template: str) -> set[str]:
+    placeholders: set[str] = set()
+    for _, field_name, _, _ in FORMATTER.parse(template):
+        if not field_name:
+            continue
+        placeholders.add(field_name.split(".", maxsplit=1)[0].split("[", maxsplit=1)[0])
+    return placeholders
+
+
+def format_template(template: str, **kwargs: object) -> str:
+    return template.format(**kwargs)
+
+
+def select_plural_category(locale: str, count: int | float | Decimal) -> str:
+    babel_locale = Locale.parse(to_babel_locale(locale))
+    return str(babel_locale.plural_form(count))
+
+
+def format_datetime_localized(value: datetime | date, locale: str = DEFAULT_LOCALE, format: str = "medium") -> str:
+    if isinstance(value, date) and not isinstance(value, datetime):
+        value = datetime.combine(value, time.min)
+    return babel_format_datetime(value, format=format, locale=to_babel_locale(locale))
+
+
+def format_number_localized(value: int | float | Decimal, locale: str = DEFAULT_LOCALE) -> str:
+    return format_decimal_localized(value, locale=locale)
+
+
+def format_decimal_localized(value: int | float | Decimal, locale: str = DEFAULT_LOCALE) -> str:
+    return babel_format_decimal(value, locale=to_babel_locale(locale))

src/common/i18n/loaders.py

@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import json
+
+from .exceptions import (
+    DuplicateTranslationKeyError,
+    InvalidLocaleError,
+    InvalidTranslationFileError,
+    LocaleNotFoundError,
+)
+
+DEFAULT_LOCALE = "zh-CN"
+PLURAL_CATEGORIES = {"zero", "one", "two", "few", "many", "other"}
+TranslationValue = str | dict[str, str]
+
+
+def get_project_root() -> Path:
+    return Path(__file__).resolve().parents[3]
+
+
+def get_locales_root(locales_root: Path | None = None) -> Path:
+    if locales_root is not None:
+        return locales_root.resolve()
+    return (get_project_root() / "locales").resolve()
+
+
+def normalize_locale(locale: str) -> str:
+    cleaned_locale = locale.strip().replace("_", "-")
+    if not cleaned_locale:
+        raise InvalidLocaleError("Locale 不能为空")
+
+    parts = [part for part in cleaned_locale.split("-") if part]
+    if not parts:
+        raise InvalidLocaleError(f"Locale 非法: {locale}")
+
+    normalized_parts: list[str] = []
+    for index, part in enumerate(parts):
+        if index == 0:
+            normalized_parts.append(part.lower())
+        elif len(part) == 2:
+            normalized_parts.append(part.upper())
+        elif len(part) == 4:
+            normalized_parts.append(part.title())
+        else:
+            normalized_parts.append(part)
+    return "-".join(normalized_parts)
+
+
+def to_babel_locale(locale: str) -> str:
+    return normalize_locale(locale).replace("-", "_")
+
+
+def discover_locales(locales_root: Path | None = None) -> list[str]:
+    root = get_locales_root(locales_root)
+    if not root.exists():
+        return []
+
+    locale_names = [path.name for path in root.iterdir() if path.is_dir()]
+    return sorted(locale_names)
+
+
+def iter_locale_files(locale_dir: Path) -> list[Path]:
+    return sorted(path for path in locale_dir.glob("*.json") if path.is_file())
+
+
+def validate_translation_value(key: str, value: object, file_path: Path) -> TranslationValue:
+    if isinstance(value, str):
+        return value
+
+    if not isinstance(value, dict):
+        raise InvalidTranslationFileError(f"{file_path} 中的 key '{key}' 必须是字符串或 plural 对象")
+
+    if not value:
+        raise InvalidTranslationFileError(f"{file_path} 中的 key '{key}' 不能为空对象")
+
+    validated_value: dict[str, str] = {}
+    for category, category_value in value.items():
+        if category not in PLURAL_CATEGORIES:
+            raise InvalidTranslationFileError(f"{file_path} 中的 key '{key}' 使用了非法 plural category: '{category}'")
+        if not isinstance(category_value, str):
+            raise InvalidTranslationFileError(
+                f"{file_path} 中的 key '{key}' 的 plural category '{category}' 必须是字符串"
+            )
+        validated_value[category] = category_value
+    return validated_value
+
+
+def load_translation_file(file_path: Path) -> dict[str, TranslationValue]:
+    try:
+        raw_payload = json.loads(file_path.read_text(encoding="utf-8"))
+    except json.JSONDecodeError as exc:
+        raise InvalidTranslationFileError(f"{file_path} 不是合法 JSON: {exc}") from exc
+
+    if not isinstance(raw_payload, dict):
+        raise InvalidTranslationFileError(f"{file_path} 顶层必须是 JSON object")
+
+    translations: dict[str, TranslationValue] = {}
+    for raw_key, raw_value in raw_payload.items():
+        if not isinstance(raw_key, str):
+            raise InvalidTranslationFileError(f"{file_path} 中存在非字符串 key")
+        if not raw_key.strip():
+            raise InvalidTranslationFileError(f"{file_path} 中存在空字符串 key")
+        translations[raw_key] = validate_translation_value(raw_key, raw_value, file_path)
+    return translations
+
+
+def load_locale_catalog(locale: str, locales_root: Path | None = None) -> dict[str, TranslationValue]:
+    normalized_locale = normalize_locale(locale)
+    locale_dir = get_locales_root(locales_root) / normalized_locale
+    if not locale_dir.exists():
+        raise LocaleNotFoundError(f"未找到 locale 目录: {locale_dir}")
+
+    merged_translations: dict[str, TranslationValue] = {}
+    for file_path in iter_locale_files(locale_dir):
+        file_translations = load_translation_file(file_path)
+        for key, value in file_translations.items():
+            if key in merged_translations:
+                raise DuplicateTranslationKeyError(
+                    f"locale '{normalized_locale}' 中存在重复 key: '{key}'，冲突文件包含 {file_path.name}"
+                )
+            merged_translations[key] = value
+    return merged_translations

src/common/i18n/manager.py

@@ -0,0 +1,198 @@
+from __future__ import annotations
+
+from contextlib import contextmanager
+from contextvars import ContextVar
+from pathlib import Path
+from typing import Iterator
+
+import logging
+import os
+import threading
+
+from .exceptions import I18nError, InvalidLocaleError
+from .formatting import format_template, select_plural_category
+from .loaders import DEFAULT_LOCALE, TranslationValue, get_locales_root, load_locale_catalog, normalize_locale
+
+logger = logging.getLogger("maibot.i18n")
+
+
+class I18nManager:
+    """基于 JSON 的轻量级国际化管理器。"""
+
+    def __init__(self, default_locale: str = DEFAULT_LOCALE, locales_root: Path | None = None):
+        self._default_locale = normalize_locale(default_locale)
+        self._locales_root = get_locales_root(locales_root)
+        self._catalog_cache: dict[str, dict[str, TranslationValue]] = {}
+        self._locale_override: ContextVar[str | None] = ContextVar("maibot_locale", default=None)
+        self._warning_cache: set[tuple[str, str, str]] = set()
+        self._cache_lock = threading.RLock()
+
+    def set_locale(self, locale: str) -> str:
+        self._default_locale = normalize_locale(locale)
+        return self._default_locale
+
+    def get_locale(self) -> str:
+        override_locale = self._locale_override.get()
+        if override_locale:
+            return override_locale
+
+        env_locale = os.getenv("MAIBOT_LOCALE")
+        if env_locale:
+            try:
+                return normalize_locale(env_locale)
+            except InvalidLocaleError:
+                self._log_once(
+                    ("invalid_env_locale", "env", env_locale),
+                    logging.WARNING,
+                    "检测到非法 MAIBOT_LOCALE=%s，已回退到默认 locale %s",
+                    env_locale,
+                    self._default_locale,
+                )
+        return self._default_locale
+
+    @contextmanager
+    def use_locale(self, locale: str) -> Iterator[None]:
+        token = self._locale_override.set(normalize_locale(locale))
+        try:
+            yield
+        finally:
+            self._locale_override.reset(token)
+
+    def reload(self, locale: str | None = None) -> None:
+        with self._cache_lock:
+            if locale is None:
+                self._catalog_cache.clear()
+                return
+            self._catalog_cache.pop(normalize_locale(locale), None)
+
+    def t(self, key: str, locale: str | None = None, **kwargs: object) -> str:
+        translation_value, _ = self._get_translation_value(key, locale)
+        if translation_value is None:
+            return key
+
+        if isinstance(translation_value, dict):
+            template = translation_value.get("other")
+            if template is None:
+                self._log_once(
+                    ("plural_missing_other", self.get_locale(), key),
+                    logging.WARNING,
+                    "翻译 key '%s' 缺少 other plural category，已回退到 key 本身",
+                    key,
+                )
+                return key
+            return self._format_translation(key, template, kwargs)
+        return self._format_translation(key, translation_value, kwargs)
+
+    def tn(self, key: str, count: int | float, locale: str | None = None, **kwargs: object) -> str:
+        translation_value, translation_locale = self._get_translation_value(key, locale)
+        if translation_value is None:
+            return key
+
+        if not isinstance(translation_value, dict):
+            self._log_once(
+                ("non_plural_key", translation_locale, key),
+                logging.WARNING,
+                "翻译 key '%s' 不是 plural 节点，已回退到普通 t()",
+                key,
+            )
+            return self.t(key, locale=translation_locale, count=count, **kwargs)
+
+        try:
+            plural_category = select_plural_category(translation_locale, count)
+        except Exception as exc:
+            logger.warning("为 key '%s' 选择 plural category 失败: %s，已回退到 other", key, exc)
+            plural_category = "other"
+
+        template = translation_value.get(plural_category) or translation_value.get("other")
+        if template is None:
+            self._log_once(
+                ("plural_missing_template", translation_locale, key),
+                logging.WARNING,
+                "翻译 key '%s' 缺少 plural 模板，已回退到 key 本身",
+                key,
+            )
+            return key
+
+        formatting_kwargs = dict(kwargs)
+        formatting_kwargs["count"] = count
+        return self._format_translation(key, template, formatting_kwargs)
+
+    def _format_translation(self, key: str, template: str, kwargs: dict[str, object]) -> str:
+        try:
+            return format_template(template, **kwargs)
+        except Exception as exc:
+            logger.error("翻译 key '%s' 格式化失败: %s", key, exc)
+            return template or key
+
+    def _get_translation_value(self, key: str, locale: str | None) -> tuple[TranslationValue | None, str]:
+        target_locale = self._resolve_locale(locale)
+        target_catalog = self._get_catalog(target_locale)
+        if key in target_catalog:
+            return target_catalog[key], target_locale
+
+        if target_locale != self._default_locale:
+            default_catalog = self._get_catalog(self._default_locale)
+            if key in default_catalog:
+                self._log_once(
+                    ("missing_key_fallback", target_locale, key),
+                    logging.WARNING,
+                    "翻译 key '%s' 在 locale '%s' 中缺失，已回退到默认 locale '%s'",
+                    key,
+                    target_locale,
+                    self._default_locale,
+                )
+                return default_catalog[key], self._default_locale
+
+        self._log_once(
+            ("missing_key", target_locale, key),
+            logging.WARNING,
+            "翻译 key '%s' 缺失，locale='%s'，默认 locale='%s'",
+            key,
+            target_locale,
+            self._default_locale,
+        )
+        return None, target_locale
+
+    def _resolve_locale(self, locale: str | None) -> str:
+        if locale is None:
+            return self.get_locale()
+
+        try:
+            return normalize_locale(locale)
+        except InvalidLocaleError:
+            self._log_once(
+                ("invalid_locale", "explicit", locale),
+                logging.WARNING,
+                "检测到非法 locale='%s'，已回退到当前默认 locale %s",
+                locale,
+                self.get_locale(),
+            )
+            return self.get_locale()
+
+    def _get_catalog(self, locale: str) -> dict[str, TranslationValue]:
+        normalized_locale = normalize_locale(locale)
+        with self._cache_lock:
+            if normalized_locale in self._catalog_cache:
+                return self._catalog_cache[normalized_locale]
+
+            try:
+                catalog = load_locale_catalog(normalized_locale, self._locales_root)
+            except I18nError as exc:
+                self._log_once(
+                    ("load_failed", normalized_locale, exc.__class__.__name__),
+                    logging.WARNING,
+                    "加载 locale '%s' 失败: %s",
+                    normalized_locale,
+                    exc,
+                )
+                catalog = {}
+
+            self._catalog_cache[normalized_locale] = catalog
+            return catalog
+
+    def _log_once(self, cache_key: tuple[str, str, str], level: int, message: str, *args: object) -> None:
+        with self._cache_lock:
+            if cache_key in self._warning_cache:
+                return
+            self._warning_cache.add(cache_key)
+        logger.log(level, message, *args)