nocobase-mcp/nocobase_crm_mcp/config.py

from __future__ import annotations

import json
import os
from dataclasses import dataclass, field, fields, 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 = field(
        default_factory=lambda: ["id", "name", "code", "status", "source", "owner", "createdAt", "updatedAt"]
    )
    search_fields: list[str] | None = field(default_factory=lambda: ["name", "code"])
    name_fields: list[str] | None = field(
        default_factory=lambda: [
            "f_q2q60q7bqbg",
            "f_kh_english_name",
            "f_territorial_name",
            "name",
            "title",
            "code",
        ]
    )
    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 = field(default_factory=lambda: ["-updatedAt"])


@dataclass(frozen=True)
class ContactConfig:
    resource: str = "contacts"
    id_field: str = "id"
    fields: list[str] | None = field(
        default_factory=lambda: ["id", "name", "phone", "email", "title", "customer", "createdAt", "updatedAt"]
    )
    search_fields: list[str] | None = field(default_factory=lambda: ["name", "phone", "email"])
    name_fields: list[str] | None = field(default_factory=lambda: ["f_2zpg91gtmwp", "name", "title"])
    title_field: str = "f_tn8zdm2f938"
    phone_field: str = "khlxr_phone_number"
    email_field: str = "khlxr_email"
    customer_field: str = "customer"
    sort: list[str] | None = field(default_factory=lambda: ["-updatedAt"])


@dataclass(frozen=True)
class FollowupConfig:
    resource: str = "followups"
    id_field: str = "id"
    fields: list[str] | None = field(
        default_factory=lambda: [
            "id",
            "customer",
            "contact",
            "owner",
            "content",
            "followedAt",
            "nextFollowAt",
            "createdAt",
            "updatedAt",
        ]
    )
    search_fields: list[str] | None = field(default_factory=lambda: ["content"])
    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 = field(default_factory=lambda: ["-followedAt", "-createdAt"])

    def __post_init__(self) -> None:
        if self.owner_fields is None:
            object.__setattr__(self, "owner_fields", [self.owner_field] if self.owner_field else [])


@dataclass(frozen=True)
class UserConfig:
    resource: str = "users"
    id_field: str = "id"
    fields: list[str] | None = field(
        default_factory=lambda: ["id", "username", "nickname", "email", "f_yh_english_name"]
    )
    match_fields: list[str] | None = field(
        default_factory=lambda: ["username", "nickname", "email", "f_yh_english_name"]
    )
    label_fields: list[str] | None = field(
        default_factory=lambda: ["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 = field(default_factory=CustomerConfig)
    contact: ContactConfig = field(default_factory=ContactConfig)
    followup: FollowupConfig = field(default_factory=FollowupConfig)
    user: UserConfig = field(default_factory=UserConfig)
    team: TeamConfig = field(default_factory=TeamConfig)
    project: ProjectConfig = field(default_factory=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 _camel_case(name: str) -> str:
    head, *tail = name.split("_")
    return head + "".join(part.capitalize() for part in tail)


def _convert_like(default: Any, value: Any) -> Any:
    if isinstance(default, list):
        return _string_list(value)
    if isinstance(default, dict):
        return _string_map(value)
    return str(value)


def _section_overrides(base: Any, data: dict[str, Any], *, exclude: set[str] | None = None) -> dict[str, Any]:
    excluded = exclude or set()
    overrides: dict[str, Any] = {}
    for config_field in fields(base):
        name = config_field.name
        if name in excluded:
            continue
        value = _pick(data, _camel_case(name), name)
        if not value:
            continue
        overrides[name] = _convert_like(getattr(base, name), value)
    return overrides


def _simple_section[SectionConfig](config_type: type[SectionConfig], data: dict[str, Any] | None) -> SectionConfig:
    base = config_type()
    if not data:
        return base
    return replace(base, **_section_overrides(base, data))  # type: ignore[type-var]


def _customer_config(data: dict[str, Any] | None) -> CustomerConfig:
    return _simple_section(CustomerConfig, data)


def _contact_config(data: dict[str, Any] | None) -> ContactConfig:
    return _simple_section(ContactConfig, data)


def _followup_config(data: dict[str, Any] | None) -> FollowupConfig:
    if not data:
        return FollowupConfig()
    base = FollowupConfig()
    overrides = _section_overrides(base, data, exclude={"owner_fields"})
    owner_field = str(overrides.get("owner_field", base.owner_field))
    overrides["owner_fields"] = _string_list(_pick(data, "ownerFields", "owner_fields")) or (
        [owner_field] if owner_field else []
    )
    return replace(base, **overrides)


def _user_config(data: dict[str, Any] | None) -> UserConfig:
    return _simple_section(UserConfig, data)


def _team_config(data: dict[str, Any] | None) -> TeamConfig:
    return _simple_section(TeamConfig, data)


def _project_config(data: dict[str, Any] | None) -> ProjectConfig:
    if not data:
        return ProjectConfig()
    base = ProjectConfig()
    overrides = _section_overrides(base, data, exclude={"name_fields", "progress_values"})
    # Keep this as None when unset so __post_init__ derives it from a custom name_field.
    overrides["name_fields"] = _string_list(_pick(data, "nameFields", "name_fields"))
    return replace(base, **overrides)


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")),
    )