From edabf70287ecf8036bde6c8d8fddf078f0920ac4 Mon Sep 17 00:00:00 2001 From: zlwu <704541015@qq.com> Date: Tue, 13 Oct 2026 10:50:36 +0800 Subject: [PATCH] feat(backend): upsert sample customer-service fields with empty-skip merge. Co-authored-by: Cursor --- backend/app/crud/sample_customer_service.py | 149 ++++++++++++++++++ .../sample_customer_service_register.py | 42 ++++- .../test_sample_customer_service_register.py | 56 +++++++ 3 files changed, 246 insertions(+), 1 deletion(-) create mode 100644 backend/app/crud/sample_customer_service.py create mode 100644 backend/tests/test_sample_customer_service_register.py diff --git a/backend/app/crud/sample_customer_service.py b/backend/app/crud/sample_customer_service.py new file mode 100644 index 00000000..8f48ead5 --- /dev/null +++ b/backend/app/crud/sample_customer_service.py @@ -0,0 +1,149 @@ +"""样本客户服务扩展 CRUD。""" +from __future__ import annotations + +from datetime import date, datetime +from typing import Any, Dict, List, Optional + +from sqlalchemy.orm import Session + +from app.core.utils import generate_uuid +from app.crud.customer_service_dict import ensure_dict_value +from app.models.models import BizSampleCustomerService +from app.services.sample_customer_service_register import ( + CUSTOMER_SERVICE_FIELD_KEYS, + merge_customer_service_fields, + normalize_dict_value, +) + +DICT_TYPE_BY_FIELD = { + "sms_send_status": "sms_send_status", + "microbe_followup_status": "microbe_followup_status", + "fullmoon_followup_status": "fullmoon_followup_status", + "household_registered": "household_registered", +} + + +def _format_date(value: Optional[date]) -> Optional[str]: + if value is None: + return None + if isinstance(value, datetime): + return value.strftime("%Y-%m-%d") + return value.strftime("%Y-%m-%d") + + +def _format_datetime(value: Optional[datetime]) -> Optional[str]: + if value is None: + return None + return value.strftime("%Y-%m-%d %H:%M:%S") + + +def _row_field_dict(row: BizSampleCustomerService) -> Dict[str, Any]: + return {key: getattr(row, key) for key in CUSTOMER_SERVICE_FIELD_KEYS} + + +def get_by_sample_code( + db: Session, tenant_no: str, sample_code: str +) -> Optional[BizSampleCustomerService]: + code = (sample_code or "").strip() + if not code: + return None + return ( + db.query(BizSampleCustomerService) + .filter( + BizSampleCustomerService.tenant_no == tenant_no, + BizSampleCustomerService.sample_code == code, + BizSampleCustomerService.del_flag == 0, + ) + .first() + ) + + +def get_map_by_sample_codes( + db: Session, tenant_no: str, sample_codes: List[str] +) -> Dict[str, BizSampleCustomerService]: + codes = list({(c or "").strip() for c in sample_codes if (c or "").strip()}) + if not codes: + return {} + rows = ( + db.query(BizSampleCustomerService) + .filter( + BizSampleCustomerService.tenant_no == tenant_no, + BizSampleCustomerService.sample_code.in_(codes), + BizSampleCustomerService.del_flag == 0, + ) + .all() + ) + return {row.sample_code: row for row in rows} + + +def row_to_dict(row: BizSampleCustomerService) -> Dict[str, Any]: + return { + "id": row.id, + "sampleCode": row.sample_code, + "collectionProtocolId": row.collection_protocol_id, + "smsSendDate": _format_date(row.sms_send_date), + "smsSendStatus": row.sms_send_status, + "microbeFollowupDate": _format_date(row.microbe_followup_date), + "microbeFollowupStatus": row.microbe_followup_status, + "fullmoonFollowupDate": _format_date(row.fullmoon_followup_date), + "fullmoonFollowupStatus": row.fullmoon_followup_status, + "householdRegistered": row.household_registered, + "einvoiceSendTime": _format_datetime(row.einvoice_send_time), + "effectiveAddress": row.effective_address, + } + + +def upsert_by_sample_code( + db: Session, + *, + tenant_no: str, + sample_code: str, + collection_protocol_id: Optional[str], + user_id: Optional[str], + incoming: Dict[str, Any], +) -> BizSampleCustomerService: + code = (sample_code or "").strip() + if not code: + raise ValueError("sample_code is required") + + row = get_by_sample_code(db, tenant_no, code) + now = datetime.now() + if row is None: + row = BizSampleCustomerService( + id=generate_uuid(), + sample_code=code, + collection_protocol_id=collection_protocol_id, + 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) + else: + if collection_protocol_id and not row.collection_protocol_id: + row.collection_protocol_id = collection_protocol_id + row.update_by = user_id + row.update_time = now + + existing = _row_field_dict(row) + merged = merge_customer_service_fields(existing, incoming) + + for field, dict_type in DICT_TYPE_BY_FIELD.items(): + if field not in merged: + continue + value = merged[field] + if value is None or normalize_dict_value(value) is None: + continue + merged[field] = ensure_dict_value( + db, tenant_no, dict_type, value, user_id=user_id + ) + + for key in CUSTOMER_SERVICE_FIELD_KEYS: + if key in merged: + setattr(row, key, merged[key]) + + db.flush() + return row diff --git a/backend/app/services/sample_customer_service_register.py b/backend/app/services/sample_customer_service_register.py index e05a2cf6..72b1638b 100644 --- a/backend/app/services/sample_customer_service_register.py +++ b/backend/app/services/sample_customer_service_register.py @@ -1,5 +1,19 @@ """客户服务快捷登记：payload 规范化与字典回写（逐步扩展）。""" -from typing import Optional +from typing import Any, Dict, FrozenSet, Optional + +CUSTOMER_SERVICE_FIELD_KEYS: FrozenSet[str] = frozenset( + { + "sms_send_date", + "sms_send_status", + "microbe_followup_date", + "microbe_followup_status", + "fullmoon_followup_date", + "fullmoon_followup_status", + "household_registered", + "einvoice_send_time", + "effective_address", + } +) def normalize_dict_value(raw) -> Optional[str]: @@ -8,3 +22,29 @@ def normalize_dict_value(raw) -> Optional[str]: return None text = str(raw).strip() return text or None + + +def _is_empty_merge_value(value: Any) -> bool: + if value is None: + return True + if isinstance(value, str): + return not value.strip() + return False + + +def merge_customer_service_fields( + existing: Optional[Dict[str, Any]], + incoming: Optional[Dict[str, Any]], +) -> Dict[str, Any]: + """Merge incoming onto existing; empty/None/whitespace incoming values do not overwrite.""" + existing = existing or {} + incoming = incoming or {} + keys = (set(existing.keys()) | set(incoming.keys())) & CUSTOMER_SERVICE_FIELD_KEYS + merged: Dict[str, Any] = {k: existing.get(k) for k in keys} + for key, value in incoming.items(): + if key not in CUSTOMER_SERVICE_FIELD_KEYS: + continue + if _is_empty_merge_value(value): + continue + merged[key] = value + return merged diff --git a/backend/tests/test_sample_customer_service_register.py b/backend/tests/test_sample_customer_service_register.py new file mode 100644 index 00000000..eb9111ae --- /dev/null +++ b/backend/tests/test_sample_customer_service_register.py @@ -0,0 +1,56 @@ +"""样本客户服务登记：空不覆盖 merge 规则。""" +import unittest +from datetime import date, datetime + +from app.services.sample_customer_service_register import merge_customer_service_fields + + +class TestMergeCustomerServiceFields(unittest.TestCase): + def test_merge_skips_empty_and_keeps_existing(self): + existing = {"sms_send_status": "已发", "effective_address": "合肥"} + incoming = {"sms_send_status": "", "effective_address": "亳州", "household_registered": "是"} + merged = merge_customer_service_fields(existing, incoming) + self.assertEqual(merged["sms_send_status"], "已发") + self.assertEqual(merged["effective_address"], "亳州") + self.assertEqual(merged["household_registered"], "是") + + def test_merge_all_sections_partial(self): + existing = {"sms_send_status": "已发", "microbe_followup_status": None} + incoming = {"microbe_followup_status": "回访成功"} + merged = merge_customer_service_fields(existing, incoming) + self.assertEqual(merged["sms_send_status"], "已发") + self.assertEqual(merged["microbe_followup_status"], "回访成功") + + def test_merge_skips_missing_keys(self): + existing = {"sms_send_date": None, "effective_address": "A"} + incoming = {"effective_address": "B"} + merged = merge_customer_service_fields(existing, incoming) + self.assertEqual(merged["effective_address"], "B") + self.assertIsNone(merged.get("sms_send_date")) + + def test_merge_skips_none_incoming(self): + existing = {"sms_send_status": "已发", "effective_address": "合肥"} + incoming = {"sms_send_status": None, "effective_address": "亳州"} + merged = merge_customer_service_fields(existing, incoming) + self.assertEqual(merged["sms_send_status"], "已发") + self.assertEqual(merged["effective_address"], "亳州") + + def test_merge_skips_whitespace_only_incoming(self): + existing = {"microbe_followup_status": "回访成功", "household_registered": "是"} + incoming = {"microbe_followup_status": " ", "household_registered": "否"} + merged = merge_customer_service_fields(existing, incoming) + self.assertEqual(merged["microbe_followup_status"], "回访成功") + self.assertEqual(merged["household_registered"], "否") + + def test_merge_passes_through_dates(self): + d = date(2026, 10, 13) + dt = datetime(2026, 10, 13, 14, 30, 0) + existing = {"sms_send_date": None} + incoming = {"sms_send_date": d, "einvoice_send_time": dt} + merged = merge_customer_service_fields(existing, incoming) + self.assertEqual(merged["sms_send_date"], d) + self.assertEqual(merged["einvoice_send_time"], dt) + + +if __name__ == "__main__": + unittest.main()