(self, config: dict, context: ExecutionContext)
| 77 | return "js_script" |
| 78 | |
| 79 | async def execute(self, config: dict, context: ExecutionContext) -> ModuleResult: |
| 80 | import os as _os_pkg |
| 81 | if _os_pkg.environ.get('WEBRPA_PACKAGED'): |
| 82 | from app.services.packaged_ui import request_js_script_sync |
| 83 | else: |
| 84 | from app.main import request_js_script_sync |
| 85 | |
| 86 | code = context.resolve_value(config.get('code', '')) |
| 87 | result_variable = config.get('resultVariable', '') |
| 88 | |
| 89 | if not code: |
| 90 | return ModuleResult(success=False, error="JavaScript代码不能为空") |
| 91 | |
| 92 | try: |
| 93 | variables = dict(context.variables) |
| 94 | |
| 95 | loop = asyncio.get_running_loop() |
| 96 | result = await loop.run_in_executor( |
| 97 | None, |
| 98 | lambda: request_js_script_sync(code=code, variables=variables, timeout=30) |
| 99 | ) |
| 100 | |
| 101 | if result.get('success'): |
| 102 | script_result = result.get('result') |
| 103 | |
| 104 | if result_variable: |
| 105 | context.set_variable(result_variable, script_result) |
| 106 | |
| 107 | result_str = str(script_result) |
| 108 | if len(result_str) > 100: |
| 109 | result_str = result_str[:100] + '...' |
| 110 | |
| 111 | return ModuleResult(success=True, message=f"JS脚本执行成功,返回值: {result_str}", |
| 112 | data={'result': script_result}) |
| 113 | else: |
| 114 | error = result.get('error', '未知错误') |
| 115 | return ModuleResult(success=False, error=f"JS脚本执行失败: {error}") |
| 116 | except Exception as e: |
| 117 | return ModuleResult(success=False, error=f"JS脚本执行异常: {str(e)}") |
| 118 | |
| 119 | |
| 120 | @register_executor |
nothing calls this directly
no test coverage detected