feat: add customer relations API for family and protocols

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

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,

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,
+    }

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")