tenant_config_loader.py
5.33 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
"""
租户配置加载器。
从统一配置文件(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",
}
DEFAULT_INDEX_LANGUAGES: List[str] = ["en", "zh"]
def normalize_index_languages(value: Any, primary_language: str = "en") -> List[str]:
"""
将 index_languages 配置规范化为合法语言代码列表。
None 或空时返回 DEFAULT_INDEX_LANGUAGES。
"""
if value is None:
return list(DEFAULT_INDEX_LANGUAGES)
if not isinstance(value, (list, tuple)):
return list(DEFAULT_INDEX_LANGUAGES)
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 if valid else list(DEFAULT_INDEX_LANGUAGES)
def resolve_index_languages(tenant_config: Dict[str, Any]) -> List[str]:
"""
从租户配置解析 index_languages。
若存在 index_languages 则用之;否则按旧配置 translate_to_en / translate_to_zh 推导。
"""
if "index_languages" in tenant_config:
return normalize_index_languages(
tenant_config["index_languages"],
tenant_config.get("primary_language") or "en",
)
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()
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": "en", "index_languages": ["en", "zh"]},
"tenants": {},
}
return self._config
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.get("default", {"primary_language": "en", "index_languages": ["en", "zh"]})
tenants = config.get("tenants", {})
raw = tenants[tenant_id_str] if tenant_id_str in tenants else default
if tenant_id_str not in tenants:
logger.debug(f"Tenant {tenant_id} not found in config, using default")
out = dict(raw)
out["index_languages"] = resolve_index_languages(raw)
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