from __future__ import annotations import argparse import sys from typing import Any from ..context import CliContext from ..interaction import cancelled_result, resolve_interactive_selection, resolve_interactive_text_input from ..terminal_ui import SelectionOption from .common import load_object_arg, raise_config_error TASK_ACTION_LABELS = { "approve": "通过", "reject": "驳回", "rollback": "退回", "transfer": "转交", "save_only": "仅保存", "urge": "催办", } TASK_ACTION_SUCCESS_LABELS = { "approve": "已通过", "reject": "已驳回", "rollback": "已退回", "transfer": "已转交", "save_only": "已保存", "urge": "已催办", } def register(subparsers: argparse._SubParsersAction[argparse.ArgumentParser]) -> None: parser = subparsers.add_parser("task", help="待办与流程上下文") task_subparsers = parser.add_subparsers(dest="task_command", required=False) parser.set_defaults(handler=_handle_task_workbench, format_hint="task_workbench", _task_parser=parser) list_parser = task_subparsers.add_parser("list", help="列出待办") list_parser.add_argument("--task-box", default="todo") list_parser.add_argument("--flow-status", default="all") list_parser.add_argument("--app-key") list_parser.add_argument("--workflow-node-id", type=int) list_parser.add_argument( "--query", help="先走后端待办检索；当后端返回零结果时，公开 task_list 会回退到本地匹配 app_name / workflow_node_name / app_key / record_id。", ) list_parser.add_argument("--page", type=int, default=1) list_parser.add_argument("--page-size", type=int, default=20) list_parser.set_defaults(handler=_handle_list, format_hint="task_list") get = task_subparsers.add_parser("get", help="读取待办详情；--task-id 必须来自 task list 的 data.items[].task_id") get.add_argument("--task-id", help="来自 `qingflow --json task list` 的 data.items[].task_id；不是列表序号") get.add_argument("--include-candidates", action=argparse.BooleanOptionalAction, default=True) get.add_argument("--include-associated-reports", action=argparse.BooleanOptionalAction, default=True) get.set_defaults(handler=_handle_get, format_hint="task_get", app_key="", record_id="", workflow_node_id=0) action = task_subparsers.add_parser("action", help="执行待办动作；--task-id 必须来自 task list 的 data.items[].task_id") action.add_argument("--task-id", help="来自 `qingflow --json task list` 的 data.items[].task_id；不是列表序号") action.add_argument("--action", help="不传时在交互终端中选择当前待办可执行动作") action.add_argument("--payload-file") action.add_argument("--fields-file") action.set_defaults( handler=_handle_action, format_hint="task_action_execute", hide_effective_context_line=True, app_key="", record_id="", workflow_node_id=0, ) report = task_subparsers.add_parser("report", help="读取待办关联报表详情；--task-id 必须来自 task list 的 data.items[].task_id") report.add_argument("--task-id", help="来自 `qingflow --json task list` 的 data.items[].task_id；不是列表序号") report.add_argument("--report-id", type=int, help="不传时在交互终端中选择关联报表") report.add_argument("--page", type=int, default=1) report.add_argument("--page-size", type=int, default=20) report.set_defaults( handler=_handle_report, format_hint="task_associated_report_detail_get", app_key="", record_id="", workflow_node_id=0, ) log = task_subparsers.add_parser("log", help="读取流程日志；--task-id 必须来自 task list 的 data.items[].task_id") log.add_argument("--task-id", help="来自 `qingflow --json task list` 的 data.items[].task_id；不是列表序号") log.set_defaults(handler=_handle_log, format_hint="task_workflow_log_get", app_key="", record_id="", workflow_node_id=0) def _handle_list(args: argparse.Namespace, context: CliContext) -> dict: return context.task.task_list( profile=args.profile, task_box=args.task_box, flow_status=args.flow_status, app_key=args.app_key, workflow_node_id=args.workflow_node_id, query=args.query, page=args.page, page_size=args.page_size, ) def _handle_task_workbench(args: argparse.Namespace, context: CliContext) -> dict: if bool(args.json) or not _has_interactive_terminal(args): _raise_task_subcommand_required(args) banner: str | None = None while True: title = _build_task_workbench_list_title(banner) selection = _choose_todo_task_interactively(args, context, tool_name="task", title=title) banner = None if selection.status == "empty": return {"status": "success", "message": "当前没有待办"} if selection.status == "cancelled": return cancelled_result("已退出") if selection.status != "selected": _raise_task_subcommand_required(args) args.task_id = str(selection.value or "") args.app_key = "" args.record_id = "" args.workflow_node_id = 0 outcome, payload = _run_task_workbench_task_loop(args, context) if outcome == "refresh": banner = str(payload or "").strip() or None continue if outcome == "back": continue return payload if isinstance(payload, dict) else cancelled_result("已退出") def _handle_get(args: argparse.Namespace, context: CliContext) -> dict: selection_result = _resolve_task_locator_or_select(args, context, tool_name="task get") if isinstance(selection_result, dict): return selection_result return context.task.task_get( profile=args.profile, task_id=args.task_id, app_key=args.app_key or "", record_id=args.record_id or "", workflow_node_id=int(args.workflow_node_id or 0), include_candidates=bool(args.include_candidates), include_associated_reports=bool(args.include_associated_reports), ) def _handle_action(args: argparse.Namespace, context: CliContext) -> dict: selection_result = _resolve_task_locator_or_select(args, context, tool_name="task action") if isinstance(selection_result, dict): return selection_result payload = load_object_arg(args.payload_file, option_name="--payload-file") or {} fields = load_object_arg(args.fields_file, option_name="--fields-file") or {} task_context: dict[str, Any] | None = None if not str(args.action or "").strip(): task_context = _load_task_action_context(args, context) action_selection = _resolve_task_action_or_select(args, task_context, fields=fields) if isinstance(action_selection, dict): return action_selection payload_selection = _resolve_action_payload_or_select(args, task_context, payload=payload) if _is_cancelled_result(payload_selection): return payload_selection payload = payload_selection return context.task.task_action_execute( profile=args.profile, task_id=args.task_id, action=args.action, payload=payload, fields=fields, ) def _handle_log(args: argparse.Namespace, context: CliContext) -> dict: selection_result = _resolve_task_locator_or_select(args, context, tool_name="task log") if isinstance(selection_result, dict): return selection_result return context.task.task_workflow_log_get( profile=args.profile, task_id=args.task_id, app_key=args.app_key or "", record_id=args.record_id or "", workflow_node_id=int(args.workflow_node_id or 0), ) def _handle_report(args: argparse.Namespace, context: CliContext) -> dict: selection_result = _resolve_task_locator_or_select(args, context, tool_name="task report") if isinstance(selection_result, dict): return selection_result report_selection = _resolve_report_id_or_select(args, context) if isinstance(report_selection, dict): return report_selection return context.task.task_associated_report_detail_get( profile=args.profile, task_id=args.task_id, app_key=args.app_key or "", record_id=args.record_id or "", workflow_node_id=int(args.workflow_node_id or 0), report_id=int(args.report_id), page=int(args.page), page_size=int(args.page_size), ) def _run_task_workbench_task_loop(args: argparse.Namespace, context: CliContext) -> tuple[str, dict | str]: while True: task_context = _load_task_workbench_context(args, context) detail_selection = _show_task_workbench_detail(args, task_context) if detail_selection.status == "cancelled": return "back", "" selected_value = str(detail_selection.value or "").strip() if selected_value == "exit": return "result", cancelled_result("已退出") if selected_value == "back": return "back", "" if not selected_value: continue args.action = selected_value payload_selection = _resolve_action_payload_or_select(args, task_context, payload={}) if _is_cancelled_result(payload_selection): continue result = context.task.task_action_execute( profile=args.profile, task_id=args.task_id, action=args.action, payload=payload_selection, fields={}, ) if _is_success_result(result): return "refresh", _task_action_success_label(str(args.action or "").strip().lower()) return "result", result def _resolve_task_locator_or_select(args: argparse.Namespace, context: CliContext, *, tool_name: str) -> dict | None: if (args.task_id or "").strip(): return None selection = _choose_todo_task_interactively(args, context, tool_name=tool_name) if selection.status == "unavailable": raise_config_error( f"{tool_name} requires --task-id from task list data.items[].task_id", fix_hint=( "Retry in an interactive terminal to choose from current todo tasks, " "or pass `--task-id` from `task list` explicitly." ), ) if selection.status == "empty": raise_config_error( selection.message or f"{tool_name} could not open a selector because no current todo tasks are available.", fix_hint="Run `task list` to confirm current todo tasks, then pass the selected `data.items[].task_id`.", ) if selection.status == "cancelled": return cancelled_result(selection.message or "已取消") args.task_id = str(selection.value or "") return None def _choose_todo_task_interactively( args: argparse.Namespace, context: CliContext, *, tool_name: str, title: str = "选择待办", ): def load_options() -> list[SelectionOption[str]]: result = context.task.task_list( profile=args.profile, task_box="todo", flow_status="all", app_key=None, workflow_node_id=None, query=None, page=1, page_size=100, ) data = result.get("data") if isinstance(result, dict) and isinstance(result.get("data"), dict) else {} items = data.get("items") if isinstance(data.get("items"), list) else [] options: list[SelectionOption[str]] = [] for item in items: if not isinstance(item, dict): continue task_id = str(item.get("task_id") or "").strip() if not task_id: continue node_name = str(item.get("workflow_node_name") or "未命名节点").strip() or "未命名节点" app_name = str(item.get("app_name") or item.get("app_key") or "未知应用").strip() or "未知应用" label = f"{node_name} / {app_name}" hint_parts = [f"task_id={task_id}"] apply_time = str(item.get("apply_time") or "").strip() if apply_time: hint_parts.append(apply_time) summary_fields = item.get("summary_fields") if isinstance(item.get("summary_fields"), list) else [] preview_parts: list[str] = [] for summary in summary_fields[:2]: if not isinstance(summary, dict): continue title = str(summary.get("title") or "").strip() answer = str(summary.get("answer") or "").strip() if title and answer: preview_parts.append(f"{title}: {answer}") if preview_parts: hint_parts.append("; ".join(preview_parts)) options.append(SelectionOption(value=task_id, label=label, hint=" · ".join(hint_parts))) return options return resolve_interactive_selection( args, title=title, unavailable_message=( f"{tool_name} requires --task-id from task list data.items[].task_id" ), empty_message=f"{tool_name} could not open a selector because no current todo tasks are available.", load_options=load_options, ) def _resolve_report_id_or_select(args: argparse.Namespace, context: CliContext) -> dict | None: if int(args.report_id or 0) > 0: return None selection = _choose_associated_report_interactively(args, context) if selection.status == "unavailable": raise_config_error( "task report requires --report-id, or an interactive terminal to choose a visible associated report", fix_hint="Pass `--report-id REPORT_ID`, or run in an interactive terminal and choose from visible associated reports.", ) if selection.status == "empty": raise_config_error( selection.message or "task report could not open a selector because the selected task has no visible associated reports.", fix_hint="Run `task get --task-id ` to inspect `extras.associated_reports`, or choose another task.", ) if selection.status == "cancelled": return cancelled_result(selection.message or "已取消") args.report_id = int(selection.value or 0) return None def _load_task_action_context(args: argparse.Namespace, context: CliContext) -> dict[str, Any]: return context.task.task_get( profile=args.profile, task_id=args.task_id, app_key=args.app_key or "", record_id=args.record_id or "", workflow_node_id=int(args.workflow_node_id or 0), include_candidates=True, include_associated_reports=False, ) def _load_task_workbench_context(args: argparse.Namespace, context: CliContext) -> dict[str, Any]: return context.task.task_get( profile=args.profile, task_id=args.task_id, app_key=args.app_key or "", record_id=args.record_id or "", workflow_node_id=int(args.workflow_node_id or 0), include_candidates=True, include_associated_reports=True, ) def _resolve_task_action_or_select( args: argparse.Namespace, task_context: dict[str, Any], *, fields: dict[str, Any], ) -> dict | None: selection = _choose_task_action_interactively(args, task_context, fields=fields) if selection.status == "unavailable": raise_config_error( "task action requires --action, or an interactive terminal to choose the current task action", fix_hint="Pass `--action ACTION`, or run `qingflow task action` in an interactive terminal and choose from available actions.", ) if selection.status == "empty": raise_config_error( selection.message or "task action could not open an action selector because no interactive actions are available.", fix_hint="Run `task get --task-id ` to inspect available_actions, or pass a supported `--action` explicitly.", ) if selection.status == "cancelled": return cancelled_result(selection.message or "已取消") args.action = str(selection.value or "").strip() return None def _choose_task_action_interactively( args: argparse.Namespace, task_context: dict[str, Any], *, fields: dict[str, Any], title: str = "选择操作", ): def load_options() -> list[SelectionOption[str]]: return _task_interactive_action_options(task_context, fields=fields) return resolve_interactive_selection( args, title=title, unavailable_message="task action requires --action, or an interactive terminal to choose the current task action", empty_message="task action could not open an action selector because no interactive actions are available.", load_options=load_options, ) def _show_task_workbench_detail(args: argparse.Namespace, task_context: dict[str, Any]): def load_options() -> list[SelectionOption[str]]: options = _task_interactive_action_options(task_context, fields={}) if options: options.append(SelectionOption(value="back", label="返回列表", hint="回到待办列表")) return options return [ SelectionOption(value="back", label="返回列表", hint="当前节点没有可执行动作"), SelectionOption(value="exit", label="退出工作台", hint="结束当前待办工作台"), ] return resolve_interactive_selection( args, title=_build_task_detail_title(task_context), unavailable_message="task workbench requires an interactive terminal", empty_message="task workbench could not open the task detail view.", load_options=load_options, ) def _resolve_action_payload_or_select( args: argparse.Namespace, task_context: dict[str, Any], *, payload: dict[str, Any], ) -> dict[str, Any] | dict[str, str]: action = str(args.action or "").strip().lower() if action == "reject" and not _extract_audit_feedback(payload): feedback = resolve_interactive_text_input( args, prompt="请输入驳回原因并回车，留空取消: ", unavailable_message="task reject requires payload.audit_feedback, or an interactive terminal to enter a reject reason", ) if feedback.status == "unavailable": raise_config_error( "task reject requires payload.audit_feedback, or an interactive terminal to enter a reject reason", fix_hint="Pass `--payload-file` with `{\"audit_feedback\": \"...\"}`, or retry in an interactive terminal.", ) if feedback.status == "cancelled": return cancelled_result(feedback.message or "已取消") payload["audit_feedback"] = str(feedback.value or "").strip() return payload if action == "rollback" and _extract_positive_int(payload, "target_workflow_node_id", aliases=("targetAuditNodeId", "targetWorkflowNodeId")) <= 0: selection = _choose_rollback_candidate_interactively(args, task_context) if selection.status == "unavailable": raise_config_error( "task rollback requires target_workflow_node_id, or an interactive terminal to choose a rollback node", fix_hint="Pass `--payload-file` with `{\"target_workflow_node_id\": NODE_ID}`, or retry in an interactive terminal.", ) if selection.status == "empty": raise_config_error( selection.message or "task rollback could not open a selector because no rollback candidates are visible.", fix_hint="Run `task get --task-id ` to inspect rollback candidates, or choose another action.", ) if selection.status == "cancelled": return cancelled_result(selection.message or "已取消") payload["target_workflow_node_id"] = int(selection.value or 0) return payload if action == "transfer" and _extract_positive_int(payload, "target_member_id", aliases=("uid", "targetMemberId")) <= 0: selection = _choose_transfer_candidate_interactively(args, task_context) if selection.status == "unavailable": raise_config_error( "task transfer requires target_member_id, or an interactive terminal to choose a transfer member", fix_hint="Pass `--payload-file` with `{\"target_member_id\": UID}`, or retry in an interactive terminal.", ) if selection.status == "empty": raise_config_error( selection.message or "task transfer could not open a selector because no transfer candidates are visible.", fix_hint="Run `task get --task-id ` to inspect transfer candidates, or choose another action.", ) if selection.status == "cancelled": return cancelled_result(selection.message or "已取消") payload["target_member_id"] = int(selection.value or 0) return payload return payload def _choose_rollback_candidate_interactively(args: argparse.Namespace, task_context: dict[str, Any]): extras = _task_context_extras(task_context) def load_options() -> list[SelectionOption[int]]: rollback_candidates = extras.get("rollback_candidates") if isinstance(extras.get("rollback_candidates"), dict) else {} items = rollback_candidates.get("items") if isinstance(rollback_candidates.get("items"), list) else [] options: list[SelectionOption[int]] = [] for item in items: if not isinstance(item, dict): continue workflow_node_id = int(item.get("workflow_node_id") or 0) if workflow_node_id <= 0: continue workflow_node_name = str(item.get("workflow_node_name") or f"节点 {workflow_node_id}").strip() or f"节点 {workflow_node_id}" options.append( SelectionOption( value=workflow_node_id, label=workflow_node_name, hint=f"workflow_node_id={workflow_node_id}", ) ) return options return resolve_interactive_selection( args, title="选择退回节点", unavailable_message="task rollback requires target_workflow_node_id, or an interactive terminal to choose a rollback node", empty_message="task rollback could not open a selector because no rollback candidates are visible.", load_options=load_options, ) def _choose_transfer_candidate_interactively(args: argparse.Namespace, task_context: dict[str, Any]): extras = _task_context_extras(task_context) def load_options() -> list[SelectionOption[int]]: transfer_candidates = extras.get("transfer_candidates") if isinstance(extras.get("transfer_candidates"), dict) else {} items = transfer_candidates.get("items") if isinstance(transfer_candidates.get("items"), list) else [] options: list[SelectionOption[int]] = [] for item in items: if not isinstance(item, dict): continue uid = int(item.get("uid") or 0) if uid <= 0: continue name = str(item.get("name") or f"成员 {uid}").strip() or f"成员 {uid}" hint_parts = [f"uid={uid}"] department_name = str(item.get("department_name") or "").strip() email = str(item.get("email") or "").strip() if department_name: hint_parts.append(department_name) if email: hint_parts.append(email) options.append(SelectionOption(value=uid, label=name, hint=" · ".join(hint_parts))) return options return resolve_interactive_selection( args, title="选择转交成员", unavailable_message="task transfer requires target_member_id, or an interactive terminal to choose a transfer member", empty_message="task transfer could not open a selector because no transfer candidates are visible.", load_options=load_options, ) def _task_context_extras(task_context: dict[str, Any]) -> dict[str, Any]: data = task_context.get("data") if isinstance(task_context.get("data"), dict) else {} return data.get("extras") if isinstance(data.get("extras"), dict) else {} def _count_candidate_items(value: Any) -> int: if not isinstance(value, dict): return 0 items = value.get("items") return len(items) if isinstance(items, list) else 0 def _extract_audit_feedback(payload: dict[str, Any]) -> str | None: for key in ("audit_feedback", "auditFeedback"): value = payload.get(key) if isinstance(value, str) and value.strip(): return value.strip() return None def _extract_positive_int(payload: dict[str, Any], key: str, *, aliases: tuple[str, ...] = ()) -> int: for candidate in (key, *aliases): value = payload.get(candidate) if isinstance(value, int) and value > 0: return value return 0 def _is_cancelled_result(value: Any) -> bool: return isinstance(value, dict) and str(value.get("status") or "").strip().lower() == "cancelled" def _is_success_result(result: dict[str, Any]) -> bool: if not isinstance(result, dict): return False if result.get("ok") is False: return False return str(result.get("status") or "").strip().lower() == "success" def _task_action_success_label(action: str) -> str: return TASK_ACTION_SUCCESS_LABELS.get(action, "已完成") def _build_task_workbench_list_title(banner: str | None) -> str: lines = ["待办工作台", "选择一条待办后查看详情并执行操作"] if banner: lines.extend(["", f"上一条结果：{banner}"]) return "\n".join(lines) def _build_task_detail_title(task_context: dict[str, Any]) -> str: data = task_context.get("data") if isinstance(task_context.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 {} available_actions = data.get("available_actions") if isinstance(data.get("available_actions"), list) else [] editable_fields = data.get("editable_fields") if isinstance(data.get("editable_fields"), list) else [] extras = data.get("extras") if isinstance(data.get("extras"), dict) else {} initiator = task.get("initiator") if isinstance(task.get("initiator"), dict) else {} 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 {} field_map = all_fields or core_fields action_labels = ", ".join(TASK_ACTION_LABELS.get(str(item), str(item)) for item in available_actions if str(item).strip()) or "-" lines = [ "待办详情", f"Task ID: {task.get('task_id') or '-'}", f"应用: {task.get('app_name') or '-'}", f"节点: {task.get('workflow_node_name') or '-'}", f"发起人: {_task_initiator_label(initiator)}", f"状态: {record_summary.get('apply_status') or '-'}", f"申请编号: {record_summary.get('custom_apply_num') or record_summary.get('apply_num') or '-'}", f"提交时间: {record_summary.get('apply_time') or '-'}", f"可执行动作: {action_labels}", f"可编辑字段: {len(editable_fields)}", ( "附加信息: " f"报表 {_count_candidate_items(extras.get('associated_reports'))} / " f"退回 {_count_candidate_items(extras.get('rollback_candidates'))} / " f"转交 {_count_candidate_items(extras.get('transfer_candidates'))}" ), ] if field_map: lines.append("") lines.append(f"字段值（{len(field_map)}）:") for key, value in field_map.items(): lines.append(f"- {key}: {_full_display(value)}") return "\n".join(lines) def _task_initiator_label(initiator: dict[str, Any]) -> str: for key in ("display_name", "displayName", "name", "email", "uid"): value = initiator.get(key) if value not in (None, ""): return str(value) return "-" def _task_interactive_action_options(task_context: dict[str, Any], *, fields: dict[str, Any]) -> list[SelectionOption[str]]: data = task_context.get("data") if isinstance(task_context.get("data"), dict) else {} available_actions = [str(item).strip() for item in (data.get("available_actions") or []) if str(item).strip()] action_metadata = data.get("action_metadata") if isinstance(data.get("action_metadata"), dict) else {} extras = data.get("extras") if isinstance(data.get("extras"), dict) else {} feedback_required = { str(item).strip() for item in (action_metadata.get("feedback_required_for") or []) if str(item).strip() } rollback_count = _count_candidate_items(extras.get("rollback_candidates")) transfer_count = _count_candidate_items(extras.get("transfer_candidates")) options: list[SelectionOption[str]] = [] for action in available_actions: if action == "save_only" and not fields: continue label = TASK_ACTION_LABELS.get(action, action) hint_parts = [action] if action in feedback_required: hint_parts.append("需要理由") if action == "rollback": hint_parts.append(f"可退回节点 {rollback_count}") if action == "transfer": hint_parts.append(f"可转交成员 {transfer_count}") if action == "save_only": hint_parts.append("仅保存字段，不推进流程") options.append(SelectionOption(value=action, label=label, hint=" · ".join(hint_parts))) return options def _full_display(value: Any) -> str: if value in (None, ""): return "-" if isinstance(value, list): normalized = [str(item) for item in value if item not in (None, "")] return " / ".join(normalized) if normalized else "-" return str(value) def _has_interactive_terminal(args: argparse.Namespace) -> bool: input_stream = getattr(args, "_stdin", None) return bool(getattr(input_stream, "isatty", lambda: False)()) def _raise_task_subcommand_required(args: argparse.Namespace) -> None: stream = getattr(args, "_stderr_stream", None) or sys.stderr task_parser = getattr(args, "_task_parser", None) if isinstance(task_parser, argparse.ArgumentParser): task_parser.print_usage(stream) stream.write(f"{task_parser.prog}: error: the following arguments are required: task_command\n") else: stream.write("qingflow task: error: the following arguments are required: task_command\n") raise SystemExit(2) def _choose_associated_report_interactively(args: argparse.Namespace, context: CliContext): def load_options() -> list[SelectionOption[int]]: result = context.task.task_get( profile=args.profile, task_id=args.task_id, app_key=args.app_key or "", record_id=args.record_id or "", workflow_node_id=int(args.workflow_node_id or 0), include_candidates=False, include_associated_reports=True, ) data = result.get("data") if isinstance(result, dict) and isinstance(result.get("data"), dict) else {} extras = data.get("extras") if isinstance(data.get("extras"), dict) else {} associated_reports = extras.get("associated_reports") if isinstance(extras.get("associated_reports"), dict) else {} items = associated_reports.get("items") if isinstance(associated_reports.get("items"), list) else [] options: list[SelectionOption[int]] = [] for item in items: if not isinstance(item, dict): continue report_id = int(item.get("report_id") or 0) if report_id <= 0: continue chart_name = str(item.get("chart_name") or f"报表 {report_id}").strip() or f"报表 {report_id}" graph_type = str(item.get("graph_type") or "chart").strip() or "chart" options.append( SelectionOption( value=report_id, label=chart_name, hint=f"report_id={report_id} · {graph_type}", ) ) return options return resolve_interactive_selection( args, title="选择关联报表", unavailable_message="task report requires --report-id, or an interactive terminal to choose a visible associated report", empty_message="task report could not open a selector because the selected task has no visible associated reports.", load_options=load_options, )