From 66fc507b1a1fcfdee835aea7493a5a828c0c5141 Mon Sep 17 00:00:00 2001
From: zlwu <704541015@qq.com>
Date: Wed, 7 Oct 2026 23:18:56 +0800
Subject: [PATCH] feat(backend): resolve sample transport mode dict with
 auto-create

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 backend/app/services/sample_transport_mode.py | 113 ++++++++++++++++++
 backend/tests/test_sample_freight.py          |  43 +++++++
 2 files changed, 156 insertions(+)
 create mode 100644 backend/app/services/sample_transport_mode.py
 create mode 100644 backend/tests/test_sample_freight.py

diff --git a/backend/app/services/sample_transport_mode.py b/backend/app/services/sample_transport_mode.py
new file mode 100644
index 00000000..25a9ed31
--- /dev/null
+++ b/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
diff --git a/backend/tests/test_sample_freight.py b/backend/tests/test_sample_freight.py
new file mode 100644
index 00000000..4fa2d3ed
--- /dev/null
+++ b/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()