""" 租户配置加载器。 从统一配置文件(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(供商家勾选主市场语言等场景使用) # 语言代码与展示名的双向映射(供翻译/LLM 提示等统一使用) SOURCE_LANG_CODE_MAP: 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", } TARGET_LANG_CODE_MAP: Dict[str, str] = {v: k for k, v in SOURCE_LANG_CODE_MAP.items()} 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 SOURCE_LANG_CODE_MAP: 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 SOURCE_LANG_CODE_MAP: 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 SOURCE_LANG_CODE_MAP: 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