""" 租户配置加载器。 从统一配置文件(config.yaml)加载租户配置,包括主语言和翻译配置。 """ import logging from typing import Dict, Any, Optional logger = logging.getLogger(__name__) 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() self._config = search_config.tenant_config 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) # 返回默认配置 self._config = { "default": { "primary_language": "zh", "translate_to_en": True, "translate_to_zh": False }, "tenants": {} } return self._config def get_tenant_config(self, tenant_id: str) -> Dict[str, Any]: """ 获取指定租户的配置。 Args: tenant_id: 租户ID Returns: 租户配置字典,如果租户不存在则返回默认配置 """ config = self.load_config() tenant_id_str = str(tenant_id) tenants = config.get("tenants", {}) if tenant_id_str in tenants: return tenants[tenant_id_str] else: logger.debug(f"Tenant {tenant_id} not found in config, using default") return config.get("default", { "primary_language": "zh", "translate_to_en": True, "translate_to_zh": False }) 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