File size: 3,540 Bytes
acf77ab | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | from __future__ import annotations
import shutil
import tempfile
from datetime import UTC, datetime
from pathlib import Path
from typing import TYPE_CHECKING
from codeforge.sandbox.imports import scan_imports
from codeforge.sandbox.metric import composite_score
from codeforge.sandbox.models import (
ImportReport,
ParsedResult,
SandboxResult,
ToolResult,
)
from codeforge.sandbox.runner import run_tool
from codeforge.sandbox.tools import (
DEFAULT_TOOLS,
argv_for,
is_available,
parse,
)
if TYPE_CHECKING:
from collections.abc import Iterable
def run_sandbox(
*,
project_dir: Path | None = None,
files: dict[str, str] | None = None,
tools: Iterable[str] = DEFAULT_TOOLS,
timeout_per_tool: float = 60.0,
) -> SandboxResult:
if (project_dir is None) == (files is None):
msg = "exactly one of project_dir / files must be set"
raise ValueError(msg)
tmp_root: Path | None = None
try:
if files is not None:
tmp_root = Path(tempfile.mkdtemp(prefix="codeforge_sandbox_"))
tmp_root_resolved = tmp_root.resolve()
for name, content in files.items():
target = (tmp_root / name).resolve()
if not target.is_relative_to(tmp_root_resolved):
msg = f"path escapes sandbox root: {name!r}"
raise ValueError(msg)
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(content, encoding="utf-8")
project_dir = tmp_root
assert project_dir is not None
tool_list = tuple(tools)
tool_results: dict[str, ToolResult] = {}
parsed_results: dict[str, ParsedResult] = {}
imports_report: ImportReport | None = None
for name in tool_list:
if name == "imports":
imports_report = scan_imports(project_dir)
parsed_results[name] = ParsedResult(
ok=len(imports_report.unresolved) == 0,
count=len(imports_report.unresolved),
details={"unresolved": list(imports_report.unresolved)},
)
continue
if not is_available(name):
tool_results[name] = ToolResult(
name=name, argv=(name,), exit_code=-1,
stdout="", stderr="binary not found",
duration_ms=0, timed_out=False,
)
parsed_results[name] = ParsedResult(
ok=False, count=0, details={"unavailable": True},
)
continue
argv = argv_for(name, project_dir)
tr = run_tool(name, argv, cwd=project_dir, timeout=timeout_per_tool)
tool_results[name] = tr
parsed_results[name] = parse(name, tr)
if imports_report is None:
imports_report = ImportReport(total=0, unresolved=(), by_file={})
result = SandboxResult(
project_dir=str(project_dir),
tools_run=tool_list,
tool_results=tool_results,
parsed=parsed_results,
imports=imports_report,
composite_score=0.0,
generated_at=datetime.now(UTC).isoformat(timespec="seconds"),
)
score = composite_score(result)
return result.model_copy(update={"composite_score": score})
finally:
if tmp_root is not None and tmp_root.exists():
shutil.rmtree(tmp_root, ignore_errors=True)
|