(
booter: ComputerBooter,
script: str,
*,
action: str,
)
| 241 | |
| 242 | |
| 243 | async def _exec_python_json( |
| 244 | booter: ComputerBooter, |
| 245 | script: str, |
| 246 | *, |
| 247 | action: str, |
| 248 | ) -> dict: |
| 249 | result = await booter.python.exec(script) |
| 250 | data = result.get("data") if isinstance(result.get("data"), dict) else {} |
| 251 | if not isinstance(data, dict): |
| 252 | raise RuntimeError(f"{action} failed: invalid result format") |
| 253 | output = data.get("output") if isinstance(data.get("output"), dict) else {} |
| 254 | if not isinstance(output, dict): |
| 255 | raise RuntimeError(f"{action} failed: invalid output format") |
| 256 | error_text = str(data.get("error", "") or result.get("error", "") or "").strip() |
| 257 | if error_text: |
| 258 | raise RuntimeError(f"{action} failed: {error_text}") |
| 259 | |
| 260 | text = str(output.get("text", "") or "").strip() |
| 261 | if not text: |
| 262 | raise RuntimeError(f"{action} failed: empty output") |
| 263 | |
| 264 | try: |
| 265 | payload = json.loads(text) |
| 266 | except json.JSONDecodeError as exc: |
| 267 | raise RuntimeError(f"{action} failed: invalid JSON output") from exc |
| 268 | |
| 269 | if not isinstance(payload, dict): |
| 270 | raise RuntimeError(f"{action} failed: invalid JSON payload") |
| 271 | return payload |
| 272 | |
| 273 | |
| 274 | async def _probe_local_file(path: str) -> dict[str, str | int]: |
no test coverage detected