from __future__ import annotations import re from typing import Any CONTACT_SEPARATOR = "、" MAX_CONTACTS = 3 def _clean_text(value: Any) -> str: return str(value or "").strip() def _entry_value(entry: Any, *keys: str) -> str: for key in keys: if isinstance(entry, dict): value = entry.get(key) else: value = getattr(entry, key, None) cleaned = _clean_text(value) if cleaned: return cleaned return "" def _split_contacts(value: str | None) -> list[str]: text = _clean_text(value) if not text: return [] return [part.strip() for part in re.split(r"[、,,;;\n]+", text) if part.strip()] def normalize_contacts( contacts: list[Any] | None, *, fallback_name: str | None = None, fallback_phone: str | None = None, max_contacts: int = MAX_CONTACTS, ) -> list[dict[str, str]]: normalized: list[dict[str, str]] = [] for entry in contacts or []: contact_name = _entry_value(entry, "contact_name", "name") contact_phone = _entry_value(entry, "contact_phone", "phone") if not contact_name and not contact_phone: continue normalized.append({"contact_name": contact_name, "contact_phone": contact_phone}) if not normalized: contact_name = _clean_text(fallback_name) contact_phone = _clean_text(fallback_phone) if contact_name or contact_phone: normalized.append({"contact_name": contact_name, "contact_phone": contact_phone}) if len(normalized) > max_contacts: raise ValueError(f"联系人最多维护 {max_contacts} 位") return normalized def serialize_contacts(contacts: list[dict[str, str]]) -> tuple[str | None, str | None]: names = [item["contact_name"] for item in contacts if item.get("contact_name")] phones = [item["contact_phone"] for item in contacts if item.get("contact_phone")] return ( CONTACT_SEPARATOR.join(names) if names else None, CONTACT_SEPARATOR.join(phones) if phones else None, ) def parse_contacts(contact_name: str | None, contact_phone: str | None, *, max_contacts: int = MAX_CONTACTS) -> list[dict[str, str]]: names = _split_contacts(contact_name) phones = _split_contacts(contact_phone) total = min(max(len(names), len(phones)), max_contacts) return [ { "contact_name": names[index] if index < len(names) else "", "contact_phone": phones[index] if index < len(phones) else "", } for index in range(total) ]