|
1 | | -"""Safe process runner for user-supplied Xbox command-line tools.""" |
| 1 | +"""Compatibility wrapper for package-owned external tool execution.""" |
2 | 2 |
|
3 | 3 | from __future__ import annotations |
4 | 4 |
|
5 | | -import os |
6 | | -import shlex |
7 | | -import subprocess |
8 | | -import threading |
9 | | -import time |
10 | | -from dataclasses import dataclass |
11 | | -from pathlib import Path |
12 | | -from typing import Iterable |
13 | | - |
14 | | - |
15 | | -class ExternalToolError(RuntimeError): |
16 | | - """Raised when an external tool cannot be configured or launched.""" |
17 | | - |
18 | | - |
19 | | -@dataclass(frozen=True) |
20 | | -class ToolResult: |
21 | | - """Captured result from one external tool invocation.""" |
22 | | - |
23 | | - command: tuple[str, ...] |
24 | | - returncode: int |
25 | | - stdout: str |
26 | | - stderr: str |
27 | | - duration_seconds: float |
28 | | - cancelled: bool |
29 | | - |
30 | | - |
31 | | -@dataclass(frozen=True) |
32 | | -class ToolLaunch: |
33 | | - """Details for a detached graphical tool launch.""" |
34 | | - |
35 | | - command: tuple[str, ...] |
36 | | - pid: int |
37 | | - |
38 | | - |
39 | | -def split_arguments(value: str, *, windows: bool | None = None) -> list[str]: |
40 | | - """Split an editable argument template without passing it through a shell.""" |
41 | | - use_windows_rules = os.name == "nt" if windows is None else windows |
42 | | - arguments = shlex.split(value, posix=not use_windows_rules) |
43 | | - if use_windows_rules: |
44 | | - return [ |
45 | | - argument[1:-1] |
46 | | - if len(argument) >= 2 and argument[0] == argument[-1] == '"' |
47 | | - else argument |
48 | | - for argument in arguments |
49 | | - ] |
50 | | - return arguments |
51 | | - |
52 | | - |
53 | | -def format_command(command: Iterable[str], *, windows: bool | None = None) -> str: |
54 | | - """Format an argument vector for display only.""" |
55 | | - values = list(command) |
56 | | - use_windows_rules = os.name == "nt" if windows is None else windows |
57 | | - return subprocess.list2cmdline(values) if use_windows_rules else shlex.join(values) |
58 | | - |
59 | | - |
60 | | -class ExternalToolRunner: |
61 | | - """Run one selected executable at a time without shell interpretation.""" |
62 | | - |
63 | | - def __init__(self) -> None: |
64 | | - self._lock = threading.Lock() |
65 | | - self._process: subprocess.Popen[str] | None = None |
66 | | - self._cancel_requested = False |
67 | | - |
68 | | - def build_command( |
69 | | - self, |
70 | | - executable: str | Path, |
71 | | - argument_template: Iterable[str], |
72 | | - *, |
73 | | - input_path: str | Path | None = None, |
74 | | - output_path: str | Path | None = None, |
75 | | - input_kind: str = "file", |
76 | | - output_kind: str = "optional", |
77 | | - ) -> tuple[str, ...]: |
78 | | - tool = Path(executable).expanduser().resolve() |
79 | | - if not tool.is_file(): |
80 | | - raise ExternalToolError(f"Tool executable was not found: {tool}") |
81 | | - |
82 | | - source = self._resolve_input(input_path, input_kind) |
83 | | - output = self._resolve_output(output_path, output_kind) |
84 | | - arguments: list[str] = [] |
85 | | - for value in argument_template: |
86 | | - if "{input}" in value and source is None: |
87 | | - raise ExternalToolError("This command requires an input file") |
88 | | - if "{output}" in value and output is None: |
89 | | - raise ExternalToolError("This command requires an output path") |
90 | | - arguments.append( |
91 | | - value.replace("{input}", str(source) if source else "") |
92 | | - .replace("{output}", str(output) if output else "") |
93 | | - ) |
94 | | - return (str(tool), *arguments) |
95 | | - |
96 | | - def launch_detached( |
97 | | - self, |
98 | | - executable: str | Path, |
99 | | - argument_template: Iterable[str] = (), |
100 | | - *, |
101 | | - input_path: str | Path | None = None, |
102 | | - output_path: str | Path | None = None, |
103 | | - input_kind: str = "none", |
104 | | - output_kind: str = "none", |
105 | | - ) -> ToolLaunch: |
106 | | - """Launch a GUI utility without waiting for it to exit.""" |
107 | | - command = self.build_command( |
108 | | - executable, |
109 | | - argument_template, |
110 | | - input_path=input_path, |
111 | | - output_path=output_path, |
112 | | - input_kind=input_kind, |
113 | | - output_kind=output_kind, |
114 | | - ) |
115 | | - creation_flags = 0 |
116 | | - if os.name == "nt": |
117 | | - creation_flags = subprocess.CREATE_NEW_PROCESS_GROUP |
118 | | - try: |
119 | | - process = subprocess.Popen( |
120 | | - command, |
121 | | - cwd=Path(command[0]).parent, |
122 | | - shell=False, |
123 | | - creationflags=creation_flags, |
124 | | - close_fds=os.name != "nt", |
125 | | - ) |
126 | | - except OSError as exc: |
127 | | - raise ExternalToolError(f"Could not start external tool: {exc}") from exc |
128 | | - return ToolLaunch(command, process.pid) |
129 | | - |
130 | | - def run( |
131 | | - self, |
132 | | - executable: str | Path, |
133 | | - argument_template: Iterable[str], |
134 | | - *, |
135 | | - input_path: str | Path | None = None, |
136 | | - output_path: str | Path | None = None, |
137 | | - timeout: float = 300, |
138 | | - input_kind: str = "file", |
139 | | - output_kind: str = "optional", |
140 | | - ) -> ToolResult: |
141 | | - command = self.build_command( |
142 | | - executable, |
143 | | - argument_template, |
144 | | - input_path=input_path, |
145 | | - output_path=output_path, |
146 | | - input_kind=input_kind, |
147 | | - output_kind=output_kind, |
148 | | - ) |
149 | | - source = self._resolve_input(input_path, input_kind) |
150 | | - working_directory = source.parent if source else Path(command[0]).parent |
151 | | - creation_flags = subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0 |
152 | | - started = time.monotonic() |
153 | | - |
154 | | - with self._lock: |
155 | | - if self._process is not None: |
156 | | - raise ExternalToolError("Another external tool is already running") |
157 | | - self._cancel_requested = False |
158 | | - try: |
159 | | - self._process = subprocess.Popen( |
160 | | - command, |
161 | | - cwd=working_directory, |
162 | | - stdout=subprocess.PIPE, |
163 | | - stderr=subprocess.PIPE, |
164 | | - text=True, |
165 | | - shell=False, |
166 | | - creationflags=creation_flags, |
167 | | - ) |
168 | | - except OSError as exc: |
169 | | - raise ExternalToolError(f"Could not start external tool: {exc}") from exc |
170 | | - process = self._process |
171 | | - |
172 | | - try: |
173 | | - stdout, stderr = process.communicate(timeout=max(1, timeout)) |
174 | | - except subprocess.TimeoutExpired as exc: |
175 | | - process.kill() |
176 | | - stdout, stderr = process.communicate() |
177 | | - raise ExternalToolError( |
178 | | - f"External tool exceeded the {timeout:g}-second timeout" |
179 | | - ) from exc |
180 | | - finally: |
181 | | - with self._lock: |
182 | | - cancelled = self._cancel_requested |
183 | | - self._process = None |
184 | | - |
185 | | - return ToolResult( |
186 | | - command, |
187 | | - process.returncode, |
188 | | - stdout, |
189 | | - stderr, |
190 | | - time.monotonic() - started, |
191 | | - cancelled, |
192 | | - ) |
193 | | - |
194 | | - def cancel(self) -> bool: |
195 | | - """Terminate the active process, returning whether one was running.""" |
196 | | - with self._lock: |
197 | | - if self._process is None: |
198 | | - return False |
199 | | - self._cancel_requested = True |
200 | | - self._process.terminate() |
201 | | - return True |
202 | | - |
203 | | - @staticmethod |
204 | | - def _resolve_input(value: str | Path | None, kind: str = "file") -> Path | None: |
205 | | - if kind not in {"file", "directory", "any", "optional", "none"}: |
206 | | - raise ExternalToolError(f"Unsupported input path kind: {kind}") |
207 | | - if kind == "none": |
208 | | - return None |
209 | | - if value is None or not str(value).strip(): |
210 | | - if kind not in {"none", "optional"}: |
211 | | - raise ExternalToolError("This command requires an input path") |
212 | | - return None |
213 | | - path = Path(value).expanduser().resolve() |
214 | | - if kind == "file" and not path.is_file(): |
215 | | - raise ExternalToolError(f"Input file was not found: {path}") |
216 | | - if kind == "directory" and not path.is_dir(): |
217 | | - raise ExternalToolError(f"Input folder was not found: {path}") |
218 | | - if kind in {"any", "optional"} and not path.exists(): |
219 | | - raise ExternalToolError(f"Input path was not found: {path}") |
220 | | - return path |
221 | | - |
222 | | - @staticmethod |
223 | | - def _resolve_output(value: str | Path | None, kind: str = "file") -> Path | None: |
224 | | - if kind not in {"file", "directory", "optional", "none"}: |
225 | | - raise ExternalToolError(f"Unsupported output path kind: {kind}") |
226 | | - if kind == "none": |
227 | | - return None |
228 | | - if value is None or not str(value).strip(): |
229 | | - if kind not in {"none", "optional"}: |
230 | | - raise ExternalToolError("This command requires an output path") |
231 | | - return None |
232 | | - path = Path(value).expanduser().resolve() |
233 | | - if kind == "directory" and not path.is_dir(): |
234 | | - raise ExternalToolError(f"Output folder was not found: {path}") |
235 | | - if kind != "directory" and not path.parent.is_dir(): |
236 | | - raise ExternalToolError(f"Output folder was not found: {path.parent}") |
237 | | - return path |
| 5 | +from unityscraper.domains.tools.models import ToolLaunch, ToolResult |
| 6 | +from unityscraper.domains.tools.runner import ( |
| 7 | + ExternalToolError, |
| 8 | + ExternalToolRunner, |
| 9 | + format_command, |
| 10 | + split_arguments, |
| 11 | +) |
| 12 | + |
| 13 | +__all__ = [ |
| 14 | + "ExternalToolError", |
| 15 | + "ExternalToolRunner", |
| 16 | + "ToolLaunch", |
| 17 | + "ToolResult", |
| 18 | + "format_command", |
| 19 | + "split_arguments", |
| 20 | +] |
0 commit comments