tenant_config_loader.py
5.82 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
"""
租户配置加载器。
从统一配置文件(config.yaml)加载租户配置,包括主语言和索引语言(index_languages)。
"""
import logging
from typing import Dict, Any, Optional, List
from config.loader import get_app_config
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。
若配置缺失或非法,则回退到默认配置。
"""
normalized = normalize_index_languages(
tenant_config.get("index_languages"),
tenant_config.get("primary_language") or "en",
)
return normalized if normalized 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:
tenant_cfg = get_app_config().tenants
default_cfg = tenant_cfg.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.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