refactor: simplify CRM MCP architecture

nocobase_crm_mcp/crm_schema_helpers.py

@@ -0,0 +1,145 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+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
+
+
+__all__ = [
+    "_configured_or_search_fields",
+    "_date_field",
+    "_existing_field_names",
+    "_field_list",
+    "_field_names",
+    "_field_path",
+    "_field_text",
+    "_find_field",
+    "_find_named_field",
+    "_metadata_error",
+    "_relation_field",
+    "_search_fields",
+    "_template_sort",
+]