99 lines
3.4 KiB
Python
99 lines
3.4 KiB
Python
|
import subprocess
|
|||
|
|
import time
|
|||
|
|
import os
|
|||
|
|
from pathlib import Path
|
|||
|
|
from typing import Dict, Any
|
|||
|
|
|
|||
|
|
import yaml
|
|||
|
|
|
|||
|
|
from .models import RunRequest, RunResult
|
|||
|
|
|
|||
|
|
|
|||
|
|
def load_config(config_path: str | Path = None) -> Dict[str, Any]:
|
|||
|
|
"""加载配置，默认使用 AGENT_CONFIG 或 ./config.yaml。"""
|
|||
|
|
if config_path is None:
|
|||
|
|
env_path = os.environ.get("AGENT_CONFIG")
|
|||
|
|
config_path = Path(env_path) if env_path else Path(__file__).resolve().parent.parent / "config.yaml"
|
|||
|
|
config_path = Path(config_path)
|
|||
|
|
with open(config_path, "r", encoding="utf-8") as f:
|
|||
|
|
return yaml.safe_load(f) or {}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def resolve_tool(config: Dict[str, Any], tool_name: str) -> Path:
|
|||
|
|
tools = config.get("tools", {})
|
|||
|
|
if tool_name not in tools:
|
|||
|
|
raise ValueError(f"未知工具: {tool_name}")
|
|||
|
|
tool_path = Path(tools[tool_name]["path"])
|
|||
|
|
if not tool_path.exists():
|
|||
|
|
raise FileNotFoundError(f"工具不存在: {tool_path}")
|
|||
|
|
return tool_path
|
|||
|
|
|
|||
|
|
|
|||
|
|
def run_tool(config: Dict[str, Any], req: RunRequest) -> RunResult:
|
|||
|
|
start = time.time()
|
|||
|
|
defaults = config.get("run_defaults", {})
|
|||
|
|
|
|||
|
|
tool_path = resolve_tool(config, req.tool)
|
|||
|
|
workdir = Path(req.workdir or config.get("base_workdir", ".")).resolve()
|
|||
|
|
workdir.mkdir(parents=True, exist_ok=True)
|
|||
|
|
|
|||
|
|
# 1. 先构建参数列表 (List)
|
|||
|
|
args_list = [subprocess.list2cmdline([str(tool_path)])] + req.args
|
|||
|
|
|
|||
|
|
# 2. 【关键修改】转换为安全的命令行字符串
|
|||
|
|
cmd_str = " ".join(args_list)
|
|||
|
|
|
|||
|
|
# 处理环境
|
|||
|
|
env = os.environ.copy()
|
|||
|
|
if config.get("env"):
|
|||
|
|
env.update({k: str(v) for k, v in config["env"].items()})
|
|||
|
|
if req.env:
|
|||
|
|
env.update({k: str(v) for k, v in req.env.items()})
|
|||
|
|
|
|||
|
|
timeout = req.timeout
|
|||
|
|
capture_output = req.capture_output if req.capture_output is not None else defaults.get("capture_output", True)
|
|||
|
|
strip_output = req.strip_output if req.strip_output is not None else defaults.get("strip_output", True)
|
|||
|
|
|
|||
|
|
print(f"Running tool in {workdir} :: timeout={timeout}s command:")
|
|||
|
|
#print(f"Command: {cmd_str}") # 现在的打印结果是完全真实的执行指令
|
|||
|
|
print(cmd_str)
|
|||
|
|
try:
|
|||
|
|
# 3. 执行：传入字符串 cmd_str
|
|||
|
|
completed = subprocess.run(
|
|||
|
|
cmd_str, # <--- 这里现在是字符串
|
|||
|
|
cwd=str(workdir),
|
|||
|
|
env=env,
|
|||
|
|
timeout=timeout,
|
|||
|
|
capture_output=capture_output,
|
|||
|
|
text=True,
|
|||
|
|
check=False,
|
|||
|
|
shell=False # Windows 上执行 exe 字符串通常不需要 shell=True，除非用到管道/重定向
|
|||
|
|
)
|
|||
|
|
stdout = completed.stdout if capture_output else None
|
|||
|
|
stderr = completed.stderr if capture_output else None
|
|||
|
|
if strip_output:
|
|||
|
|
stdout = stdout.strip() if stdout is not None else None
|
|||
|
|
stderr = stderr.strip() if stderr is not None else None
|
|||
|
|
|
|||
|
|
elapsed = time.time() - start
|
|||
|
|
return RunResult(
|
|||
|
|
success=completed.returncode == 0,
|
|||
|
|
return_code=completed.returncode,
|
|||
|
|
stdout=stdout,
|
|||
|
|
stderr=stderr,
|
|||
|
|
elapsed=elapsed,
|
|||
|
|
tool_path=str(tool_path),
|
|||
|
|
workdir=str(workdir),
|
|||
|
|
)
|
|||
|
|
except Exception as exc:
|
|||
|
|
elapsed = time.time() - start
|
|||
|
|
return RunResult(
|
|||
|
|
success=False,
|
|||
|
|
return_code=None,
|
|||
|
|
stdout=None,
|
|||
|
|
stderr=str(exc),
|
|||
|
|
elapsed=elapsed,
|
|||
|
|
tool_path=str(tool_path),
|
|||
|
|
workdir=str(workdir),
|
|||
|
|
)
|