280 lines
10 KiB
Python
280 lines
10 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",
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CustomerConfig:
|
|
resource: str = "customers"
|
|
id_field: str = "id"
|
|
fields: list[str] | None = None
|
|
search_fields: list[str] | None = None
|
|
status_field: str = "status"
|
|
source_field: str = "source"
|
|
owner_field: str = "owner"
|
|
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.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
|
|
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.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"
|
|
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
|
|
|
|
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"])
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ProjectConfig:
|
|
resource: str = "t_projects"
|
|
id_field: str = "id"
|
|
name_field: str = "f_project_name"
|
|
customer_field: str = "client_ID"
|
|
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"
|
|
progress_values: dict[str, str] = field(default_factory=lambda: PROJECT_PROGRESS_VALUES.copy())
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CrmConfig:
|
|
env_name: str = "nb-mcp"
|
|
data_source: str = "main"
|
|
customer: CustomerConfig = CustomerConfig()
|
|
contact: ContactConfig = ContactConfig()
|
|
followup: FollowupConfig = FollowupConfig()
|
|
user: UserConfig = UserConfig()
|
|
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 _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,
|
|
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),
|
|
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,
|
|
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),
|
|
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,
|
|
)
|
|
|
|
|
|
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),
|
|
customer_field=str(_pick(data, "customerField", "customer_field") or base.customer_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),
|
|
)
|
|
|
|
|
|
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"
|
|
|
|
return CrmConfig(
|
|
env_name=str(env_name),
|
|
data_source=str(data_source),
|
|
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")),
|
|
project=_project_config(raw.get("projects")),
|
|
)
|