zkxw/backend/app/services/customer_relations.py

"""客户详情：关联协议与同单家庭成员聚合。"""
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,
    }