104 lines
4.1 KiB
Python
104 lines
4.1 KiB
Python
"""导入 Excel 时把中文 dict_label 翻译为 sys_dict 的英码 dict_value。
|
|
|
|
背景：breeding 模块的枚举字段在方案 B 下统一以英码(dict_value)落库、以中文(dict_label)
|
|
展示。Excel 模板里用户填的是中文 label，写入前必须经 sys_dict 翻译为英码，否则库里会混进
|
|
中文、与字典对不齐，下拉/回显也会错乱。
|
|
|
|
本解析器按 auth+db 直接查 sys_dict_data（带 label->value 缓存，单次导入只查一次），
|
|
避免了 redis 依赖、也无需改动 import 控制器。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.base_schema import AuthSchema
|
|
from app.core.exceptions import CustomException
|
|
|
|
|
|
_value2label_cache: dict[str, dict[str, str]] | None = None
|
|
|
|
|
|
def dict_value_to_label(dict_type: str, value: Any) -> Any:
|
|
"""把英码 dict_value 翻译为中文 dict_label（导出用）。
|
|
|
|
与 DictLabelResolver 方向相反：导出时把库内的英码还原为中文，便于用户阅读；
|
|
值空或无法识别时原样返回，保证"导出的文件再导入"仍可解析。
|
|
使用同步引擎读取 sys_dict_data 并进程内缓存，避免每次导出都查库。
|
|
"""
|
|
if value is None:
|
|
return value
|
|
text = str(value)
|
|
if text == "":
|
|
return value
|
|
global _value2label_cache
|
|
if _value2label_cache is None:
|
|
from sqlalchemy import select
|
|
|
|
from app.api.v1.module_system.dict.model import DictDataModel
|
|
from app.core.database import db_session
|
|
|
|
mapping: dict[str, dict[str, str]] = {}
|
|
with db_session() as session:
|
|
rows = session.execute(
|
|
select(DictDataModel.dict_type, DictDataModel.dict_value, DictDataModel.dict_label)
|
|
.where(DictDataModel.is_deleted.is_(False))
|
|
).all()
|
|
for dt, dv, dl in rows:
|
|
mapping.setdefault(dt, {})[dv] = dl
|
|
_value2label_cache = mapping
|
|
return _value2label_cache.get(dict_type, {}).get(text, value)
|
|
|
|
|
|
class DictLabelResolver:
|
|
"""把一组字典类型的中文 label 解析为英码 value，支持批量复用。"""
|
|
|
|
def __init__(self, auth: AuthSchema, db: AsyncSession, dict_types: list[str]) -> None:
|
|
self._auth = auth
|
|
self._db = db
|
|
self._dict_types = dict_types
|
|
self._loaded = False
|
|
# dict_type -> {label: value}
|
|
self._label2value: dict[str, dict[str, str]] = {}
|
|
# dict_type -> {value} (支持再导入"导出文件"——文件里已是英码)
|
|
self._values: dict[str, set[str]] = {}
|
|
|
|
async def _ensure_loaded(self) -> None:
|
|
if self._loaded:
|
|
return
|
|
from app.api.v1.module_system.dict.crud import DictDataCRUD
|
|
|
|
crud = DictDataCRUD(self._auth, self._db)
|
|
for dt in self._dict_types:
|
|
rows = await crud.get_list(search={"dict_type": dt})
|
|
self._label2value[dt] = {row.dict_label: row.dict_value for row in rows}
|
|
self._values[dt] = {row.dict_value for row in rows}
|
|
self._loaded = True
|
|
|
|
async def resolve(self, dict_type: str, label: Any, *, required: bool = True) -> Any:
|
|
"""把中文 label 解析为英码 value；空值返回 None，已为英码则原样返回。
|
|
|
|
Args:
|
|
dict_type: 字典类型，如 "cross_method"
|
|
label: Excel 单元格原始值（可能为中文 label，也可能已是英码）
|
|
required: 非空白且无法识别时是否抛错（默认 True，便于发现脏数据）
|
|
"""
|
|
if label is None:
|
|
return None
|
|
text = str(label).strip()
|
|
if text == "":
|
|
return None
|
|
await self._ensure_loaded()
|
|
mapping = self._label2value.get(dict_type, {})
|
|
if text in mapping:
|
|
return mapping[text]
|
|
if text in self._values.get(dict_type, set()):
|
|
# 已是英码（如重新导入导出的文件），直接落库
|
|
return text
|
|
if required:
|
|
raise CustomException(
|
|
msg=f"字典类型「{dict_type}」中不存在标签或编码「{text}」，请核对导入模板"
|
|
)
|
|
return text
|