zkxw/backend/app/services/sample_transport_mode.py

"""样本运输方式字典 sample_transport_mode：解析 / 自动入库（flush，不单独 commit）"""

from __future__ import annotations

from datetime import datetime
from typing import Optional

from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.models.models import SysDictData, SysDictType

DICT_TYPE = "sample_transport_mode"


def ensure_dict_type(db: Session, tenant_no: str, user_id: Optional[str]) -> None:
    exists = (
        db.query(SysDictType.id)
        .filter(
            SysDictType.dict_type == DICT_TYPE,
            SysDictType.tenant_no == tenant_no,
            SysDictType.del_flag == 0,
        )
        .first()
    )
    if exists:
        return
    db.add(
        SysDictType(
            id=generate_uuid(),
            dict_name="样本运输方式",
            dict_type=DICT_TYPE,
            tenant_no=tenant_no,
            order_num=0,
            remark="协议单运费批次运输方式",
            create_by=user_id,
            create_dept_id="",
            create_time=datetime.now(),
            update_by=user_id,
            update_time=datetime.now(),
            del_flag=0,
        )
    )
    db.flush()


def resolve_transport_mode_value(
    db: Session, *, tenant_no: str, raw: str, user_id: Optional[str]
) -> str:
    text = (raw or "").strip()
    if not text:
        raise ValueError("运输方式不能为空")
    ensure_dict_type(db, tenant_no, user_id)

    by_value = (
        db.query(SysDictData)
        .filter(
            SysDictData.dict_type == DICT_TYPE,
            SysDictData.tenant_no == tenant_no,
            SysDictData.dict_value == text,
            SysDictData.del_flag == 0,
        )
        .first()
    )
    if by_value:
        return by_value.dict_value

    by_label = (
        db.query(SysDictData)
        .filter(
            SysDictData.dict_type == DICT_TYPE,
            SysDictData.tenant_no == tenant_no,
            SysDictData.dict_label == text,
            SysDictData.del_flag == 0,
        )
        .first()
    )
    if by_label:
        return by_label.dict_value

    value = text
    clash = (
        db.query(SysDictData.id)
        .filter(
            SysDictData.dict_type == DICT_TYPE,
            SysDictData.tenant_no == tenant_no,
            SysDictData.dict_value == value,
            SysDictData.del_flag == 0,
        )
        .first()
    )
    if clash:
        value = f"{text}_{generate_uuid()[:8]}"

    db.add(
        SysDictData(
            id=generate_uuid(),
            dict_label=text,
            dict_value=value,
            dict_type=DICT_TYPE,
            tenant_no=tenant_no,
            order_num=99,
            remark=None,
            create_by=user_id,
            create_dept_id="",
            create_time=datetime.now(),
            update_by=user_id,
            update_time=datetime.now(),
            del_flag=0,
        )
    )
    db.flush()
    return value