from __future__ import annotations import json import os from typing import Any from mcp.server.fastmcp import FastMCP from ..backend_client import BackendRequestContext, BackendResponse from ..cloud_context import current_cloud_context from ..config import ( DEFAULT_PROFILE, get_default_base_url, get_default_qf_version, get_mcporter_config_path, normalize_base_url, ) from ..errors import QingflowApiError, backend_code_int, is_auth_like_error, raise_tool_error from ..session_store import SessionStore from .base import ToolBase class AuthTools(ToolBase): """认证类工具（中文名：身份与会话工具）。 主要职责： 1. 使用 credential 调用 /mcp/auth/context 建立会话； 2. 查询当前登录身份与工作区成员信息； 3. 退出并清理当前 profile 会话。 """ def __init__(self, sessions: SessionStore, backend) -> None: """执行内部辅助逻辑。""" super().__init__(sessions, backend) self.sessions.set_profile_refresher(self._refresh_profile_via_credential) def register(self, mcp: FastMCP) -> None: """注册当前工具到 MCP 服务。""" @mcp.tool( description=( "类型：认证工具；中文名：凭证登录。" "用途：使用 createClaw 提供的 credential 交换上下文并建立本地会话。" ) ) def auth_use_credential( profile: str = DEFAULT_PROFILE, base_url: str | None = None, qf_version: str | None = None, credential: str = "", persist: bool = False, ) -> dict[str, Any]: return self.auth_use_credential( profile=profile, base_url=base_url, qf_version=qf_version, credential=credential, persist=persist, ) @mcp.tool( description=( "类型：认证工具；中文名：我的身份。" "用途：查看当前 profile 的登录身份、工作区与权限信息。" ) ) def auth_whoami(profile: str = DEFAULT_PROFILE) -> dict[str, Any]: return self.auth_whoami(profile=profile) @mcp.tool( description=( "类型：认证工具；中文名：退出登录。" "用途：退出当前 profile，并可选清理持久化会话。" ) ) def auth_logout(profile: str = DEFAULT_PROFILE, forget_persisted: bool = False) -> dict[str, Any]: return self.auth_logout(profile=profile, forget_persisted=forget_persisted) def auth_use_credential( self, *, profile: str = DEFAULT_PROFILE, base_url: str | None = None, qf_version: str | None = None, credential: str | None = None, persist: bool = False, ) -> dict[str, Any]: """执行认证与会话相关逻辑。""" resolved_base_url, resolved_credential = self._resolve_auth_inputs( profile=profile, base_url=base_url, credential=credential, ) normalized_base_url = self._normalize_base_url(resolved_base_url) normalized_qf_version, qf_version_source = self._resolve_qf_version_input(qf_version) normalized_credential = str(resolved_credential).strip() if not normalized_credential: raise_tool_error( QingflowApiError.config_error( "credential is required or configure ~/.openclaw/workspace/config/mcporter.json " "with mcpServers.qingflow.headers.x-qingflow-client-id" ) ) context_payload, detected_qf_version = self._fetch_auth_context( normalized_base_url, normalized_credential, qf_version=normalized_qf_version, ) token = self._normalize_text(context_payload.get("token")) if not token: raise_tool_error(QingflowApiError(category="auth", message="Credential context did not return a valid Qingflow token")) response_qf_version = self._normalize_text(context_payload.get("qfVersion")) resolved_qf_version, resolved_qf_version_source = self._resolve_backend_qf_version( response_qf_version or detected_qf_version, fallback_qf_version=normalized_qf_version, fallback_source=qf_version_source, ) resolved_base_url = self._normalize_text(context_payload.get("baseUrl")) or normalized_base_url selected_ws_id = self._coerce_int(context_payload.get("wsId")) if selected_ws_id is None or selected_ws_id <= 0: raise_tool_error(QingflowApiError(category="auth", message="Credential context did not return a valid wsId")) selected_ws_name = self._normalize_text(context_payload.get("wsName")) uid = self._coerce_int(context_payload.get("uid")) if uid is None: raise_tool_error(QingflowApiError(category="auth", message="Credential context did not return valid user info")) session_profile = self.sessions.save_session( profile=profile, base_url=resolved_base_url, qf_version=resolved_qf_version, qf_version_source=resolved_qf_version_source, token=token, login_token=None, credential=normalized_credential, uid=uid, email=self._normalize_text(context_payload.get("email")), nick_name=self._normalize_text( context_payload.get("nickName") or context_payload.get("displayName") or context_payload.get("name") ), persist=persist, ) session_profile = self.sessions.select_workspace(profile, ws_id=selected_ws_id, ws_name=selected_ws_name) return { "profile": session_profile.profile, "base_url": session_profile.base_url, "qf_version": session_profile.qf_version, "qf_version_source": session_profile.qf_version_source, "uid": session_profile.uid, "email": session_profile.email, "nick_name": session_profile.nick_name, "selected_ws_id": session_profile.selected_ws_id, "selected_ws_name": session_profile.selected_ws_name, "suggested_ws_id": session_profile.selected_ws_id, "suggested_ws_name": session_profile.selected_ws_name, "persisted": session_profile.persisted, "request_route": self._request_route_payload( BackendRequestContext( base_url=session_profile.base_url, token=token, ws_id=session_profile.selected_ws_id, qf_version=session_profile.qf_version, qf_version_source=session_profile.qf_version_source, ) ), } def auth_use_token( self, *, profile: str = DEFAULT_PROFILE, base_url: str | None = None, qf_version: str | None = None, token: str | None = None, login_token: str | None = None, persist: bool = False, user_info: dict[str, Any] | None = None, ) -> dict[str, Any]: """使用已获得的 Qingflow token 建立本地会话。""" normalized_base_url = self._normalize_base_url(base_url) normalized_qf_version, qf_version_source = self._resolve_qf_version_input(qf_version) normalized_token = self._normalize_text(token) normalized_login_token = self._normalize_text(login_token) if not normalized_token: raise_tool_error(QingflowApiError.config_error("token is required")) resolved_user_info = user_info if isinstance(user_info, dict) else None response_qf_version: str | None = None if resolved_user_info is None: resolved_user_info, response_qf_version = self._fetch_user_info( normalized_base_url, normalized_token, None, qf_version=normalized_qf_version, qf_version_source=qf_version_source, ) last_workspace = resolved_user_info.get("lastWsInfo") selected_ws_id = self._coerce_positive_int( last_workspace.get("wsId") if isinstance(last_workspace, dict) else None ) selected_ws_name = self._normalize_text( (last_workspace.get("wsName") if isinstance(last_workspace, dict) else None) or (last_workspace.get("workspaceName") if isinstance(last_workspace, dict) else None) or (last_workspace.get("remark") if isinstance(last_workspace, dict) else None) ) workspace_qf_version = ( self._workspace_system_version(last_workspace) if isinstance(last_workspace, dict) else None ) if selected_ws_id is None: fallback_workspace, fallback_qf_version = self._fetch_first_workspace( normalized_base_url, normalized_token, qf_version=normalized_qf_version, qf_version_source=qf_version_source, ) if isinstance(fallback_workspace, dict): selected_ws_id = self._coerce_positive_int(fallback_workspace.get("wsId")) selected_ws_name = self._normalize_text( fallback_workspace.get("workspaceName") or fallback_workspace.get("wsName") or fallback_workspace.get("remark") ) or selected_ws_name workspace_qf_version = self._workspace_system_version(fallback_workspace) or fallback_qf_version elif selected_ws_name is None or workspace_qf_version is None: workspace = self._fetch_workspace_with_name_fallback( normalized_base_url, normalized_token, selected_ws_id, qf_version=normalized_qf_version, qf_version_source=qf_version_source, ) if isinstance(workspace, dict): selected_ws_name = self._normalize_text( workspace.get("workspaceName") or workspace.get("wsName") or workspace.get("remark") ) or selected_ws_name workspace_qf_version = self._workspace_system_version(workspace) or workspace_qf_version if workspace_qf_version is not None: resolved_qf_version, resolved_qf_version_source = workspace_qf_version, "workspace_system_version" else: resolved_qf_version, resolved_qf_version_source = self._resolve_backend_qf_version( response_qf_version, fallback_qf_version=normalized_qf_version, fallback_source=qf_version_source, ) uid = self._coerce_positive_int(resolved_user_info.get("uid")) if uid is None: raise_tool_error(QingflowApiError(category="auth", message="Token validation did not return valid user info")) session_profile = self.sessions.save_session( profile=profile, base_url=normalized_base_url, qf_version=resolved_qf_version, qf_version_source=resolved_qf_version_source, token=normalized_token, login_token=normalized_login_token, credential=None, uid=uid, email=self._normalize_text(resolved_user_info.get("email")), nick_name=self._normalize_text( resolved_user_info.get("nickName") or resolved_user_info.get("displayName") or resolved_user_info.get("name") ), persist=persist, ) if selected_ws_id is not None: session_profile = self.sessions.select_workspace(profile, ws_id=selected_ws_id, ws_name=selected_ws_name) backend_session = self.sessions.get_backend_session(profile) permission_level = ( self._workspace_permission_level( session_profile=session_profile, backend_session=backend_session, ) if backend_session is not None else None ) return { "profile": session_profile.profile, "base_url": session_profile.base_url, "qf_version": session_profile.qf_version, "qf_version_source": session_profile.qf_version_source, "uid": session_profile.uid, "email": session_profile.email, "nick_name": session_profile.nick_name, "selected_ws_id": session_profile.selected_ws_id, "selected_ws_name": session_profile.selected_ws_name, "suggested_ws_id": session_profile.selected_ws_id, "suggested_ws_name": session_profile.selected_ws_name, "permission_level": permission_level, "persisted": session_profile.persisted, "request_route": self._request_route_payload( BackendRequestContext( base_url=session_profile.base_url, token=normalized_token, ws_id=session_profile.selected_ws_id, qf_version=session_profile.qf_version, qf_version_source=session_profile.qf_version_source, ) ), } def _resolve_auth_inputs( self, *, profile: str, base_url: str | None, credential: str | None, ) -> tuple[str | None, str]: """按显式参数、环境变量、mcporter、profile 的顺序解析认证输入。""" normalized_base_url = self._normalize_text(base_url) normalized_credential = self._normalize_text(credential) if not normalized_credential: normalized_credential = self._normalize_text(os.getenv("QINGFLOW_CREDENTIAL")) if normalized_base_url and normalized_credential: return normalized_base_url, normalized_credential mcporter_context = self._read_mcporter_qingflow_context() if not normalized_base_url: normalized_base_url = self._normalize_text(mcporter_context.get("base_url")) if not normalized_base_url: normalized_base_url = self._normalize_text(os.getenv("QINGFLOW_BASE_URL")) if not normalized_credential: normalized_credential = self._normalize_text(mcporter_context.get("credential")) if not normalized_credential: session_profile = self.sessions.peek_profile(profile) if session_profile is not None: normalized_credential = self._normalize_text(session_profile.credential) return normalized_base_url, normalized_credential or "" def _read_mcporter_qingflow_context(self) -> dict[str, str]: """读取 OpenClaw mcporter 中的 Qingflow MCP 上下文。""" path = get_mcporter_config_path() if not path.exists(): return {} try: with path.open("r", encoding="utf-8") as handle: payload = json.load(handle) except (OSError, json.JSONDecodeError) as exc: raise_tool_error(QingflowApiError.config_error(f"failed to read mcporter config '{path}': {exc}")) if not isinstance(payload, dict): raise_tool_error(QingflowApiError.config_error(f"mcporter config '{path}' must be a JSON object")) mcp_servers = payload.get("mcpServers") qingflow = mcp_servers.get("qingflow") if isinstance(mcp_servers, dict) else None if not isinstance(qingflow, dict): return {} headers = qingflow.get("headers") credential = None if isinstance(headers, dict): credential = headers.get("x-qingflow-client-id") return { "base_url": str(qingflow.get("url") or "").strip(), "credential": str(credential or "").strip(), } def _refresh_profile_via_credential( self, profile: str, session_profile, ) -> bool: base_url = session_profile.base_url if session_profile is not None else None persist = bool(session_profile.persisted) if session_profile is not None else False try: self.auth_use_credential( profile=profile, base_url=base_url, persist=persist, ) except (QingflowApiError, RuntimeError): return False return True def auth_whoami(self, *, profile: str = DEFAULT_PROFILE) -> dict[str, Any]: """执行认证与会话相关逻辑。""" def build_response( session_profile, # type: ignore[no-untyped-def] backend_session, # type: ignore[no-untyped-def] context: BackendRequestContext, ) -> dict[str, Any]: cloud_context = current_cloud_context.get() workspace, workspace_qf_version = self._selected_workspace_snapshot( session_profile=session_profile, backend_session=backend_session, ) resolved_qf_version = workspace_qf_version or session_profile.qf_version resolved_qf_version_source = ( "workspace_system_version" if workspace_qf_version is not None else session_profile.qf_version_source ) if ( cloud_context is None and workspace_qf_version is not None and ( workspace_qf_version != session_profile.qf_version or session_profile.qf_version_source != "workspace_system_version" ) ): session_profile = self.sessions.update_route( profile, qf_version=workspace_qf_version, qf_version_source="workspace_system_version", ) backend_session = self.sessions.get_backend_session(profile) or backend_session context = BackendRequestContext( base_url=backend_session.base_url, token=backend_session.token, ws_id=session_profile.selected_ws_id, qf_version=backend_session.qf_version, qf_version_source=backend_session.qf_version_source, ) if cloud_context is None and self._should_refresh_identity_metadata(session_profile): refreshed_profile = self._refresh_identity_metadata( profile=profile, session_profile=session_profile, backend_session=backend_session, context=context, ) if refreshed_profile is not None: session_profile = refreshed_profile response = { "profile": session_profile.profile, "base_url": session_profile.base_url, "qf_version": resolved_qf_version, "qf_version_source": resolved_qf_version_source, "uid": session_profile.uid, "email": session_profile.email, "nick_name": session_profile.nick_name, "selected_ws_id": session_profile.selected_ws_id, "selected_ws_name": session_profile.selected_ws_name, "persisted": session_profile.persisted, "request_route": self._request_route_payload(context), } member_info, member_warnings = self._workspace_member_info( session_profile=session_profile, backend_session=backend_session, ) response.update(member_info) if member_warnings: response["warnings"] = member_warnings return response session_profile = None backend_session = None try: session_profile, backend_session, context = self._require_context(profile, require_workspace=False) if backend_session.credential and current_cloud_context.get() is None: self._probe_token_validity( session_profile=session_profile, backend_session=backend_session, ) return build_response(session_profile, backend_session, context) except QingflowApiError as error: if error.looks_like_invalid_token() and self._refresh_current_cloud_context(): try: refreshed_profile, refreshed_backend_session, refreshed_context = self._require_context( profile, require_workspace=False, ) return build_response(refreshed_profile, refreshed_backend_session, refreshed_context) except QingflowApiError as refreshed_error: self._handle_error(profile, refreshed_error) if ( error.looks_like_invalid_token() and session_profile is not None and backend_session is not None and self._refresh_session_from_credential( profile, session_profile=session_profile, backend_session=backend_session, ) ): try: refreshed_profile, refreshed_backend_session, refreshed_context = self._require_context( profile, require_workspace=False, ) return build_response(refreshed_profile, refreshed_backend_session, refreshed_context) except QingflowApiError as refreshed_error: self._handle_error(profile, refreshed_error) self._handle_error(profile, error) raise AssertionError("unreachable") def _probe_token_validity( self, *, session_profile, # type: ignore[no-untyped-def] backend_session, # type: ignore[no-untyped-def] ) -> None: """执行内部辅助逻辑。""" probe_context = BackendRequestContext( base_url=backend_session.base_url, token=backend_session.token, ws_id=session_profile.selected_ws_id, qf_version=backend_session.qf_version, qf_version_source=backend_session.qf_version_source, ) try: self.backend.request("GET", probe_context, "/user") except QingflowApiError as error: if error.looks_like_invalid_token(): raise def auth_logout(self, *, profile: str = DEFAULT_PROFILE, forget_persisted: bool = False) -> dict[str, Any]: """执行认证与会话相关逻辑。""" if not self.sessions.has_profile(profile): raise_tool_error(QingflowApiError.auth_required(profile)) self.sessions.logout(profile, forget_persisted=forget_persisted) return { "profile": profile, "logged_out": True, "forgot_persisted": forget_persisted, } def _normalize_base_url(self, base_url: str | None) -> str: """执行内部辅助逻辑。""" normalized_base_url = normalize_base_url(base_url) or get_default_base_url() if not normalized_base_url: raise_tool_error( QingflowApiError.config_error( "base_url is required or configure default_base_url / QINGFLOW_MCP_DEFAULT_BASE_URL" ) ) return normalized_base_url def _normalize_qf_version(self, qf_version: str | None) -> str | None: """执行内部辅助逻辑。""" if qf_version is not None: normalized = str(qf_version).strip() return normalized or None return get_default_qf_version() def _resolve_qf_version_input(self, qf_version: str | None) -> tuple[str | None, str]: """执行内部辅助逻辑。""" if qf_version is not None: normalized = self._normalize_qf_version(qf_version) return normalized, "explicit" if normalized else "unset" normalized = self._normalize_qf_version(None) if normalized: return normalized, "default_config" return None, "unset" def _resolve_backend_qf_version( self, backend_qf_version: str | None, *, fallback_qf_version: str | None, fallback_source: str, ) -> tuple[str | None, str]: """执行内部辅助逻辑。""" if backend_qf_version: return backend_qf_version, "backend_response" return fallback_qf_version, fallback_source def _fetch_auth_context( self, base_url: str, credential: str, *, qf_version: str | None, ) -> tuple[dict[str, Any], str | None]: """执行内部辅助逻辑。""" response = self.backend.public_request_with_meta( "POST", base_url, "/mcp/auth/context", json_body={"credential": credential}, qf_version=qf_version, ) payload = self._unwrap_auth_context_payload(response.data) return payload, response.qf_response_version def _unwrap_auth_context_payload(self, payload: Any) -> dict[str, Any]: """执行内部辅助逻辑。""" if not isinstance(payload, dict): raise_tool_error(QingflowApiError(category="auth", message="Credential context did not return a valid result")) for key in ("data", "result"): nested = payload.get(key) if isinstance(nested, dict): return nested return payload def _workspace_system_version(self, workspace: Any) -> str | None: """执行内部辅助逻辑。""" if not isinstance(workspace, dict): return None value = workspace.get("systemVersion") if value is None: return None normalized = str(value).strip() return normalized or None def _fetch_user_info( self, base_url: str, token: str, ws_id: int | None, *, qf_version: str | None, qf_version_source: str | None, ) -> tuple[dict[str, Any], str | None]: """执行内部辅助逻辑。""" request_context = BackendRequestContext( base_url=base_url, token=token, ws_id=ws_id, qf_version=qf_version, qf_version_source=qf_version_source, ) try: user_response = self.backend.request_with_meta("GET", request_context, "/user") user_info = user_response.data if isinstance(user_info, dict): return user_info, user_response.qf_response_version except QingflowApiError as original_error: if ws_id is not None: raise original_error first_workspace, workspace_qf_version = self._fetch_first_workspace( base_url, token, qf_version=qf_version, qf_version_source=qf_version_source, ) if not first_workspace: raise original_error first_ws_id = first_workspace.get("wsId") if not first_ws_id: raise original_error effective_qf_version = workspace_qf_version or qf_version effective_qf_version_source = "backend_response" if workspace_qf_version else qf_version_source fallback_context = BackendRequestContext( base_url=base_url, token=token, ws_id=int(first_ws_id), qf_version=effective_qf_version, qf_version_source=effective_qf_version_source, ) user_response = self.backend.request_with_meta("GET", fallback_context, "/user") user_info = user_response.data if isinstance(user_info, dict): return user_info, user_response.qf_response_version or effective_qf_version raise original_error raise_tool_error(QingflowApiError(category="auth", message="Token validation did not return valid user info")) def _try_fetch_user_info( self, base_url: str, token: str, *, qf_version: str | None, qf_version_source: str | None, ) -> tuple[dict[str, Any] | None, str | None]: """执行内部辅助逻辑。""" try: return self._fetch_user_info( base_url, token, None, qf_version=qf_version, qf_version_source=qf_version_source, ) except QingflowApiError as exc: if is_auth_like_error(exc): raise return None, None def _fetch_first_workspace( self, base_url: str, token: str, *, qf_version: str | None, qf_version_source: str | None, ) -> tuple[dict[str, Any] | None, str | None]: """执行内部辅助逻辑。""" page_response = self.backend.request_with_meta( "POST", BackendRequestContext( base_url=base_url, token=token, ws_id=None, qf_version=qf_version, qf_version_source=qf_version_source, ), "/user/workspaceList/pageQuery", json_body={"pageNum": 1, "pageSize": 1}, ) page = page_response.data if not isinstance(page, dict): return None, page_response.qf_response_version workspaces = page.get("list") or [] if not workspaces: return None, page_response.qf_response_version first_workspace = workspaces[0] return (first_workspace if isinstance(first_workspace, dict) else None), page_response.qf_response_version def _fetch_workspace( self, base_url: str, token: str, ws_id: int, *, qf_version: str | None, qf_version_source: str | None, ) -> dict[str, Any]: """执行内部辅助逻辑。""" workspace = self.backend.request( "GET", BackendRequestContext( base_url=base_url, token=token, ws_id=None, qf_version=qf_version, qf_version_source=qf_version_source, ), f"/user/workspace/{ws_id}", ) if not isinstance(workspace, dict): raise_tool_error(QingflowApiError(category="workspace", message=f"Workspace {ws_id} is not accessible")) return workspace def _selected_workspace_snapshot( self, *, session_profile, # type: ignore[no-untyped-def] backend_session, # type: ignore[no-untyped-def] ) -> tuple[dict[str, Any] | None, str | None]: ws_id = session_profile.selected_ws_id if ws_id is None: return None, None workspace = self._fetch_workspace_with_name_fallback( session_profile.base_url, backend_session.token, ws_id, qf_version=session_profile.qf_version, qf_version_source=session_profile.qf_version_source, ) return workspace, self._workspace_system_version(workspace) def _request_route_payload(self, context: BackendRequestContext) -> dict[str, Any]: """执行内部辅助逻辑。""" 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": context.base_url, "qf_version": context.qf_version, "qf_version_source": context.qf_version_source or ("context" if context.qf_version else "unknown"), } def _should_refresh_identity_metadata(self, session_profile) -> bool: # type: ignore[no-untyped-def] """执行内部辅助逻辑。""" return ( session_profile.uid == 0 or session_profile.email is None or session_profile.nick_name is None or session_profile.selected_ws_name is None ) def _refresh_identity_metadata( self, *, profile: str, session_profile, # type: ignore[no-untyped-def] backend_session, # type: ignore[no-untyped-def] context: BackendRequestContext, ): """执行内部辅助逻辑。""" try: user_info, _ = self._fetch_user_info( session_profile.base_url, backend_session.token, session_profile.selected_ws_id, qf_version=session_profile.qf_version, qf_version_source=session_profile.qf_version_source, ) except QingflowApiError as exc: if is_auth_like_error(exc): raise return None ws_name = session_profile.selected_ws_name if session_profile.selected_ws_id is not None: workspace = self._fetch_workspace_with_name_fallback( session_profile.base_url, backend_session.token, session_profile.selected_ws_id, qf_version=session_profile.qf_version, qf_version_source=session_profile.qf_version_source, ) if isinstance(workspace, dict): ws_name = ( str(workspace.get("workspaceName") or workspace.get("wsName") or workspace.get("remark") or "").strip() or ws_name ) email = user_info["email"] if "email" in user_info else session_profile.email nick_name = ( user_info.get("nickName") or user_info.get("displayName") or user_info.get("name") or session_profile.nick_name ) uid = user_info.get("uid") refreshed = self.sessions.update_profile_metadata( profile, uid=int(uid) if uid is not None else session_profile.uid, email=email, nick_name=nick_name, selected_ws_id=session_profile.selected_ws_id, selected_ws_name=ws_name, ) return refreshed def _workspace_member_info( self, *, session_profile, # type: ignore[no-untyped-def] backend_session, # type: ignore[no-untyped-def] ) -> tuple[dict[str, Any], list[dict[str, Any]]]: """执行内部辅助逻辑。""" default_payload = { "departments": [], "roles": [], "permission_level": None, } ws_id = session_profile.selected_ws_id if ws_id is None: return default_payload, [] permission_level = self._workspace_permission_level( session_profile=session_profile, backend_session=backend_session, ) payload = dict(default_payload) payload["permission_level"] = permission_level warnings: list[dict[str, Any]] = [] if permission_level is None: warnings.append( { "code": "WORKSPACE_PERMISSION_LEVEL_UNAVAILABLE", "message": ( "auth_whoami could not resolve the selected workspace permission level; " "do not infer the user is a basic member or an administrator from this null value." ), "ws_id": ws_id, } ) context = BackendRequestContext( base_url=backend_session.base_url, token=backend_session.token, ws_id=ws_id, qf_version=backend_session.qf_version, qf_version_source=backend_session.qf_version_source, ) member_lookup_warnings: list[dict[str, Any]] = [] member = self._lookup_current_member( context=context, uid=session_profile.uid, email=session_profile.email, nick_name=session_profile.nick_name, warnings=member_lookup_warnings, ) warnings.extend(member_lookup_warnings) if member is None: warnings.append( { "code": "CURRENT_MEMBER_PROFILE_UNAVAILABLE", "message": ( "auth_whoami could not resolve current member departments and roles " f"in workspace {ws_id}." ), } ) return payload, warnings payload["departments"] = self._compact_departments(member) payload["roles"] = self._compact_roles(member) return payload, warnings def _workspace_permission_level( self, *, session_profile, # type: ignore[no-untyped-def] backend_session, # type: ignore[no-untyped-def] ) -> str | None: """Resolve the selected workspace permission label without requiring member lookup.""" ws_id = session_profile.selected_ws_id if ws_id is None: return None context = BackendRequestContext( base_url=backend_session.base_url, token=backend_session.token, ws_id=ws_id, qf_version=backend_session.qf_version, qf_version_source=backend_session.qf_version_source, ) return self._resolve_permission_level(self._workspace_auth(context, ws_id=ws_id)) def _workspace_auth(self, context: BackendRequestContext, *, ws_id: int) -> int | None: """执行内部辅助逻辑。""" workspace = self._fetch_workspace_auth_from_detail(context, ws_id=ws_id) if workspace is not None: return workspace return self._fetch_workspace_auth_from_list(context, ws_id=ws_id) def _fetch_workspace_auth_from_detail(self, context: BackendRequestContext, *, ws_id: int) -> int | None: """执行内部辅助逻辑。""" try: workspace = self.backend.request("GET", context, f"/user/workspace/{ws_id}") except QingflowApiError as exc: if not _is_optional_auth_lookup_error(exc): raise return None if not isinstance(workspace, dict): return None return self._coerce_auth_value(workspace.get("auth")) def _fetch_workspace_auth_from_list(self, context: BackendRequestContext, *, ws_id: int) -> int | None: """执行内部辅助逻辑。""" try: payload = self.backend.request( "POST", context, "/user/workspaceList/pageQuery", json_body={"pageNum": 1, "pageSize": 100, "authList": [0, 1, 2, 3]}, ) except QingflowApiError as exc: if not _is_optional_auth_lookup_error(exc): raise return None workspaces = payload.get("list") if isinstance(payload, dict) else [] if not isinstance(workspaces, list): return None for item in workspaces: if not isinstance(item, dict) or item.get("wsId") != ws_id: continue return self._coerce_auth_value(item.get("auth")) return None def _lookup_current_member( self, *, context: BackendRequestContext, uid: int | None, email: str | None, nick_name: str | None, warnings: list[dict[str, Any]] | None = None, ) -> dict[str, Any] | None: """执行内部辅助逻辑。""" candidates: list[dict[str, Any]] = [] for keyword in (email, nick_name): member = self._search_member_once(context, uid=uid, keyword=keyword, warnings=warnings) if member is not None: return member if keyword: candidates.extend(self._search_member_items(context, keyword=keyword, warnings=warnings)) if uid is not None and uid > 0: for item in candidates: if self._same_member(item, uid=uid): return item return self._search_member_once(context, uid=uid, keyword=None, warnings=warnings) return None def _search_member_once( self, context: BackendRequestContext, *, uid: int | None, keyword: str | None, warnings: list[dict[str, Any]] | None = None, ) -> dict[str, Any] | None: """执行内部辅助逻辑。""" for item in self._search_member_items(context, keyword=keyword, warnings=warnings): if self._same_member(item, uid=uid): return item return None def _search_member_items( self, context: BackendRequestContext, *, keyword: str | None, warnings: list[dict[str, Any]] | None = None, ) -> list[dict[str, Any]]: """执行内部辅助逻辑。""" params: dict[str, Any] = {"pageNum": 1, "pageSize": 100, "containDisable": True} normalized_keyword = str(keyword or "").strip() if normalized_keyword: params["keyword"] = normalized_keyword try: payload = self.backend.request("GET", context, "/contact", params=params) except QingflowApiError as exc: if not _is_optional_auth_lookup_error(exc): raise if warnings is not None and _is_contact_directory_permission_denied(exc): self._append_unique_warning(warnings, self._contact_directory_permission_warning(exc)) return [] items = self._extract_items(payload) return [item for item in items if isinstance(item, dict)] def _append_unique_warning(self, warnings: list[dict[str, Any]], warning: dict[str, Any]) -> None: """执行内部辅助逻辑。""" code = self._normalize_text(warning.get("code")) if code is not None and any(item.get("code") == code for item in warnings): return warnings.append(warning) def _contact_directory_permission_warning(self, error: QingflowApiError) -> dict[str, Any]: """执行内部辅助逻辑。""" warning: dict[str, Any] = { "code": "CONTACT_DIRECTORY_PERMISSION_DENIED", "message": ( "auth_whoami could not read current member departments and roles because " "the contact directory is not readable in this permission context; " "permission_level still comes from the workspace auth route." ), "category": error.category, "backend_code": backend_code_int(error), "http_status": error.http_status, "request_id": error.request_id, } return {key: value for key, value in warning.items() if value is not None} def _same_member(self, item: dict[str, Any], *, uid: int | None) -> bool: """执行内部辅助逻辑。""" if uid is None or uid <= 0: return False for key in ("uid", "id", "userId"): value = item.get(key) if value is None: continue coerced = self._coerce_int(value) if coerced is not None and coerced == uid: return True if str(value).strip() == str(uid): return True return False def _compact_departments(self, member: dict[str, Any]) -> list[dict[str, Any]]: """执行内部辅助逻辑。""" items: list[dict[str, Any]] = [] seen: set[tuple[int | None, str | None]] = set() for depart in self._walk_nested_items(member.get("departs")): if not isinstance(depart, dict): continue dept_id = self._coerce_int( depart.get("deptId", depart.get("departId", depart.get("id"))) ) dept_name = self._normalize_text( depart.get("deptName", depart.get("departName", depart.get("name"))) ) key = (dept_id, dept_name) if key in seen or (dept_id is None and dept_name is None): continue seen.add(key) item = {"dept_id": dept_id, "dept_name": dept_name} items.append({k: v for k, v in item.items() if v is not None}) return items def _compact_roles(self, member: dict[str, Any]) -> list[dict[str, Any]]: """执行内部辅助逻辑。""" items: list[dict[str, Any]] = [] seen: set[tuple[int | None, str | None]] = set() for role in self._walk_nested_items(member.get("roles")): if not isinstance(role, dict): continue role_id = self._coerce_int(role.get("roleId", role.get("id"))) role_name = self._normalize_text(role.get("roleName", role.get("name"))) key = (role_id, role_name) if key in seen or (role_id is None and role_name is None): continue seen.add(key) item = {"role_id": role_id, "role_name": role_name} items.append({k: v for k, v in item.items() if v is not None}) return items def _resolve_permission_level(self, auth_code: int | None) -> str | None: """执行内部辅助逻辑。""" mapping = { 2: "超级管理", 1: "系统管理员", 3: "子管理员", 0: "基本成员", } return mapping.get(auth_code) def _coerce_auth_value(self, value: Any) -> int | None: """执行内部辅助逻辑。""" coerced = self._coerce_int(value) if coerced is not None: return coerced normalized = self._normalize_text(value) if normalized is None: return None lowered = normalized.lower() if lowered in {"creator", "workspaccreator", "workspacecreator"}: return 2 if lowered in {"admin", "administrator"}: return 1 if lowered in {"subadmin", "dataadmin"}: return 3 if lowered in {"member", "visitor", "normal"}: return 0 return None def _extract_items(self, payload: Any) -> list[Any]: """执行内部辅助逻辑。""" if isinstance(payload, list): return payload if not isinstance(payload, dict): return [] for key in ("list", "items", "rows", "result"): value = payload.get(key) if isinstance(value, list): return value for key in ("data", "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 _walk_nested_items(self, value: Any) -> list[Any]: """执行内部辅助逻辑。""" if isinstance(value, list): items: list[Any] = [] for item in value: items.extend(self._walk_nested_items(item)) return items return [value] def _coerce_int(self, value: Any) -> int | None: """执行内部辅助逻辑。""" if isinstance(value, bool) or value is None: return None if isinstance(value, int): return value try: return int(str(value).strip()) except (TypeError, ValueError): return None def _normalize_text(self, value: Any) -> str | None: """执行内部辅助逻辑。""" if value is None: return None text = str(value).strip() return text or None def _fetch_workspace_with_name_fallback( self, base_url: str, token: str, ws_id: int, *, qf_version: str | None, qf_version_source: str | None, ) -> dict[str, Any] | None: """执行内部辅助逻辑。""" try: workspace = self._fetch_workspace( base_url, token, ws_id, qf_version=qf_version, qf_version_source=qf_version_source, ) except QingflowApiError as exc: if is_auth_like_error(exc): raise workspace = None if isinstance(workspace, dict): workspace_name = str(workspace.get("workspaceName") or workspace.get("wsName") or "").strip() if workspace_name: return workspace try: fallback = self._fetch_workspace_from_list( base_url, token, ws_id, qf_version=qf_version, qf_version_source=qf_version_source, ) except QingflowApiError as exc: if is_auth_like_error(exc): raise fallback = None return fallback or workspace def _fetch_workspace_from_list( self, base_url: str, token: str, ws_id: int, *, qf_version: str | None, qf_version_source: str | None, ) -> dict[str, Any] | None: """执行内部辅助逻辑。""" payload = self.backend.request( "POST", BackendRequestContext( base_url=base_url, token=token, ws_id=ws_id, qf_version=qf_version, qf_version_source=qf_version_source, ), "/user/workspaceList/pageQuery", json_body={"pageNum": 1, "pageSize": 100, "authList": [0, 1, 2]}, ) workspaces = payload.get("list") if isinstance(payload, dict) else [] if not isinstance(workspaces, list): return None found = next( ( item for item in workspaces if isinstance(item, dict) and item.get("wsId") == ws_id ), None, ) return found if isinstance(found, dict) else None def _is_contact_directory_permission_denied(error: QingflowApiError) -> bool: if is_auth_like_error(error): return False return backend_code_int(error) in {40002, 40027} def _is_optional_auth_lookup_error(error: QingflowApiError) -> bool: if is_auth_like_error(error): return False backend_code = backend_code_int(error) return backend_code in {40002, 40027, 404} or error.http_status == 404