diff --git a/backend/app/api/basic_info.py b/backend/app/api/basic_info.py index 9f5c7ff8..e969d646 100644 --- a/backend/app/api/basic_info.py +++ b/backend/app/api/basic_info.py @@ -24,6 +24,7 @@ from app.models.models import ( BizSampleCode, SysUser, ) +from app.services.customer_relations import build_customer_relations router = APIRouter(prefix="/web/admin", tags=["基础信息管理"]) @@ -119,6 +120,19 @@ async def list_customers( return success_response({"list": result_list, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp}) +@router.get("/customers/{customer_id}/relations") +async def get_customer_relations( + customer_id: str, + user: dict = Depends(require_menu_code("biz:customer:list")), + db: Session = Depends(get_db), +): + tenant_no = _get_tenant_no(db, user) + data = build_customer_relations(db, customer_id, tenant_no) + if not data: + return error_response(404, "客户不存在") + return success_response(data) + + @router.post("/customers") async def create_customer( body: CustomerCreateBody, diff --git a/backend/app/services/customer_relations.py b/backend/app/services/customer_relations.py new file mode 100644 index 00000000..1c59793c --- /dev/null +++ b/backend/app/services/customer_relations.py @@ -0,0 +1,209 @@ +"""客户详情：关联协议与同单家庭成员聚合。""" +from __future__ import annotations + +from datetime import datetime +from typing import Any, Dict, List, Optional, Tuple + +from sqlalchemy import or_ +from sqlalchemy.orm import Session + +from app.crud import customer as customer_crud +from app.models.models import BizCollectionProtocol, BizProtocolPartyCustomer +from app.models.research_donation import BizResearchDonation, BizResearchDonationParty + +_ROLE_SPECS = ( + { + "role": "mother", + "id_attr": "customer_id", + "name_attr": "mother_name", + "phone_attr": "mother_phone", + "id_card_attr": "mother_id_card", + }, + { + "role": "father", + "id_attr": "father_customer_id", + "name_attr": "father_name", + "phone_attr": "father_phone", + "id_card_attr": "father_id_card", + }, + { + "role": "newborn", + "id_attr": "newborn_customer_id", + "name_attr": "newborn_name", + "phone_attr": None, + "id_card_attr": "newborn_id_card", + }, +) + + +def resolve_my_role(party: Any, customer_id: str) -> str: + cid = (customer_id or "").strip() + if not cid: + return "" + if (getattr(party, "customer_id", None) or "").strip() == cid: + return "mother" + if (getattr(party, "father_customer_id", None) or "").strip() == cid: + return "father" + if (getattr(party, "newborn_customer_id", None) or "").strip() == cid: + return "newborn" + return "" + + +def party_members(party: Any) -> List[Dict[str, str]]: + out: List[Dict[str, str]] = [] + for spec in _ROLE_SPECS: + name = (getattr(party, spec["name_attr"], None) or "").strip() + if not name: + continue + phone_attr = spec["phone_attr"] + phone = (getattr(party, phone_attr, None) or "").strip() if phone_attr else "" + out.append( + { + "role": spec["role"], + "customerId": (getattr(party, spec["id_attr"], None) or "").strip(), + "name": name, + "phone": phone, + "idCardNo": (getattr(party, spec["id_card_attr"], None) or "").strip(), + } + ) + return out + + +def _fmt_date(val: Any) -> str: + if val is None: + return "" + if isinstance(val, datetime): + return val.strftime("%Y%m%d") + return str(val).strip() + + +def _group_sort_key(group: Dict[str, Any]) -> datetime: + ct = group.get("_createTime") + if isinstance(ct, datetime): + return ct + return datetime.min + + +def _storage_group(party: Any, protocol: Any, customer_id: str) -> Dict[str, Any]: + return { + "bizType": "cell_storage", + "protocolId": protocol.id, + "protocolNo": (protocol.protocol_no or "").strip(), + "protocolStatus": (protocol.protocol_status or "").strip(), + "registerDate": _fmt_date(protocol.register_date), + "myRole": resolve_my_role(party, customer_id), + "members": party_members(party), + "_createTime": protocol.create_time, + } + + +def _donation_group(party: Any, donation: Any, customer_id: str) -> Dict[str, Any]: + return { + "bizType": "research_donation", + "protocolId": donation.id, + "protocolNo": (donation.donation_no or "").strip(), + "protocolStatus": (donation.donation_status or "").strip(), + "registerDate": _fmt_date(donation.donation_date), + "myRole": resolve_my_role(party, customer_id), + "members": party_members(party), + "_createTime": donation.create_time, + } + + +def _fetch_storage_party_rows( + db: Session, customer_id: str, tenant_no: str +) -> List[Tuple[Any, Any]]: + cid = (customer_id or "").strip() + rows = ( + db.query(BizProtocolPartyCustomer, BizCollectionProtocol) + .join( + BizCollectionProtocol, + BizCollectionProtocol.id == BizProtocolPartyCustomer.collection_protocol_id, + ) + .filter( + BizProtocolPartyCustomer.tenant_no == tenant_no, + BizProtocolPartyCustomer.del_flag == 0, + BizCollectionProtocol.tenant_no == tenant_no, + BizCollectionProtocol.del_flag == 0, + or_( + BizProtocolPartyCustomer.customer_id == cid, + BizProtocolPartyCustomer.father_customer_id == cid, + BizProtocolPartyCustomer.newborn_customer_id == cid, + ), + ) + .all() + ) + return list(rows) + + +def _fetch_donation_party_rows( + db: Session, customer_id: str, tenant_no: str +) -> List[Tuple[Any, Any]]: + cid = (customer_id or "").strip() + rows = ( + db.query(BizResearchDonationParty, BizResearchDonation) + .join( + BizResearchDonation, + BizResearchDonation.id == BizResearchDonationParty.research_donation_id, + ) + .filter( + BizResearchDonationParty.tenant_no == tenant_no, + BizResearchDonationParty.del_flag == 0, + BizResearchDonation.tenant_no == tenant_no, + BizResearchDonation.del_flag == 0, + or_( + BizResearchDonationParty.customer_id == cid, + BizResearchDonationParty.father_customer_id == cid, + BizResearchDonationParty.newborn_customer_id == cid, + ), + ) + .all() + ) + return list(rows) + + +def build_customer_relations( + db: Session, customer_id: str, tenant_no: str +) -> Optional[Dict[str, Any]]: + customer = customer_crud.get_customer(db, customer_id, tenant_no=tenant_no) + if not customer: + return None + + groups: List[Dict[str, Any]] = [] + for party, protocol in _fetch_storage_party_rows(db, customer_id, tenant_no): + groups.append(_storage_group(party, protocol, customer_id)) + for party, donation in _fetch_donation_party_rows(db, customer_id, tenant_no): + groups.append(_donation_group(party, donation, customer_id)) + + groups.sort(key=_group_sort_key, reverse=True) + + related_groups: List[Dict[str, Any]] = [] + protocols: List[Dict[str, Any]] = [] + for g in groups: + related_groups.append( + { + "bizType": g["bizType"], + "protocolId": g["protocolId"], + "protocolNo": g["protocolNo"], + "protocolStatus": g["protocolStatus"], + "registerDate": g["registerDate"], + "myRole": g["myRole"], + "members": g["members"], + } + ) + protocols.append( + { + "bizType": g["bizType"], + "protocolId": g["protocolId"], + "protocolNo": g["protocolNo"], + "protocolStatus": g["protocolStatus"], + "registerDate": g["registerDate"], + "myRole": g["myRole"], + } + ) + + return { + "customer": customer_crud._row_to_dict(customer), + "relatedGroups": related_groups, + "protocols": protocols, + } diff --git a/backend/tests/test_customer_relations.py b/backend/tests/test_customer_relations.py new file mode 100644 index 00000000..fd0dad9f --- /dev/null +++ b/backend/tests/test_customer_relations.py @@ -0,0 +1,183 @@ +"""客户关联协议 / 同单家庭成员聚合""" +import unittest +from datetime import datetime +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from app.services.customer_relations import ( + build_customer_relations, + party_members, + resolve_my_role, +) + + +class TestResolveMyRole(unittest.TestCase): + def test_mother_father_newborn(self): + party = SimpleNamespace( + customer_id="m1", + father_customer_id="f1", + newborn_customer_id="n1", + ) + self.assertEqual(resolve_my_role(party, "m1"), "mother") + self.assertEqual(resolve_my_role(party, "f1"), "father") + self.assertEqual(resolve_my_role(party, "n1"), "newborn") + + def test_unknown_returns_empty(self): + party = SimpleNamespace( + customer_id="m1", + father_customer_id=None, + newborn_customer_id="", + ) + self.assertEqual(resolve_my_role(party, "x"), "") + + +class TestPartyMembers(unittest.TestCase): + def test_includes_named_roles_only(self): + party = SimpleNamespace( + customer_id="m1", + mother_name="妈", + mother_phone="1", + mother_id_card="idm", + father_customer_id="f1", + father_name="爸", + father_phone="2", + father_id_card="idf", + newborn_customer_id=None, + newborn_name="", + newborn_id_card="", + ) + members = party_members(party) + roles = [m["role"] for m in members] + self.assertEqual(roles, ["mother", "father"]) + self.assertEqual(members[0]["customerId"], "m1") + self.assertEqual(members[0]["name"], "妈") + self.assertEqual(members[1]["name"], "爸") + + def test_name_without_customer_id_still_listed(self): + party = SimpleNamespace( + customer_id=None, + mother_name="妈", + mother_phone="", + mother_id_card="", + father_customer_id=None, + father_name="", + father_phone="", + father_id_card="", + newborn_customer_id=None, + newborn_name="宝", + newborn_id_card="", + ) + members = party_members(party) + self.assertEqual(len(members), 2) + newborn = next(m for m in members if m["role"] == "newborn") + self.assertEqual(newborn["customerId"], "") + self.assertEqual(newborn["name"], "宝") + + +class TestBuildCustomerRelations(unittest.TestCase): + def test_missing_customer_returns_none(self): + db = MagicMock() + with patch("app.services.customer_relations.customer_crud.get_customer", return_value=None): + self.assertIsNone(build_customer_relations(db, "c1", "zkxw")) + + def test_two_protocols_sorted_and_roles(self): + db = MagicMock() + customer = SimpleNamespace(id="m1", customer_name="妈") + older = ( + SimpleNamespace( + customer_id="m1", + father_customer_id="f1", + newborn_customer_id="n1", + mother_name="妈", + mother_phone="", + mother_id_card="", + father_name="爸", + father_phone="", + father_id_card="", + newborn_name="宝", + newborn_id_card="", + ), + SimpleNamespace( + id="p-old", + protocol_no="P001", + protocol_status="sxz", + register_date="20260101", + create_time=datetime(2026, 1, 1, 10, 0, 0), + ), + ) + newer = ( + SimpleNamespace( + customer_id="m1", + father_customer_id=None, + newborn_customer_id=None, + mother_name="妈", + mother_phone="", + mother_id_card="", + father_name="", + father_phone="", + father_id_card="", + newborn_name="", + newborn_id_card="", + ), + SimpleNamespace( + id="p-new", + protocol_no="P002", + protocol_status="dsx", + register_date="20260301", + create_time=datetime(2026, 3, 1, 10, 0, 0), + ), + ) + donation = ( + SimpleNamespace( + customer_id=None, + father_customer_id="m1", + newborn_customer_id=None, + mother_name="他妈", + mother_phone="", + mother_id_card="", + father_name="爸本人", + father_phone="", + father_id_card="", + newborn_name="", + newborn_id_card="", + ), + SimpleNamespace( + id="d1", + donation_no="D001", + donation_status="sxz", + donation_date="20260201", + create_time=datetime(2026, 2, 1, 10, 0, 0), + ), + ) + + with patch("app.services.customer_relations.customer_crud.get_customer", return_value=customer): + with patch("app.services.customer_relations.customer_crud._row_to_dict", return_value={"id": "m1"}): + with patch( + "app.services.customer_relations._fetch_storage_party_rows", + return_value=[older, newer], + ): + with patch( + "app.services.customer_relations._fetch_donation_party_rows", + return_value=[donation], + ): + data = build_customer_relations(db, "m1", "zkxw") + + self.assertIsNotNone(data) + groups = data["relatedGroups"] + self.assertEqual(len(groups), 3) + # create_time desc: p-new, d1, p-old + self.assertEqual(groups[0]["protocolId"], "p-new") + self.assertEqual(groups[0]["bizType"], "cell_storage") + self.assertEqual(groups[0]["myRole"], "mother") + self.assertEqual(groups[1]["protocolId"], "d1") + self.assertEqual(groups[1]["bizType"], "research_donation") + self.assertEqual(groups[1]["myRole"], "father") + self.assertEqual(groups[2]["protocolNo"], "P001") + self.assertEqual(len(groups[2]["members"]), 3) + + protocols = data["protocols"] + self.assertEqual(len(protocols), 3) + self.assertEqual(protocols[0]["protocolNo"], "P002") + self.assertEqual(protocols[0]["registerDate"], "20260301") + self.assertEqual(protocols[1]["protocolNo"], "D001") + self.assertEqual(protocols[1]["myRole"], "father")