config/tenant_config_loader.py

"""
租户配置加载器。
从统一配置文件（config.yaml）加载租户配置，包括主语言和索引语言（index_languages）。
支持旧配置 translate_to_en / translate_to_zh 的兼容解析。
"""
import logging
from typing import Dict, Any, Optional, List
logger = logging.getLogger(__name__)
# 支持的索引语言：code -> display name（供商家勾选主市场语言等场景使用）
SUPPORTED_INDEX_LANGUAGES: Dict[str, str] = {
    "en": "English",
    "zh": "Chinese",
    "zh_tw": "Traditional Chinese",
    "ru": "Russian",
    "ja": "Japanese",
    "ko": "Korean",
    "es": "Spanish",
    "fr": "French",
    "pt": "Portuguese",
    "de": "German",
    "it": "Italian",
    "th": "Thai",
    "vi": "Vietnamese",
    "id": "Indonesian",
    "ms": "Malay",
    "ar": "Arabic",
    "hi": "Hindi",
    "he": "Hebrew",
    "my": "Burmese",
    "ta": "Tamil",
    "ur": "Urdu",
    "bn": "Bengali",
    "pl": "Polish",
    "nl": "Dutch",
    "ro": "Romanian",
    "tr": "Turkish",
    "km": "Khmer",
    "lo": "Lao",
    "yue": "Cantonese",
    "cs": "Czech",
    "el": "Greek",
    "sv": "Swedish",
    "hu": "Hungarian",
    "da": "Danish",
    "fi": "Finnish",
    "uk": "Ukrainian",
    "bg": "Bulgarian",
}
def normalize_index_languages(value: Any, primary_language: str = "en") -> List[str]:
    """
    将 index_languages 配置规范化为合法语言代码列表。
    仅做规范化，不做默认值兜底。
    """
    del primary_language
    if value is None:
        return []
    if not isinstance(value, (list, tuple)):
        return []
    valid: List[str] = []
    seen: set = set()
    for item in value:
        code = (item or "").strip().lower()
        if not code or code in seen:
            continue
        if code in SUPPORTED_INDEX_LANGUAGES:
            valid.append(code)
            seen.add(code)
    return valid
def resolve_index_languages(
    tenant_config: Dict[str, Any],
    default_index_languages: List[str],
) -> List[str]:
    """
    从租户配置解析 index_languages。
    若存在 index_languages 则用之；否则按旧配置 translate_to_en / translate_to_zh 推导。
    """
    if "index_languages" in tenant_config:
        normalized = normalize_index_languages(
            tenant_config["index_languages"],
            tenant_config.get("primary_language") or "en",
        )
        return normalized if normalized else list(default_index_languages)
    primary = (tenant_config.get("primary_language") or "en").strip().lower()
    to_en = bool(tenant_config.get("translate_to_en"))
    to_zh = bool(tenant_config.get("translate_to_zh"))
    langs: List[str] = []
    if primary and primary in SUPPORTED_INDEX_LANGUAGES:
        langs.append(primary)
    for code in ("en", "zh"):
        if code not in langs and ((code == "en" and to_en) or (code == "zh" and to_zh)):
            if code in SUPPORTED_INDEX_LANGUAGES:
                langs.append(code)
    return langs if langs else list(default_index_languages)
class TenantConfigLoader:
    """租户配置加载器。"""
    def __init__(self):
        """初始化租户配置加载器。"""
        self._config: Optional[Dict[str, Any]] = None
    def load_config(self) -> Dict[str, Any]:
        """
        加载租户配置（从统一配置文件）。
        Returns:
            租户配置字典，格式：{"tenants": {...}, "default": {...}}
        """
        if self._config is not None:
            return self._config
        try:
            from config import ConfigLoader
            config_loader = ConfigLoader()
            search_config = config_loader.load_config()
            tenant_cfg = search_config.tenant_config
            if not isinstance(tenant_cfg, dict):
                raise RuntimeError("tenant_config must be an object")
            default_cfg = tenant_cfg.get("default")
            if not isinstance(default_cfg, dict):
                raise RuntimeError("tenant_config.default must be configured in config.yaml")
            default_primary = (default_cfg.get("primary_language") or "en").strip().lower()
            default_index_languages = normalize_index_languages(
                default_cfg.get("index_languages"),
                default_primary,
            )
            if not default_index_languages:
                raise RuntimeError(
                    "tenant_config.default.index_languages must include at least one supported language"
                )
            tenants_cfg = tenant_cfg.get("tenants", {})
            if not isinstance(tenants_cfg, dict):
                raise RuntimeError("tenant_config.tenants must be an object")
            normalized_default = dict(default_cfg)
            normalized_default["primary_language"] = default_primary
            normalized_default["index_languages"] = default_index_languages
            self._config = {
                "default": normalized_default,
                "tenants": tenants_cfg,
            }
            logger.info("Loaded tenant config from unified config.yaml")
            return self._config
        except Exception as e:
            logger.error(f"Failed to load tenant config: {e}", exc_info=True)
            raise
    def get_tenant_config(self, tenant_id: str) -> Dict[str, Any]:
        """
        获取指定租户的配置。
        Args:
            tenant_id: 租户ID
        Returns:
            租户配置字典，若租户不存在则用默认配置。始终包含已解析的 index_languages。
        """
        config = self.load_config()
        tenant_id_str = str(tenant_id)
        default = config["default"]
        tenants = config.get("tenants", {})
        raw = tenants[tenant_id_str] if tenant_id_str in tenants else {}
        if raw and not isinstance(raw, dict):
            raise RuntimeError(f"tenant_config.tenants.{tenant_id_str} must be an object")
        if tenant_id_str not in tenants:
            logger.debug(f"Tenant {tenant_id} not found in config, using default")
        merged = dict(default)
        merged.update(raw)
        out = dict(merged)
        out["index_languages"] = resolve_index_languages(
            merged,
            default_index_languages=default["index_languages"],
        )
        return out
    def reload(self):
        """重新加载配置（用于配置更新）。"""
        self._config = None
        return self.load_config()
# 全局实例
_tenant_config_loader: Optional[TenantConfigLoader] = None
def get_tenant_config_loader() -> TenantConfigLoader:
    """获取全局租户配置加载器实例。"""
    global _tenant_config_loader
    if _tenant_config_loader is None:
        _tenant_config_loader = TenantConfigLoader()
    return _tenant_config_loader