from __future__ import annotations

from collections import deque
from typing import Any

from mcp.server.fastmcp import FastMCP

from ..config import DEFAULT_PROFILE
from ..errors import QingflowApiError, backend_code_int, is_auth_like_error, raise_tool_error
from .base import ToolBase


ALLOWED_DIRECTORY_SEARCH_SCOPES = {"MEMBER", "DEPT"}


class DirectoryTools(ToolBase):
    """组织架构工具（中文名：通讯录与组织目录）。

    类型：组织目录工具。
    主要职责：
    1. 搜索用户、部门、外部联系人；
    2. 列举内部用户与部门层级；
    3. 为 builder 配置、审批处理或联系人确认提供目录侧数据来源。

    注意：记录成员/部门字段候选应优先使用 record_member_candidates /
    record_department_candidates，避免把普通记录读写误带入通讯录目录链路。
    """

    def register(self, mcp: FastMCP) -> None:
        """注册当前工具到 MCP 服务。"""
        self.register_frontend_search(mcp)

        @mcp.tool(description="List internal workspace members through the contact directory route. This is a directory/admin-style lookup, not the record field candidate path.")
        def directory_list_internal_users(
            profile: str = DEFAULT_PROFILE,
            keyword: str | None = None,
            department_id: int | None = None,
            role_id: int | None = None,
            page_num: int = 1,
            page_size: int = 20,
            include_disabled: bool = False,
        ) -> dict[str, Any]:
            return self.directory_list_internal_users(
                profile=profile,
                keyword=keyword,
                dept_id=department_id,
                role_id=role_id,
                page_num=page_num,
                page_size=page_size,
                contain_disable=include_disabled,
            )

        @mcp.tool(description="Page through internal workspace members through the contact directory route. This is a directory/admin-style lookup, not the record field candidate path.")
        def directory_list_all_internal_users(
            profile: str = DEFAULT_PROFILE,
            keyword: str | None = None,
            department_id: int | None = None,
            role_id: int | None = None,
            page_size: int = 200,
            include_disabled: bool = False,
            max_pages: int = 100,
        ) -> dict[str, Any]:
            return self.directory_list_all_internal_users(
                profile=profile,
                keyword=keyword,
                dept_id=department_id,
                role_id=role_id,
                page_size=page_size,
                contain_disable=include_disabled,
                max_pages=max_pages,
            )

        @mcp.tool(description="List internal departments. With keyword, uses frontend directory search; without keyword, browses the contact department tree and may require contact-directory permission.")
        def directory_list_internal_departments(
            profile: str = DEFAULT_PROFILE,
            keyword: str = "",
            page_num: int = 1,
            page_size: int = 20,
        ) -> dict[str, Any]:
            return self.directory_list_internal_departments(
                profile=profile,
                keyword=keyword,
                page_num=page_num,
                page_size=page_size,
            )

        @mcp.tool(description="Walk the internal department tree through the contact directory route. This may require contact-directory permission and should not be used for record department field candidates.")
        def directory_list_all_departments(
            profile: str = DEFAULT_PROFILE,
            parent_department_id: int | None = None,
            max_depth: int = 20,
            max_items: int = 2000,
        ) -> dict[str, Any]:
            return self.directory_list_all_departments(
                profile=profile,
                parent_dept_id=parent_department_id,
                max_depth=max_depth,
                max_items=max_items,
            )

        @mcp.tool(description="List child departments through the contact directory route. This may require contact-directory permission and should not be used for record department field candidates.")
        def directory_list_sub_departments(
            profile: str = DEFAULT_PROFILE,
            parent_department_id: int | None = None,
        ) -> dict[str, Any]:
            return self.directory_list_sub_departments(profile=profile, parent_dept_id=parent_department_id)

        @mcp.tool(description="Search external contacts. Defaults to the frontend simple external-member route; set simple=false only for full external-contact management data, which may require contact permission.")
        def directory_list_external_members(
            profile: str = DEFAULT_PROFILE,
            keyword: str | None = None,
            page_num: int = 1,
            page_size: int = 20,
            simple: bool = True,
        ) -> dict[str, Any]:
            return self.directory_list_external_members(
                profile=profile,
                keyword=keyword,
                page_num=page_num,
                page_size=page_size,
                simple=simple,
            )

    def register_frontend_search(self, mcp: FastMCP) -> None:
        """Register only the member-visible frontend search route."""
        @mcp.tool(description="Search internal workspace members/departments by keyword via the frontend directory search route. Do not use for record member/department field candidates; use record_member_candidates or record_department_candidates instead.")
        def directory_search(
            profile: str = DEFAULT_PROFILE,
            query: str = "",
            scopes: list[str] | None = None,
            page_num: int = 1,
            page_size: int = 20,
        ) -> dict[str, Any]:
            return self.directory_search(
                profile=profile,
                query=query,
                scopes=scopes,
                page_num=page_num,
                page_size=page_size,
            )

    def directory_search(
        self,
        *,
        profile: str,
        query: str,
        scopes: list[str] | None,
        page_num: int,
        page_size: int,
    ) -> dict[str, Any]:
        """执行组织目录相关逻辑。"""
        normalized_scopes = scopes or ["MEMBER", "DEPT"]
        invalid_scopes = [scope for scope in normalized_scopes if scope not in ALLOWED_DIRECTORY_SEARCH_SCOPES]
        if invalid_scopes:
            raise_tool_error(QingflowApiError.not_supported(f"directory_search only supports internal scopes {sorted(ALLOWED_DIRECTORY_SEARCH_SCOPES)}; got {invalid_scopes}"))
        if not query:
            raise_tool_error(QingflowApiError.config_error("query is required"))

        def runner(session_profile, context):
            try:
                result = self.backend.request(
                    "POST",
                    context,
                    "/member/search",
                    json_body={
                        "dimensions": normalized_scopes,
                        "searchKey": query,
                        "pageNum": page_num,
                        "pageSize": page_size,
                    },
                )
            except QingflowApiError as exc:
                if _is_directory_permission_denied(exc):
                    return self._directory_permission_denied_payload(
                        profile=profile,
                        session_profile=session_profile,
                        context=context,
                        error=exc,
                        operation="directory_search",
                    )
                raise
            return {
                "profile": profile,
                "ws_id": session_profile.selected_ws_id,
                "request_route": self._request_route_payload(context),
                "result": result,
            }

        raw = self._run(profile, runner, tool_name='通讯录搜索')
        items = [item for item in _directory_items(raw.get("result")) if isinstance(item, dict)]
        return self._public_directory_response(
            raw,
            items=items,
            pagination={
                "page": page_num,
                "page_size": page_size,
                "returned_items": len(items),
                "reported_total": _coerce_int(_payload_value(raw.get("result"), "total")),
                "page_amount": _coerce_int(_payload_value(raw.get("result"), "pageAmount")),
            },
            selection={"query": query, "scopes": normalized_scopes},
        )

    def directory_list_internal_users(
        self,
        *,
        profile: str,
        keyword: str | None,
        dept_id: int | None,
        role_id: int | None,
        page_num: int,
        page_size: int,
        contain_disable: bool,
    ) -> dict[str, Any]:
        """执行组织目录相关逻辑。"""
        def runner(session_profile, context):
            params: dict[str, Any] = {
                "pageNum": page_num,
                "pageSize": page_size,
                "containDisable": contain_disable,
            }
            if keyword:
                params["keyword"] = keyword
            if dept_id is not None:
                params["deptId"] = dept_id
            if role_id is not None:
                params["roleId"] = role_id
            try:
                result = self.backend.request("GET", context, "/contact", params=params)
            except QingflowApiError as exc:
                if _is_directory_permission_denied(exc):
                    return self._directory_permission_denied_payload(
                        profile=profile,
                        session_profile=session_profile,
                        context=context,
                        error=exc,
                        operation="directory_list_internal_users",
                    )
                raise
            return {
                "profile": profile,
                "ws_id": session_profile.selected_ws_id,
                "request_route": self._request_route_payload(context),
                "result": result,
            }

        raw = self._run(profile, runner, tool_name='内部成员列表')
        items = [item for item in _directory_items(raw.get("result")) if isinstance(item, dict)]
        return self._public_directory_response(
            raw,
            items=items,
            pagination={
                "page": page_num,
                "page_size": page_size,
                "returned_items": len(items),
                "reported_total": _coerce_int(_payload_value(raw.get("result"), "total")),
                "page_amount": _coerce_int(_payload_value(raw.get("result"), "pageAmount")),
            },
            selection={"keyword": keyword, "department_id": dept_id, "role_id": role_id, "include_disabled": contain_disable},
        )

    def directory_list_all_internal_users(
        self,
        *,
        profile: str,
        keyword: str | None,
        dept_id: int | None,
        role_id: int | None,
        page_size: int,
        contain_disable: bool,
        max_pages: int,
    ) -> dict[str, Any]:
        """执行组织目录相关逻辑。"""
        if page_size <= 0:
            raise_tool_error(QingflowApiError.config_error("page_size must be positive"))
        if max_pages <= 0:
            raise_tool_error(QingflowApiError.config_error("max_pages must be positive"))

        def runner(session_profile, context):
            current_page = 1
            fetched_pages = 0
            has_more = False
            reported_total: int | None = None
            seen_keys: set[str] = set()
            items: list[dict[str, Any]] = []
            while fetched_pages < max_pages:
                params: dict[str, Any] = {
                    "pageNum": current_page,
                    "pageSize": page_size,
                    "containDisable": contain_disable,
                }
                if keyword:
                    params["keyword"] = keyword
                if dept_id is not None:
                    params["deptId"] = dept_id
                if role_id is not None:
                    params["roleId"] = role_id
                try:
                    result = self.backend.request("GET", context, "/contact", params=params)
                except QingflowApiError as exc:
                    if _is_directory_permission_denied(exc):
                        return self._directory_permission_denied_payload(
                            profile=profile,
                            session_profile=session_profile,
                            context=context,
                            error=exc,
                            operation="directory_list_all_internal_users",
                        )
                    raise
                page_items = _directory_items(result)
                if reported_total is None:
                    reported_total = _coerce_int(_payload_value(result, "total"))
                for item in page_items:
                    if not isinstance(item, dict):
                        continue
                    member_key = _directory_member_key(item)
                    if member_key in seen_keys:
                        continue
                    seen_keys.add(member_key)
                    items.append(dict(item))
                fetched_pages += 1
                has_more = _directory_has_more(result, current_page=current_page, page_size=page_size, returned_items=len(page_items))
                if not has_more:
                    break
                current_page += 1
            return {
                "profile": profile,
                "ws_id": session_profile.selected_ws_id,
                "request_route": self._request_route_payload(context),
                "items": items,
                "pagination": {
                    "page_size": page_size,
                    "fetched_pages": fetched_pages,
                    "returned_items": len(items),
                    "reported_total": reported_total,
                    "is_complete": not has_more,
                    "has_more": has_more,
                    "next_page_num": current_page + 1 if has_more else None,
                    "max_pages": max_pages,
                },
            }

        raw = self._run(profile, runner, tool_name='内部成员全量')
        return self._public_directory_response(
            raw,
            items=[item for item in raw.get("items", []) if isinstance(item, dict)],
            pagination=raw.get("pagination", {}),
            selection={"keyword": keyword, "department_id": dept_id, "role_id": role_id, "include_disabled": contain_disable},
        )

    def directory_list_internal_departments(
        self,
        *,
        profile: str,
        keyword: str,
        page_num: int,
        page_size: int,
    ) -> dict[str, Any]:
        """执行组织目录相关逻辑。"""
        if page_num <= 0:
            raise_tool_error(QingflowApiError.config_error("page_num must be positive"))
        if page_size <= 0:
            raise_tool_error(QingflowApiError.config_error("page_size must be positive"))
        normalized_keyword = keyword.strip()

        if not normalized_keyword:
            def runner(session_profile, context):
                try:
                    fetch_limit = max((page_num + 1) * page_size + 1, page_size + 1)
                    items, truncated, deepest_depth = self._walk_department_tree(
                        context,
                        parent_dept_id=None,
                        max_depth=20,
                        max_items=fetch_limit,
                    )
                except QingflowApiError as exc:
                    if _is_directory_permission_denied(exc):
                        return self._directory_permission_denied_payload(
                            profile=profile,
                            session_profile=session_profile,
                            context=context,
                            error=exc,
                            operation="directory_list_internal_departments",
                        )
                    raise
                start = (page_num - 1) * page_size
                page_items = items[start : start + page_size]
                reported_total = None if truncated else len(items)
                page_amount = None if truncated else ((len(items) + page_size - 1) // page_size if items else 0)
                if truncated and page_items:
                    page_amount = max(page_num + 1, (start + len(page_items) + page_size - 1) // page_size)
                return {
                    "profile": profile,
                    "ws_id": session_profile.selected_ws_id,
                    "request_route": self._request_route_payload(context),
                    "items": page_items,
                    "pagination": {
                        "page": page_num,
                        "page_size": page_size,
                        "returned_items": len(page_items),
                        "reported_total": reported_total,
                        "page_amount": page_amount,
                        "depth_scanned": deepest_depth + 1 if page_items else 0,
                    },
                }

            raw = self._run(profile, runner, tool_name='内部部门列表')
            items = [item for item in raw.get("items", []) if isinstance(item, dict)]
            return self._public_directory_response(
                raw,
                items=items,
                pagination=raw.get("pagination", {}),
                selection={"keyword": None},
            )

        def runner(session_profile, context):
            try:
                result = self.backend.request(
                    "POST",
                    context,
                    "/member/search",
                    json_body={
                        "dimensions": ["DEPT"],
                        "searchKey": normalized_keyword,
                        "pageNum": page_num,
                        "pageSize": page_size,
                    },
                )
            except QingflowApiError as exc:
                if _is_directory_permission_denied(exc):
                    return self._directory_permission_denied_payload(
                        profile=profile,
                        session_profile=session_profile,
                        context=context,
                        error=exc,
                        operation="directory_list_internal_departments",
                    )
                raise
            dept_bucket = _payload_value(result, "dept") or result
            return {
                "profile": profile,
                "ws_id": session_profile.selected_ws_id,
                "request_route": self._request_route_payload(context),
                "page": dept_bucket,
                "result": result,
            }

        raw = self._run(profile, runner, tool_name='内部部门列表')
        items = [item for item in _directory_items(raw.get("page")) if isinstance(item, dict)]
        return self._public_directory_response(
            raw,
            items=items,
            pagination={
                "page": page_num,
                "page_size": page_size,
                "returned_items": len(items),
                "reported_total": _coerce_int(_payload_value(raw.get("page"), "total")),
                "page_amount": _coerce_int(_payload_value(raw.get("page"), "pageAmount")),
            },
            selection={"keyword": normalized_keyword},
        )

    def directory_list_all_departments(
        self,
        *,
        profile: str,
        parent_dept_id: int | None,
        max_depth: int,
        max_items: int,
    ) -> dict[str, Any]:
        """执行组织目录相关逻辑。"""
        if max_depth < 0:
            raise_tool_error(QingflowApiError.config_error("max_depth must be non-negative"))
        if max_items <= 0:
            raise_tool_error(QingflowApiError.config_error("max_items must be positive"))

        def runner(session_profile, context):
            try:
                items, truncated, deepest_depth = self._walk_department_tree(
                    context,
                    parent_dept_id=parent_dept_id,
                    max_depth=max_depth,
                    max_items=max_items,
                )
                if not items and parent_dept_id is None:
                    items, truncated, deepest_depth = self._walk_department_tree(
                        context,
                        parent_dept_id=0,
                        max_depth=max_depth,
                        max_items=max_items,
                    )
            except QingflowApiError as exc:
                if _is_directory_permission_denied(exc):
                    return self._directory_permission_denied_payload(
                        profile=profile,
                        session_profile=session_profile,
                        context=context,
                        error=exc,
                        operation="directory_list_all_departments",
                    )
                raise
            return {
                "profile": profile,
                "ws_id": session_profile.selected_ws_id,
                "request_route": self._request_route_payload(context),
                "items": items,
                "pagination": {
                    "root_parent_department_id": parent_dept_id,
                    "returned_items": len(items),
                    "fetched_pages": deepest_depth + 1 if items else 0,
                    "is_complete": not truncated,
                    "has_more": truncated,
                    "max_items": max_items,
                },
            }

        raw = self._run(profile, runner, tool_name='部门树全量')
        return self._public_directory_response(
            raw,
            items=[item for item in raw.get("items", []) if isinstance(item, dict)],
            pagination=raw.get("pagination", {}),
            selection={"parent_department_id": parent_dept_id, "max_depth": max_depth, "max_items": max_items},
        )

    def directory_list_sub_departments(self, *, profile: str, parent_dept_id: int | None) -> dict[str, Any]:
        """执行组织目录相关逻辑。"""
        def runner(session_profile, context):
            params: dict[str, Any] = {}
            if parent_dept_id is not None:
                params["parentDeptId"] = parent_dept_id
            try:
                result = self.backend.request("GET", context, "/contact/subDeptList", params=params)
            except QingflowApiError as exc:
                if _is_directory_permission_denied(exc):
                    return self._directory_permission_denied_payload(
                        profile=profile,
                        session_profile=session_profile,
                        context=context,
                        error=exc,
                        operation="directory_list_sub_departments",
                    )
                raise
            return {
                "profile": profile,
                "ws_id": session_profile.selected_ws_id,
                "request_route": self._request_route_payload(context),
                "items": result,
            }

        raw = self._run(profile, runner, tool_name='子部门列表')
        items = [item for item in raw.get("items", []) if isinstance(item, dict)]
        return self._public_directory_response(
            raw,
            items=items,
            pagination={"returned_items": len(items)},
            selection={"parent_department_id": parent_dept_id},
        )

    def directory_list_external_members(
        self,
        *,
        profile: str,
        keyword: str | None,
        page_num: int,
        page_size: int,
        simple: bool = True,
    ) -> dict[str, Any]:
        """执行组织目录相关逻辑。"""
        def runner(session_profile, context):
            if simple:
                try:
                    result = self.backend.request(
                        "POST",
                        context,
                        "/external/member/simple/pageList",
                        json_body={"keyword": keyword, "pageNum": page_num, "pageSize": page_size},
                    )
                except QingflowApiError as exc:
                    if _is_directory_permission_denied(exc):
                        return self._directory_permission_denied_payload(
                            profile=profile,
                            session_profile=session_profile,
                            context=context,
                            error=exc,
                            operation="directory_list_external_members_simple",
                        ) | {"simple": simple}
                    raise
            else:
                try:
                    result = self.backend.request(
                        "POST",
                        context,
                        "/external/member/pageList",
                        json_body={"keyword": keyword, "pageNum": page_num, "pageSize": page_size},
                    )
                except QingflowApiError as exc:
                    if _is_directory_permission_denied(exc):
                        return self._directory_permission_denied_payload(
                            profile=profile,
                            session_profile=session_profile,
                            context=context,
                            error=exc,
                            operation="directory_list_external_members_full",
                        ) | {"simple": simple}
                    raise
            return {
                "profile": profile,
                "ws_id": session_profile.selected_ws_id,
                "request_route": self._request_route_payload(context),
                "page": result,
                "simple": simple,
            }

        raw = self._run(profile, runner, tool_name='外部联系人列表')
        items = [item for item in _directory_items(raw.get("page")) if isinstance(item, dict)]
        return self._public_directory_response(
            raw,
            items=items,
            pagination={
                "page": page_num,
                "page_size": page_size,
                "returned_items": len(items),
                "reported_total": _coerce_int(_payload_value(raw.get("page"), "total")),
                "page_amount": _coerce_int(_payload_value(raw.get("page"), "pageAmount")),
            },
            selection={"keyword": keyword, "simple": simple},
        )

    def _request_route_payload(self, context) -> dict[str, Any]:  # type: ignore[no-untyped-def]
        """执行内部辅助逻辑。"""
        describe_route = getattr(self.backend, "describe_route", None)
        if callable(describe_route):
            payload = describe_route(context)
            if isinstance(payload, dict):
                return payload
        return {
            "base_url": getattr(context, "base_url", None),
            "qf_version": getattr(context, "qf_version", None),
            "qf_version_source": getattr(context, "qf_version_source", None) or ("context" if getattr(context, "qf_version", None) else "unknown"),
        }

    def _public_directory_response(
        self,
        raw: dict[str, Any],
        *,
        items: list[dict[str, Any]],
        pagination: dict[str, Any],
        selection: dict[str, Any],
    ) -> dict[str, Any]:
        """执行内部辅助逻辑。"""
        response = dict(raw)
        response["ok"] = bool(raw.get("ok", True))
        response["warnings"] = list(raw.get("warnings") or [])
        response["output_profile"] = "normal"
        response["data"] = {
            "items": items,
            "pagination": pagination,
            "selection": selection,
        }
        return response

    def _directory_permission_denied_payload(
        self,
        *,
        profile: str,
        session_profile,  # type: ignore[no-untyped-def]
        context,  # type: ignore[no-untyped-def]
        error: QingflowApiError,
        operation: str,
    ) -> dict[str, Any]:
        candidate_hint = (
            "For record member/department fields, use record_member_candidates or "
            "record_department_candidates."
            if operation == "directory_search"
            else (
                "For record member/department fields, use record_member_candidates, "
                "record_department_candidates, or directory_search with a keyword instead."
            )
        )
        return {
            "profile": profile,
            "ws_id": session_profile.selected_ws_id,
            "ok": False,
            "status": "failed",
            "error_code": "CONTACT_DIRECTORY_PERMISSION_DENIED",
            "message": (
                "Contact-directory management data is not readable for the current user. "
                f"{candidate_hint}"
            ),
            "backend_code": error.backend_code,
            "request_id": error.request_id,
            "http_status": error.http_status,
            "request_route": self._request_route_payload(context),
            "items": [],
            "pagination": {"returned_items": 0},
            "warnings": [
                {
                    "code": "CONTACT_DIRECTORY_PERMISSION_DENIED",
                    "message": "This is a contact-directory permission boundary, not proof that record field candidates are unavailable.",
                    "operation": operation,
                }
            ],
        }

    def _walk_department_tree(
        self,
        context,  # type: ignore[no-untyped-def]
        *,
        parent_dept_id: int | None,
        max_depth: int,
        max_items: int,
    ) -> tuple[list[dict[str, Any]], bool, int]:
        """执行内部辅助逻辑。"""
        queue: deque[tuple[int | None, int]] = deque([(parent_dept_id, 0)])
        seen_ids: set[int] = set()
        requested_parents: set[int | None] = set()
        items: list[dict[str, Any]] = []
        truncated = False
        deepest_depth = 0
        while queue:
            current_parent, depth = queue.popleft()
            if current_parent in requested_parents:
                continue
            requested_parents.add(current_parent)
            params: dict[str, Any] = {}
            if current_parent is not None:
                params["parentDeptId"] = current_parent
            result = self.backend.request("GET", context, "/contact/subDeptList", params=params)
            page_items = _directory_items(result)
            for item in page_items:
                if not isinstance(item, dict):
                    continue
                dept_id = _department_id(item)
                if dept_id is None or dept_id in seen_ids:
                    continue
                seen_ids.add(dept_id)
                entry = dict(item)
                if current_parent is not None and "parentDeptId" not in entry:
                    entry["parentDeptId"] = current_parent
                entry.setdefault("depth", depth)
                items.append(entry)
                deepest_depth = max(deepest_depth, depth)
                if len(items) >= max_items:
                    truncated = True
                    return items, truncated, deepest_depth
                if depth < max_depth:
                    queue.append((dept_id, depth + 1))
                else:
                    truncated = True
        return items, truncated, deepest_depth


def _payload_value(payload: Any, key: str) -> Any:
    if isinstance(payload, dict):
        if key in payload:
            return payload.get(key)
        for container_key in ("data", "result", "page"):
            nested = payload.get(container_key)
            if isinstance(nested, dict) and key in nested:
                return nested.get(key)
    return None


def _directory_items(payload: Any) -> list[Any]:
    if isinstance(payload, list):
        return payload
    if not isinstance(payload, dict):
        return []
    for key in ("list", "items", "rows"):
        value = payload.get(key)
        if isinstance(value, list):
            return value
    for key in ("data", "result", "page"):
        nested = payload.get(key)
        if isinstance(nested, list):
            return nested
        if isinstance(nested, dict):
            for nested_key in ("list", "items", "rows", "result"):
                value = nested.get(nested_key)
                if isinstance(value, list):
                    return value
    return []


def _directory_has_more(payload: Any, *, current_page: int, page_size: int, returned_items: int) -> bool:
    page_amount = _coerce_int(_payload_value(payload, "pageAmount"))
    if page_amount is not None:
        return current_page < page_amount
    total = _coerce_int(_payload_value(payload, "total"))
    if total is not None:
        return current_page * page_size < total
    return returned_items >= page_size and returned_items > 0


def _is_directory_permission_denied(error: QingflowApiError) -> bool:
    if is_auth_like_error(error):
        return False
    return backend_code_int(error) in {40002, 40027}


def _directory_member_key(item: dict[str, Any]) -> str:
    for key in ("uid", "id", "userId"):
        value = item.get(key)
        if value is not None:
            return f"{key}:{value}"
    for key in ("email", "nickName", "name"):
        value = item.get(key)
        if isinstance(value, str) and value:
            return f"{key}:{value}"
    return repr(sorted(item.items()))


def _department_id(item: dict[str, Any]) -> int | None:
    return _coerce_int(item.get("deptId", item.get("id")))


def _coerce_int(value: Any) -> int | None:
    if isinstance(value, bool) or value is None:
        return None
    if isinstance(value, int):
        return value
    if isinstance(value, float):
        return int(value)
    if isinstance(value, str):
        text = value.strip()
        if not text:
            return None
        try:
            return int(text)
        except ValueError:
            return None
    return None