mirror of
https://gitee.com/wanwujie/deer-flow
synced 2026-04-24 14:44:46 +08:00
feat: add clarification feature (#13)
This commit is contained in:
@@ -42,9 +42,102 @@ class LocalSandbox(Sandbox):
|
||||
# No mapping found, return original path
|
||||
return path_str
|
||||
|
||||
def _reverse_resolve_path(self, path: str) -> str:
|
||||
"""
|
||||
Reverse resolve local path back to container path using mappings.
|
||||
|
||||
Args:
|
||||
path: Local path that might need to be mapped to container path
|
||||
|
||||
Returns:
|
||||
Container path if mapping exists, otherwise original path
|
||||
"""
|
||||
path_str = str(Path(path).resolve())
|
||||
|
||||
# Try each mapping (longest local path first for more specific matches)
|
||||
for container_path, local_path in sorted(self.path_mappings.items(), key=lambda x: len(x[1]), reverse=True):
|
||||
local_path_resolved = str(Path(local_path).resolve())
|
||||
if path_str.startswith(local_path_resolved):
|
||||
# Replace the local path prefix with container path
|
||||
relative = path_str[len(local_path_resolved) :].lstrip("/")
|
||||
resolved = f"{container_path}/{relative}" if relative else container_path
|
||||
return resolved
|
||||
|
||||
# No mapping found, return original path
|
||||
return path_str
|
||||
|
||||
def _reverse_resolve_paths_in_output(self, output: str) -> str:
|
||||
"""
|
||||
Reverse resolve local paths back to container paths in output string.
|
||||
|
||||
Args:
|
||||
output: Output string that may contain local paths
|
||||
|
||||
Returns:
|
||||
Output with local paths resolved to container paths
|
||||
"""
|
||||
import re
|
||||
|
||||
# Sort mappings by local path length (longest first) for correct prefix matching
|
||||
sorted_mappings = sorted(self.path_mappings.items(), key=lambda x: len(x[1]), reverse=True)
|
||||
|
||||
if not sorted_mappings:
|
||||
return output
|
||||
|
||||
# Create pattern that matches absolute paths
|
||||
# Match paths like /Users/... or other absolute paths
|
||||
result = output
|
||||
for container_path, local_path in sorted_mappings:
|
||||
local_path_resolved = str(Path(local_path).resolve())
|
||||
# Escape the local path for use in regex
|
||||
escaped_local = re.escape(local_path_resolved)
|
||||
# Match the local path followed by optional path components
|
||||
pattern = re.compile(escaped_local + r"(?:/[^\s\"';&|<>()]*)?")
|
||||
|
||||
def replace_match(match: re.Match) -> str:
|
||||
matched_path = match.group(0)
|
||||
return self._reverse_resolve_path(matched_path)
|
||||
|
||||
result = pattern.sub(replace_match, result)
|
||||
|
||||
return result
|
||||
|
||||
def _resolve_paths_in_command(self, command: str) -> str:
|
||||
"""
|
||||
Resolve container paths to local paths in a command string.
|
||||
|
||||
Args:
|
||||
command: Command string that may contain container paths
|
||||
|
||||
Returns:
|
||||
Command with container paths resolved to local paths
|
||||
"""
|
||||
import re
|
||||
|
||||
# Sort mappings by length (longest first) for correct prefix matching
|
||||
sorted_mappings = sorted(self.path_mappings.items(), key=lambda x: len(x[0]), reverse=True)
|
||||
|
||||
# Build regex pattern to match all container paths
|
||||
# Match container path followed by optional path components
|
||||
if not sorted_mappings:
|
||||
return command
|
||||
|
||||
# Create pattern that matches any of the container paths
|
||||
patterns = [re.escape(container_path) + r"(?:/[^\s\"';&|<>()]*)??" for container_path, _ in sorted_mappings]
|
||||
pattern = re.compile("|".join(f"({p})" for p in patterns))
|
||||
|
||||
def replace_match(match: re.Match) -> str:
|
||||
matched_path = match.group(0)
|
||||
return self._resolve_path(matched_path)
|
||||
|
||||
return pattern.sub(replace_match, command)
|
||||
|
||||
def execute_command(self, command: str) -> str:
|
||||
# Resolve container paths in command before execution
|
||||
resolved_command = self._resolve_paths_in_command(command)
|
||||
|
||||
result = subprocess.run(
|
||||
command,
|
||||
resolved_command,
|
||||
executable="/bin/zsh",
|
||||
shell=True,
|
||||
capture_output=True,
|
||||
@@ -56,11 +149,16 @@ class LocalSandbox(Sandbox):
|
||||
output += f"\nStd Error:\n{result.stderr}" if output else result.stderr
|
||||
if result.returncode != 0:
|
||||
output += f"\nExit Code: {result.returncode}"
|
||||
return output if output else "(no output)"
|
||||
|
||||
final_output = output if output else "(no output)"
|
||||
# Reverse resolve local paths back to container paths in output
|
||||
return self._reverse_resolve_paths_in_output(final_output)
|
||||
|
||||
def list_dir(self, path: str, max_depth=2) -> list[str]:
|
||||
resolved_path = self._resolve_path(path)
|
||||
return list_dir(resolved_path, max_depth)
|
||||
entries = list_dir(resolved_path, max_depth)
|
||||
# Reverse resolve local paths back to container paths in output
|
||||
return [self._reverse_resolve_paths_in_output(entry) for entry in entries]
|
||||
|
||||
def read_file(self, path: str) -> str:
|
||||
resolved_path = self._resolve_path(path)
|
||||
|
||||
Reference in New Issue
Block a user