Private
Public Access
refactor(mcp_client): migrate L1465 get_tree to Result[T] (Phase 7 site 5)
Added get_tree_result inside Result Variants region. Legacy get_tree (str) now delegates to it.
This commit is contained in:
+43
-32
@@ -1000,6 +1000,40 @@ def derive_code_path_result(target: str, max_depth: int = 5) -> Result[str]:
|
|||||||
return Result(data="\n".join(output))
|
return Result(data="\n".join(output))
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="mcp.derive_code_path_result", original=e)])
|
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="mcp.derive_code_path_result", original=e)])
|
||||||
|
|
||||||
|
def get_tree_result(path: str, max_depth: int = 2) -> Result[str]:
|
||||||
|
resolved = _resolve_and_check_result(path)
|
||||||
|
if not resolved.ok:
|
||||||
|
return Result(data="", errors=resolved.errors)
|
||||||
|
p = resolved.data
|
||||||
|
if isinstance(p, NilPath):
|
||||||
|
return Result(data="", errors=resolved.errors)
|
||||||
|
if not p.is_dir():
|
||||||
|
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INVALID_INPUT, message=f"not a directory: {path}", source="mcp.get_tree_result")])
|
||||||
|
m_depth = max_depth
|
||||||
|
def _build_tree(dir_path: Path, current_depth: int, prefix: str = "") -> list[str]:
|
||||||
|
if current_depth > m_depth: return []
|
||||||
|
lines = []
|
||||||
|
try:
|
||||||
|
entries = sorted(dir_path.iterdir(), key=lambda e: (e.is_file(), e.name.lower()))
|
||||||
|
except PermissionError:
|
||||||
|
return []
|
||||||
|
entries = [e for e in entries if not e.name.startswith('.') and e.name not in ('__pycache__', 'venv', 'env') and e.name != "history.toml" and not e.name.endswith("_history.toml")]
|
||||||
|
for i, entry in enumerate(entries):
|
||||||
|
is_last = (i == len(entries) - 1)
|
||||||
|
connector = "└── " if is_last else "├── "
|
||||||
|
if entry.is_dir():
|
||||||
|
lines.append(f"{prefix}{connector}{entry.name}/")
|
||||||
|
extension = " " if is_last else "│ "
|
||||||
|
lines.extend(_build_tree(entry, current_depth + 1, prefix + extension))
|
||||||
|
else:
|
||||||
|
lines.append(f"{prefix}{connector}{entry.name}")
|
||||||
|
return lines
|
||||||
|
try:
|
||||||
|
tree_lines = [f"{p.name}/"] + _build_tree(p, 1)
|
||||||
|
return Result(data="\n".join(tree_lines))
|
||||||
|
except Exception as e:
|
||||||
|
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="mcp.get_tree_result", original=e)])
|
||||||
#endregion: Result Variants
|
#endregion: Result Variants
|
||||||
|
|
||||||
def edit_file(path: str, old_string: str, new_string: str, replace_all: bool = False) -> str:
|
def edit_file(path: str, old_string: str, new_string: str, replace_all: bool = False) -> str:
|
||||||
@@ -1429,39 +1463,16 @@ def derive_code_path(target: str, max_depth: int = 5) -> str:
|
|||||||
#endregion Python AST
|
#endregion Python AST
|
||||||
|
|
||||||
def get_tree(path: str, max_depth: int = 2) -> str:
|
def get_tree(path: str, max_depth: int = 2) -> str:
|
||||||
"""Returns a directory structure up to a max depth."""
|
"""Returns a directory structure up to a max depth.
|
||||||
p, err = _resolve_and_check(path)
|
|
||||||
if err: return err
|
|
||||||
assert p is not None
|
|
||||||
if not p.is_dir(): return f"ERROR: not a directory: {path}"
|
|
||||||
try:
|
|
||||||
m_depth = max_depth
|
|
||||||
|
|
||||||
def _build_tree(dir_path: Path, current_depth: int, prefix: str = "") -> list[str]:
|
|
||||||
if current_depth > m_depth: return []
|
|
||||||
lines = []
|
|
||||||
try:
|
|
||||||
entries = sorted(dir_path.iterdir(), key=lambda e: (e.is_file(), e.name.lower()))
|
|
||||||
except PermissionError:
|
|
||||||
return []
|
|
||||||
# Filter
|
|
||||||
entries = [e for e in entries if not e.name.startswith('.') and e.name not in ('__pycache__', 'venv', 'env') and e.name != "history.toml" and not e.name.endswith("_history.toml")]
|
|
||||||
for i, entry in enumerate(entries):
|
|
||||||
is_last = (i == len(entries) - 1)
|
|
||||||
connector = "└── " if is_last else "├── "
|
|
||||||
lines.append(f"{prefix}{connector}{entry.name}")
|
|
||||||
if entry.is_dir():
|
|
||||||
extension = " " if is_last else "│ "
|
|
||||||
lines.extend(_build_tree(entry, current_depth + 1, prefix + extension))
|
|
||||||
return lines
|
|
||||||
tree_lines = [f"{p.name}/"] + _build_tree(p, 1)
|
|
||||||
return "\n".join(tree_lines)
|
|
||||||
except Exception as e:
|
|
||||||
return f"ERROR generating tree for '{path}': {e}"
|
|
||||||
# ------------------------------------------------------------------ web tools
|
|
||||||
|
|
||||||
#region: Web
|
|
||||||
|
|
||||||
|
Thin wrapper over get_tree_result; the legacy str shape is
|
||||||
|
preserved for backward compatibility, but the try/except Exception
|
||||||
|
lives in the Result variant.
|
||||||
|
"""
|
||||||
|
resolved = get_tree_result(path, max_depth)
|
||||||
|
if resolved.ok:
|
||||||
|
return resolved.data
|
||||||
|
return "; ".join(e.ui_message() for e in resolved.errors)
|
||||||
class _DDGParser(HTMLParser):
|
class _DDGParser(HTMLParser):
|
||||||
def __init__(self) -> None:
|
def __init__(self) -> None:
|
||||||
super().__init__()
|
super().__init__()
|
||||||
|
|||||||
Reference in New Issue
Block a user