(self, config: dict, context: ExecutionContext)
| 126 | return "input_prompt" |
| 127 | |
| 128 | async def execute(self, config: dict, context: ExecutionContext) -> ModuleResult: |
| 129 | import os as _os_pkg |
| 130 | if _os_pkg.environ.get('WEBRPA_PACKAGED'): |
| 131 | from app.services.packaged_ui import request_input_prompt_sync |
| 132 | else: |
| 133 | from app.main import request_input_prompt_sync |
| 134 | import json |
| 135 | |
| 136 | variable_name = config.get('variableName', '') |
| 137 | prompt_title = context.resolve_value(config.get('promptTitle', '输入')) |
| 138 | prompt_message = context.resolve_value(config.get('promptMessage', '请输入值:')) |
| 139 | default_value = context.resolve_value(config.get('defaultValue', '')) |
| 140 | input_mode = context.resolve_value(config.get('inputMode', 'single')) |
| 141 | min_value = config.get('minValue') |
| 142 | max_value = config.get('maxValue') |
| 143 | max_length = config.get('maxLength') |
| 144 | required_raw = config.get('required', True) |
| 145 | if isinstance(required_raw, str): |
| 146 | required_raw = context.resolve_value(required_raw) |
| 147 | required = required_raw in [True, 'true', 'True', '1', 1] |
| 148 | |
| 149 | # 列表选择模式:解析选项列表 |
| 150 | select_options = None |
| 151 | if input_mode in ('select_single', 'select_multiple'): |
| 152 | select_options_var = config.get('selectOptions', '') |
| 153 | if select_options_var: |
| 154 | resolved_options = context.resolve_value(select_options_var) |
| 155 | if isinstance(resolved_options, list): |
| 156 | select_options = [] |
| 157 | for item in resolved_options: |
| 158 | if isinstance(item, dict): |
| 159 | select_options.append(json.dumps(item, ensure_ascii=False)) |
| 160 | elif isinstance(item, (list, tuple)): |
| 161 | select_options.append(str(item)) |
| 162 | else: |
| 163 | select_options.append(str(item)) |
| 164 | elif isinstance(resolved_options, str): |
| 165 | var_name = select_options_var.strip('{}') |
| 166 | raw_list = context.variables.get(var_name, []) |
| 167 | if isinstance(raw_list, list): |
| 168 | select_options = [] |
| 169 | for item in raw_list: |
| 170 | if isinstance(item, dict): |
| 171 | select_options.append(json.dumps(item, ensure_ascii=False)) |
| 172 | elif isinstance(item, (list, tuple)): |
| 173 | select_options.append(str(item)) |
| 174 | else: |
| 175 | select_options.append(str(item)) |
| 176 | else: |
| 177 | select_options = [] |
| 178 | |
| 179 | if not variable_name: |
| 180 | return ModuleResult(success=False, error="变量名不能为空") |
| 181 | |
| 182 | try: |
| 183 | loop = asyncio.get_running_loop() |
| 184 | user_input = await loop.run_in_executor( |
| 185 | None, |
nothing calls this directly
no test coverage detected