feat: sync mother/father/newborn customers on protocol party save

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

backend/app/services/party_customer_sync.py

@@ -0,0 +1,262 @@
+"""协议当事人三角色客户同步：新建、补证、撞库合并改挂。"""
+from __future__ import annotations
+
+from datetime import date
+from types import SimpleNamespace
+from typing import Any, Dict, Optional
+
+from sqlalchemy.orm import Session
+
+from app.crud import customer as customer_crud
+from app.models.models import BizCustomer, BizProtocolPartyCustomer
+from app.models.research_donation import BizResearchDonationParty
+
+ROLE_MOTHER = "mother"
+ROLE_FATHER = "father"
+ROLE_NEWBORN = "newborn"
+
+_ROLE_SPECS = (
+    {
+        "role": ROLE_MOTHER,
+        "out_key": "customer_id",
+        "id_keys": ("customer_id", "customerId"),
+        "name_keys": ("mother_name", "motherName"),
+        "phone_keys": ("mother_phone", "motherPhone"),
+        "id_card_keys": ("mother_id_card", "motherIdCard"),
+        "sex_keys": (),
+        "default_sex": "1",
+    },
+    {
+        "role": ROLE_FATHER,
+        "out_key": "father_customer_id",
+        "id_keys": ("father_customer_id", "fatherCustomerId"),
+        "name_keys": ("father_name", "fatherName"),
+        "phone_keys": ("father_phone", "fatherPhone"),
+        "id_card_keys": ("father_id_card", "fatherIdCard"),
+        "sex_keys": (),
+        "default_sex": "0",
+    },
+    {
+        "role": ROLE_NEWBORN,
+        "out_key": "newborn_customer_id",
+        "id_keys": ("newborn_customer_id", "newbornCustomerId"),
+        "name_keys": ("newborn_name", "newbornName"),
+        "phone_keys": ("newborn_phone", "newbornPhone"),
+        "id_card_keys": ("newborn_id_card", "newbornIdCard"),
+        "sex_keys": ("newborn_sex", "newbornSex"),
+        "default_sex": "2",
+    },
+)
+
+
+def _is_empty(val: Any) -> bool:
+    if val is None:
+        return True
+    if isinstance(val, str) and not val.strip():
+        return True
+    return False
+
+
+def _pick(party: Dict[str, Any], keys: tuple) -> str:
+    for key in keys:
+        if key not in party or party[key] is None:
+            continue
+        return str(party[key]).strip()
+    return ""
+
+
+def _union_tags(left: Any, right: Any) -> str:
+    seen = []
+    for raw in (left, right):
+        if _is_empty(raw):
+            continue
+        for part in str(raw).split(","):
+            tag = part.strip()
+            if tag and tag not in seen:
+                seen.append(tag)
+    return ",".join(seen)
+
+
+def _age_from_id_card(id_card: str) -> Optional[int]:
+    s = (id_card or "").strip()
+    if len(s) != 18:
+        return None
+    try:
+        birth = date(int(s[6:10]), int(s[10:12]), int(s[12:14]))
+    except ValueError:
+        return None
+    today = date.today()
+    age = today.year - birth.year - ((today.month, today.day) < (birth.month, birth.day))
+    if age < 0 or age > 150:
+        return None
+    return age
+
+
+def merge_customer_fields(target: Any, source: Any) -> None:
+    """目标为空才补；性别在空或未知(2)时补；标签逗号并集；姓名保留目标。"""
+    if _is_empty(getattr(target, "phone_no", None)):
+        src_phone = getattr(source, "phone_no", None)
+        if not _is_empty(src_phone):
+            target.phone_no = src_phone
+    if _is_empty(getattr(target, "age", None)):
+        src_age = getattr(source, "age", None)
+        if not _is_empty(src_age):
+            target.age = src_age
+    target_sex = getattr(target, "sex", None)
+    if _is_empty(target_sex) or str(target_sex).strip() == "2":
+        src_sex = getattr(source, "sex", None)
+        if not _is_empty(src_sex):
+            target.sex = src_sex
+    if _is_empty(getattr(target, "remark", None)):
+        src_remark = getattr(source, "remark", None)
+        if not _is_empty(src_remark):
+            target.remark = src_remark
+    target.customer_tags = _union_tags(
+        getattr(target, "customer_tags", None),
+        getattr(source, "customer_tags", None),
+    )
+
+
+def find_customer_by_id_card(db: Session, id_card: str, tenant_no: str) -> Optional[BizCustomer]:
+    ic = (id_card or "").strip()
+    if not ic:
+        return None
+    return (
+        db.query(BizCustomer)
+        .filter(
+            BizCustomer.id_card_no == ic,
+            BizCustomer.tenant_no == tenant_no,
+            BizCustomer.del_flag == 0,
+        )
+        .first()
+    )
+
+
+def get_customer(db: Session, customer_id: str, tenant_no: Optional[str] = None) -> Optional[BizCustomer]:
+    cid = (customer_id or "").strip()
+    if not cid:
+        return None
+    return customer_crud.get_customer(db, cid, tenant_no=tenant_no)
+
+
+def repoint_customer_refs(db: Session, from_id: str, to_id: str, tenant_no: str) -> None:
+    if not from_id or not to_id or from_id == to_id:
+        return
+    for model in (BizProtocolPartyCustomer, BizResearchDonationParty):
+        for col_name in ("customer_id", "father_customer_id", "newborn_customer_id"):
+            col = getattr(model, col_name)
+            db.query(model).filter(
+                col == from_id,
+                model.del_flag == 0,
+                model.tenant_no == tenant_no,
+            ).update({col_name: to_id}, synchronize_session=False)
+
+
+def _create_role_customer(
+    db: Session,
+    *,
+    name: str,
+    sex: str,
+    id_card: str,
+    phone: str,
+    tenant_no: str,
+    user_id: str,
+) -> str:
+    obj = customer_crud.create_customer(
+        db,
+        customer_name=name,
+        sex=sex,
+        id_card_no=id_card or None,
+        age=_age_from_id_card(id_card),
+        phone_no=phone or None,
+        tenant_no=tenant_no,
+        user_id=user_id,
+    )
+    return obj.id
+
+
+def _apply_direct_identity(obj: Any, *, name: str, phone: str, id_card: str) -> None:
+    obj.customer_name = name
+    if phone:
+        obj.phone_no = phone
+    if id_card:
+        obj.id_card_no = id_card
+    age = _age_from_id_card(id_card)
+    if age is not None and _is_empty(getattr(obj, "age", None)):
+        obj.age = age
+
+
+def _merge_and_repoint(
+    db: Session,
+    *,
+    from_id: str,
+    to_customer: Any,
+    tenant_no: str,
+) -> str:
+    source = get_customer(db, from_id, tenant_no=tenant_no)
+    if source is not None:
+        merge_customer_fields(to_customer, source)
+        db.add(to_customer)
+    repoint_customer_refs(db, from_id, to_customer.id, tenant_no)
+    customer_crud.soft_delete_customer(db, from_id, tenant_no=tenant_no)
+    return to_customer.id
+
+
+def _sync_one_role(
+    db: Session,
+    party: Dict[str, Any],
+    spec: Dict[str, Any],
+    tenant_no: str,
+    user_id: str,
+) -> str:
+    name = _pick(party, spec["name_keys"])
+    if not name:
+        return ""
+    phone = _pick(party, spec["phone_keys"])
+    id_card = _pick(party, spec["id_card_keys"])
+    linked_id = _pick(party, spec["id_keys"])
+    sex = _pick(party, spec["sex_keys"]) or spec["default_sex"]
+
+    found = find_customer_by_id_card(db, id_card, tenant_no) if id_card else None
+
+    if found is not None:
+        if linked_id and linked_id != found.id:
+            return _merge_and_repoint(db, from_id=linked_id, to_customer=found, tenant_no=tenant_no)
+        merge_customer_fields(
+            found,
+            SimpleNamespace(
+                phone_no=phone or None,
+                age=_age_from_id_card(id_card),
+                sex=sex,
+                remark=None,
+                customer_tags=None,
+            ),
+        )
+        db.add(found)
+        return found.id
+
+    if linked_id:
+        linked = get_customer(db, linked_id, tenant_no=tenant_no)
+        if linked is not None:
+            _apply_direct_identity(linked, name=name, phone=phone, id_card=id_card)
+            db.add(linked)
+            return linked.id
+
+    return _create_role_customer(
+        db,
+        name=name,
+        sex=sex,
+        id_card=id_card,
+        phone=phone,
+        tenant_no=tenant_no,
+        user_id=user_id,
+    )
+
+
+def sync_party_customers(db: Session, party_dict: Dict[str, Any], tenant_no: str, user_id: str) -> Dict[str, Any]:
+    out = dict(party_dict or {})
+    for spec in _ROLE_SPECS:
+        role_id = _sync_one_role(db, out, spec, tenant_no, user_id)
+        for key in spec["id_keys"]:
+            out[key] = role_id
+    return out

backend/tests/test_party_customer_sync.py

@@ -0,0 +1,177 @@
+"""当事人三角色客户同步 / 撞库合并"""
+import unittest
+from unittest.mock import MagicMock, patch
+
+from app.services.party_customer_sync import (
+    merge_customer_fields,
+    sync_party_customers,
+    ROLE_MOTHER,
+    ROLE_FATHER,
+    ROLE_NEWBORN,
+)
+
+
+class TestMergeCustomerFields(unittest.TestCase):
+    def test_fill_empty_only_and_union_tags(self):
+        target = MagicMock(
+            customer_name="已有",
+            phone_no="111",
+            age=None,
+            sex="2",
+            remark="",
+            customer_tags="vip",
+        )
+        source = MagicMock(
+            phone_no="222",
+            age=30,
+            sex="1",
+            remark="备注A",
+            customer_tags="a,vip",
+        )
+        merge_customer_fields(target, source)
+        self.assertEqual(target.phone_no, "111")  # 不覆盖
+        self.assertEqual(target.age, 30)
+        self.assertEqual(target.sex, "1")  # 目标为未知才补
+        self.assertEqual(target.remark, "备注A")
+        tags = set((target.customer_tags or "").split(","))
+        self.assertEqual(tags, {"vip", "a"})
+        self.assertEqual(target.customer_name, "已有")
+
+
+class TestSyncPartyCustomers(unittest.TestCase):
+    def test_role_constants(self):
+        self.assertEqual(ROLE_MOTHER, "mother")
+        self.assertEqual(ROLE_FATHER, "father")
+        self.assertEqual(ROLE_NEWBORN, "newborn")
+
+    def test_no_name_clears_role_id(self):
+        db = MagicMock()
+        party = {
+            "motherName": "",
+            "customerId": "old-m",
+            "fatherName": "",
+            "fatherCustomerId": "old-f",
+        }
+        out = sync_party_customers(db, party, tenant_no="zkxw", user_id="u1")
+        self.assertEqual(out.get("customer_id") or out.get("customerId"), "")
+        self.assertEqual(out.get("father_customer_id") or out.get("fatherCustomerId"), "")
+
+    def test_three_names_without_id_card_creates_three(self):
+        db = MagicMock()
+        created = []
+
+        def fake_create(db_, **kwargs):
+            obj = MagicMock()
+            obj.id = f"c{len(created)+1}"
+            created.append(kwargs)
+            return obj
+
+        party = {
+            "motherName": "妈",
+            "fatherName": "爸",
+            "newbornName": "宝",
+            "newbornSex": "0",
+        }
+        with patch("app.services.party_customer_sync.customer_crud.create_customer", side_effect=fake_create):
+            with patch("app.services.party_customer_sync.find_customer_by_id_card", return_value=None):
+                with patch("app.services.party_customer_sync.get_customer", return_value=None):
+                    out = sync_party_customers(db, party, tenant_no="zkxw", user_id="u1")
+        self.assertEqual(len(created), 3)
+        self.assertTrue(out.get("customer_id") or out.get("customerId"))
+        self.assertTrue(out.get("father_customer_id") or out.get("fatherCustomerId"))
+        self.assertTrue(out.get("newborn_customer_id") or out.get("newbornCustomerId"))
+        sexes = {c["customer_name"]: c.get("sex") for c in created}
+        self.assertEqual(sexes["妈"], "1")
+        self.assertEqual(sexes["爸"], "0")
+        self.assertEqual(sexes["宝"], "0")
+
+    def test_snake_case_input_clears_and_keeps_other_fields(self):
+        db = MagicMock()
+        party = {
+            "mother_name": "",
+            "customer_id": "old-m",
+            "remark": "keep-me",
+        }
+        out = sync_party_customers(db, party, tenant_no="zkxw", user_id="u1")
+        self.assertEqual(out.get("customer_id"), "")
+        self.assertEqual(out.get("remark"), "keep-me")
+
+    def test_name_and_id_card_keeps_linked_id_when_not_found(self):
+        db = MagicMock()
+        linked = MagicMock()
+        linked.id = "cust-a"
+        linked.customer_name = "旧名"
+        linked.phone_no = None
+        linked.id_card_no = None
+        party = {
+            "mother_name": "妈",
+            "mother_phone": "13900000000",
+            "mother_id_card": "110101199001011234",
+            "customer_id": "cust-a",
+        }
+        with patch("app.services.party_customer_sync.find_customer_by_id_card", return_value=None):
+            with patch("app.services.party_customer_sync.get_customer", return_value=linked):
+                with patch("app.services.party_customer_sync.customer_crud.create_customer") as create:
+                    out = sync_party_customers(db, party, tenant_no="zkxw", user_id="u1")
+        create.assert_not_called()
+        self.assertEqual(out.get("customer_id"), "cust-a")
+        self.assertEqual(linked.customer_name, "妈")
+        self.assertEqual(linked.phone_no, "13900000000")
+        self.assertEqual(linked.id_card_no, "110101199001011234")
+
+    def test_id_card_collision_merges_and_repoints(self):
+        db = MagicMock()
+        customer_a = MagicMock(
+            id="A",
+            customer_name="临时",
+            phone_no="111",
+            age=None,
+            sex="2",
+            remark="",
+            customer_tags="tmp",
+        )
+        customer_b = MagicMock(
+            id="B",
+            customer_name="已有",
+            phone_no="",
+            age=None,
+            sex="2",
+            remark="",
+            customer_tags="vip",
+        )
+        party = {
+            "motherName": "妈",
+            "motherIdCard": "110101199001011234",
+            "customerId": "A",
+        }
+
+        def fake_get(db_, customer_id, tenant_no=None):
+            mapping = {"A": customer_a, "B": customer_b}
+            return mapping.get(customer_id)
+
+        with patch("app.services.party_customer_sync.find_customer_by_id_card", return_value=customer_b):
+            with patch("app.services.party_customer_sync.get_customer", side_effect=fake_get):
+                with patch("app.services.party_customer_sync.repoint_customer_refs") as repoint:
+                    with patch(
+                        "app.services.party_customer_sync.customer_crud.soft_delete_customer"
+                    ) as soft_delete:
+                        out = sync_party_customers(db, party, tenant_no="zkxw", user_id="u1")
+        self.assertEqual(out.get("customer_id") or out.get("customerId"), "B")
+        repoint.assert_called_once()
+        args, kwargs = repoint.call_args
+        if args and len(args) >= 4:
+            self.assertEqual(args[1], "A")
+            self.assertEqual(args[2], "B")
+            self.assertEqual(args[3], "zkxw")
+        else:
+            self.assertEqual(kwargs.get("from_id"), "A")
+            self.assertEqual(kwargs.get("to_id"), "B")
+            self.assertEqual(kwargs.get("tenant_no"), "zkxw")
+        soft_delete.assert_called_once()
+        sd_args, sd_kwargs = soft_delete.call_args
+        deleted_id = sd_args[1] if len(sd_args) > 1 else sd_kwargs.get("customer_id")
+        self.assertEqual(deleted_id, "A")
+        self.assertEqual(customer_b.phone_no, "111")
+        self.assertEqual(customer_b.customer_name, "已有")
+        tags = set((customer_b.customer_tags or "").split(","))
+        self.assertEqual(tags, {"vip", "tmp"})