from __future__ import annotations from collections import Counter from typing import Any from mcp.server.fastmcp import FastMCP from ..backend_client import BackendRequestContext from ..config import DEFAULT_PROFILE from ..errors import QingflowApiError, backend_code_int, is_auth_like_error, raise_tool_error from ..json_types import JSONObject from ..list_type_labels import SYSTEM_VIEW_DEFINITIONS, get_app_publish_status_label from ..solution.compiler.icon_utils import workspace_icon_config from .base import ToolBase class AppTools(ToolBase): """应用工具（中文名：应用与表单管理）。 类型：应用元数据与配置工具。 主要职责： 1. 查询应用列表、读取应用详情； 2. 读取与更新应用基础配置、表单结构与发布状态； 3. 提供应用创建、删除、发布等管理能力。 """ def register(self, mcp: FastMCP) -> None: """注册当前工具到 MCP 服务。""" @mcp.tool() def app_list(profile: str = DEFAULT_PROFILE, query: str = "", keyword: str = "", ship_auth: bool = False) -> JSONObject: return self.app_list(profile=profile, query=query, keyword=keyword, ship_auth=ship_auth) @mcp.tool() def app_get(profile: str = DEFAULT_PROFILE, app_key: str = "") -> JSONObject: return self.app_get(profile=profile, app_key=app_key) @mcp.tool() def app_get_base(profile: str = DEFAULT_PROFILE, app_key: str = "", include_raw: bool = False) -> JSONObject: return self.app_get_base(profile=profile, app_key=app_key, include_raw=include_raw) @mcp.tool(description=self._high_risk_tool_description(operation="update", target="app base settings")) def app_update_base(profile: str = DEFAULT_PROFILE, app_key: str = "", payload: JSONObject | None = None) -> JSONObject: return self.app_update_base(profile=profile, app_key=app_key, payload=payload or {}) @mcp.tool() def app_get_form_schema( profile: str = DEFAULT_PROFILE, app_key: str = "", form_type: int | str = 1, being_draft: bool | None = None, being_apply: bool | None = None, audit_node_id: int | None = None, include_raw: bool = False, ) -> JSONObject: return self.app_get_form_schema( profile=profile, app_key=app_key, form_type=form_type, being_draft=being_draft, being_apply=being_apply, audit_node_id=audit_node_id, include_raw=include_raw, ) @mcp.tool() def app_get_edit_version_no(profile: str = DEFAULT_PROFILE, app_key: str = "") -> JSONObject: return self.app_get_edit_version_no(profile=profile, app_key=app_key) @mcp.tool() def app_edit_finished(profile: str = DEFAULT_PROFILE, app_key: str = "", payload: JSONObject | None = None) -> JSONObject: return self.app_edit_finished(profile=profile, app_key=app_key, payload=payload or {}) @mcp.tool() def app_create(profile: str = DEFAULT_PROFILE, payload: JSONObject | None = None) -> JSONObject: return self.app_create(profile=profile, payload=payload or {}) @mcp.tool(description=self._high_risk_tool_description(operation="delete", target="app configuration")) def app_delete(profile: str = DEFAULT_PROFILE, app_key: str = "") -> JSONObject: return self.app_delete(profile=profile, app_key=app_key) @mcp.tool() def app_publish( profile: str = DEFAULT_PROFILE, app_key: str = "", payload: JSONObject | None = None, ) -> JSONObject: return self.app_publish(profile=profile, app_key=app_key, payload=payload or {}) def app_list(self, *, profile: str, query: str = "", keyword: str = "", ship_auth: bool = False) -> JSONObject: """List current-user visible apps in the selected workspace.""" def runner(session_profile, context): result = self.backend.request("GET", context, "/tag/apps") items, source_shape = self._extract_visible_apps(result) normalized_query, warnings = _normalize_app_list_query(query=query, keyword=keyword) unfiltered_count = len(items) if normalized_query: items = self._filter_visible_apps(items, normalized_query) response = { "profile": profile, "ws_id": session_profile.selected_ws_id, "items": items, "count": len(items), "source_shape": source_shape, "query": normalized_query, "matched_count": len(items), "unfiltered_count": unfiltered_count, "filter_mode": "local_visible_apps", } if warnings: response["warnings"] = warnings if ship_auth: response["raw"] = result return response return self._run(profile, runner, tool_name='应用列表') def _filter_visible_apps(self, items: list[JSONObject], query: str) -> list[JSONObject]: """Filter current-user visible apps locally without calling admin search APIs.""" needle = query.casefold() matched: list[JSONObject] = [] for item in items: haystacks = ( item.get("app_key"), item.get("app_name"), item.get("title"), item.get("package_name"), item.get("tag_name"), item.get("group_name"), ) if any(needle in str(value).casefold() for value in haystacks if value not in (None, "")): matched.append(item) return matched def _resolve_visible_app(self, context: BackendRequestContext, app_key: str) -> JSONObject | None: """Resolve app display metadata from the current user's visible app tree.""" try: result = self.backend.request("GET", context, "/tag/apps") except QingflowApiError as exc: if not _is_optional_app_metadata_error(exc): raise return None items, _source_shape = self._extract_visible_apps(result) for item in items: if str(item.get("app_key") or "").strip() == app_key: return item return None def app_search(self, *, profile: str, keyword: str = "", page_num: int = 1, page_size: int = 50) -> JSONObject: """Search apps by keyword in name/title using backend search API. Useful for finding BUG-related apps across all packages.""" def runner(session_profile, context): # Use GET /app/item which supports queryKey search across all apps params: JSONObject = {"pageNum": page_num, "pageSize": page_size} if keyword: params["queryKey"] = keyword warnings: list[JSONObject] = [] source = "app_item_search" try: result = self.backend.request("GET", context, "/app/item", params=params) except QingflowApiError as exc: if is_auth_like_error(exc): raise if not _is_optional_app_metadata_error(exc): raise visible_payload = self.backend.request("GET", context, "/tag/apps") all_apps, source_shape = self._extract_visible_apps(visible_payload) matched_apps = self._filter_visible_apps(all_apps, keyword) if keyword else all_apps page_start = max(page_num - 1, 0) * page_size page_end = page_start + page_size apps = matched_apps[page_start:page_end] warnings.append( { "code": "APP_SEARCH_FALLBACK_VISIBLE_APPS", "message": ( "app_search used the current user's visible app tree because the global app search " "endpoint is unavailable in this permission context." ), "backend_code": exc.backend_code, "http_status": exc.http_status, "request_id": exc.request_id, } ) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "keyword": keyword, "page_num": page_num, "page_size": page_size, "total": len(matched_apps), "items": apps, "apps": apps, "source": "visible_apps_fallback", "source_shape": source_shape, "unfiltered_count": len(all_apps), "matched_count": len(matched_apps), "warnings": warnings, } apps = [] if isinstance(result, dict): items = result.get("list", []) for item in items: if isinstance(item, dict): normalized = self._normalize_visible_app( item, package_tag_id=_coerce_positive_int(item.get("tagId")), package_name=str(item.get("tagName") or "").strip() or None, group_id=_coerce_positive_int(item.get("groupId")), group_name=str(item.get("groupName") or "").strip() or None, ) if normalized is not None: apps.append(normalized) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "keyword": keyword, "page_num": page_num, "page_size": page_size, "total": result.get("total") if isinstance(result, dict) else len(apps), "items": apps, "apps": apps, "source": source, } return self._run(profile, runner, tool_name='应用搜索') def app_get(self, *, profile: str, app_key: str) -> JSONObject: """执行应用相关逻辑。""" self._require_app_key(app_key) def runner(session_profile, context): warnings: list[JSONObject] = [] app_name = app_key base_info: JSONObject | None = None app_icon = None visible_app: JSONObject | None = None try: base_info = self.backend.request("GET", context, f"/app/{app_key}/baseInfo") if isinstance(base_info, dict): app_name = ( str(base_info.get("formTitle") or base_info.get("title") or base_info.get("appName") or app_key).strip() or app_key ) app_icon = str(base_info.get("appIcon") or "").strip() or None except QingflowApiError as exc: if not _is_optional_app_metadata_error(exc): raise visible_app = self._resolve_visible_app(context, app_key) if visible_app is not None: app_name = str(visible_app.get("app_name") or visible_app.get("title") or app_key).strip() or app_key app_icon = str(visible_app.get("app_icon") or "").strip() or None warnings.append( _with_api_error_fields( { "code": "APP_BASE_INFO_UNAVAILABLE", "message": ( f"app_get could not load base info for {app_key}; " "using the current user's visible app tree when available." ), }, exc, ) ) can_create = self._probe_create_access(context, app_key) accessible_views, system_view_warnings = self._resolve_accessible_system_views(context, app_key) warnings.extend(system_view_warnings) custom_views, custom_view_warnings = self._resolve_accessible_custom_views(context, app_key) warnings.extend(custom_view_warnings) accessible_views.extend(custom_views) import_capability, import_warnings = _derive_import_capability(base_info) warnings.extend(import_warnings) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "ok": True, "request_route": {"base_url": context.base_url, "qf_version": context.qf_version}, "warnings": warnings, "data": { "app_key": app_key, "app_name": app_name, "app_icon": app_icon, "icon_config": workspace_icon_config(app_icon), **({"visible_app": visible_app} if base_info is None and visible_app is not None else {}), "can_create": can_create, "import_capability": import_capability, "accessible_views": accessible_views, }, } return self._run(profile, runner, tool_name='应用详情') def app_get_base(self, *, profile: str, app_key: str, include_raw: bool = False) -> JSONObject: """执行应用相关逻辑。""" self._require_app_key(app_key) def runner(session_profile, context): warnings: list[JSONObject] = [] verification: JSONObject = {"base_info_verified": True, "fallback_visible_app_used": False} try: result = self.backend.request("GET", context, f"/app/{app_key}/baseInfo") except QingflowApiError as exc: if not _is_optional_app_metadata_error(exc): raise visible_app = self._resolve_visible_app(context, app_key) if visible_app is None: raise result = self._base_info_from_visible_app(app_key, visible_app) warnings.append( _with_api_error_fields( { "code": "APP_BASE_INFO_UNAVAILABLE", "message": ( f"app_get_base could not load base info for {app_key}; " "using the current user's visible app tree." ), }, exc, ) ) verification = {"base_info_verified": False, "fallback_visible_app_used": True} publish_status = result.get("appPublishStatus") if isinstance(result, dict) else None compact = self._compact_base_info(result if isinstance(result, dict) else {}) response = { "profile": profile, "ws_id": session_profile.selected_ws_id, "app_key": app_key, "result": result if include_raw else compact, "app_publish_status": publish_status, "app_publish_status_label": get_app_publish_status_label(publish_status if isinstance(publish_status, int) else None), "compact": not include_raw, "verification": verification, } if warnings: response["warnings"] = warnings if include_raw: response["summary"] = compact return response return self._run(profile, runner, tool_name='应用基础信息') def app_update_base(self, *, profile: str, app_key: str, payload: JSONObject) -> JSONObject: """执行应用相关逻辑。""" self._require_app_key(app_key) body = self._require_dict(payload) def runner(session_profile, context): result = self.backend.request("POST", context, f"/app/{app_key}/baseInfo", json_body=body) return self._attach_human_review_notice( {"profile": profile, "ws_id": session_profile.selected_ws_id, "app_key": app_key, "result": result}, operation="update", target="app base settings", ) return self._run(profile, runner, tool_name='更新应用基础信息') def app_get_form_schema( self, *, profile: str, app_key: str, form_type: int | str, being_draft: bool | None, being_apply: bool | None, audit_node_id: int | None, include_raw: bool = False, ) -> JSONObject: """执行应用相关逻辑。""" self._require_app_key(app_key) resolved_form_type = _normalize_form_type(form_type) def runner(session_profile, context): params: JSONObject = {"type": resolved_form_type} if being_draft is not None: params["beingDraft"] = being_draft if being_apply is not None: params["beingApply"] = being_apply if audit_node_id is not None: params["auditNodeId"] = audit_node_id result = self.backend.request("GET", context, f"/app/{app_key}/form", params=params) compact = self._compact_form_schema(result if isinstance(result, dict) else {}) response = { "profile": profile, "ws_id": session_profile.selected_ws_id, "app_key": app_key, "result": result if include_raw else compact, "compact": not include_raw, "requested_form_type": form_type, "resolved_form_type": resolved_form_type, } if include_raw: response["summary"] = compact return response return self._run(profile, runner, tool_name='应用表单结构') def app_get_edit_version_no(self, *, profile: str, app_key: str) -> JSONObject: """执行应用相关逻辑。""" self._require_app_key(app_key) def runner(session_profile, context): result = self.backend.request("GET", context, f"/app/{app_key}/editVersionNo") return {"profile": profile, "ws_id": session_profile.selected_ws_id, "app_key": app_key, "result": result} return self._run(profile, runner, tool_name='应用编辑版本号') def app_get_apply_base_info(self, *, profile: str, app_key: str, list_type: int) -> JSONObject: """执行应用相关逻辑。""" self._require_app_key(app_key) if not isinstance(list_type, int) or isinstance(list_type, bool) or list_type <= 0: raise_tool_error(QingflowApiError.config_error("list_type must be a positive integer")) def runner(session_profile, context): result = self.backend.request("GET", context, f"/app/{app_key}/apply/baseInfo", params={"type": list_type}) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "app_key": app_key, "list_type": list_type, "result": result, } return self._run(profile, runner, tool_name='应用流程基础信息') def app_update_apply_config(self, *, profile: str, app_key: str, payload: JSONObject) -> JSONObject: """执行应用相关逻辑。""" self._require_app_key(app_key) body = self._require_dict(payload) def runner(session_profile, context): result = self.backend.request("POST", context, f"/app/{app_key}/apply/config", json_body=body) return self._attach_human_review_notice( { "profile": profile, "ws_id": session_profile.selected_ws_id, "app_key": app_key, "result": result, }, operation="update", target="app apply list configuration", ) return self._run(profile, runner, tool_name='更新应用流程配置') def app_update_form_schema(self, *, profile: str, app_key: str, payload: JSONObject) -> JSONObject: """执行应用相关逻辑。""" self._require_app_key(app_key) body = self._require_dict(payload) def runner(session_profile, context): result = self.backend.request("POST", context, f"/app/{app_key}/form", json_body=body) return self._attach_human_review_notice( {"profile": profile, "ws_id": session_profile.selected_ws_id, "app_key": app_key, "result": result}, operation="update", target="app form schema", ) return self._run(profile, runner, tool_name='更新应用表单结构') def app_edit_finished(self, *, profile: str, app_key: str, payload: JSONObject) -> JSONObject: """执行应用相关逻辑。""" self._require_app_key(app_key) body = self._require_dict(payload) def runner(session_profile, context): result = self.backend.request("POST", context, f"/app/{app_key}/editFinished", json_body=body) return {"profile": profile, "ws_id": session_profile.selected_ws_id, "app_key": app_key, "result": result} return self._run(profile, runner, tool_name='应用编辑完成') def app_create(self, *, profile: str, payload: JSONObject) -> JSONObject: """执行应用相关逻辑。""" body = self._require_dict(payload) def runner(session_profile, context): attempted_contexts = [context] if context.qf_version is not None: attempted_contexts.append( BackendRequestContext( base_url=context.base_url, token=context.token, ws_id=context.ws_id, qf_version=None, qf_version_source="app_create_retry_without_qf_version", ) ) last_error: QingflowApiError | None = None request_with_meta = getattr(self.backend, "request_with_meta", None) for call_context in attempted_contexts: try: if callable(request_with_meta): response = request_with_meta("POST", call_context, "/app", json_body=body) result = response.data request_id = response.request_id else: result = self.backend.request("POST", call_context, "/app", json_body=body) request_id = None return { "profile": profile, "ws_id": session_profile.selected_ws_id, "result": result, "request_route": self.backend.describe_route(call_context), "request_id": request_id, } except QingflowApiError as error: last_error = error is_retryable_404 = ( error.http_status == 404 and call_context.qf_version is not None ) if not is_retryable_404: raise assert last_error is not None raise last_error return self._run(profile, runner, tool_name='创建应用') def app_delete(self, *, profile: str, app_key: str) -> JSONObject: """执行应用相关逻辑。""" self._require_app_key(app_key) def runner(session_profile, context): result = self.backend.request("DELETE", context, f"/app/{app_key}") return self._attach_human_review_notice( {"profile": profile, "ws_id": session_profile.selected_ws_id, "app_key": app_key, "result": result}, operation="delete", target="app configuration", ) return self._run(profile, runner, tool_name='删除应用') def app_publish(self, *, profile: str, app_key: str, payload: JSONObject) -> JSONObject: """执行应用相关逻辑。""" self._require_app_key(app_key) body = self._require_dict(payload) def runner(session_profile, context): result = self.backend.request("POST", context, f"/app/{app_key}/publish", json_body=body) return {"profile": profile, "ws_id": session_profile.selected_ws_id, "app_key": app_key, "result": result} return self._run(profile, runner, tool_name='发布应用') def _require_app_key(self, app_key: str) -> None: """执行内部辅助逻辑。""" if not app_key: raise_tool_error(QingflowApiError.config_error("app_key is required")) def _probe_create_access(self, context: BackendRequestContext, app_key: str) -> bool: """执行内部辅助逻辑。""" try: self.backend.request( "GET", context, f"/app/{app_key}/form", params={"type": 2, "beingApply": True}, ) return True except QingflowApiError as exc: if is_auth_like_error(exc): raise if backend_code_int(exc) in {40002, 40027, 404} or exc.http_status == 404: return False raise def _probe_list_type_access(self, context: BackendRequestContext, app_key: str, list_type: int) -> bool: """执行内部辅助逻辑。""" try: self.backend.request( "POST", context, f"/app/{app_key}/apply/filter", json_body={"type": list_type, "pageNum": 1, "pageSize": 1}, ) return True except QingflowApiError as exc: if is_auth_like_error(exc): raise if backend_code_int(exc) in {40002, 40027, 404} or exc.http_status == 404: return False raise def _resolve_accessible_system_views(self, context: BackendRequestContext, app_key: str) -> tuple[list[JSONObject], list[JSONObject]]: """执行内部辅助逻辑。""" items: list[JSONObject] = [] warnings: list[JSONObject] = [] for view_id, list_type, name in SYSTEM_VIEW_DEFINITIONS: try: can_access = self._probe_list_type_access(context, app_key, list_type) except QingflowApiError as exc: raise if not can_access: continue items.append({"view_id": view_id, "name": name, "kind": "system", "analysis_supported": True}) return items, warnings def _resolve_accessible_custom_views(self, context: BackendRequestContext, app_key: str) -> tuple[list[JSONObject], list[JSONObject]]: """执行内部辅助逻辑。""" warnings: list[JSONObject] = [] try: payload = self.backend.request("GET", context, f"/app/{app_key}/view/viewList") except QingflowApiError as exc: if _is_optional_app_metadata_error(exc): warnings.append( _with_api_error_fields( { "code": "CUSTOM_VIEW_LIST_UNAVAILABLE", "message": ( f"app_get could not load custom views for {app_key}; " "system views and other app metadata may still be usable." ), }, exc, ) ) return [], warnings raise items: list[JSONObject] = [] for item in _normalize_view_list(payload): view_key = str(item.get("viewKey") or item.get("viewgraphKey") or "").strip() if not view_key: continue normalized: JSONObject = { "view_id": f"custom:{view_key}", "name": str(item.get("viewName") or item.get("viewgraphName") or view_key).strip() or view_key, "kind": "custom", } view_type = str(item.get("viewType") or item.get("viewgraphType") or "").strip() if view_type: normalized["view_type"] = view_type normalized["analysis_supported"] = _analysis_supported_for_view_type(view_type or None) items.append(normalized) return items, warnings def _compact_base_info(self, result: dict[str, Any]) -> JSONObject: """执行内部辅助逻辑。""" publish_status = result.get("appPublishStatus") auth = result.get("auth") if isinstance(result.get("auth"), dict) else {} contact_auth = auth.get("contactAuth") if isinstance(auth, dict) and isinstance(auth.get("contactAuth"), dict) else {} external_auth = auth.get("externalMemberAuth") if isinstance(auth, dict) and isinstance(auth.get("externalMemberAuth"), dict) else {} tag_items = result.get("tagItems") if isinstance(result.get("tagItems"), list) else [] tag_ids = result.get("tagIds") if isinstance(result.get("tagIds"), list) else [] return { "appKey": result.get("appKey"), "formId": result.get("formId"), "formTitle": result.get("formTitle"), "appIcon": result.get("appIcon"), "appPublishStatus": publish_status, "appPublishStatusLabel": get_app_publish_status_label(publish_status if isinstance(publish_status, int) else None), "appOpenStatus": result.get("appOpenStatus"), "applyOpenStatus": result.get("applyOpenStatus"), "dataManageStatus": result.get("dataManageStatus"), "editItemStatus": result.get("editItemStatus"), "deleteItemStatus": result.get("deleteItemStatus"), "copyAppStatus": result.get("copyAppStatus"), "flowStatus": result.get("flowStatus"), "createTime": result.get("createTime"), "creator": self._compact_user(result.get("creator")), "tagIds": tag_ids, "tagCount": len(tag_ids), "tagItemCount": len(tag_items), "tagItemsPreview": [self._compact_tag_item(item) for item in tag_items[:10] if isinstance(item, dict)], "authSummary": { "type": auth.get("type") if isinstance(auth, dict) else None, "contactAuthType": contact_auth.get("type") if isinstance(contact_auth, dict) else None, "contactDepartments": self._count_auth_members(contact_auth, "depart"), "contactMembers": self._count_auth_members(contact_auth, "member"), "contactRoles": self._count_auth_members(contact_auth, "role"), "contactIncludeSubDeparts": self._extract_include_sub_departs(contact_auth), "externalAuthType": external_auth.get("type") if isinstance(external_auth, dict) else None, "externalDepartments": self._count_auth_members(external_auth, "depart"), "externalMembers": self._count_auth_members(external_auth, "member"), "externalRoles": self._count_auth_members(external_auth, "role"), }, } def _base_info_from_visible_app(self, app_key: str, visible_app: JSONObject) -> JSONObject: """Build a base-info-shaped fallback from a current-user visible app item.""" tag_ids: list[int] = [] tag_id = _coerce_positive_int(visible_app.get("tag_id")) if tag_id is not None: tag_ids.append(tag_id) for value in visible_app.get("tag_ids") if isinstance(visible_app.get("tag_ids"), list) else []: tag_id = _coerce_positive_int(value) if tag_id is not None and tag_id not in tag_ids: tag_ids.append(tag_id) fallback: JSONObject = { "appKey": app_key, "formId": visible_app.get("form_id"), "formTitle": visible_app.get("app_name") or visible_app.get("title") or app_key, "appIcon": visible_app.get("app_icon"), "tagIds": tag_ids, "visibleApp": visible_app, } package_name = visible_app.get("package_name") or visible_app.get("tag_name") or visible_app.get("group_name") if package_name: fallback["tagItems"] = [{"tagId": tag_id, "tagName": package_name} for tag_id in tag_ids] return fallback def _compact_form_schema(self, result: dict[str, Any]) -> JSONObject: """执行内部辅助逻辑。""" base_questions_raw = result.get("baseQues") if isinstance(result.get("baseQues"), list) else [] form_questions_raw = result.get("formQues") if isinstance(result.get("formQues"), list) else [] base_questions = [self._compact_question(question) for question in base_questions_raw if isinstance(question, dict)] form_questions = [self._compact_question(question) for question in form_questions_raw if isinstance(question, dict)] all_questions = base_questions + form_questions type_counts = Counter( question["queType"] for question in all_questions if isinstance(question.get("queType"), int) ) required_count = sum(1 for question in all_questions if question.get("required") is True) return { "appKey": result.get("appKey"), "formId": result.get("formId"), "formTitle": result.get("formTitle"), "editVersionNo": result.get("editVersionNo"), "serialNumType": result.get("serialNumType"), "hideCopyright": result.get("hideCopyright"), "questionCounts": { "baseQuestions": len(base_questions), "formQuestions": len(form_questions), "totalQuestions": len(all_questions), "requiredQuestions": required_count, "questionRelations": len(result.get("questionRelations") or []), "selectScopeRelations": len(result.get("selectScopeRelations") or []), "serialNumConfigs": len(result.get("serialNumConfig") or []), }, "questionTypeCounts": [ {"queType": que_type, "count": count} for que_type, count in sorted(type_counts.items()) ], "baseQuestions": base_questions, "formQuestions": form_questions, } def _compact_user(self, user: Any) -> JSONObject | None: """执行内部辅助逻辑。""" if not isinstance(user, dict): return None return { "uid": user.get("uid"), "nickName": user.get("nickName"), "email": user.get("email"), "remark": user.get("remark"), } def _compact_tag_item(self, item: dict[str, Any]) -> JSONObject: """执行内部辅助逻辑。""" return { "itemType": item.get("itemType"), "title": item.get("title"), "appKey": item.get("appKey"), "formId": item.get("formId"), "dashKey": item.get("dashKey"), "pageKey": item.get("pageKey"), } def _compact_question(self, question: dict[str, Any]) -> JSONObject: """执行内部辅助逻辑。""" options = question.get("options") inner_questions = question.get("innerQuestions") sub_questions = question.get("subQuestions") compact = { "queId": question.get("queId"), "queTitle": question.get("queTitle"), "queType": question.get("queType"), "required": question.get("required") is True, "ordinal": question.get("ordinal"), "inlineOrdinal": question.get("inlineOrdinal"), "queWidth": question.get("queWidth"), "sectionId": question.get("sectionId"), "supId": question.get("supId"), "relatedQueId": question.get("relatedQueId"), "pluginStatus": question.get("pluginStatus"), "optionCount": len(options) if isinstance(options, list) else 0, "innerQuestionCount": len(inner_questions) if isinstance(inner_questions, list) else 0, "subQuestionCount": len(sub_questions) if isinstance(sub_questions, list) else 0, } return {key: value for key, value in compact.items() if value is not None} def _extract_visible_apps(self, result: Any) -> tuple[list[JSONObject], str]: """执行内部辅助逻辑。""" apps: list[JSONObject] = [] seen: set[str] = set() def walk( node: Any, *, package_tag_id: int | None = None, package_name: str | None = None, group_id: int | None = None, group_name: str | None = None, ) -> None: if isinstance(node, list): for item in node: walk( item, package_tag_id=package_tag_id, package_name=package_name, group_id=group_id, group_name=group_name, ) return if not isinstance(node, dict): return next_package_tag_id = _coerce_positive_int(node.get("tagId")) or package_tag_id next_package_name = str(node.get("tagName") or "").strip() or package_name next_group_id = _coerce_positive_int(node.get("groupId")) or group_id next_group_name = str(node.get("groupName") or node.get("groupTitle") or "").strip() or group_name normalized = self._normalize_visible_app( node, package_tag_id=next_package_tag_id, package_name=next_package_name, group_id=next_group_id, group_name=next_group_name, ) if normalized is not None: app_key = str(normalized.get("app_key") or "").strip() if app_key and app_key not in seen: seen.add(app_key) apps.append(normalized) for value in node.values(): if isinstance(value, (list, dict)): walk( value, package_tag_id=next_package_tag_id, package_name=next_package_name, group_id=next_group_id, group_name=next_group_name, ) walk(result) return apps, type(result).__name__ def _normalize_visible_app( self, item: dict[str, Any], *, package_tag_id: int | None, package_name: str | None, group_id: int | None, group_name: str | None, ) -> JSONObject | None: """执行内部辅助逻辑。""" app_key = str(item.get("appKey") or item.get("app_key") or "").strip() if not app_key: return None title = str(item.get("title") or item.get("formTitle") or item.get("appName") or item.get("name") or app_key).strip() or app_key tag_ids = item.get("tagIds") if isinstance(item.get("tagIds"), list) else [] compact = { "app_key": app_key, "app_name": title, "title": title, "app_icon": str(item.get("appIcon") or "").strip() or None, "form_id": item.get("formId"), "tag_id": package_tag_id, "package_name": package_name, "group_id": group_id, "group_name": group_name, "tag_ids": [value for value in (_coerce_positive_int(tag_id) for tag_id in tag_ids) if value is not None], } return {key: value for key, value in compact.items() if value not in (None, [], "", {})} def _count_auth_members(self, auth_payload: Any, member_key: str) -> int: """执行内部辅助逻辑。""" if not isinstance(auth_payload, dict): return 0 auth_members = auth_payload.get("authMembers") if not isinstance(auth_members, dict): return 0 members = auth_members.get(member_key) return len(members) if isinstance(members, list) else 0 def _extract_include_sub_departs(self, auth_payload: Any) -> bool | None: """执行内部辅助逻辑。""" if not isinstance(auth_payload, dict): return None auth_members = auth_payload.get("authMembers") if isinstance(auth_members, dict): include_sub_departs = auth_members.get("includeSubDeparts") if isinstance(include_sub_departs, bool): return include_sub_departs include_sub_departs = auth_payload.get("includeSubDeparts") if isinstance(include_sub_departs, bool): return include_sub_departs return None FORM_TYPE_ALIASES = { "default": 1, "form": 1, "schema": 1, "new": 1, "draft": 1, "edit": 1, } def _normalize_form_type(value: int | str) -> int: if isinstance(value, bool): raise_tool_error(QingflowApiError.config_error("form_type must be a positive integer or supported alias")) if isinstance(value, int): return value text = str(value or "").strip().lower() if not text: raise_tool_error(QingflowApiError.config_error("form_type is required")) if text.isdigit(): return int(text) if text in FORM_TYPE_ALIASES: return FORM_TYPE_ALIASES[text] raise_tool_error(QingflowApiError.config_error("form_type must be a positive integer or one of: default, form, schema, new, draft, edit")) def _normalize_view_list(payload: Any) -> list[JSONObject]: if isinstance(payload, dict): raw_list = payload.get("list") if isinstance(raw_list, list): payload = raw_list else: payload = [payload] if not isinstance(payload, list): return [] flattened: list[JSONObject] = [] for group in payload: if not isinstance(group, dict): continue if group.get("viewKey") or group.get("viewgraphKey"): flattened.append(group) continue view_list = group.get("viewList") if not isinstance(view_list, list): continue for item in view_list: if isinstance(item, dict) and (item.get("viewKey") or item.get("viewgraphKey")): flattened.append(item) return flattened def _analysis_supported_for_view_type(view_type: str | None) -> bool: normalized = str(view_type or "").strip().lower() if not normalized: return True return normalized not in {"boardview", "ganttview"} def _is_optional_app_metadata_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 def _with_api_error_fields(payload: JSONObject, error: QingflowApiError) -> JSONObject: if error.backend_code is not None: payload["backend_code"] = error.backend_code if error.http_status is not None: payload["http_status"] = error.http_status if error.request_id: payload["request_id"] = error.request_id return payload def _derive_import_capability(base_info: Any) -> tuple[JSONObject, list[JSONObject]]: warnings: list[JSONObject] = [] if not isinstance(base_info, dict): warnings.append( { "code": "IMPORT_CAPABILITY_UNAVAILABLE", "message": "app_get could not determine import capability because baseInfo was unavailable.", } ) return _unknown_import_capability(), warnings has_data_import_status = "dataImportStatus" in base_info has_data_manage_status = "dataManageStatus" in base_info applicant_import_enabled = _coerce_optional_bool(base_info.get("dataImportStatus")) if has_data_import_status else None data_manage_status = _coerce_optional_bool(base_info.get("dataManageStatus")) if has_data_manage_status else None if applicant_import_enabled is True: return { "can_import": True, "auth_source": "apply_auth", "applicant_import_enabled": True, "data_manage_status": data_manage_status, "runtime_checks_required": ["user_disabled", "function_demoted"], "confidence": "preflight", }, warnings if data_manage_status is True: return { "can_import": True, "auth_source": "data_manage_auth", "applicant_import_enabled": applicant_import_enabled, "data_manage_status": True, "runtime_checks_required": ["user_disabled", "function_demoted"], "confidence": "preflight", }, warnings if applicant_import_enabled is False and data_manage_status is False: return { "can_import": False, "auth_source": "none", "applicant_import_enabled": False, "data_manage_status": False, "runtime_checks_required": [], "confidence": "preflight", }, warnings warnings.append( { "code": "IMPORT_CAPABILITY_UNAVAILABLE", "message": "app_get could not fully determine import capability because baseInfo did not include a complete import permission summary.", } ) return _unknown_import_capability( applicant_import_enabled=applicant_import_enabled, data_manage_status=data_manage_status, ), warnings def _unknown_import_capability( *, applicant_import_enabled: bool | None = None, data_manage_status: bool | None = None, ) -> JSONObject: return { "can_import": None, "auth_source": "unknown", "applicant_import_enabled": applicant_import_enabled, "data_manage_status": data_manage_status, "runtime_checks_required": [], "confidence": "unknown", } def _derive_editability(base_info: Any) -> tuple[JSONObject, list[JSONObject]]: warnings: list[JSONObject] = [] if not isinstance(base_info, dict): warnings.append( { "code": "APP_EDITABILITY_UNAVAILABLE", "message": "app_get could not determine editability because baseInfo was unavailable.", } ) return { "can_edit_form": None, "can_edit_flow": None, "can_edit_views": None, "can_edit_charts": None, }, warnings edit_item_status = _coerce_optional_bool(base_info.get("editItemStatus")) data_manage_status = _coerce_optional_bool(base_info.get("dataManageStatus")) return { "can_edit_form": edit_item_status, "can_edit_flow": edit_item_status, "can_edit_views": data_manage_status, "can_edit_charts": data_manage_status, }, warnings def _coerce_positive_int(value: Any) -> int | None: try: number = int(value) except (TypeError, ValueError): return None return number if number > 0 else None def _coerce_optional_bool(value: Any) -> bool | None: if isinstance(value, bool): return value return None def _normalize_app_list_query(*, query: str = "", keyword: str = "") -> tuple[str, list[JSONObject]]: normalized_query = str(query or "").strip() normalized_keyword = str(keyword or "").strip() warnings: list[JSONObject] = [] if normalized_query and normalized_keyword: warnings.append( { "code": "APP_LIST_QUERY_TAKES_PRECEDENCE", "message": "Both query and keyword were provided; query was used and keyword was ignored.", "ignored_keyword": normalized_keyword, } ) if normalized_query: return normalized_query, warnings return normalized_keyword, warnings