dpb/backend/app/utils/dict_util.py

"""导入 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 _rows_to_mapping(rows) -> dict[str, dict[str, str]]:
    mapping: dict[str, dict[str, str]] = {}
    for dt, dv, dl in rows:
        mapping.setdefault(dt, {})[dv] = dl
    return mapping


async def preload_value2label_cache() -> None:
    """应用启动时用异步引擎预载全量字典映射（value→label）。

    batch_export 均为同步 staticmethod（控制器同步调用、bytes 序列化），无法 await，
    故启动时预热一次进程内缓存，导出请求路径只做内存查找、不触达同步引擎。
    """
    global _value2label_cache
    if _value2label_cache is not None:
        return
    from sqlalchemy import select

    from app.api.v1.module_system.dict.model import DictDataModel
    from app.core.database import async_db_session

    async with async_db_session() as session:
        rows = (
            await session.execute(
                select(DictDataModel.dict_type, DictDataModel.dict_value, DictDataModel.dict_label)
                .where(DictDataModel.is_deleted.is_(False))
            )
        ).all()
    _value2label_cache = _rows_to_mapping(rows)


def dict_value_to_label(dict_type: str, value: Any) -> Any:
    """把英码 dict_value 翻译为中文 dict_label（导出用）。

    与 DictLabelResolver 方向相反：导出时把库内的英码还原为中文，便于用户阅读；
    值空或无法识别时原样返回，保证"导出的文件再导入"仍可解析。
    应用启动经 preload_value2label_cache 预热后纯内存查找；脚本等未走 lifespan 的
    上下文兜底用同步会话载入（见 create_engine_and_session 的调度器孤岛注释）。
    """
    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

        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()
        _value2label_cache = _rows_to_mapping(rows)
    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