MCPcopy
hub / github.com/pmh1314520/WebRPA / execute

Method execute

backend/app/executors/basic_page.py:128–263  ·  view source on GitHub ↗
(self, config: dict, context: ExecutionContext)

Source from the content-addressed store, hash-verified

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,

Callers

nothing calls this directly

Calls 5

ModuleResultClass · 0.85
resolve_valueMethod · 0.80
set_variableMethod · 0.80
getMethod · 0.45

Tested by

no test coverage detected