409 lines
17 KiB
Python
409 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from dataclasses import dataclass, field, replace
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
PROJECT_PROGRESS_VALUES = {
|
|
"询价阶段": "fk9xj8akn46",
|
|
"已报价": "k4w2vcz819w",
|
|
"送样": "g95vqeetu83",
|
|
"已定点": "fe0peg505d7",
|
|
"SOP": "a9t9awhtt3r",
|
|
"终止": "a9t9awhtsfasg3r",
|
|
}
|
|
|
|
# Output key -> NocoBase field code, used to render `get_customer` detail with
|
|
# human-readable names instead of internal field codes. Entries whose field is
|
|
# absent from the configured `fields` list are simply skipped.
|
|
CUSTOMER_DETAIL_FIELDS = {
|
|
"address": "f_y8iyf2jhana",
|
|
"intro": "f_spgoie0arjq",
|
|
"tags": "kh_tags",
|
|
"followup_interval_days": "f_kyf51uh0ky7",
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CustomerConfig:
|
|
resource: str = "customers"
|
|
id_field: str = "id"
|
|
fields: list[str] | None = None
|
|
search_fields: list[str] | None = None
|
|
name_fields: list[str] | None = None
|
|
english_name_field: str = "f_kh_english_name"
|
|
territorial_name_field: str = "f_territorial_name"
|
|
status_field: str = "status"
|
|
source_field: str = "source"
|
|
owner_field: str = "owner"
|
|
manager_field: str = "kehu_manage_person"
|
|
detail_fields: dict[str, str] = field(default_factory=lambda: CUSTOMER_DETAIL_FIELDS.copy())
|
|
sort: list[str] | None = None
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.fields is None:
|
|
object.__setattr__(
|
|
self,
|
|
"fields",
|
|
["id", "name", "code", "status", "source", "owner", "createdAt", "updatedAt"],
|
|
)
|
|
if self.search_fields is None:
|
|
object.__setattr__(self, "search_fields", ["name", "code"])
|
|
if self.name_fields is None:
|
|
object.__setattr__(
|
|
self,
|
|
"name_fields",
|
|
["f_q2q60q7bqbg", "f_kh_english_name", "f_territorial_name", "name", "title", "code"],
|
|
)
|
|
if self.sort is None:
|
|
object.__setattr__(self, "sort", ["-updatedAt"])
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ContactConfig:
|
|
resource: str = "contacts"
|
|
id_field: str = "id"
|
|
fields: list[str] | None = None
|
|
search_fields: list[str] | None = None
|
|
name_fields: list[str] | None = None
|
|
title_field: str = "f_tn8zdm2f938"
|
|
phone_field: str = "khlxr_phone_number"
|
|
email_field: str = "khlxr_email"
|
|
customer_field: str = "customer"
|
|
sort: list[str] | None = None
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.fields is None:
|
|
object.__setattr__(
|
|
self,
|
|
"fields",
|
|
["id", "name", "phone", "email", "title", "customer", "createdAt", "updatedAt"],
|
|
)
|
|
if self.search_fields is None:
|
|
object.__setattr__(self, "search_fields", ["name", "phone", "email"])
|
|
if self.name_fields is None:
|
|
object.__setattr__(self, "name_fields", ["f_2zpg91gtmwp", "name", "title"])
|
|
if self.sort is None:
|
|
object.__setattr__(self, "sort", ["-updatedAt"])
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FollowupConfig:
|
|
resource: str = "followups"
|
|
id_field: str = "id"
|
|
fields: list[str] | None = None
|
|
search_fields: list[str] | None = None
|
|
customer_field: str = "customer"
|
|
contact_field: str = "contact"
|
|
owner_field: str = "owner"
|
|
owner_fields: list[str] | None = None
|
|
date_field: str = "followedAt"
|
|
content_field: str = "f_19iplxz0nds"
|
|
type_field: str = "gjjl_followup_type"
|
|
status_field: str = "f_hskl7gjvos9"
|
|
project_field: str = "f_gjjl_projects"
|
|
location_field: str = "f_b5c4hfi5og6"
|
|
sort: list[str] | None = None
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.owner_fields is None:
|
|
object.__setattr__(self, "owner_fields", [self.owner_field] if self.owner_field else [])
|
|
if self.fields is None:
|
|
object.__setattr__(
|
|
self,
|
|
"fields",
|
|
[
|
|
"id",
|
|
"customer",
|
|
"contact",
|
|
"owner",
|
|
"content",
|
|
"followedAt",
|
|
"nextFollowAt",
|
|
"createdAt",
|
|
"updatedAt",
|
|
],
|
|
)
|
|
if self.search_fields is None:
|
|
object.__setattr__(self, "search_fields", ["content"])
|
|
if self.sort is None:
|
|
object.__setattr__(self, "sort", ["-followedAt", "-createdAt"])
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class UserConfig:
|
|
resource: str = "users"
|
|
id_field: str = "id"
|
|
fields: list[str] | None = None
|
|
match_fields: list[str] | None = None
|
|
label_fields: list[str] | None = None
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.fields is None:
|
|
object.__setattr__(
|
|
self,
|
|
"fields",
|
|
["id", "username", "nickname", "email", "f_yh_english_name"],
|
|
)
|
|
if self.match_fields is None:
|
|
object.__setattr__(self, "match_fields", ["username", "nickname", "email", "f_yh_english_name"])
|
|
if self.label_fields is None:
|
|
object.__setattr__(self, "label_fields", ["nickname", "f_yh_english_name", "username", "email"])
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TeamConfig:
|
|
resource: str = "t_team_management"
|
|
id_field: str = "id"
|
|
name_field: str = "f_team_name"
|
|
manager_field: str = "tdgl_manager"
|
|
member_field: str = "tdgl_team_members"
|
|
user_status_field: str = "yh_status"
|
|
formal_user_status_value: str = "nkluuumb1r1"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ProjectConfig:
|
|
resource: str = "t_projects"
|
|
id_field: str = "id"
|
|
name_field: str = "f_project_name"
|
|
name_fields: list[str] | None = None
|
|
customer_field: str = "client_ID"
|
|
customer_relation_field: str = "f_projects_client"
|
|
inquiry_date_field: str = "xm_project_inquiry_date"
|
|
progress_field: str = "f_projects_progress"
|
|
applications_field: str = "f_projects_applications"
|
|
part_numbers_field: str = "f_projects_part_numbers"
|
|
period_usage_field: str = "f_projects_period_usage"
|
|
details_field: str = "f_projects_details"
|
|
attachment_field: str = "f_projects_attachments"
|
|
contact_field: str = "f_projects_contacts"
|
|
followup_field: str = "f_projects_followup_records"
|
|
update_field: str = "f_u28gm9h2uzl"
|
|
update_resource: str = "t_qxeepjvi39s"
|
|
update_project_field: str = "f_k37rbpi4vye"
|
|
update_date_field: str = "f_6g9fw5oqef9"
|
|
update_content_field: str = "f_ei16g0ihwwd"
|
|
update_attachment_field: str = "f_6pk5pv17i27"
|
|
progress_values: dict[str, str] = field(default_factory=lambda: PROJECT_PROGRESS_VALUES.copy())
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.name_fields is None:
|
|
object.__setattr__(self, "name_fields", [self.name_field, "name", "title"])
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CrmConfig:
|
|
env_name: str = "nb-mcp"
|
|
data_source: str = "main"
|
|
timezone: str = "Asia/Shanghai"
|
|
customer: CustomerConfig = CustomerConfig()
|
|
contact: ContactConfig = ContactConfig()
|
|
followup: FollowupConfig = FollowupConfig()
|
|
user: UserConfig = UserConfig()
|
|
team: TeamConfig = TeamConfig()
|
|
project: ProjectConfig = ProjectConfig()
|
|
|
|
|
|
def _load_json(path: Path) -> dict[str, Any]:
|
|
with path.open("r", encoding="utf-8") as handle:
|
|
data = json.load(handle)
|
|
if not isinstance(data, dict):
|
|
raise ValueError(f"CRM config must be a JSON object: {path}")
|
|
return data
|
|
|
|
|
|
def _string_list(value: Any) -> list[str] | None:
|
|
if value is None:
|
|
return None
|
|
if not isinstance(value, list):
|
|
raise ValueError("Expected a JSON array of strings")
|
|
return [str(item) for item in value]
|
|
|
|
|
|
def _string_map(value: Any) -> dict[str, str] | None:
|
|
if value is None:
|
|
return None
|
|
if not isinstance(value, dict):
|
|
raise ValueError("Expected a JSON object mapping strings to strings")
|
|
return {str(key): str(item) for key, item in value.items()}
|
|
|
|
|
|
def _pick(data: dict[str, Any], *names: str) -> Any:
|
|
for name in names:
|
|
if name in data:
|
|
return data[name]
|
|
return None
|
|
|
|
|
|
def _customer_config(data: dict[str, Any] | None) -> CustomerConfig:
|
|
if not data:
|
|
return CustomerConfig()
|
|
base = CustomerConfig()
|
|
return replace(
|
|
base,
|
|
resource=str(_pick(data, "resource") or base.resource),
|
|
id_field=str(_pick(data, "idField", "id_field") or base.id_field),
|
|
fields=_string_list(_pick(data, "fields")) or base.fields,
|
|
search_fields=_string_list(_pick(data, "searchFields", "search_fields")) or base.search_fields,
|
|
name_fields=_string_list(_pick(data, "nameFields", "name_fields")) or base.name_fields,
|
|
english_name_field=str(
|
|
_pick(data, "englishNameField", "english_name_field") or base.english_name_field
|
|
),
|
|
territorial_name_field=str(
|
|
_pick(data, "territorialNameField", "territorial_name_field") or base.territorial_name_field
|
|
),
|
|
status_field=str(_pick(data, "statusField", "status_field") or base.status_field),
|
|
source_field=str(_pick(data, "sourceField", "source_field") or base.source_field),
|
|
owner_field=str(_pick(data, "ownerField", "owner_field") or base.owner_field),
|
|
manager_field=str(_pick(data, "managerField", "manager_field") or base.manager_field),
|
|
detail_fields=_string_map(_pick(data, "detailFields", "detail_fields")) or base.detail_fields,
|
|
sort=_string_list(_pick(data, "sort")) or base.sort,
|
|
)
|
|
|
|
|
|
def _contact_config(data: dict[str, Any] | None) -> ContactConfig:
|
|
if not data:
|
|
return ContactConfig()
|
|
base = ContactConfig()
|
|
return replace(
|
|
base,
|
|
resource=str(_pick(data, "resource") or base.resource),
|
|
id_field=str(_pick(data, "idField", "id_field") or base.id_field),
|
|
fields=_string_list(_pick(data, "fields")) or base.fields,
|
|
search_fields=_string_list(_pick(data, "searchFields", "search_fields")) or base.search_fields,
|
|
name_fields=_string_list(_pick(data, "nameFields", "name_fields")) or base.name_fields,
|
|
title_field=str(_pick(data, "titleField", "title_field") or base.title_field),
|
|
phone_field=str(_pick(data, "phoneField", "phone_field") or base.phone_field),
|
|
email_field=str(_pick(data, "emailField", "email_field") or base.email_field),
|
|
customer_field=str(_pick(data, "customerField", "customer_field") or base.customer_field),
|
|
sort=_string_list(_pick(data, "sort")) or base.sort,
|
|
)
|
|
|
|
|
|
def _followup_config(data: dict[str, Any] | None) -> FollowupConfig:
|
|
if not data:
|
|
return FollowupConfig()
|
|
base = FollowupConfig()
|
|
owner_field = str(_pick(data, "ownerField", "owner_field") or base.owner_field)
|
|
owner_fields = _string_list(_pick(data, "ownerFields", "owner_fields")) or ([owner_field] if owner_field else [])
|
|
return replace(
|
|
base,
|
|
resource=str(_pick(data, "resource") or base.resource),
|
|
id_field=str(_pick(data, "idField", "id_field") or base.id_field),
|
|
fields=_string_list(_pick(data, "fields")) or base.fields,
|
|
search_fields=_string_list(_pick(data, "searchFields", "search_fields")) or base.search_fields,
|
|
customer_field=str(_pick(data, "customerField", "customer_field") or base.customer_field),
|
|
contact_field=str(_pick(data, "contactField", "contact_field") or base.contact_field),
|
|
owner_field=owner_field,
|
|
owner_fields=owner_fields,
|
|
date_field=str(_pick(data, "dateField", "date_field") or base.date_field),
|
|
content_field=str(_pick(data, "contentField", "content_field") or base.content_field),
|
|
type_field=str(_pick(data, "typeField", "type_field") or base.type_field),
|
|
status_field=str(_pick(data, "statusField", "status_field") or base.status_field),
|
|
project_field=str(_pick(data, "projectField", "project_field") or base.project_field),
|
|
location_field=str(_pick(data, "locationField", "location_field") or base.location_field),
|
|
sort=_string_list(_pick(data, "sort")) or base.sort,
|
|
)
|
|
|
|
|
|
def _user_config(data: dict[str, Any] | None) -> UserConfig:
|
|
if not data:
|
|
return UserConfig()
|
|
base = UserConfig()
|
|
return replace(
|
|
base,
|
|
resource=str(_pick(data, "resource") or base.resource),
|
|
id_field=str(_pick(data, "idField", "id_field") or base.id_field),
|
|
fields=_string_list(_pick(data, "fields")) or base.fields,
|
|
match_fields=_string_list(_pick(data, "matchFields", "match_fields")) or base.match_fields,
|
|
label_fields=_string_list(_pick(data, "labelFields", "label_fields")) or base.label_fields,
|
|
)
|
|
|
|
|
|
def _team_config(data: dict[str, Any] | None) -> TeamConfig:
|
|
if not data:
|
|
return TeamConfig()
|
|
base = TeamConfig()
|
|
return replace(
|
|
base,
|
|
resource=str(_pick(data, "resource") or base.resource),
|
|
id_field=str(_pick(data, "idField", "id_field") or base.id_field),
|
|
name_field=str(_pick(data, "nameField", "name_field") or base.name_field),
|
|
manager_field=str(_pick(data, "managerField", "manager_field") or base.manager_field),
|
|
member_field=str(_pick(data, "memberField", "member_field") or base.member_field),
|
|
user_status_field=str(_pick(data, "userStatusField", "user_status_field") or base.user_status_field),
|
|
formal_user_status_value=str(
|
|
_pick(data, "formalUserStatusValue", "formal_user_status_value") or base.formal_user_status_value
|
|
),
|
|
)
|
|
|
|
|
|
def _project_config(data: dict[str, Any] | None) -> ProjectConfig:
|
|
if not data:
|
|
return ProjectConfig()
|
|
base = ProjectConfig()
|
|
return replace(
|
|
base,
|
|
resource=str(_pick(data, "resource") or base.resource),
|
|
id_field=str(_pick(data, "idField", "id_field") or base.id_field),
|
|
name_field=str(_pick(data, "nameField", "name_field") or base.name_field),
|
|
# Left as None when unset so __post_init__ re-derives it from the configured name_field.
|
|
name_fields=_string_list(_pick(data, "nameFields", "name_fields")),
|
|
customer_field=str(_pick(data, "customerField", "customer_field") or base.customer_field),
|
|
customer_relation_field=str(
|
|
_pick(data, "customerRelationField", "customer_relation_field") or base.customer_relation_field
|
|
),
|
|
inquiry_date_field=str(_pick(data, "inquiryDateField", "inquiry_date_field") or base.inquiry_date_field),
|
|
progress_field=str(_pick(data, "progressField", "progress_field") or base.progress_field),
|
|
applications_field=str(_pick(data, "applicationsField", "applications_field") or base.applications_field),
|
|
part_numbers_field=str(_pick(data, "partNumbersField", "part_numbers_field") or base.part_numbers_field),
|
|
period_usage_field=str(_pick(data, "periodUsageField", "period_usage_field") or base.period_usage_field),
|
|
details_field=str(_pick(data, "detailsField", "details_field") or base.details_field),
|
|
attachment_field=str(_pick(data, "attachmentField", "attachment_field") or base.attachment_field),
|
|
contact_field=str(_pick(data, "contactField", "contact_field") or base.contact_field),
|
|
followup_field=str(_pick(data, "followupField", "followup_field") or base.followup_field),
|
|
update_field=str(_pick(data, "updateField", "update_field") or base.update_field),
|
|
update_resource=str(_pick(data, "updateResource", "update_resource") or base.update_resource),
|
|
update_project_field=str(
|
|
_pick(data, "updateProjectField", "update_project_field") or base.update_project_field
|
|
),
|
|
update_date_field=str(_pick(data, "updateDateField", "update_date_field") or base.update_date_field),
|
|
update_content_field=str(
|
|
_pick(data, "updateContentField", "update_content_field") or base.update_content_field
|
|
),
|
|
update_attachment_field=str(
|
|
_pick(data, "updateAttachmentField", "update_attachment_field") or base.update_attachment_field
|
|
),
|
|
)
|
|
|
|
|
|
def load_config(path: str | Path | None = None) -> CrmConfig:
|
|
configured_path = path or os.environ.get("NOCOBASE_MCP_CONFIG")
|
|
raw: dict[str, Any] = {}
|
|
if configured_path:
|
|
raw = _load_json(Path(configured_path).expanduser())
|
|
|
|
collections = raw.get("collections") or {}
|
|
if not isinstance(collections, dict):
|
|
raise ValueError("CRM config collections must be a JSON object")
|
|
|
|
env_name = os.environ.get("NOCOBASE_MCP_ENV") or raw.get("env") or raw.get("envName") or "nb-mcp"
|
|
data_source = os.environ.get("NOCOBASE_MCP_DATA_SOURCE") or raw.get("dataSource") or raw.get("data_source") or "main"
|
|
timezone = os.environ.get("NOCOBASE_MCP_TIMEZONE") or raw.get("timezone") or "Asia/Shanghai"
|
|
|
|
return CrmConfig(
|
|
env_name=str(env_name),
|
|
data_source=str(data_source),
|
|
timezone=str(timezone),
|
|
customer=_customer_config(collections.get("customers")),
|
|
contact=_contact_config(collections.get("contacts")),
|
|
followup=_followup_config(collections.get("followups")),
|
|
user=_user_config(raw.get("users") or raw.get("user")),
|
|
team=_team_config(raw.get("teams") or raw.get("team")),
|
|
project=_project_config(raw.get("projects")),
|
|
)
|