feat(backend): resolve sample transport mode dict with auto-create

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/services/sample_transport_mode.py

@@ -0,0 +1,113 @@
+"""样本运输方式字典 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

backend/tests/test_sample_freight.py

@@ -0,0 +1,43 @@
+"""样本运费：运输方式字典解析与关联规则"""
+
+import unittest
+from unittest.mock import MagicMock
+
+from app.services.sample_transport_mode import resolve_transport_mode_value
+
+
+class TestResolveTransportMode(unittest.TestCase):
+    def test_empty_raises(self):
+        db = MagicMock()
+        with self.assertRaises(ValueError):
+            resolve_transport_mode_value(db, tenant_no="zkxw", raw="", user_id=None)
+
+    def test_hit_by_value(self):
+        db = MagicMock()
+        type_exists = MagicMock()
+        row = MagicMock(dict_value="sfkd", dict_label="顺丰快递", del_flag=0)
+        db.query.return_value.filter.return_value.first.side_effect = [type_exists, row]
+        self.assertEqual(
+            resolve_transport_mode_value(db, tenant_no="zkxw", raw="sfkd", user_id=None),
+            "sfkd",
+        )
+
+    def test_hit_by_label_then_create_path_not_called(self):
+        db = MagicMock()
+        type_exists = MagicMock()
+        miss = None
+        hit = MagicMock(dict_value="zxc", dict_label="专线车", del_flag=0)
+        db.query.return_value.filter.return_value.first.side_effect = [
+            type_exists,
+            miss,
+            hit,
+        ]
+        self.assertEqual(
+            resolve_transport_mode_value(db, tenant_no="zkxw", raw="专线车", user_id=None),
+            "zxc",
+        )
+        db.add.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()