diff --git a/backend/app/services/party_customer_sync.py b/backend/app/services/party_customer_sync.py new file mode 100644 index 00000000..963b5b57 --- /dev/null +++ b/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 diff --git a/backend/tests/test_party_customer_sync.py b/backend/tests/test_party_customer_sync.py new file mode 100644 index 00000000..42d846bc --- /dev/null +++ b/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"})