51 lines
1.5 KiB
Python
51 lines
1.5 KiB
Python
"""客户服务快捷登记：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
|