from __future__ import annotations import json import textwrap from typing import Any, TextIO def emit_text_result(result: dict[str, Any], *, hint: str, stream: TextIO) -> None: text = _format_cancelled_result(result) if text is None: if _is_failed_result(result) and hint != "task_action_execute": text = _format_failed_result(result) else: formatter = _FORMATTERS.get(hint, _format_generic) text = formatter(result) stream.write(text) if not text.endswith("\n"): stream.write("\n") def _format_cancelled_result(result: dict[str, Any]) -> str | None: if str(result.get("status") or "").lower() != "cancelled": return None return str(result.get("message") or "已取消") + "\n" def _is_failed_result(result: dict[str, Any]) -> bool: if _is_executed_nonfatal_result(result): return False if result.get("ok") is False: return True return str(result.get("status") or "").lower() in {"failed", "blocked"} def _is_executed_nonfatal_result(result: dict[str, Any]) -> bool: status = str(result.get("status") or "").lower() executed = bool( result.get("write_executed") or result.get("write_may_have_succeeded") or result.get("delete_executed") or result.get("action_executed") or result.get("export_executed") ) return executed and status in {"partial_success", "verification_failed", "running", "queued", "unknown"} def _format_failed_result(result: dict[str, Any]) -> str: lines = [f"Status: {result.get('status') or 'failed'}"] for key, label in ( ("error_code", "Error Code"), ("message", "Message"), ("backend_code", "Backend Code"), ("request_id", "Request ID"), ("http_status", "HTTP Status"), ("category", "Category"), ): value = result.get(key) if value not in (None, ""): lines.append(f"{label}: {value}") data = result.get("data") if isinstance(result.get("data"), dict) else {} selection = data.get("selection") if isinstance(data.get("selection"), dict) else {} if selection: lines.append("Selection:") lines.extend(f"- {line}" for line in _dict_scalar_lines(selection)) _append_warnings(lines, result.get("warnings")) _append_verification(lines, result.get("verification")) return "\n".join(lines) + "\n" def _format_generic(result: dict[str, Any]) -> str: lines: list[str] = [] title = _first_present(result, "status", "message") if title: lines.append(str(title)) data = result.get("data") if isinstance(data, dict): scalar_lines = _dict_scalar_lines(data) if scalar_lines: lines.extend(scalar_lines) elif result: scalar_lines = _dict_scalar_lines(result) if scalar_lines: lines.extend(scalar_lines) if not lines: lines.append(json.dumps(result, ensure_ascii=False, indent=2)) _append_warnings(lines, result.get("warnings")) _append_verification(lines, result.get("verification")) return "\n".join(lines) + "\n" def _format_whoami(result: dict[str, Any]) -> str: lines = [ f"User: {result.get('nick_name') or '-'} ({result.get('email') or '-'})", f"UID: {result.get('uid')}", f"Workspace: {result.get('selected_ws_name') or '-'} ({result.get('selected_ws_id') or '-'})", f"Workspace QF Version: {result.get('qf_version') or '-'}", ] cli_auth = result.get("cli_auth") if isinstance(result.get("cli_auth"), dict) else {} if cli_auth: lines.append(f"Login Flow: {cli_auth.get('flow') or '-'}") if cli_auth.get("verification_uri"): lines.append(f"Verification URL: {cli_auth.get('verification_uri')}") if cli_auth.get("user_code"): lines.append(f"User Code: {cli_auth.get('user_code')}") request_route = result.get("request_route") if isinstance(result.get("request_route"), dict) else {} route_qf_version = request_route.get("qf_version") if route_qf_version and route_qf_version != result.get("qf_version"): lines.append(f"Request Route QF Version: {route_qf_version}") lines.append(f"Permission Level: {result.get('permission_level') or '-'}") departments = result.get("departments") if isinstance(result.get("departments"), list) else [] roles = result.get("roles") if isinstance(result.get("roles"), list) else [] if departments: lines.append( "Departments: " + ", ".join( str(item.get("dept_name") or item.get("dept_id")) for item in departments if isinstance(item, dict) ) ) if roles: lines.append( "Roles: " + ", ".join( str(item.get("role_name") or item.get("role_id")) for item in roles if isinstance(item, dict) ) ) _append_warnings(lines, result.get("warnings")) return "\n".join(lines) + "\n" def _format_workspace_list(result: dict[str, Any]) -> str: page = result.get("page") if isinstance(result.get("page"), dict) else {} items = page.get("list") if isinstance(page.get("list"), list) else [] rows = [] for item in items: if not isinstance(item, dict): continue rows.append( [ str(item.get("wsId") or ""), str(item.get("workspaceName") or item.get("wsName") or ""), str(item.get("remark") or ""), ] ) return _render_titled_table("Workspaces", ["ws_id", "name", "remark"], rows) def _format_workspace_get(result: dict[str, Any]) -> str: workspace = result.get("workspace") if isinstance(result.get("workspace"), dict) else {} lines = [ f"Workspace: {workspace.get('workspaceName') or workspace.get('wsName') or '-'} ({workspace.get('wsId') or result.get('ws_id') or '-'})", f"QF Version: {result.get('qf_version') or workspace.get('systemVersion') or '-'}", f"Identity: {workspace.get('identity') or '-'}", f"Auth: {workspace.get('auth') if workspace.get('auth') is not None else '-'}", f"State: {workspace.get('state') if workspace.get('state') is not None else '-'}", ] _append_warnings(lines, result.get("warnings")) return "\n".join(lines) + "\n" def _format_workspace_select(result: dict[str, Any]) -> str: status = str(result.get("status") or "").lower() if status == "cancelled": return str(result.get("message") or "已取消") + "\n" workspace = result.get("workspace") if isinstance(result.get("workspace"), dict) else {} workspace_name = workspace.get("workspaceName") or workspace.get("wsName") or result.get("selected", {}).get("workspace_name") if isinstance(result.get("selected"), dict) else None lines = [ f"已切换到: {workspace_name or '-'} ({workspace.get('wsId') or result.get('ws_id') or '-'})", f"QF Version: {result.get('qf_version') or workspace.get('systemVersion') or '-'}", ] _append_warnings(lines, result.get("warnings")) return "\n".join(lines) + "\n" def _format_app_items(result: dict[str, Any]) -> str: items = result.get("items") if not isinstance(items, list): items = result.get("apps") rows = [] for item in items or []: if not isinstance(item, dict): continue rows.append( [ str(item.get("app_key") or item.get("appKey") or ""), str(item.get("app_name") or item.get("name") or item.get("title") or ""), str(item.get("package_name") or item.get("tag_name") or ""), ] ) return _render_titled_table("Apps", ["app_key", "app_name", "package"], rows) def _format_app_get(result: dict[str, Any]) -> str: data = result.get("data") if isinstance(result.get("data"), dict) else {} lines = [ f"App: {data.get('app_name') or '-'}", f"App Key: {data.get('app_key') or '-'}", f"Can Create: {data.get('can_create')}", ] import_capability = data.get("import_capability") if isinstance(import_capability, dict): lines.append( "Import Capability: " f"{import_capability.get('auth_source') or 'unknown'} / " f"can_import={import_capability.get('can_import')}" ) editability = data.get("editability") if isinstance(data.get("editability"), dict) else {} if editability: lines.append( "Editability: " f"app_base={editability.get('can_edit_app_base')} / " f"form={editability.get('can_edit_form')} / " f"flow={editability.get('can_edit_flow')} / " f"views={editability.get('can_edit_views')} / " f"charts={editability.get('can_edit_charts')}" ) views = data.get("accessible_views") if isinstance(data.get("accessible_views"), list) else [] lines.append(f"Accessible Views: {len(views)}") for item in views[:10]: if isinstance(item, dict): lines.append(f"- {item.get('view_id')}: {item.get('name')}") _append_warnings(lines, result.get("warnings")) return "\n".join(lines) + "\n" def _format_portal_list(result: dict[str, Any]) -> str: data = result.get("data") if isinstance(result.get("data"), dict) else {} items = data.get("items") if isinstance(data.get("items"), list) else [] rows: list[list[str]] = [] for item in items: if not isinstance(item, dict): continue rows.append( [ str(item.get("dash_key") or ""), str(item.get("dash_name") or ""), ",".join(str(tag_id) for tag_id in item.get("package_tag_ids") or []), ] ) text = _render_titled_table("Portals", ["dash_key", "name", "package_tag_ids"], rows).rstrip("\n") lines = [text, f"Total: {data.get('total') if data.get('total') is not None else len(items)}"] _append_warnings(lines, result.get("warnings")) _append_verification(lines, result.get("verification")) return "\n".join(lines) + "\n" def _format_portal_get(result: dict[str, Any]) -> str: data = result.get("data") if isinstance(result.get("data"), dict) else {} components = data.get("components") if isinstance(data.get("components"), list) else [] lines = [ f"Portal: {data.get('dash_name') or '-'}", f"Dash Key: {data.get('dash_key') or '-'}", f"Components: {data.get('component_count') if data.get('component_count') is not None else len(components)}", ] package_tag_ids = data.get("package_tag_ids") if isinstance(data.get("package_tag_ids"), list) else [] if package_tag_ids: lines.append("Package Tag IDs: " + ", ".join(str(item) for item in package_tag_ids)) if components: lines.append("Component Refs:") for item in components[:12]: if not isinstance(item, dict): continue prefix = f"- {item.get('source_type') or 'unknown'}" title = item.get("title") chart_ref = item.get("chart_ref") if isinstance(item.get("chart_ref"), dict) else {} view_ref = item.get("view_ref") if isinstance(item.get("view_ref"), dict) else {} if chart_ref: lines.append(f"{prefix}: {title or chart_ref.get('chart_name') or '-'} / chart_id={chart_ref.get('chart_id') or '-'}") elif view_ref: lines.append( f"{prefix}: {title or view_ref.get('view_name') or '-'} / " f"view_id={view_ref.get('view_id') or '-'} / app_key={view_ref.get('app_key') or '-'}" ) else: lines.append(f"{prefix}: {title or '-'}") if len(components) > 12: lines.append(f"... {len(components) - 12} more") _append_warnings(lines, result.get("warnings")) _append_verification(lines, result.get("verification")) return "\n".join(lines) + "\n" def _format_view_get(result: dict[str, Any]) -> str: data = result.get("data") if isinstance(result.get("data"), dict) else {} export_capability = data.get("export_capability") if isinstance(data.get("export_capability"), dict) else {} visible_columns = data.get("visible_columns") if isinstance(data.get("visible_columns"), list) else [] lines = [ f"View: {data.get('view_name') or '-'}", f"View ID: {data.get('view_id') or '-'}", f"View Key: {data.get('view_key') or '-'}", f"App Key: {data.get('app_key') or '-'}", f"View Type: {data.get('view_type') or '-'}", f"Analysis Supported: {data.get('analysis_supported')}", ] if export_capability: lines.append( "Export: " f"supported={export_capability.get('supported')} / " f"tool={export_capability.get('tool') or '-'} / " f"requires_app_key={export_capability.get('requires_app_key')}" ) lines.append(f"Visible Columns: {len(visible_columns)}") for column in visible_columns[:20]: lines.append(f"- {column}") if len(visible_columns) > 20: lines.append(f"... {len(visible_columns) - 20} more") _append_warnings(lines, result.get("warnings")) _append_verification(lines, result.get("verification")) return "\n".join(lines) + "\n" def _format_chart_get(result: dict[str, Any]) -> str: data = result.get("data") if isinstance(result.get("data"), dict) else {} chart_data = data.get("data") if isinstance(data.get("data"), dict) else {} config = data.get("config") if isinstance(data.get("config"), dict) else {} lines = [ f"Chart: {data.get('chart_name') or '-'}", f"Chart ID: {data.get('chart_id') or '-'}", f"Chart Type: {data.get('chart_type') or '-'}", f"Data Source: {data.get('data_source_type') or '-'} / {data.get('data_source_id') or '-'}", f"Data Loaded: {bool(chart_data)}", f"Config Loaded: {bool(config)}", ] summary = chart_data.get("summary") if isinstance(chart_data.get("summary"), dict) else {} rows = chart_data.get("rows") if isinstance(chart_data.get("rows"), list) else [] if summary: lines.append("Summary:") lines.extend(f"- {line}" for line in _dict_scalar_lines(summary)) if rows: lines.append(f"Rows: {len(rows)}") lines.append(json.dumps(rows[0], ensure_ascii=False)) elif config: lines.append("Config Keys: " + ", ".join(str(key) for key in list(config.keys())[:12])) _append_warnings(lines, result.get("warnings")) _append_verification(lines, result.get("verification")) return "\n".join(lines) + "\n" def _format_record_list(result: dict[str, Any]) -> str: data = result.get("data") if isinstance(result.get("data"), dict) else {} items = data.get("items") if isinstance(data.get("items"), list) else [] lines: list[str] = [] lookup = result.get("lookup") if isinstance(result.get("lookup"), dict) else {} if lookup: lines.append("Lookup:") lines.append(f"- query: {lookup.get('query')}") lines.append(f"- confidence: {lookup.get('confidence')}") lines.append(f"- next_action: {lookup.get('next_action')}") lines.append(f"- total_count: {lookup.get('total_count')}") lines.append(f"- returned_count: {lookup.get('returned_count')}") lines.append(f"- truncated: {lookup.get('truncated')}") lines.append(f"Returned Records: {len(items)}") for item in items[:10]: if isinstance(item, dict): lines.append(json.dumps(item, ensure_ascii=False)) if len(items) > 10: lines.append(f"... {len(items) - 10} more") _append_warnings(lines, result.get("warnings")) _append_verification(lines, result.get("verification")) return "\n".join(lines) + "\n" def _format_record_access(result: dict[str, Any]) -> str: status = result.get("status") or "-" lines = [ f"Status: {status}", f"Rows: {result.get('row_count')}", f"Complete: {result.get('complete')}", f"Safe for final conclusion: {result.get('safe_for_final_conclusion')}", ] if result.get("local_dir"): lines.append(f"Local dir: {result.get('local_dir')}") files = result.get("files") if isinstance(result.get("files"), list) else [] if files: lines.append("Files:") for item in files: if isinstance(item, dict): lines.append(f"- part {item.get('part')}: {item.get('local_path')} ({item.get('row_count')} rows)") scope = result.get("scope") if isinstance(result.get("scope"), dict) else {} if status == "needs_scope" and scope: lines.append("Scope required:") lines.append(f"- reported_total: {scope.get('reported_total')}") lines.append(f"- estimated_pages: {scope.get('estimated_pages')}") suggested = scope.get("suggested_time_fields") if isinstance(scope.get("suggested_time_fields"), list) else [] if suggested: names = ", ".join(str(item.get("title") or item.get("field_id")) for item in suggested if isinstance(item, dict)) lines.append(f"- suggested_time_fields: {names}") _append_warnings(lines, result.get("warnings")) _append_verification(lines, result.get("verification")) return "\n".join(lines) + "\n" def _format_record_candidates(result: dict[str, Any]) -> str: data = result.get("data") if isinstance(result.get("data"), dict) else {} selection = data.get("selection") if isinstance(data.get("selection"), dict) else {} pagination = data.get("pagination") if isinstance(data.get("pagination"), dict) else {} items = data.get("items") if isinstance(data.get("items"), list) else [] lines = [ f"Field: {selection.get('field_title') or '-'} ({selection.get('field_id') or '-'})", f"App Key: {selection.get('app_key') or '-'}", f"Scope Source: {data.get('scope_source') or '-'}", f"Keyword: {selection.get('keyword') if selection.get('keyword') not in (None, '') else '-'}", ] if selection.get("record_id") not in (None, "") or selection.get("workflow_node_id") not in (None, "") or selection.get("fields_present"): lines.append( "Runtime Context: " f"record_id={selection.get('record_id') or '-'} / " f"workflow_node_id={selection.get('workflow_node_id') or '-'} / " f"fields_present={selection.get('fields_present')}" ) if pagination: lines.append( "Pagination: " f"page={pagination.get('page')} / " f"page_size={pagination.get('page_size')} / " f"returned={pagination.get('returned_items')} / " f"total={pagination.get('reported_total')} / " f"pages={pagination.get('page_amount')}" ) lines.append(f"Candidates: {len(items)}") for item in items[:20]: if not isinstance(item, dict): lines.append(f"- {item}") continue parts = [str(item.get("value") or item.get("name") or item.get("id") or "-")] if item.get("id") not in (None, ""): parts.append(f"id={item.get('id')}") if item.get("userId") not in (None, ""): parts.append(f"userId={item.get('userId')}") if item.get("email") not in (None, ""): parts.append(f"email={item.get('email')}") if item.get("path") not in (None, ""): parts.append(f"path={item.get('path')}") sources = item.get("sources") if isinstance(item.get("sources"), list) else [] source_labels = [ str(source.get("kind")) for source in sources if isinstance(source, dict) and source.get("kind") not in (None, "") ] if source_labels: parts.append("sources=" + ",".join(source_labels)) lines.append("- " + " / ".join(parts)) if len(items) > 20: lines.append(f"... {len(items) - 20} more") _append_warnings(lines, result.get("warnings")) _append_verification(lines, result.get("verification")) return "\n".join(lines) + "\n" def _format_record_schema(result: dict[str, Any]) -> str: lines = [ f"Status: {result.get('status') or '-'}", f"App Key: {result.get('app_key') or '-'}", f"Schema Scope: {result.get('schema_scope') or '-'}", ] if result.get("record_id") not in (None, ""): lines.append(f"Record ID: {result.get('record_id')}") view_resolution = result.get("view_resolution") if isinstance(result.get("view_resolution"), dict) else {} view_id = view_resolution.get("view_id") or view_resolution.get("id") if view_id: lines.append(f"View ID: {view_id}") if result.get("schema_fingerprint") not in (None, ""): lines.append(f"Schema Fingerprint: {result.get('schema_fingerprint')}") sections: list[tuple[str, list[Any]]] = [] for key, label in ( ("fields", "Fields"), ("required_fields", "Required Fields"), ("optional_fields", "Optional Fields"), ("runtime_linked_required_fields", "Runtime Linked Required Fields"), ("writable_fields", "Writable Fields"), ("columns", "Import Columns"), ("input_fields", "Input Fields"), ("code_block_fields", "Code Block Fields"), ("ambiguous_fields", "Ambiguous Fields"), ): value = result.get(key) if isinstance(value, list): sections.append((label, value)) for label, items in sections: lines.append(f"{label}: {len(items)}") for item in items[:20]: if isinstance(item, dict): lines.append("- " + _schema_item_summary(item)) else: lines.append(f"- {item}") if len(items) > 20: lines.append(f"... {len(items) - 20} more") payload_template = result.get("payload_template") if isinstance(result.get("payload_template"), dict) else {} if payload_template: lines.append("Payload Template Keys: " + ", ".join(str(key) for key in list(payload_template.keys())[:20])) available_routes = result.get("available_update_routes") if isinstance(result.get("available_update_routes"), list) else [] if available_routes: route_labels = [ str(item.get("route_type") or item.get("endpoint_kind") or item.get("view_id") or "-") for item in available_routes[:8] if isinstance(item, dict) ] lines.append("Available Update Routes: " + ", ".join(route_labels)) suggested_next_call = result.get("suggested_next_call") if isinstance(result.get("suggested_next_call"), dict) else {} if suggested_next_call.get("tool_name"): lines.append(f"Suggested Next Call: {suggested_next_call.get('tool_name')}") _append_warnings(lines, result.get("warnings")) _append_verification(lines, result.get("verification")) return "\n".join(lines) + "\n" def _schema_item_summary(item: dict[str, Any]) -> str: field_id = ( item.get("field_id") or item.get("que_id") or item.get("queId") or item.get("id") or item.get("selector") or "-" ) title = item.get("title") or item.get("que_title") or item.get("queTitle") or item.get("name") or item.get("selector") or "-" kind = item.get("kind") or item.get("write_kind") or item.get("que_type") or item.get("queType") or item.get("type") or "-" parts = [f"{title} ({field_id})", f"kind={kind}"] for key in ("required", "writable", "may_become_required", "requires_lookup", "requires_upload"): if key in item: parts.append(f"{key}={item.get(key)}") if item.get("import_value_format") not in (None, ""): parts.append(f"import_value_format={item.get('import_value_format')}") if item.get("target_app_key") not in (None, ""): parts.append(f"target_app_key={item.get('target_app_key')}") options = item.get("options") if isinstance(item.get("options"), list) else [] if options: parts.append(f"options={len(options)}") bound_outputs = item.get("bound_output_fields") if isinstance(item.get("bound_output_fields"), list) else [] if bound_outputs: parts.append("bound_outputs=" + ",".join(str(value) for value in bound_outputs[:5])) aliases = item.get("configured_aliases") if isinstance(item.get("configured_aliases"), list) else [] if aliases: parts.append("aliases=" + ",".join(str(value) for value in aliases[:5])) if item.get("format_hint") not in (None, ""): parts.append(f"hint={item.get('format_hint')}") return " / ".join(parts) def _format_record_get(result: dict[str, Any]) -> str: record = result.get("record") if isinstance(result.get("record"), dict) else {} app = result.get("app") if isinstance(result.get("app"), dict) else {} view = result.get("view") if isinstance(result.get("view"), dict) else {} fields = result.get("fields") if isinstance(result.get("fields"), list) else [] data_logs = result.get("data_logs") if isinstance(result.get("data_logs"), dict) else {} workflow_logs = result.get("workflow_logs") if isinstance(result.get("workflow_logs"), dict) else {} media_assets = result.get("media_assets") if isinstance(result.get("media_assets"), dict) else {} media_items = media_assets.get("items") if isinstance(media_assets.get("items"), list) else [] downloaded_media = [item for item in media_items if isinstance(item, dict) and item.get("access_status") == "downloaded"] failed_media = [item for item in media_items if isinstance(item, dict) and item.get("access_status") != "downloaded"] file_assets = result.get("file_assets") if isinstance(result.get("file_assets"), dict) else {} file_items = file_assets.get("items") if isinstance(file_assets.get("items"), list) else [] downloaded_files = [item for item in file_items if isinstance(item, dict) and item.get("access_status") == "downloaded"] failed_files = [item for item in file_items if isinstance(item, dict) and item.get("access_status") != "downloaded"] extracted_files = [ item for item in downloaded_files if isinstance(item.get("extraction"), dict) and item["extraction"].get("status") == "ok" ] associated_resources = result.get("associated_resources") if isinstance(result.get("associated_resources"), list) else [] unavailable_context = result.get("unavailable_context") if isinstance(result.get("unavailable_context"), list) else [] lines = [ f"Status: {result.get('status') or '-'}", f"App: {app.get('app_name') or app.get('app_key') or '-'} ({app.get('app_key') or '-'})", f"View: {view.get('name') or view.get('view_id') or '-'}", f"Record: {record.get('title') or '-'} ({record.get('record_id') or '-'})", f"Fields: {len(fields)}", f"Data logs: {data_logs.get('status') or '-'} / loaded={data_logs.get('items_loaded')}", f"Workflow logs: {workflow_logs.get('status') or '-'} / loaded={workflow_logs.get('items_loaded')}", f"Media assets: {media_assets.get('status') or '-'} / downloaded={len(downloaded_media)} / failed={len(failed_media)}", f"File assets: {file_assets.get('status') or '-'} / downloaded={len(downloaded_files)} / extracted={len(extracted_files)} / failed={len(failed_files)}", f"Associated resources: {len(associated_resources)}", f"Unavailable contexts: {len(unavailable_context)}", ] if media_assets.get("local_dir"): lines.append(f"Media dir: {media_assets.get('local_dir')}") if file_assets.get("local_dir"): lines.append(f"File dir: {file_assets.get('local_dir')}") if failed_media: failure_counts: dict[str, int] = {} for item in failed_media: key = f"{item.get('access_status') or 'unknown'} via {item.get('download_strategy') or 'unknown'}" failure_counts[key] = failure_counts.get(key, 0) + 1 lines.append("Media failures: " + ", ".join(f"{key}={count}" for key, count in sorted(failure_counts.items()))) if failed_files: failure_counts = {} for item in failed_files: key = f"{item.get('access_status') or 'unknown'} via {item.get('download_strategy') or 'unknown'}" failure_counts[key] = failure_counts.get(key, 0) + 1 lines.append("File failures: " + ", ".join(f"{key}={count}" for key, count in sorted(failure_counts.items()))) summary = result.get("summary") if isinstance(result.get("summary"), dict) else {} if summary.get("text"): lines.append(f"Summary: {summary.get('text')}") _append_warnings(lines, result.get("warnings")) return "\n".join(lines) + "\n" def _format_record_logs(result: dict[str, Any]) -> str: app = result.get("app") if isinstance(result.get("app"), dict) else {} view = result.get("view") if isinstance(result.get("view"), dict) else {} record = result.get("record") if isinstance(result.get("record"), dict) else {} data_logs = result.get("data_logs") if isinstance(result.get("data_logs"), dict) else {} workflow_logs = result.get("workflow_logs") if isinstance(result.get("workflow_logs"), dict) else {} integrity = result.get("context_integrity") if isinstance(result.get("context_integrity"), dict) else {} lines = [ f"Status: {result.get('status') or '-'}", f"App: {app.get('app_name') or app.get('app_key') or '-'} ({app.get('app_key') or '-'})", f"View: {view.get('name') or view.get('view_id') or '-'}", f"Record: {record.get('title') or '-'} ({record.get('record_id') or '-'})", f"Data logs: {data_logs.get('status') or '-'} / count={data_logs.get('items_count')} / pages={data_logs.get('pages_fetched')} / complete={data_logs.get('complete')}", f"Workflow logs: {workflow_logs.get('status') or '-'} / count={workflow_logs.get('items_count')} / pages={workflow_logs.get('pages_fetched')} / complete={workflow_logs.get('complete')}", f"Safe for full log conclusion: {integrity.get('safe_for_full_log_conclusion')}", ] if result.get("local_dir"): lines.append(f"Local dir: {result.get('local_dir')}") if data_logs.get("local_path"): lines.append(f"Data logs file: {data_logs.get('local_path')}") if workflow_logs.get("local_path"): lines.append(f"Workflow logs file: {workflow_logs.get('local_path')}") if result.get("summary_path"): lines.append(f"Summary file: {result.get('summary_path')}") _append_warnings(lines, result.get("warnings")) unavailable = result.get("unavailable_context") if isinstance(result.get("unavailable_context"), list) else [] if unavailable: lines.append(f"Unavailable contexts: {len(unavailable)}") return "\n".join(lines) + "\n" def _format_record_write(result: dict[str, Any]) -> str: status = str(result.get("status") or "").strip().lower() write_executed = bool(result.get("write_executed")) verification_status = str(result.get("verification_status") or "").strip().lower() if write_executed and verification_status == "failed": title = "写入已提交（回读验证未完成）" elif write_executed and status in {"success", "completed"}: title = "写入成功" elif status: title = status else: title = "写入结果" lines = [title] if result.get("record_id") not in (None, ""): lines.append(f"Record ID: {result.get('record_id')}") if result.get("apply_id") not in (None, "") and result.get("apply_id") != result.get("record_id"): lines.append(f"Apply ID: {result.get('apply_id')}") update_route = result.get("update_route") if isinstance(result.get("update_route"), dict) else {} route_name = update_route.get("route") or update_route.get("type") or update_route.get("label") if route_name: lines.append(f"Update Route: {route_name}") if verification_status: lines.append(f"Verification: {verification_status}") if write_executed: lines.append("Safe To Retry: false") if write_executed and verification_status == "failed": verification = result.get("data", {}).get("verification") if isinstance(result.get("data"), dict) else None if isinstance(verification, dict): warning_codes = [ str(item.get("code")) for item in verification.get("warnings", []) if isinstance(item, dict) and item.get("code") ] if warning_codes: lines.append("Verification Note: " + ", ".join(warning_codes[:3])) lines.append("说明：写请求已执行；当前结果只表示后置回读未能确认字段值，不等同于写入被拒绝。") _append_warnings(lines, result.get("warnings")) return "\n".join(lines) + "\n" def _format_record_delete(result: dict[str, Any]) -> str: status = str(result.get("status") or "").strip().lower() deleted_ids = [str(item) for item in result.get("deleted_ids", []) if item not in (None, "")] failed_ids = [str(item) for item in result.get("failed_ids", []) if item not in (None, "")] write_executed = bool(result.get("write_executed")) if deleted_ids and failed_ids: title = "删除部分完成" elif deleted_ids: title = "删除完成" elif status == "failed" or result.get("ok") is False: title = "删除未执行" else: title = "删除结果" lines = [title] lines.append(f"Deleted: {len(deleted_ids)}") if deleted_ids: lines.append("Deleted IDs: " + ", ".join(deleted_ids[:20])) lines.append(f"Failed: {len(failed_ids)}") if failed_ids: lines.append("Failed IDs: " + ", ".join(failed_ids[:20])) if write_executed: lines.append("Safe To Retry: false") elif result.get("safe_to_retry") is not None: lines.append(f"Safe To Retry: {str(bool(result.get('safe_to_retry'))).lower()}") _append_warnings(lines, result.get("warnings")) return "\n".join(lines) + "\n" def _format_code_block_run(result: dict[str, Any]) -> str: status = str(result.get("status") or "").strip().lower() execution = result.get("execution") if isinstance(result.get("execution"), dict) else {} writeback = result.get("writeback") if isinstance(result.get("writeback"), dict) else {} executed = bool(execution.get("executed")) writeback_applied = bool(writeback.get("applied")) write_verified = writeback.get("write_verified") if executed and writeback_applied and status == "verification_failed": title = "代码块已执行，回写已提交（验证未完成）" elif executed and writeback_applied: title = "代码块已执行，回写成功" elif executed: title = "代码块已执行" elif status: title = status else: title = "代码块结果" lines = [title] if result.get("record_id") not in (None, ""): lines.append(f"Record ID: {result.get('record_id')}") code_block_field = result.get("code_block_field") if isinstance(result.get("code_block_field"), dict) else {} if code_block_field.get("title"): lines.append(f"Code Block Field: {code_block_field.get('title')}") if execution: lines.append(f"Result Count: {execution.get('result_count', 0)}") if writeback: lines.append(f"Writeback: attempted={writeback.get('attempted')} applied={writeback_applied} verified={write_verified}") if executed and writeback_applied and status == "verification_failed": lines.append("说明：代码块执行和回写请求已完成；当前结果只表示后置回读未能确认字段值，不等同于回写被拒绝。") _append_warnings(lines, result.get("warnings")) return "\n".join(lines) + "\n" def _format_task_list(result: dict[str, Any]) -> str: data = result.get("data") if isinstance(result.get("data"), dict) else {} items = data.get("items") if isinstance(data.get("items"), list) else [] lines = ["Tasks"] for item in items: if not isinstance(item, dict): continue header_parts = [ str(item.get("task_id") or "-"), str(item.get("app_name") or item.get("app_key") or "-"), str(item.get("workflow_node_name") or "-"), ] apply_time = item.get("apply_time") if apply_time not in (None, ""): header_parts.append(str(apply_time)) lines.append("- " + " / ".join(header_parts)) summary_fields = item.get("summary_fields") if isinstance(item.get("summary_fields"), list) else [] for summary in summary_fields: if not isinstance(summary, dict): continue lines.append(f" {summary.get('title') or '-'}: {summary.get('answer') or '-'}") _append_warnings(lines, result.get("warnings")) return "\n".join(lines) + "\n" def _format_task_get(result: dict[str, Any]) -> str: data = result.get("data") if isinstance(result.get("data"), dict) else {} task = data.get("task") if isinstance(data.get("task"), dict) else {} record_summary = data.get("record_summary") if isinstance(data.get("record_summary"), dict) else {} editable_fields = data.get("editable_fields") if isinstance(data.get("editable_fields"), list) else [] available_actions = data.get("available_actions") if isinstance(data.get("available_actions"), list) else [] extras = data.get("extras") if isinstance(data.get("extras"), dict) else {} initiator = task.get("initiator") if isinstance(task.get("initiator"), dict) else {} initiator_label = initiator.get("displayName") or initiator.get("email") or "-" lines = [] if task.get("task_id") not in (None, ""): lines.append(f"Task ID: {task.get('task_id')}") lines.extend( [ f"Locator: {task.get('app_key') or '-'} / {task.get('record_id') or '-'} / {task.get('workflow_node_id') or '-'}", f"Node: {task.get('workflow_node_name') or '-'}", f"App: {task.get('app_name') or '-'}", f"Initiator: {initiator_label}", f"Apply Status: {record_summary.get('apply_status')}", f"Apply Number: {record_summary.get('custom_apply_num') or record_summary.get('apply_num') or '-'}", f"Apply Time: {record_summary.get('apply_time') or '-'}", f"Available Actions: {', '.join(str(item) for item in available_actions) or '-'}", f"Editable Fields: {len(editable_fields)}", ] ) all_fields = record_summary.get("all_fields") if isinstance(record_summary.get("all_fields"), dict) else {} core_fields = record_summary.get("core_fields") if isinstance(record_summary.get("core_fields"), dict) else {} if all_fields: lines.append("Fields:") for key, value in all_fields.items(): lines.extend(_format_field_line(key, value)) elif core_fields: lines.append("Core Fields:") for key, value in list(core_fields.items())[:12]: lines.extend(_format_field_line(key, value)) if editable_fields: lines.append("Editable Fields:") for item in editable_fields[:10]: if isinstance(item, dict): lines.append(f"- {item.get('title') or '-'} ({item.get('kind') or 'field'})") associated_reports = extras.get("associated_reports") if isinstance(extras.get("associated_reports"), dict) else {} rollback_candidates = extras.get("rollback_candidates") if isinstance(extras.get("rollback_candidates"), dict) else {} transfer_candidates = extras.get("transfer_candidates") if isinstance(extras.get("transfer_candidates"), dict) else {} lines.append( "Extras: " f"reports={associated_reports.get('count', 0)}, " f"rollback={rollback_candidates.get('count', 0)}, " f"transfer={transfer_candidates.get('count', 0)}" ) transfer_items = transfer_candidates.get("items") if isinstance(transfer_candidates.get("items"), list) else [] if transfer_items: lines.append("Transfer Candidates:") for item in transfer_items: if isinstance(item, dict): display = item.get("name") or item.get("uid") or item suffix = f" <{item.get('email')}>" if item.get("email") else "" lines.append(f"- {display}{suffix} (uid={item.get('uid') or '-'})") _append_warnings(lines, result.get("warnings")) return "\n".join(lines) + "\n" def _format_task_workflow_log(result: dict[str, Any]) -> str: data = result.get("data") if isinstance(result.get("data"), dict) else {} selection = data.get("selection") if isinstance(data.get("selection"), dict) else {} visibility = data.get("visibility") if isinstance(data.get("visibility"), dict) else {} items = data.get("items") if isinstance(data.get("items"), list) else [] lines: list[str] = [] if selection.get("task_id") not in (None, ""): lines.append(f"Task ID: {selection.get('task_id')}") locator = " / ".join( str(value or "-") for value in ( selection.get("app_key"), selection.get("record_id"), selection.get("workflow_node_id"), ) ) if locator != "- / - / -": lines.append(f"Locator: {locator}") if visibility: lines.append( "Visibility: " f"audit_record={visibility.get('audit_record_visible')} / " f"qrobot_record={visibility.get('qrobot_record_visible')}" ) lines.append(f"Workflow Logs: {len(items)}") for item in items[:20]: if not isinstance(item, dict): continue operator = item.get("operator") if isinstance(item.get("operator"), dict) else {} operator_name = operator.get("name") or operator.get("email") or operator.get("uid") or "-" parts = [ str(item.get("operation_time") or "-"), str(item.get("node_name") or item.get("node_id") or "-"), str(operator_name), str(item.get("remark") or item.get("operation") or "-"), ] lines.append("- " + " / ".join(parts)) if len(items) > 20: lines.append(f"... {len(items) - 20} more") _append_warnings(lines, result.get("warnings")) _append_verification(lines, result.get("verification")) return "\n".join(lines) + "\n" def _format_task_action(result: dict[str, Any]) -> str: data = result.get("data") if isinstance(result.get("data"), dict) else {} action = str(data.get("action") or "").strip().lower() status = str(result.get("status") or "").strip().lower() if status == "failed" or result.get("ok") is False: lines = [_task_action_failure_label(action)] reason = _task_action_failure_reason(result) if reason: lines.append(f"原因：{reason}") debug_lines = _task_action_debug_lines(result) if debug_lines: lines.append("调试信息：") lines.extend(f"- {line}" for line in debug_lines) return "\n".join(lines) + "\n" if status == "partial_success": lines = [_task_action_success_label(action)] lines.append(f"说明：{_task_action_partial_success_message(result)}") return "\n".join(lines) + "\n" return _task_action_success_label(action) + "\n" def _format_task_workbench(result: dict[str, Any]) -> str: message = str(result.get("message") or "").strip() if message: return message + "\n" return "" def _format_field_line(key: Any, value: Any) -> list[str]: if isinstance(value, list): text = " / ".join(str(item) for item in value if item not in (None, "")) else: text = str(value if value not in (None, "") else "-") wrapped = textwrap.wrap( text, width=120, initial_indent=f"- {key}: ", subsequent_indent=" ", replace_whitespace=False, drop_whitespace=False, break_long_words=True, ) return wrapped or [f"- {key}: -"] def _format_task_associated_report_detail(result: dict[str, Any]) -> str: data = result.get("data") if isinstance(result.get("data"), dict) else {} selection = data.get("selection") if isinstance(data.get("selection"), dict) else {} result_type = str(data.get("result_type") or "-") context = data.get("context") if isinstance(data.get("context"), dict) else {} lines = [] if selection.get("task_id") not in (None, ""): lines.append(f"Task ID: {selection.get('task_id')}") lines.extend( [ f"Report: {selection.get('chart_name') or '-'} ({selection.get('report_id') or '-'})", f"Type: {result_type}", ] ) if result_type == "view_list": result_payload = data.get("result") if isinstance(data.get("result"), dict) else {} items = result_payload.get("items") if isinstance(result_payload.get("items"), list) else [] lines.append(f"Returned Records: {len(items)}") for item in items[:10]: if isinstance(item, dict): lines.append(json.dumps(item, ensure_ascii=False)) if len(items) > 10: lines.append(f"... {len(items) - 10} more") elif result_type == "chart_data": result_payload = data.get("result") if isinstance(data.get("result"), dict) else {} summary = result_payload.get("summary") if isinstance(result_payload.get("summary"), dict) else {} rows = result_payload.get("rows") if isinstance(result_payload.get("rows"), list) else [] if summary: lines.append(f"Summary: {json.dumps(summary, ensure_ascii=False)}") lines.append(f"Rows: {len(rows)}") for row in rows[:10]: if isinstance(row, dict): lines.append(json.dumps(row, ensure_ascii=False)) if len(rows) > 10: lines.append(f"... {len(rows) - 10} more") resolved_filters = context.get("resolved_filters") if isinstance(context.get("resolved_filters"), list) else [] if resolved_filters: lines.append(f"Resolved Filters: {len(resolved_filters)}") _append_warnings(lines, result.get("warnings")) return "\n".join(lines) + "\n" def _format_import_verify(result: dict[str, Any]) -> str: lines = [ f"App Key: {result.get('app_key') or '-'}", f"File: {result.get('file_name') or result.get('file_path') or '-'}", f"Can Import: {result.get('can_import')}", f"Verification ID: {result.get('verification_id') or '-'}", ] issue_summary = result.get("issue_summary") if isinstance(result.get("issue_summary"), dict) else {} if issue_summary: lines.append( "Issues: " f"total={issue_summary.get('total', 0)}, " f"errors={issue_summary.get('errors', 0)}, " f"warnings={issue_summary.get('warnings', 0)}" ) sample = issue_summary.get("sample") if isinstance(issue_summary.get("sample"), list) else [] if sample: lines.append("Issue Samples:") for item in sample: if isinstance(item, dict): lines.append(f"- {item.get('code') or 'ISSUE'}: {item.get('message') or ''}".rstrip()) _append_warnings(lines, result.get("warnings")) _append_verification(lines, result.get("verification")) return "\n".join(lines) + "\n" def _format_import_template(result: dict[str, Any]) -> str: verification = result.get("verification") if isinstance(result.get("verification"), dict) else {} template_source = verification.get("template_source") if not template_source: template_source = "official" if result.get("template_url") else ("local_generated" if result.get("downloaded_to_path") else "unknown") if result.get("downloaded_to_path"): title = "导入模板已生成" if template_source == "local_generated" else "导入模板已下载" elif result.get("template_url"): title = "导入模板已获取" else: title = "导入模板结果" lines = [ title, f"App Key: {result.get('app_key') or '-'}", f"Template Source: {template_source}", ] if result.get("downloaded_to_path"): lines.append(f"Local Path: {result.get('downloaded_to_path')}") if result.get("template_url"): lines.append(f"Template URL: {result.get('template_url')}") import_capability = result.get("import_capability") if isinstance(result.get("import_capability"), dict) else {} if import_capability: lines.append( "Import Capability: " f"{import_capability.get('auth_source') or 'unknown'} / " f"can_import={import_capability.get('can_import')}" ) expected_columns = result.get("expected_columns") if isinstance(result.get("expected_columns"), list) else [] if expected_columns: lines.append(f"Columns: {len(expected_columns)}") _append_warnings(lines, result.get("warnings")) _append_verification(lines, result.get("verification")) return "\n".join(lines) + "\n" def _format_import_status(result: dict[str, Any]) -> str: lines = [ f"Status: {result.get('status') or '-'}", f"Import ID: {result.get('import_id') or '-'}", f"Process ID: {result.get('process_id_str') or '-'}", f"Total Rows: {result.get('total') or 0}", f"Finished Rows: {result.get('finished') or 0}", f"Succeeded Rows: {result.get('succeeded') or 0}", f"Failed Rows: {result.get('failed') or 0}", f"Progress: {result.get('progress') or '-'}", ] if result.get("process_status") not in (None, ""): lines.append(f"Process Status: {result.get('process_status')}") error_file_urls = result.get("error_file_urls") if isinstance(result.get("error_file_urls"), list) else [] if error_file_urls: lines.append("Error Files:") for url in error_file_urls: lines.append(f"- {url}") _append_warnings(lines, result.get("warnings")) _append_verification(lines, result.get("verification")) return "\n".join(lines) + "\n" def _format_import_repair(result: dict[str, Any]) -> str: lines = [ f"Status: {result.get('status') or '-'}", f"Source File: {result.get('source_file_path') or '-'}", f"Repaired File: {result.get('repaired_file_path') or '-'}", f"Can Import After Repair: {result.get('can_import_after_repair')}", ] verification_id = result.get("verification_id") or result.get("new_verification_id") if verification_id: lines.append(f"Verification ID: {verification_id}") applied_repairs = result.get("applied_repairs") if isinstance(result.get("applied_repairs"), list) else [] skipped_repairs = result.get("skipped_repairs") if isinstance(result.get("skipped_repairs"), list) else [] lines.append(f"Applied Repairs: {', '.join(map(str, applied_repairs)) if applied_repairs else '-'}") if skipped_repairs: lines.append(f"Skipped Repairs: {', '.join(map(str, skipped_repairs))}") issue_summary = result.get("post_repair_issue_summary") if isinstance(result.get("post_repair_issue_summary"), dict) else {} if issue_summary: lines.append( "Post Repair Issues: " f"total={issue_summary.get('total', 0)}, " f"errors={issue_summary.get('errors', 0)}, " f"warnings={issue_summary.get('warnings', 0)}" ) _append_warnings(lines, result.get("warnings")) _append_verification(lines, result.get("verification")) return "\n".join(lines) + "\n" def _format_import_start(result: dict[str, Any]) -> str: lines = [ f"Status: {result.get('status') or '-'}", f"Accepted: {result.get('accepted')}", f"Write Executed: {result.get('write_executed')}", f"Safe To Retry: {result.get('safe_to_retry')}", f"Import ID: {result.get('import_id') or '-'}", f"Process ID: {result.get('process_id_str') or '-'}", ] if result.get("file_url"): lines.append(f"File URL: {result.get('file_url')}") for key, label in ( ("total", "Total Rows"), ("finished", "Finished Rows"), ("succeeded", "Succeeded Rows"), ("failed", "Failed Rows"), ): value = result.get(key) if value not in (None, ""): lines.append(f"{label}: {value}") _append_warnings(lines, result.get("warnings")) _append_verification(lines, result.get("verification")) return "\n".join(lines) + "\n" def _format_export_common(result: dict[str, Any], *, title: str | None = None) -> str: lines: list[str] = [] if title: lines.append(title) lines.extend( [ f"Status: {result.get('status') or '-'}", f"Export Handle: {result.get('export_handle') or '-'}", f"App Key: {result.get('app_key') or '-'}", f"View ID: {result.get('view_id') or '-'}", f"Process Status: {result.get('process_status') or '-'}", f"Rows: {result.get('num') if result.get('num') is not None else '-'}", ] ) row_scope = result.get("row_scope") if row_scope not in (None, ""): lines.append(f"Row Scope: {row_scope}") selected_record_count = result.get("selected_record_count") if selected_record_count not in (None, ""): lines.append(f"Selected Rows: {selected_record_count}") field_scope = result.get("field_scope") if field_scope not in (None, ""): lines.append(f"Field Scope: {field_scope}") selected_field_count = result.get("selected_field_count") if selected_field_count not in (None, ""): lines.append(f"Selected Fields: {selected_field_count}") include_workflow_log = result.get("include_workflow_log") if include_workflow_log not in (None, ""): lines.append(f"Include Workflow Log: {include_workflow_log}") file_names = result.get("file_names") if isinstance(result.get("file_names"), list) else [] file_urls = result.get("file_urls") if isinstance(result.get("file_urls"), list) else [] if file_names or file_urls: lines.append("Remote Files:") max_items = max(len(file_names), len(file_urls)) for index in range(max_items): name = file_names[index] if index < len(file_names) else "-" url = file_urls[index] if index < len(file_urls) else "-" lines.append(f"- {name}: {url}") downloaded_files = result.get("downloaded_files") if isinstance(result.get("downloaded_files"), list) else [] if downloaded_files: lines.append("Downloaded Files:") for item in downloaded_files: if isinstance(item, dict): lines.append(f"- {item.get('file_name') or '-'}: {item.get('path') or '-'}") else: lines.append(f"- {item}") _append_warnings(lines, result.get("warnings")) _append_verification(lines, result.get("verification")) return "\n".join(lines) + "\n" def _format_export_start(result: dict[str, Any]) -> str: return _format_export_common(result, title="Export Accepted") def _format_export_status(result: dict[str, Any]) -> str: return _format_export_common(result, title="Export Status") def _format_export_get(result: dict[str, Any]) -> str: return _format_export_common(result, title="Export Result") def _format_export_direct(result: dict[str, Any]) -> str: return _format_export_common(result, title="Export Direct") def _format_builder_summary(result: dict[str, Any]) -> str: lines = [] emitted: set[str] = set() for key, label in ( ("status", "Status"), ("message", "Message"), ("app_key", "App Key"), ("dash_key", "Dash Key"), ("view_key", "View Key"), ("chart_id", "Chart ID"), ("package_id", "Package ID"), ("verified", "Verified"), ("write_executed", "Write Executed"), ("delete_executed", "Delete Executed"), ("safe_to_retry", "Safe To Retry"), ): if key in result and result.get(key) not in (None, ""): lines.append(f"{label}: {result.get(key)}") emitted.add(key) for key, value in result.items(): if key in emitted or key in {"data", "warnings", "verification", "details", "result", "raw"}: continue if isinstance(value, (str, int, float, bool)) or value is None: lines.append(f"{key}: {value}") data = result.get("data") if isinstance(data, dict): scalar_lines = _dict_scalar_lines(data) if scalar_lines: lines.append("Data:") lines.extend(f"- {line}" for line in scalar_lines[:12]) for key in ( "items", "apps", "fields", "sections", "components", "questions", "views", "charts", "buttons", "associations", "package_ids", "app_keys", ): value = result.get(key) if not isinstance(value, list) and isinstance(data, dict): value = data.get(key) if isinstance(value, list): lines.append(f"{key}: {len(value)}") for item in value[:8]: lines.append("- " + _builder_item_summary(item)) if len(value) > 8: lines.append(f"... {len(value) - 8} more") components = result.get("components") if isinstance(result.get("components"), list) else [] if components: lines.append("Component Refs:") for item in components[:8]: if not isinstance(item, dict): continue chart_ref = item.get("chart_ref") if isinstance(item.get("chart_ref"), dict) else {} view_ref = item.get("view_ref") if isinstance(item.get("view_ref"), dict) else {} source_type = item.get("source_type") or item.get("type") or "unknown" title = item.get("title") if chart_ref: lines.append(f"- {source_type}: {title or chart_ref.get('chart_name') or '-'} / chart_id={chart_ref.get('chart_id') or '-'}") elif view_ref: lines.append( f"- {source_type}: {title or view_ref.get('view_name') or '-'} / " f"view_key={view_ref.get('view_key') or '-'} / app_key={view_ref.get('app_key') or '-'}" ) else: lines.append(f"- {source_type}: {title or '-'}") if isinstance(result.get("base"), dict): lines.append(f"Base Keys: {', '.join(str(key) for key in list(result['base'].keys())[:12])}") if isinstance(result.get("config"), dict): lines.append(f"Config Keys: {', '.join(str(key) for key in list(result['config'].keys())[:12])}") if isinstance(result.get("visibility"), dict): lines.append(f"Visibility Keys: {', '.join(str(key) for key in list(result['visibility'].keys())[:12])}") _append_warnings(lines, result.get("warnings")) _append_verification(lines, result.get("verification")) if not lines: return _format_generic(result) return "\n".join(lines) + "\n" def _builder_item_summary(item: Any) -> str: if not isinstance(item, dict): return str(item) preferred_keys = ( "app_key", "appKey", "app_name", "appName", "package_id", "packageId", "package_name", "packageName", "dash_key", "dashKey", "dash_name", "dashName", "view_key", "viewKey", "view_name", "viewName", "chart_id", "chartId", "chart_name", "chartName", "field_id", "queId", "title", "queTitle", "name", "role_id", "roleId", "role_name", "roleName", "uid", "userId", "email", ) parts = [ f"{key}={item.get(key)}" for key in preferred_keys if item.get(key) not in (None, "", [], {}) ] if not parts: parts = _dict_scalar_lines(item)[:6] return " / ".join(parts) if parts else json.dumps(item, ensure_ascii=False) def _format_file_upload_local(result: dict[str, Any]) -> str: lines = [ f"Upload Kind: {result.get('upload_kind') or result.get('requested_upload_kind') or '-'}", f"Effective Upload Kind: {result.get('effective_upload_kind') or result.get('upload_kind') or '-'}", ] if "upload_fallback_applied" in result: lines.append(f"Fallback Applied: {result.get('upload_fallback_applied')}") if result.get("upload_fallback_reason"): lines.append(f"Fallback Reason: {result.get('upload_fallback_reason')}") if result.get("file_name"): lines.append(f"File: {result.get('file_name')}") if result.get("file_size") is not None: lines.append(f"Size: {result.get('file_size')}") if result.get("content_type"): lines.append(f"Content Type: {result.get('content_type')}") if result.get("upload_protocol"): lines.append(f"Protocol: {result.get('upload_protocol')}") if result.get("download_url"): lines.append(f"Download URL: {result.get('download_url')}") attachment_value = result.get("attachment_value") if isinstance(result.get("attachment_value"), dict) else {} if attachment_value: lines.append("Attachment Value:") for key in ("value", "name", "otherInfo"): value = attachment_value.get(key) if value not in (None, ""): lines.append(f"- {key}: {value}") comment_file_info = result.get("comment_file_info") if isinstance(result.get("comment_file_info"), dict) else {} if comment_file_info: lines.append("Comment File:") for key in ("url", "name", "uploadFileSize"): value = comment_file_info.get(key) if value not in (None, ""): lines.append(f"- {key}: {value}") _append_warnings(lines, result.get("warnings")) _append_verification(lines, result.get("verification")) return "\n".join(lines) + "\n" def _format_feedback_submit(result: dict[str, Any]) -> str: submitted = bool(result.get("ok") is True or result.get("feedback_request_id") or result.get("submission_mode")) lines = [ "Feedback Submitted" if submitted else "Feedback Result", f"Submission Mode: {result.get('submission_mode') or '-'}", f"Request ID: {result.get('feedback_request_id') or '-'}", ] target = result.get("feedback_target") if isinstance(result.get("feedback_target"), dict) else {} if target: lines.append( "Target: " f"app_key={target.get('app_key') or '-'} / " f"mcp_side={target.get('mcp_side') or '-'}" ) payload = result.get("submission_summary") if isinstance(result.get("submission_summary"), dict) else {} if not payload: payload = result.get("normalized_payload") if isinstance(result.get("normalized_payload"), dict) else {} if payload: for key, label in ( ("category", "Category"), ("反馈类型", "Category"), ("title", "Title"), ("tool_name", "Tool"), ("关联工具", "Tool"), ("app_key", "App Key"), ("关联应用", "App Key"), ("record_id", "Record ID"), ("关联记录", "Record ID"), ("workflow_node_id", "Workflow Node ID"), ("关联节点", "Workflow Node ID"), ("impact_scope", "Impact Scope"), ("影响范围", "Impact Scope"), ): value = payload.get(key) if value not in (None, ""): lines.append(f"{label}: {value}") _append_warnings(lines, result.get("warnings")) _append_verification(lines, result.get("verification")) return "\n".join(lines) + "\n" def emit_json_result(result: dict[str, Any], *, stream: TextIO) -> None: json.dump(result, stream, ensure_ascii=False, indent=2) stream.write("\n") def _render_titled_table(title: str, headers: list[str], rows: list[list[str]]) -> str: lines = [title] if not rows: lines.append("(empty)") return "\n".join(lines) + "\n" widths = [len(header) for header in headers] for row in rows: for index, cell in enumerate(row): widths[index] = max(widths[index], len(cell)) header_line = " ".join(header.ljust(widths[index]) for index, header in enumerate(headers)) lines.append(header_line) lines.append(" ".join("-" * width for width in widths)) for row in rows: lines.append(" ".join(cell.ljust(widths[index]) for index, cell in enumerate(row))) return "\n".join(lines) + "\n" def _dict_scalar_lines(payload: dict[str, Any]) -> list[str]: lines: list[str] = [] for key, value in payload.items(): if isinstance(value, (str, int, float, bool)) or value is None: lines.append(f"{key}: {value}") return lines def _append_warnings(lines: list[str], warnings: Any) -> None: if not isinstance(warnings, list) or not warnings: return lines.append("Warnings:") for warning in warnings: if isinstance(warning, dict): code = warning.get("code") message = warning.get("message") if code or message: lines.append(f"- {code or 'WARNING'}: {message or ''}".rstrip()) else: lines.append(f"- {json.dumps(warning, ensure_ascii=False)}") else: lines.append(f"- {warning}") def _append_verification(lines: list[str], verification: Any) -> None: if not isinstance(verification, dict) or not verification: return lines.append("Verification:") for key, value in verification.items(): if isinstance(value, (str, int, float, bool)) or value is None: lines.append(f"- {key}: {value}") def _first_present(payload: dict[str, Any], *keys: str) -> Any: for key in keys: if key in payload and payload.get(key) is not None: return payload.get(key) return None def _task_action_success_label(action: str) -> str: return { "approve": "已通过", "reject": "已驳回", "rollback": "已退回", "transfer": "已转交", "save_only": "已保存", "urge": "已催办", }.get(action, "已执行") def _task_action_failure_label(action: str) -> str: return { "approve": "审批失败", "reject": "驳回失败", "rollback": "退回失败", "transfer": "转交失败", "save_only": "保存失败", "urge": "催办失败", }.get(action, "执行失败") def _task_action_partial_success_message(result: dict[str, Any]) -> str: warnings = result.get("warnings") if isinstance(warnings, list): for warning in warnings: if not isinstance(warning, dict): continue code = str(warning.get("code") or "").strip().upper() if code == "TASK_ACTION_VERIFICATION_PERMISSION_UNAVAILABLE": return "动作已提交；后置验证读取受当前权限限制，不能据此判断动作被拒绝。可使用 --json 查看详细信息。" if code == "TASK_ALREADY_PROCESSED_UNCONFIRMED_ACTOR": return "当前待办已不可操作，系统判断流程可能已被其他人处理。可使用 --json 查看详细信息。" if code == "WORKFLOW_CONTINUATION_UNVERIFIED": return "动作已提交，但暂未完成后续流程验证。可使用 --json 查看详细信息。" error_code = str(result.get("error_code") or "").strip().upper() if error_code == "WORKFLOW_CONTINUATION_UNVERIFIED": return "动作已提交，但暂未完成后续流程验证。可使用 --json 查看详细信息。" if error_code == "TASK_ALREADY_PROCESSED": return "当前待办已不可操作，系统判断流程可能已被其他人处理。可使用 --json 查看详细信息。" return "动作已提交，但结果验证不完整。可使用 --json 查看详细信息。" def _task_action_failure_reason(result: dict[str, Any]) -> str | None: error_code = str(result.get("error_code") or "").strip().upper() mapped_error = { "TASK_CONTEXT_VISIBILITY_UNVERIFIED": "当前待办已不可操作，且系统未能确认是否已被处理。", "TASK_SAVE_ONLY_VERIFICATION_FAILED": "保存请求已发送，但未能确认字段是否全部保存成功。", "WORKFLOW_CONTINUATION_UNVERIFIED": "动作已提交，但暂未验证到流程继续推进。", "TASK_ALREADY_PROCESSED": "当前待办已不可操作，系统判断流程可能已被其他人处理。", }.get(error_code) if mapped_error: return mapped_error warnings = result.get("warnings") if isinstance(warnings, list): for warning in warnings: if isinstance(warning, dict) and warning.get("message"): return str(warning.get("message")) data = result.get("data") if isinstance(result.get("data"), dict) else {} transport_error = data.get("transport_error") if isinstance(data.get("transport_error"), dict) else {} backend_code = transport_error.get("backend_code") http_status = transport_error.get("http_status") if backend_code not in (None, ""): return f"后端返回错误码 {backend_code}。" if http_status not in (None, ""): return f"请求返回 HTTP {http_status}。" verification = result.get("verification") if isinstance(result.get("verification"), dict) else {} record_state_error = verification.get("record_state_error") if isinstance(verification.get("record_state_error"), dict) else {} backend_code = record_state_error.get("backend_code") http_status = record_state_error.get("http_status") if backend_code not in (None, ""): return f"后端返回错误码 {backend_code}。" if http_status not in (None, ""): return f"请求返回 HTTP {http_status}。" if error_code: return f"错误码：{error_code}" return None def _task_action_debug_lines(result: dict[str, Any]) -> list[str]: lines: list[str] = [] error_code = result.get("error_code") if error_code not in (None, ""): lines.append(f"error_code: {error_code}") data = result.get("data") if isinstance(result.get("data"), dict) else {} transport_error = data.get("transport_error") if isinstance(data.get("transport_error"), dict) else {} for key in ("backend_code", "http_status", "category"): value = transport_error.get(key) if value not in (None, ""): lines.append(f"{key}: {value}") verification = result.get("verification") if isinstance(result.get("verification"), dict) else {} for key in ( "runtime_continuation_verified", "task_context_visibility_verified", "fields_saved_verified", "task_still_actionable", "workflow_not_advanced", "record_state_readable", ): if key in verification and verification.get(key) is not None: lines.append(f"{key}: {verification.get(key)}") record_state_error = verification.get("record_state_error") if isinstance(verification.get("record_state_error"), dict) else {} for key in ("backend_code", "http_status", "category"): value = record_state_error.get(key) if value not in (None, ""): entry = f"record_state_{key}: {value}" if entry not in lines: lines.append(entry) return lines _FORMATTERS = { "auth_whoami": _format_whoami, "workspace_list": _format_workspace_list, "workspace_get": _format_workspace_get, "workspace_select": _format_workspace_select, "app_list": _format_app_items, "app_get": _format_app_get, "portal_list": _format_portal_list, "portal_get": _format_portal_get, "view_get": _format_view_get, "chart_get": _format_chart_get, "record_list": _format_record_list, "record_access": _format_record_access, "record_candidates": _format_record_candidates, "record_schema": _format_record_schema, "record_get": _format_record_get, "record_logs": _format_record_logs, "record_write": _format_record_write, "record_delete": _format_record_delete, "code_block_run": _format_code_block_run, "task_list": _format_task_list, "task_workbench": _format_task_workbench, "task_get": _format_task_get, "task_workflow_log_get": _format_task_workflow_log, "task_action_execute": _format_task_action, "task_associated_report_detail_get": _format_task_associated_report_detail, "import_template": _format_import_template, "import_verify": _format_import_verify, "import_repair": _format_import_repair, "import_start": _format_import_start, "import_status": _format_import_status, "export_start": _format_export_start, "export_status": _format_export_status, "export_get": _format_export_get, "export_direct": _format_export_direct, "builder_summary": _format_builder_summary, "file_upload_local": _format_file_upload_local, "feedback_submit": _format_feedback_submit, }