from __future__ import annotations from typing import Any, Mapping from .config import CrmConfig from .filters import ( build_and_filter, build_date_range_condition, build_eq_condition, build_relation_id_condition, build_relation_id_or_condition, build_search_filter, ) from .nb_client import NbClient, NbCommandError DEFAULT_PAGE_SIZE = 20 MAX_PAGE_SIZE = 100 CONTENT_PREVIEW_CHARS = 500 CUSTOMER_STATUS_LABELS = { "kh_p_not_start": "未开始", "kh_p_inprogress": "进行中", "kh_p_ordered": "已下单", "kh_p_audited": "已审核", } CUSTOMER_NAME_KEYS = ("f_q2q60q7bqbg", "name", "title", "code", "id") CONTACT_NAME_KEYS = ("f_2zpg91gtmwp", "name", "phone", "email", "title", "id") USER_LABEL_KEYS = ("nickname", "username", "email", "f_yh_english_name", "name", "id") FOLLOWUP_CONTENT_KEYS = ("f_19iplxz0nds", "content", "description", "remark") FOLLOWUP_DATE_KEYS = ("f_srysmjkwfq1", "followedAt", "createdAt", "updatedAt") class CrmService: def __init__(self, config: CrmConfig, client: NbClient) -> None: self.config = config self.client = client def crm_health(self) -> dict[str, Any]: health = self.client.health() configured = { "env": self.config.env_name, "data_source": self.config.data_source, "collections": { "customers": self.config.customer.resource, "contacts": self.config.contact.resource, "followups": self.config.followup.resource, }, } return {"ok": bool(health.get("ok")), "nb": health, "configured": configured} def crm_schema(self, include_fields: bool = False) -> dict[str, Any]: schema: dict[str, Any] = { "env": self.config.env_name, "data_source": self.config.data_source, "collections": { "customers": self.config.customer.__dict__, "contacts": self.config.contact.__dict__, "followups": self.config.followup.__dict__, "users": self.config.user.__dict__, }, } if include_fields: schema["metadata"] = self._safe_metadata() return schema def crm_config_template(self) -> dict[str, Any]: metadata = self._safe_metadata() fields_by_kind = metadata.get("fields", {}) if isinstance(metadata, dict) else {} customer_fields = _field_list(fields_by_kind.get("customers")) contact_fields = _field_list(fields_by_kind.get("contacts")) followup_fields = _field_list(fields_by_kind.get("followups")) customer_resource = self.config.customer.resource contact_resource = self.config.contact.resource return { "env": self.config.env_name, "dataSource": self.config.data_source, "users": { "resource": self.config.user.resource, "fields": self.config.user.fields, "matchFields": self.config.user.match_fields, }, "collections": { "customers": { "resource": customer_resource, "fields": _field_names(customer_fields), "searchFields": _configured_or_search_fields( customer_fields, ["f_q2q60q7bqbg", "f_kh_english_name", "f_territorial_name"], ), "statusField": _find_named_field(customer_fields, ["kh_progress"]) or _find_field(customer_fields, ["进度", "状态", "status"]), "sourceField": _find_named_field(customer_fields, ["f_225h6n6v654"]) or _find_field(customer_fields, ["客户来源", "来源", "source"]), "ownerField": _find_named_field(customer_fields, ["kehu_followup_person"]) or self.config.customer.owner_field, "sort": self.config.customer.sort, }, "contacts": { "resource": contact_resource, "fields": _field_names(contact_fields), "searchFields": _configured_or_search_fields( contact_fields, ["f_2zpg91gtmwp", "khlxr_phone_number", "khlxr_email"], ), "customerField": _find_named_field(contact_fields, ["kehulianxiren_client"]) or _relation_field(contact_fields, ["公司", "客户"]) or self.config.contact.customer_field, "sort": self.config.contact.sort, }, "followups": { "resource": self.config.followup.resource, "fields": _field_names(followup_fields), "searchFields": _configured_or_search_fields(followup_fields, ["f_19iplxz0nds"]), "customerField": _find_named_field(followup_fields, ["gjjl_client"]) or _relation_field(followup_fields, ["客户"]) or self.config.followup.customer_field, "contactField": _find_named_field(followup_fields, ["gjjl_client_contacts"]) or _relation_field(followup_fields, ["客户联系人", "联系人"]) or self.config.followup.contact_field, "ownerField": _find_named_field(followup_fields, ["gjjl_followup_person"]) or self.config.followup.owner_field, "ownerFields": _existing_field_names( followup_fields, ["gjjl_followup_person", "createdBy"], ) or self.config.followup.owner_fields, "dateField": _find_named_field(followup_fields, ["f_srysmjkwfq1"]) or _date_field(followup_fields) or self.config.followup.date_field, "sort": _template_sort( _find_named_field(followup_fields, ["f_srysmjkwfq1"]) or _date_field(followup_fields) or self.config.followup.date_field, self.config.followup.sort, ), }, }, "metadata": { "field_errors": metadata.get("field_errors", {}) if isinstance(metadata, dict) else {}, "note": "Review inferred owner/status/source/date fields before using this template in production.", }, } def crm_overview( self, date_from: str | None = None, date_to: str | None = None, owner: str | None = None, ) -> dict[str, Any]: owner_id, owner_error = self._resolve_owner_id_or_error(owner) if owner_error: return {"error": owner_error} customer_filter = build_and_filter([build_relation_id_condition(self.config.customer.owner_field, owner_id)]) followup_filter = build_and_filter( [ build_relation_id_or_condition(self.config.followup.owner_fields, owner_id), build_date_range_condition(self.config.followup.date_field, date_from, date_to), ] ) totals = { "customers": self._count(self.config.customer.resource, self.config.customer.id_field, customer_filter), "contacts": self._count(self.config.contact.resource, self.config.contact.id_field, {}), "followups": self._count(self.config.followup.resource, self.config.followup.id_field, followup_filter), } overview = { "totals": totals, "customer_status": self._group_count( self.config.customer.resource, self.config.customer.id_field, self.config.customer.status_field, customer_filter, ), "customer_source": self._group_count( self.config.customer.resource, self.config.customer.id_field, self.config.customer.source_field, customer_filter, ), "customer_owner": self._group_count( self.config.customer.resource, self.config.customer.id_field, [self.config.customer.owner_field, "id"], customer_filter, ), "recent_followups": self._compact_followups( self._data( self.client.list_records( self.config.followup.resource, filter=followup_filter, fields=self.config.followup.fields, sort=self.config.followup.sort, page=1, page_size=10, ) ) ), } overview["customer_status"] = self._decorate_status_counts(overview["customer_status"]) overview["customer_source"] = self._decorate_source_counts(overview["customer_source"]) overview["customer_owner"] = self._decorate_owner_counts(overview["customer_owner"]) return overview def crm_filter_options(self, owner_query: str | None = None) -> dict[str, Any]: owner_candidates: list[dict[str, Any]] = [] if (owner_query or "").strip(): owner_candidates = self.resolve_owner(owner_query or "")["results"] return { "customer_status": self._decorate_status_counts( self._group_count( self.config.customer.resource, self.config.customer.id_field, self.config.customer.status_field, {}, ) ), "customer_source": self._decorate_source_counts( self._group_count( self.config.customer.resource, self.config.customer.id_field, self.config.customer.source_field, {}, ) ), "owner_candidates": owner_candidates, "max_limit": MAX_PAGE_SIZE, } def resolve_owner(self, query: str) -> dict[str, Any]: normalized = (query or "").strip() if not normalized: return {"query": "", "results": [], "count": 0} rows = self._owner_matches(normalized) results = [_format_user_candidate(row, self.config.user.id_field) for row in rows] return {"query": normalized, "results": results, "count": len(results)} def list_customers( self, query: str | None = None, status: str | None = None, owner: str | None = None, source: str | None = None, limit: int = 20, offset: int = 0, include_raw: bool = False, ) -> dict[str, Any]: limit, offset = _normalize_pagination(limit, offset) owner_id, owner_error = self._resolve_owner_id_or_error(owner) if owner_error: return {"error": owner_error} filter_value = build_and_filter( [ build_eq_condition(self.config.customer.status_field, status), build_relation_id_condition(self.config.customer.owner_field, owner_id), build_eq_condition(self.config.customer.source_field, source), build_search_filter(query, self.config.customer.search_fields), ] ) response = self.client.list_records( self.config.customer.resource, filter=filter_value, fields=self.config.customer.fields if include_raw else self._customer_compact_fields(), sort=self.config.customer.sort, page=_page_from_offset(limit, offset), page_size=limit, ) result = _paginated_response(response, limit, offset) result["results"] = self._compact_customers(result["results"], include_raw=include_raw) return result def get_customer( self, customer_id: Any, include_contacts: bool = True, include_recent_followups: bool = True, followup_limit: int = 10, include_raw_followups: bool = False, ) -> dict[str, Any]: customer = self._single( self.client.get_record( self.config.customer.resource, customer_id, fields=self.config.customer.fields, ) ) result: dict[str, Any] = {"customer": customer} if include_contacts: result["contacts"] = self.list_contacts(customer_id=customer_id, limit=50, offset=0)["results"] if include_recent_followups: result["recent_followups"] = self.list_followups( customer_id=customer_id, limit=followup_limit, offset=0, include_raw=include_raw_followups, )["results"] return result def list_contacts( self, customer_id: Any | None = None, query: str | None = None, limit: int = 20, offset: int = 0, include_raw: bool = False, ) -> dict[str, Any]: limit, offset = _normalize_pagination(limit, offset) filter_value = build_and_filter( [ build_relation_id_condition(self.config.contact.customer_field, customer_id), build_search_filter(query, self.config.contact.search_fields), ] ) response = self.client.list_records( self.config.contact.resource, filter=filter_value, fields=self.config.contact.fields if include_raw else self._contact_compact_fields(), sort=self.config.contact.sort, page=_page_from_offset(limit, offset), page_size=limit, ) result = _paginated_response(response, limit, offset) result["results"] = self._compact_contacts(result["results"], include_raw=include_raw) return result def list_followups( self, customer_id: Any | None = None, contact_id: Any | None = None, owner: str | None = None, date_from: str | None = None, date_to: str | None = None, limit: int = 20, offset: int = 0, include_raw: bool = False, ) -> dict[str, Any]: limit, offset = _normalize_pagination(limit, offset) owner_id, owner_error = self._resolve_owner_id_or_error(owner) if owner_error: return {"error": owner_error} filter_value = build_and_filter( [ build_relation_id_condition(self.config.followup.customer_field, customer_id), build_relation_id_condition(self.config.followup.contact_field, contact_id), build_relation_id_or_condition(self.config.followup.owner_fields, owner_id), build_date_range_condition(self.config.followup.date_field, date_from, date_to), ] ) response = self.client.list_records( self.config.followup.resource, filter=filter_value, fields=self.config.followup.fields if include_raw else self._followup_compact_fields(), sort=self.config.followup.sort, page=_page_from_offset(limit, offset), page_size=limit, ) result = _paginated_response(response, limit, offset) result["results"] = self._compact_followups(result["results"], include_raw=include_raw) return result def _safe_metadata(self) -> dict[str, Any]: collections = { "customers": self.config.customer.resource, "contacts": self.config.contact.resource, "followups": self.config.followup.resource, } fields: dict[str, list[dict[str, Any]]] = {} field_errors: dict[str, dict[str, str]] = {} for label, collection_name in collections.items(): try: response = self.client.list_collection_fields( collection_name, page_size=500, paginate=False, ) fields[label] = self._data(response) except Exception as error: field_errors[label] = _metadata_error(error) return {"collections": collections, "fields": fields, "field_errors": field_errors} def _owner_matches(self, owner: str) -> list[dict[str, Any]]: normalized = owner.strip() if not normalized: return [] rows = self._data( self.client.list_records( self.config.user.resource, filter=build_search_filter(normalized, self.config.user.match_fields), fields=self.config.user.fields, page=1, page_size=50, ) ) return _pick_user_matches(rows, normalized, self.config.user.match_fields or []) def _resolve_owner_id_or_error(self, owner: str | None) -> tuple[Any | None, dict[str, Any] | None]: normalized = (owner or "").strip() if not normalized: return None, None matches = self._owner_matches(normalized) candidates = [_format_user_candidate(row, self.config.user.id_field) for row in matches[:10]] if not matches: return None, { "code": "owner_not_found", "message": f"No user matched owner {normalized!r}", "query": normalized, "candidates": [], } if len(matches) > 1: return None, { "code": "owner_ambiguous", "message": f"Multiple users matched owner {normalized!r}", "query": normalized, "candidates": candidates, } user_id = matches[0].get(self.config.user.id_field) if user_id is None or user_id == "": return None, { "code": "owner_invalid", "message": f"Matched owner {normalized!r} has no {self.config.user.id_field!r}", "query": normalized, "candidates": candidates, } return user_id, None def _resolve_owner_id(self, owner: str | None) -> Any | None: owner_id, error = self._resolve_owner_id_or_error(owner) if error: raise ValueError(str(error["message"])) return owner_id def _count(self, resource: str, id_field: str, filter_value: Mapping[str, Any]) -> int: response = self.client.query_records( resource, measures=[{"field": [id_field], "aggregation": "count", "alias": "count"}], filter=filter_value, limit=1, ) return _first_numeric(self._data(response), "count") def _group_count( self, resource: str, id_field: str, dimension_field: str | list[str], filter_value: Mapping[str, Any], ) -> list[dict[str, Any]]: if not dimension_field: return [] dimension_path = _field_path(dimension_field) dimension_alias = "_".join(dimension_path) response = self.client.query_records( resource, dimensions=[{"field": dimension_path, "alias": dimension_alias}], measures=[{"field": [id_field], "aggregation": "count", "alias": "count"}], orders=[{"field": dimension_path, "alias": dimension_alias, "order": "asc"}], filter=filter_value, limit=50, ) return self._data(response) def _decorate_status_counts(self, rows: list[dict[str, Any]]) -> list[dict[str, Any]]: return _decorate_dimension_counts( rows, preferred_keys=[self.config.customer.status_field, "kh_progress", "status"], labels=CUSTOMER_STATUS_LABELS, display_key="display", ) def _decorate_source_counts(self, rows: list[dict[str, Any]]) -> list[dict[str, Any]]: return _decorate_dimension_counts( rows, preferred_keys=[self.config.customer.source_field, "f_225h6n6v654", "source"], labels={}, ) def _decorate_owner_counts(self, rows: list[dict[str, Any]]) -> list[dict[str, Any]]: owner_alias = "_".join(_field_path([self.config.customer.owner_field, "id"])) decorated: list[dict[str, Any]] = [] for row in rows: value, _key = _dimension_value(row, [owner_alias, "owner_id", "kehu_followup_person_id"]) item = dict(row) if value is not None: item["owner_id"] = value item.setdefault("value", value) decorated.append(item) return decorated def _compact_customers(self, rows: list[dict[str, Any]], *, include_raw: bool = False) -> list[dict[str, Any]]: return [self._compact_customer(row, include_raw=include_raw) for row in rows] def _customer_compact_fields(self) -> list[str]: return _compact_field_list( self.config.customer.fields, [ self.config.customer.id_field, "f_q2q60q7bqbg", "name", "title", *(self.config.customer.search_fields or []), "f_kh_english_name", "english_name", "code", "f_territorial_name", "territorial_name", self.config.customer.status_field, "kh_progress", "status", self.config.customer.source_field, "f_225h6n6v654", "source", self.config.customer.owner_field, "kehu_followup_person", "owner", "createdBy", "createdAt", "updatedAt", ], ) def _compact_customer(self, row: Mapping[str, Any], *, include_raw: bool = False) -> dict[str, Any]: status = _first_value(row, [self.config.customer.status_field, "kh_progress", "status"]) source = _first_value(row, [self.config.customer.source_field, "f_225h6n6v654", "source"]) compact: dict[str, Any] = { "id": row.get(self.config.customer.id_field), "name": _first_text(row, ["f_q2q60q7bqbg", "name", "title", *(self.config.customer.search_fields or [])]), "english_name": _first_text(row, ["f_kh_english_name", "english_name", "code"]), "territorial_name": _first_text(row, ["f_territorial_name", "territorial_name"]), "status": status, "status_display": CUSTOMER_STATUS_LABELS.get(status, str(status) if status is not None else None), "source": source, "owner_names": _relation_labels( row, [self.config.customer.owner_field, "kehu_followup_person", "owner", "createdBy"], USER_LABEL_KEYS, ), "createdAt": row.get("createdAt"), "updatedAt": row.get("updatedAt"), } if include_raw: compact["raw"] = dict(row) return compact def _compact_contacts(self, rows: list[dict[str, Any]], *, include_raw: bool = False) -> list[dict[str, Any]]: return [self._compact_contact(row, include_raw=include_raw) for row in rows] def _contact_compact_fields(self) -> list[str]: return _compact_field_list( self.config.contact.fields, [ self.config.contact.id_field, "f_2zpg91gtmwp", "name", *(self.config.contact.search_fields or []), "khlxr_phone_number", "phone", "khlxr_email", "email", "f_tn8zdm2f938", "title", "position", self.config.contact.customer_field, "kehulianxiren_client", "customer", "client", "createdAt", "updatedAt", ], ) def _compact_contact(self, row: Mapping[str, Any], *, include_raw: bool = False) -> dict[str, Any]: compact: dict[str, Any] = { "id": row.get(self.config.contact.id_field), "name": _first_text(row, [*(self.config.contact.search_fields or []), "f_2zpg91gtmwp", "name", "title"]), "phone": _first_text(row, ["khlxr_phone_number", "phone"]), "email": _first_text(row, ["khlxr_email", "email"]), "title": _first_text(row, ["f_tn8zdm2f938", "title", "position"]), "customer_names": _relation_labels_first_available( row, [self.config.contact.customer_field, "kehulianxiren_client", "customer", "client"], CUSTOMER_NAME_KEYS, ), "createdAt": row.get("createdAt"), "updatedAt": row.get("updatedAt"), } if include_raw: compact["raw"] = dict(row) return compact def _compact_followups(self, rows: list[dict[str, Any]], *, include_raw: bool = False) -> list[dict[str, Any]]: return [self._compact_followup(row, include_raw=include_raw) for row in rows] def _followup_compact_fields(self) -> list[str]: return _compact_field_list( self.config.followup.fields, [ self.config.followup.id_field, self.config.followup.date_field, "f_srysmjkwfq1", "followedAt", *(self.config.followup.search_fields or []), *FOLLOWUP_CONTENT_KEYS, self.config.followup.customer_field, "gjjl_client", "customer", "client", self.config.followup.contact_field, "gjjl_client_contacts", "contact", "contacts", *self.config.followup.owner_fields, "gjjl_followup_person", "createdBy", "owner", "createdAt", "updatedAt", ], ) def _compact_followup(self, row: Mapping[str, Any], *, include_raw: bool = False) -> dict[str, Any]: summary = self._followup_summary(row) content = summary["content"] compact: dict[str, Any] = { "id": row.get(self.config.followup.id_field), "followedAt": summary["followedAt"], "content_preview": _text_preview(content, CONTENT_PREVIEW_CHARS), "content_truncated": isinstance(content, str) and len(content) > CONTENT_PREVIEW_CHARS, "customer_names": summary["customer_names"], "contact_names": summary["contact_names"], "owner_names": summary["owner_names"], "createdAt": row.get("createdAt"), "updatedAt": row.get("updatedAt"), } if include_raw: compact["raw"] = dict(row) return compact def _followup_summary(self, row: Mapping[str, Any]) -> dict[str, Any]: content_fields = list(self.config.followup.search_fields or []) + list(FOLLOWUP_CONTENT_KEYS) date_fields = [self.config.followup.date_field, *FOLLOWUP_DATE_KEYS] customer_fields = [self.config.followup.customer_field, "gjjl_client", "customer", "client"] contact_fields = [self.config.followup.contact_field, "gjjl_client_contacts", "contact", "contacts"] owner_fields = [*self.config.followup.owner_fields, "gjjl_followup_person", "createdBy", "owner"] return { "content": _first_text(row, content_fields), "followedAt": _first_text(row, date_fields), "customer_names": _relation_labels(row, customer_fields, CUSTOMER_NAME_KEYS), "contact_names": _relation_labels(row, contact_fields, CONTACT_NAME_KEYS), "owner_names": _relation_labels(row, owner_fields, USER_LABEL_KEYS), } @staticmethod def _data(response: Mapping[str, Any]) -> list[dict[str, Any]]: return _data(response) @staticmethod def _single(response: Mapping[str, Any]) -> Any: data = response.get("data") if isinstance(data, list): return data[0] if data else None return data def _page_from_offset(limit: int, offset: int) -> int: bounded_limit = max(1, int(limit)) bounded_offset = max(0, int(offset)) return bounded_offset // bounded_limit + 1 def _normalize_pagination(limit: Any, offset: Any) -> tuple[int, int]: return min(max(1, _int_or_default(limit, DEFAULT_PAGE_SIZE)), MAX_PAGE_SIZE), max(0, _int_or_default(offset, 0)) def _compact_field_list(configured_fields: list[str] | None, candidates: list[str]) -> list[str]: available = set(configured_fields or []) selected: list[str] = [] for candidate in _unique_strings(candidates): if available and candidate not in available: continue selected.append(candidate) return selected def _int_or_default(value: Any, default: int) -> int: try: return int(value) except (TypeError, ValueError): return default def _data(response: Mapping[str, Any]) -> list[dict[str, Any]]: data = response.get("data") if isinstance(data, list): return [item for item in data if isinstance(item, dict)] if isinstance(data, dict): for key in ("data", "rows", "results"): nested = data.get(key) if isinstance(nested, list): return [item for item in nested if isinstance(item, dict)] return [] def _total(response: Mapping[str, Any], fallback: int) -> int: meta = response.get("meta") if isinstance(meta, dict): for key in ("count", "total", "totalCount"): value = meta.get(key) if isinstance(value, int): return value data = response.get("data") if isinstance(data, dict): for key in ("count", "total", "totalCount"): value = data.get(key) if isinstance(value, int): return value return fallback def _paginated_response(response: Mapping[str, Any], limit: int, offset: int) -> dict[str, Any]: rows = _data(response) total = _total(response, len(rows)) next_offset = offset + limit return { "results": rows, "total": total, "returned": len(rows), "limit": limit, "offset": offset, "has_more": next_offset < total, "next_offset": next_offset if next_offset < total else None, } def _first_numeric(rows: list[dict[str, Any]], preferred_key: str) -> int: for row in rows: value = row.get(preferred_key) if isinstance(value, (int, float)): return int(value) for nested_value in row.values(): if isinstance(nested_value, (int, float)): return int(nested_value) return 0 def _decorate_dimension_counts( rows: list[dict[str, Any]], *, preferred_keys: list[str], labels: Mapping[Any, str], display_key: str = "label", ) -> list[dict[str, Any]]: decorated: list[dict[str, Any]] = [] for row in rows: value, _key = _dimension_value(row, preferred_keys) item = dict(row) if value is not None: label = labels.get(value) or str(value) item["value"] = value item["label"] = label item[display_key] = label decorated.append(item) decorated.sort(key=lambda item: (_is_blank(item.get("value")), -_count_value(item), str(item.get("label") or ""))) return decorated def _dimension_value(row: Mapping[str, Any], preferred_keys: list[str]) -> tuple[Any | None, str | None]: for key in preferred_keys: if key in row: return row[key], key for key, value in row.items(): if key == "count": continue if not isinstance(value, (dict, list)): return value, str(key) return None, None def _count_value(row: Mapping[str, Any]) -> int: value = row.get("count") if isinstance(value, (int, float)): return int(value) return 0 def _is_blank(value: Any) -> bool: return value is None or str(value).strip() == "" def _first_value(row: Mapping[str, Any], fields: list[str]) -> Any: for field in _unique_strings(fields): value = row.get(field) if value is not None and value != "": return value return None def _first_text(row: Mapping[str, Any], fields: list[str]) -> str | None: for field in _unique_strings(fields): value = row.get(field) if value is None or value == "": continue if isinstance(value, (dict, list)): continue return str(value) return None def _text_preview(value: str | None, max_chars: int) -> str | None: if value is None: return None if len(value) <= max_chars: return value return value[: max(0, max_chars - 3)].rstrip() + "..." def _relation_labels(row: Mapping[str, Any], fields: list[str], label_keys: tuple[str, ...]) -> list[str]: labels: list[str] = [] for field in _unique_strings(fields): _append_relation_labels(labels, row.get(field), label_keys) return labels def _relation_labels_first_available(row: Mapping[str, Any], fields: list[str], label_keys: tuple[str, ...]) -> list[str]: for field in _unique_strings(fields): labels: list[str] = [] _append_relation_labels(labels, row.get(field), label_keys) if labels: return labels return [] def _append_relation_labels(labels: list[str], value: Any, label_keys: tuple[str, ...]) -> None: if value is None or value == "": return if isinstance(value, list): for item in value: _append_relation_labels(labels, item, label_keys) return if isinstance(value, dict): label = _label_from_mapping(value, label_keys) if label and label not in labels: labels.append(label) return label = str(value) if label not in labels: labels.append(label) def _label_from_mapping(row: Mapping[str, Any], label_keys: tuple[str, ...]) -> str | None: for key in label_keys: value = row.get(key) if value is not None and value != "": return str(value) return None def _unique_strings(values: list[str]) -> list[str]: unique: list[str] = [] for value in values: normalized = str(value or "").strip() if normalized and normalized not in unique: unique.append(normalized) return unique def _metadata_error(error: Exception) -> dict[str, str]: message = str(error) if "403" in message or "No permissions" in message: return { "code": "metadata_permission_required", "message": message, "hint": ( "Grant the agent role read-only permission for " "`data-modeling collections fields list`, then retry crm_schema(include_fields=true)." ), } return {"code": "metadata_error", "message": message, "hint": "Check NocoBase metadata API reachability and ACL."} def _field_list(value: Any) -> list[dict[str, Any]]: return [item for item in value or [] if isinstance(item, dict)] def _field_names(fields: list[dict[str, Any]]) -> list[str]: return [str(field["name"]) for field in fields if field.get("name")] def _field_path(field: str | list[str]) -> list[str]: if isinstance(field, str): return [part for part in field.split(".") if part] return [str(part) for part in field if str(part)] def _field_text(field: dict[str, Any]) -> str: pieces = [ str(field.get("name") or ""), str(field.get("title") or ""), str(field.get("interface") or ""), str(field.get("type") or ""), ] ui_schema = field.get("uiSchema") if isinstance(ui_schema, dict): pieces.append(str(ui_schema.get("title") or "")) return " ".join(pieces).lower() def _find_named_field(fields: list[dict[str, Any]], names: list[str]) -> str | None: available = {str(field.get("name")) for field in fields if field.get("name")} for name in names: if name in available: return name return None def _existing_field_names(fields: list[dict[str, Any]], names: list[str]) -> list[str]: available = {str(field.get("name")) for field in fields if field.get("name")} return [name for name in names if name in available] def _configured_or_search_fields(fields: list[dict[str, Any]], preferred_names: list[str]) -> list[str]: preferred = _existing_field_names(fields, preferred_names) if preferred: return preferred return _search_fields(fields) def _search_fields(fields: list[dict[str, Any]]) -> list[str]: preferred_interfaces = {"input", "textarea", "email", "phone", "url"} preferred_types = {"string", "text"} names = [ str(field["name"]) for field in fields if field.get("name") and ( str(field.get("interface") or "").lower() in preferred_interfaces or str(field.get("type") or "").lower() in preferred_types ) ] if names: return names[:4] fallback = [str(field["name"]) for field in fields if field.get("name") and str(field.get("name")) != "id"] return fallback[:2] def _find_field(fields: list[dict[str, Any]], needles: list[str]) -> str | None: lowered = [needle.lower() for needle in needles] for field in fields: name = field.get("name") if not name: continue text = _field_text(field) if any(needle in text for needle in lowered): return str(name) return None def _relation_field(fields: list[dict[str, Any]], needles: list[str]) -> str | None: lowered = [needle.lower() for needle in needles] for field in fields: if not field.get("name"): continue if str(field.get("type") or "").lower() not in {"belongsto", "belongstomany", "hasmany", "hasone"}: continue text = _field_text(field) if any(needle in text for needle in lowered): return str(field["name"]) return None def _date_field(fields: list[dict[str, Any]]) -> str | None: preferred = _find_field(fields, ["跟进日期", "跟进时间", "follow", "date", "time"]) if preferred: return preferred for field in fields: if field.get("name") and str(field.get("interface") or "").lower() in {"date", "datetime", "createdat"}: return str(field["name"]) if field.get("name") and str(field.get("type") or "").lower() in {"date", "datetime"}: return str(field["name"]) return None def _template_sort(date_field: str | None, fallback: list[str] | None) -> list[str] | None: if date_field: sort = [f"-{date_field}"] if date_field != "createdAt": sort.append("-createdAt") return sort return fallback def _pick_user_matches(rows: list[dict[str, Any]], query: str, match_fields: list[str]) -> list[dict[str, Any]]: normalized = query.casefold() exact = [ row for row in rows if any(str(row.get(field) or "").strip().casefold() == normalized for field in match_fields) ] if exact: return exact fuzzy = [ row for row in rows if any(normalized in str(row.get(field) or "").strip().casefold() for field in match_fields) ] return fuzzy def _user_label(row: dict[str, Any]) -> str: for key in ("nickname", "username", "email", "f_yh_english_name", "id"): value = row.get(key) if value is not None and value != "": return str(value) return "" def _format_user_candidate(row: dict[str, Any], id_field: str) -> dict[str, Any]: candidate: dict[str, Any] = { "id": row.get(id_field), "label": _user_label(row), } for key in ("username", "nickname", "email", "f_yh_english_name"): value = row.get(key) if value is not None and value != "": candidate[key] = value return candidate