feat(backend): add customer-service dict list and ensure-on-write helpers.

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

backend/app/api/__init__.py

@@ -42,6 +42,7 @@ from app.api.mini_auth import router as mini_auth_router
 from app.api.mini_menu import router as mini_menu_router
 from app.api.market_mobile import router as market_mobile_router
 from app.api.sample_storage_cert_qc import router as sample_storage_cert_qc_router
+from app.api.customer_service_dict import router as customer_service_dict_router
 
 __all__ = [
     "public_router",
@@ -87,4 +88,5 @@ __all__ = [
     "mini_menu_router",
     "market_mobile_router",
     "sample_storage_cert_qc_router",
+    "customer_service_dict_router",
 ]

backend/app/api/customer_service_dict.py

@@ -0,0 +1,32 @@
+"""客户服务字典 API"""
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy.orm import Session
+
+from app.core.database import get_db
+from app.core.menu_guard import require_any_menu_code
+from app.core.response import success_response
+from app.crud import customer_service_dict as svc
+from app.models.models import SysUser
+
+router = APIRouter(prefix="/web/admin", tags=["客户服务字典"])
+
+
+def _get_tenant_no(db: Session, user: dict) -> str:
+    user_id = user.get("user_id")
+    if not user_id:
+        raise PermissionError("用户身份信息缺失，无法确定租户")
+    sys_user = db.query(SysUser).filter(SysUser.id == user_id).first()
+    if not sys_user or not sys_user.tenant_no:
+        raise PermissionError("用户租户信息缺失")
+    return sys_user.tenant_no
+
+
+@router.get("/customer-service-dict")
+async def get_customer_service_dict(
+    dictType: str = Query(...),
+    user: dict = Depends(require_any_menu_code("cell-storage:sample-info:list")),
+    db: Session = Depends(get_db),
+):
+    tenant_no = _get_tenant_no(db, user)
+    values = svc.list_dict_values(db, tenant_no, dictType)
+    return success_response({"list": values})

backend/app/crud/customer_service_dict.py

@@ -0,0 +1,80 @@
+"""客户服务字典 CRUD。"""
+from __future__ import annotations
+
+from datetime import datetime
+from typing import List, Optional
+
+from sqlalchemy.orm import Session
+
+from app.core.utils import generate_uuid
+from app.models.models import BizCustomerServiceDict
+from app.services.sample_customer_service_register import normalize_dict_value
+
+
+def list_dict_values(db: Session, tenant_no: str, dict_type: str) -> List[str]:
+    dtype = (dict_type or "").strip()
+    if not dtype:
+        return []
+    rows = (
+        db.query(BizCustomerServiceDict)
+        .filter(
+            BizCustomerServiceDict.tenant_no == tenant_no,
+            BizCustomerServiceDict.dict_type == dtype,
+            BizCustomerServiceDict.del_flag == 0,
+        )
+        .order_by(
+            BizCustomerServiceDict.order_num.asc(),
+            BizCustomerServiceDict.dict_value.asc(),
+        )
+        .all()
+    )
+    return [row.dict_value for row in rows if row.dict_value]
+
+
+def ensure_dict_value(
+    db: Session,
+    tenant_no: str,
+    dict_type: str,
+    value,
+    *,
+    user_id: Optional[str],
+    is_system: bool = False,
+) -> Optional[str]:
+    normalized = normalize_dict_value(value)
+    if normalized is None:
+        return None
+
+    dtype = (dict_type or "").strip()
+    if not dtype:
+        return None
+
+    existing = (
+        db.query(BizCustomerServiceDict)
+        .filter(
+            BizCustomerServiceDict.tenant_no == tenant_no,
+            BizCustomerServiceDict.dict_type == dtype,
+            BizCustomerServiceDict.dict_value == normalized,
+            BizCustomerServiceDict.del_flag == 0,
+        )
+        .first()
+    )
+    if existing:
+        return existing.dict_value
+
+    now = datetime.now()
+    row = BizCustomerServiceDict(
+        id=generate_uuid(),
+        dict_type=dtype,
+        dict_value=normalized,
+        is_system="1" if is_system else "0",
+        tenant_no=tenant_no,
+        create_by=user_id,
+        create_dept_id="",
+        create_time=now,
+        update_by=user_id,
+        update_time=now,
+        del_flag=0,
+    )
+    db.add(row)
+    db.flush()
+    return normalized

backend/app/main.py

@@ -13,7 +13,7 @@ from app.core.database import SessionLocal
 from app.core.api_request_log_middleware import ApiRequestLogMiddleware
 from app.core.api_crypto_middleware import ApiCryptoMiddleware
 from app.models import models  # noqa: F401
-from app.api import public_router, menu_router, admin_router, workflow_router, basic_info_router, collection_protocol_router, protocol_sample_payment_router, protocol_refund_record_router, presign_order_router, sample_freight_router, leave_router, operations_router, ethnic_router, sample_reception_router, sample_production_reception_router, qc_router, research_donation_router, production_router, pilot_service_router, insurance_company_router, material_router, storage_release_review_router, sample_sms_notification_router, storage_entry_sms_notification_router, qc_sms_notification_router, protocol_sample_collection_router, research_donation_sample_collection_router, pilot_application_router, pilot_checkup_router, pilot_delivery_router, pilot_infusion_confirm_router, pilot_infusion_schedule_mgmt_router, outbound_release_review_router, clinical_payment_router, env_detect_router, cleanroom_garment_router, market_router, pre_transit_sample_router, oa_approval_router, mini_auth_router, mini_menu_router, market_mobile_router, sample_storage_cert_qc_router
+from app.api import public_router, menu_router, admin_router, workflow_router, basic_info_router, collection_protocol_router, protocol_sample_payment_router, protocol_refund_record_router, presign_order_router, sample_freight_router, leave_router, operations_router, ethnic_router, sample_reception_router, sample_production_reception_router, qc_router, research_donation_router, production_router, pilot_service_router, insurance_company_router, material_router, storage_release_review_router, sample_sms_notification_router, storage_entry_sms_notification_router, qc_sms_notification_router, protocol_sample_collection_router, research_donation_sample_collection_router, pilot_application_router, pilot_checkup_router, pilot_delivery_router, pilot_infusion_confirm_router, pilot_infusion_schedule_mgmt_router, outbound_release_review_router, clinical_payment_router, env_detect_router, cleanroom_garment_router, market_router, pre_transit_sample_router, oa_approval_router, mini_auth_router, mini_menu_router, market_mobile_router, sample_storage_cert_qc_router, customer_service_dict_router
 
 
 @asynccontextmanager
@@ -154,6 +154,7 @@ app.include_router(mini_auth_router, prefix=settings.API_V1_PREFIX)
 app.include_router(mini_menu_router, prefix=settings.API_V1_PREFIX)
 app.include_router(market_mobile_router, prefix=settings.API_V1_PREFIX)
 app.include_router(sample_storage_cert_qc_router, prefix=settings.API_V1_PREFIX)
+app.include_router(customer_service_dict_router, prefix=settings.API_V1_PREFIX)
 
 
 def _mask_secret(value: str) -> str:

backend/app/services/sample_customer_service_register.py

@@ -0,0 +1,10 @@
+"""客户服务快捷登记：payload 规范化与字典回写（逐步扩展）。"""
+from typing import Optional
+
+
+def normalize_dict_value(raw) -> Optional[str]:
+    """Trim; empty/None -> None."""
+    if raw is None:
+        return None
+    text = str(raw).strip()
+    return text or None

backend/tests/test_customer_service_dict.py

@@ -0,0 +1,43 @@
+"""客户服务字典规范化与 CRUD 辅助。"""
+import unittest
+from unittest.mock import MagicMock
+
+from app.crud.customer_service_dict import ensure_dict_value
+from app.services.sample_customer_service_register import normalize_dict_value
+
+
+class TestNormalizeDictValue(unittest.TestCase):
+    def test_strips_whitespace(self):
+        self.assertEqual(normalize_dict_value("  已发  "), "已发")
+
+    def test_empty_string_returns_none(self):
+        self.assertIsNone(normalize_dict_value(""))
+
+    def test_whitespace_only_returns_none(self):
+        self.assertIsNone(normalize_dict_value("   "))
+
+    def test_none_returns_none(self):
+        self.assertIsNone(normalize_dict_value(None))
+
+
+class TestEnsureDictValue(unittest.TestCase):
+    def test_returns_none_for_blank_value(self):
+        db = MagicMock()
+        self.assertIsNone(
+            ensure_dict_value(db, "zkxw", "sms_send_status", "  ", user_id="u1")
+        )
+        db.add.assert_not_called()
+
+    def test_returns_existing_without_insert(self):
+        db = MagicMock()
+        existing = MagicMock()
+        existing.dict_value = "已发"
+        db.query.return_value.filter.return_value.first.return_value = existing
+
+        result = ensure_dict_value(
+            db, "zkxw", "sms_send_status", "  已发  ", user_id="u1"
+        )
+
+        self.assertEqual(result, "已发")
+        db.add.assert_not_called()
+        db.flush.assert_not_called()