zkxw/backend/app/services/sample_customer_service_register.py

"""客户服务快捷登记：payload 规范化与字典回写（逐步扩展）。"""
from typing import Any, Dict, FrozenSet, Optional

CUSTOMER_SERVICE_FIELD_KEYS: FrozenSet[str] = frozenset(
    {
        "sms_send_date",
        "sms_send_status",
        "microbe_followup_date",
        "microbe_followup_status",
        "fullmoon_followup_date",
        "fullmoon_followup_status",
        "household_registered",
        "einvoice_send_time",
        "effective_address",
    }
)


def normalize_dict_value(raw) -> Optional[str]:
    """Trim; empty/None -> None."""
    if raw is None:
        return None
    text = str(raw).strip()
    return text or None


def _is_empty_merge_value(value: Any) -> bool:
    if value is None:
        return True
    if isinstance(value, str):
        return not value.strip()
    return False


def merge_customer_service_fields(
    existing: Optional[Dict[str, Any]],
    incoming: Optional[Dict[str, Any]],
) -> Dict[str, Any]:
    """Merge incoming onto existing; empty/None/whitespace incoming values do not overwrite."""
    existing = existing or {}
    incoming = incoming or {}
    keys = (set(existing.keys()) | set(incoming.keys())) & CUSTOMER_SERVICE_FIELD_KEYS
    merged: Dict[str, Any] = {k: existing.get(k) for k in keys}
    for key, value in incoming.items():
        if key not in CUSTOMER_SERVICE_FIELD_KEYS:
            continue
        if _is_empty_merge_value(value):
            continue
        merged[key] = value
    return merged


def has_any_customer_service_payload(incoming: Optional[Dict[str, Any]]) -> bool:
    """True if any known customer-service field has a non-empty value."""
    incoming = incoming or {}
    for key in CUSTOMER_SERVICE_FIELD_KEYS:
        if key not in incoming:
            continue
        if not _is_empty_merge_value(incoming.get(key)):
            return True
    return False