zkxw/backend/app/services/party_customer_sync.py

"""协议当事人三角色客户同步：新建、补证、撞库合并改挂。"""
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"),
        "birth_ym_keys": ("mother_birth_ym", "motherBirthYm"),
        "ethnic_keys": ("mother_ethnic", "motherEthnic"),
        "email_keys": ("mother_email", "motherEmail"),
        "address_keys": ("mother_address", "motherAddress"),
        "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"),
        "birth_ym_keys": (),
        "ethnic_keys": (),
        "email_keys": (),
        "address_keys": (),
        "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"),
        "birth_ym_keys": (),
        "ethnic_keys": (),
        "email_keys": (),
        "address_keys": (),
        "sex_keys": ("newborn_sex", "newbornSex"),
        "default_sex": "2",
    },
)

_PROFILE_ATTRS = (
    "customer_name",
    "phone_no",
    "id_card_no",
    "sex",
    "age",
    "birth_ym",
    "ethnic",
    "email",
    "address",
)


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 apply_role_profile_overwrite(target: Any, profile: Any) -> None:
    """协议非空字段覆盖客户主档；空串/None 跳过，不碰 tags/remark。"""
    for attr in _PROFILE_ATTRS:
        val = getattr(profile, attr, None)
        if _is_empty(val):
            continue
        if attr == "sex":
            setattr(target, attr, str(val).strip()[:1])
        elif attr == "birth_ym":
            setattr(target, attr, str(val).strip()[:6])
        elif attr == "age":
            setattr(target, attr, val)
        else:
            setattr(target, attr, str(val).strip())


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
    for attr in ("birth_ym", "ethnic", "email", "address"):
        if _is_empty(getattr(target, attr, None)):
            src_val = getattr(source, attr, None)
            if not _is_empty(src_val):
                setattr(target, attr, src_val)
    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 _build_role_profile(party: Dict[str, Any], spec: Dict[str, Any]) -> SimpleNamespace:
    name = _pick(party, spec["name_keys"])
    phone = _pick(party, spec["phone_keys"])
    id_card = _pick(party, spec["id_card_keys"])
    sex = _pick(party, spec["sex_keys"]) or spec["default_sex"]
    birth_ym = _pick(party, spec.get("birth_ym_keys") or ())
    ethnic = _pick(party, spec.get("ethnic_keys") or ())
    email = _pick(party, spec.get("email_keys") or ())
    address = _pick(party, spec.get("address_keys") or ())
    return SimpleNamespace(
        customer_name=name or None,
        phone_no=phone or None,
        id_card_no=id_card or None,
        sex=sex or None,
        age=_age_from_id_card(id_card),
        birth_ym=birth_ym or None,
        ethnic=ethnic or None,
        email=email or None,
        address=address or None,
    )


def _create_role_customer(
    db: Session,
    *,
    profile: SimpleNamespace,
    tenant_no: str,
    user_id: str,
    commit: bool = True,
    create_dept_id: str = "",
) -> str:
    obj = customer_crud.create_customer(
        db,
        customer_name=profile.customer_name,
        sex=profile.sex or "2",
        id_card_no=profile.id_card_no or None,
        age=profile.age,
        birth_ym=profile.birth_ym,
        ethnic=profile.ethnic,
        email=profile.email,
        address=profile.address,
        phone_no=profile.phone_no or None,
        tenant_no=tenant_no,
        user_id=user_id,
        commit=commit,
        create_dept_id=create_dept_id,
    )
    return obj.id


def _merge_and_repoint(
    db: Session,
    *,
    from_id: str,
    to_customer: Any,
    tenant_no: str,
    profile: Optional[SimpleNamespace] = None,
    commit: bool = True,
) -> 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)
    if profile is not None:
        apply_role_profile_overwrite(to_customer, profile)
        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, commit=commit
    )
    return to_customer.id


def _sync_one_role(
    db: Session,
    party: Dict[str, Any],
    spec: Dict[str, Any],
    tenant_no: str,
    user_id: str,
    commit: bool = True,
    create_dept_id: str = "",
) -> str:
    profile = _build_role_profile(party, spec)
    if _is_empty(profile.customer_name):
        return ""
    id_card = profile.id_card_no or ""
    linked_id = _pick(party, spec["id_keys"])

    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,
                profile=profile,
                commit=commit,
            )
        apply_role_profile_overwrite(found, profile)
        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_role_profile_overwrite(linked, profile)
            db.add(linked)
            return linked.id

    return _create_role_customer(
        db,
        profile=profile,
        tenant_no=tenant_no,
        user_id=user_id,
        commit=commit,
        create_dept_id=create_dept_id,
    )


def sync_party_customers(
    db: Session,
    party_dict: Dict[str, Any],
    tenant_no: str,
    user_id: str,
    *,
    commit: bool = True,
    create_dept_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,
            commit=commit,
            create_dept_id=create_dept_id,
        )
        for key in spec["id_keys"]:
            out[key] = role_id
    return out