|
7 | 7 | import subprocess |
8 | 8 | import sys |
9 | 9 | from dataclasses import dataclass |
10 | | -from typing import Any |
| 10 | +from typing import Any, Callable |
11 | 11 |
|
12 | 12 | from astrbot.api import logger |
13 | 13 | from astrbot.core.utils.astrbot_path import ( |
@@ -53,6 +53,45 @@ def _ensure_safe_path(path: str) -> str: |
53 | 53 | return abs_path |
54 | 54 |
|
55 | 55 |
|
| 56 | +def _resolve_working_dir(configured_path: str | None, fallback_func: Callable[[], str]) -> tuple[str, bool]: |
| 57 | + """Resolve working directory with fallback to default. |
| 58 | +
|
| 59 | + Args: |
| 60 | + configured_path: The configured working directory path, or None |
| 61 | + fallback_func: A callable that returns the fallback path (e.g., get_astrbot_root) |
| 62 | +
|
| 63 | + Returns: |
| 64 | + A tuple of (resolved_path, was_fallback) where was_fallback indicates if fallback was used |
| 65 | + """ |
| 66 | + if not configured_path: |
| 67 | + return fallback_func(), True |
| 68 | + |
| 69 | + try: |
| 70 | + abs_path = _ensure_safe_path(configured_path) |
| 71 | + except PermissionError: |
| 72 | + logger.warning( |
| 73 | + f"[Computer] Configured path '{configured_path}' is outside allowed roots, " |
| 74 | + f"falling back to default directory." |
| 75 | + ) |
| 76 | + return fallback_func(), True |
| 77 | + |
| 78 | + if not os.path.exists(abs_path): |
| 79 | + logger.warning( |
| 80 | + f"[Computer] Configured path '{configured_path}' does not exist, " |
| 81 | + f"falling back to default directory." |
| 82 | + ) |
| 83 | + return fallback_func(), True |
| 84 | + |
| 85 | + if not os.access(abs_path, os.R_OK | os.W_OK): |
| 86 | + logger.warning( |
| 87 | + f"[Computer] Configured path '{configured_path}' is not accessible (no read/write permission), " |
| 88 | + f"falling back to default directory." |
| 89 | + ) |
| 90 | + return fallback_func(), True |
| 91 | + |
| 92 | + return abs_path, False |
| 93 | + |
| 94 | + |
56 | 95 | def _decode_bytes_with_fallback( |
57 | 96 | output: bytes | None, |
58 | 97 | *, |
@@ -110,7 +149,7 @@ def _run() -> dict[str, Any]: |
110 | 149 | run_env = os.environ.copy() |
111 | 150 | if env: |
112 | 151 | run_env.update({str(k): str(v) for k, v in env.items()}) |
113 | | - working_dir = _ensure_safe_path(cwd) if cwd else get_astrbot_root() |
| 152 | + working_dir, _ = _resolve_working_dir(cwd, get_astrbot_root) |
114 | 153 | if background: |
115 | 154 | # `command` is intentionally executed through the current shell so |
116 | 155 | # local computer-use behavior matches existing tool semantics. |
@@ -146,20 +185,26 @@ def _run() -> dict[str, Any]: |
146 | 185 |
|
147 | 186 | @dataclass |
148 | 187 | class LocalPythonComponent(PythonComponent): |
| 188 | + default_cwd: str | None = None |
| 189 | + |
149 | 190 | async def exec( |
150 | 191 | self, |
151 | 192 | code: str, |
152 | 193 | kernel_id: str | None = None, |
153 | 194 | timeout: int = 30, |
154 | 195 | silent: bool = False, |
| 196 | + cwd: str | None = None, |
155 | 197 | ) -> dict[str, Any]: |
156 | 198 | def _run() -> dict[str, Any]: |
157 | 199 | try: |
| 200 | + effective_cwd = cwd if cwd else self.default_cwd |
| 201 | + working_dir, _ = _resolve_working_dir(effective_cwd, get_astrbot_root) |
158 | 202 | result = subprocess.run( |
159 | 203 | [os.environ.get("PYTHON", sys.executable), "-c", code], |
160 | 204 | timeout=timeout, |
161 | 205 | capture_output=True, |
162 | 206 | text=True, |
| 207 | + cwd=working_dir, |
163 | 208 | ) |
164 | 209 | stdout = "" if silent else result.stdout |
165 | 210 | stderr = result.stderr if result.returncode != 0 else "" |
|
0 commit comments