ForgeFlow-ERP/backend/app/services/contacts.py
2026-06-12 16:00:56 +08:00

81 lines
2.5 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

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)
]