384 lines
11 KiB
Python
384 lines
11 KiB
Python
"""File operation tools"""
|
|
import os
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Optional, Tuple
|
|
from backend.tools.factory import tool
|
|
from backend import db
|
|
from backend.models import Project
|
|
from backend.utils.workspace import get_project_path, validate_path_in_project
|
|
|
|
|
|
def _resolve_path(path: str, project_id: str = None) -> Tuple[Path, Path]:
|
|
"""
|
|
Resolve path and ensure it's within project directory
|
|
|
|
Args:
|
|
path: File path (relative or absolute)
|
|
project_id: Project ID for workspace isolation
|
|
|
|
Returns:
|
|
Tuple of (resolved absolute path, project directory)
|
|
|
|
Raises:
|
|
ValueError: If project_id is missing or path is outside project
|
|
"""
|
|
if not project_id:
|
|
raise ValueError("project_id is required for file operations")
|
|
|
|
# Get project from database
|
|
project = db.session.get(Project, project_id)
|
|
if not project:
|
|
raise ValueError(f"Project not found: {project_id}")
|
|
|
|
# Get project directory
|
|
project_dir = get_project_path(project.id, project.path)
|
|
|
|
# Validate and resolve path
|
|
return validate_path_in_project(path, project_dir), project_dir
|
|
|
|
|
|
@tool(
|
|
name="file_read",
|
|
description="Read content from a file within the project workspace. Use when you need to read file content.",
|
|
parameters={
|
|
"type": "object",
|
|
"properties": {
|
|
"path": {
|
|
"type": "string",
|
|
"description": "File path to read (relative to project root or absolute within project)"
|
|
},
|
|
"encoding": {
|
|
"type": "string",
|
|
"description": "File encoding, default utf-8",
|
|
"default": "utf-8"
|
|
}
|
|
},
|
|
"required": ["path"]
|
|
},
|
|
category="file"
|
|
)
|
|
def file_read(arguments: dict) -> dict:
|
|
"""
|
|
Read file tool
|
|
|
|
Args:
|
|
arguments: {
|
|
"path": "file.txt",
|
|
"project_id": "project-uuid",
|
|
"encoding": "utf-8"
|
|
}
|
|
|
|
Returns:
|
|
{"success": true, "content": "...", "size": 100}
|
|
"""
|
|
try:
|
|
path, project_dir = _resolve_path(arguments["path"], arguments.get("project_id"))
|
|
encoding = arguments.get("encoding", "utf-8")
|
|
|
|
if not path.exists():
|
|
return {"success": False, "error": f"File not found: {path}"}
|
|
|
|
if not path.is_file():
|
|
return {"success": False, "error": f"Path is not a file: {path}"}
|
|
|
|
content = path.read_text(encoding=encoding)
|
|
|
|
return {
|
|
"success": True,
|
|
"content": content,
|
|
"size": len(content),
|
|
"path": str(path.relative_to(project_dir))
|
|
}
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
|
|
@tool(
|
|
name="file_write",
|
|
description="Write content to a file within the project workspace. Creates the file if it doesn't exist, overwrites if it does. Use when you need to create or update a file.",
|
|
parameters={
|
|
"type": "object",
|
|
"properties": {
|
|
"path": {
|
|
"type": "string",
|
|
"description": "File path to write (relative to project root or absolute within project)"
|
|
},
|
|
"content": {
|
|
"type": "string",
|
|
"description": "Content to write to the file"
|
|
},
|
|
"encoding": {
|
|
"type": "string",
|
|
"description": "File encoding, default utf-8",
|
|
"default": "utf-8"
|
|
},
|
|
"mode": {
|
|
"type": "string",
|
|
"description": "Write mode: 'write' (overwrite) or 'append'",
|
|
"enum": ["write", "append"],
|
|
"default": "write"
|
|
}
|
|
},
|
|
"required": ["path", "content"]
|
|
},
|
|
category="file"
|
|
)
|
|
def file_write(arguments: dict) -> dict:
|
|
"""
|
|
Write file tool
|
|
|
|
Args:
|
|
arguments: {
|
|
"path": "file.txt",
|
|
"content": "Hello World",
|
|
"project_id": "project-uuid",
|
|
"encoding": "utf-8",
|
|
"mode": "write"
|
|
}
|
|
|
|
Returns:
|
|
{"success": true, "size": 11}
|
|
"""
|
|
try:
|
|
path, project_dir = _resolve_path(arguments["path"], arguments.get("project_id"))
|
|
content = arguments["content"]
|
|
encoding = arguments.get("encoding", "utf-8")
|
|
mode = arguments.get("mode", "write")
|
|
|
|
# Create parent directories if needed
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Write or append
|
|
if mode == "append":
|
|
with open(path, "a", encoding=encoding) as f:
|
|
f.write(content)
|
|
else:
|
|
path.write_text(content, encoding=encoding)
|
|
|
|
return {
|
|
"success": True,
|
|
"size": len(content),
|
|
"path": str(path.relative_to(project_dir)),
|
|
"mode": mode
|
|
}
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
|
|
@tool(
|
|
name="file_delete",
|
|
description="Delete a file within the project workspace. Use when you need to remove a file.",
|
|
parameters={
|
|
"type": "object",
|
|
"properties": {
|
|
"path": {
|
|
"type": "string",
|
|
"description": "File path to delete (relative to project root or absolute within project)"
|
|
}
|
|
},
|
|
"required": ["path"]
|
|
},
|
|
category="file"
|
|
)
|
|
def file_delete(arguments: dict) -> dict:
|
|
"""
|
|
Delete file tool
|
|
|
|
Args:
|
|
arguments: {
|
|
"path": "file.txt",
|
|
"project_id": "project-uuid"
|
|
}
|
|
|
|
Returns:
|
|
{"success": true}
|
|
"""
|
|
try:
|
|
path, project_dir = _resolve_path(arguments["path"], arguments.get("project_id"))
|
|
|
|
if not path.exists():
|
|
return {"success": False, "error": f"File not found: {path}"}
|
|
|
|
if not path.is_file():
|
|
return {"success": False, "error": f"Path is not a file: {path}"}
|
|
|
|
rel_path = str(path.relative_to(project_dir))
|
|
path.unlink()
|
|
return {"success": True, "path": rel_path}
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
|
|
@tool(
|
|
name="file_list",
|
|
description="List files and directories in a directory within the project workspace. Use when you need to see what files exist.",
|
|
parameters={
|
|
"type": "object",
|
|
"properties": {
|
|
"path": {
|
|
"type": "string",
|
|
"description": "Directory path to list (relative to project root or absolute within project)",
|
|
"default": "."
|
|
},
|
|
"pattern": {
|
|
"type": "string",
|
|
"description": "Glob pattern to filter files, e.g. '*.py'",
|
|
"default": "*"
|
|
}
|
|
},
|
|
"required": []
|
|
},
|
|
category="file"
|
|
)
|
|
def file_list(arguments: dict) -> dict:
|
|
"""
|
|
List directory contents
|
|
|
|
Args:
|
|
arguments: {
|
|
"path": ".",
|
|
"pattern": "*",
|
|
"project_id": "project-uuid"
|
|
}
|
|
|
|
Returns:
|
|
{"success": true, "files": [...], "directories": [...]}
|
|
"""
|
|
try:
|
|
path, project_dir = _resolve_path(arguments.get("path", "."), arguments.get("project_id"))
|
|
pattern = arguments.get("pattern", "*")
|
|
|
|
if not path.exists():
|
|
return {"success": False, "error": f"Directory not found: {path}"}
|
|
|
|
if not path.is_dir():
|
|
return {"success": False, "error": f"Path is not a directory: {path}"}
|
|
|
|
files = []
|
|
directories = []
|
|
|
|
for item in path.glob(pattern):
|
|
if item.is_file():
|
|
files.append({
|
|
"name": item.name,
|
|
"size": item.stat().st_size,
|
|
"path": str(item.relative_to(project_dir))
|
|
})
|
|
elif item.is_dir():
|
|
directories.append({
|
|
"name": item.name,
|
|
"path": str(item.relative_to(project_dir))
|
|
})
|
|
|
|
return {
|
|
"success": True,
|
|
"path": str(path.relative_to(project_dir)),
|
|
"files": files,
|
|
"directories": directories,
|
|
"total_files": len(files),
|
|
"total_dirs": len(directories)
|
|
}
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
|
|
@tool(
|
|
name="file_exists",
|
|
description="Check if a file or directory exists within the project workspace. Use when you need to verify file existence.",
|
|
parameters={
|
|
"type": "object",
|
|
"properties": {
|
|
"path": {
|
|
"type": "string",
|
|
"description": "Path to check (relative to project root or absolute within project)"
|
|
}
|
|
},
|
|
"required": ["path"]
|
|
},
|
|
category="file"
|
|
)
|
|
def file_exists(arguments: dict) -> dict:
|
|
"""
|
|
Check if file/directory exists
|
|
|
|
Args:
|
|
arguments: {
|
|
"path": "file.txt",
|
|
"project_id": "project-uuid"
|
|
}
|
|
|
|
Returns:
|
|
{"exists": true, "type": "file"}
|
|
"""
|
|
try:
|
|
path, project_dir = _resolve_path(arguments["path"], arguments.get("project_id"))
|
|
|
|
if not path.exists():
|
|
return {"exists": False, "path": str(path.relative_to(project_dir))}
|
|
|
|
if path.is_file():
|
|
return {
|
|
"exists": True,
|
|
"type": "file",
|
|
"path": str(path.relative_to(project_dir)),
|
|
"size": path.stat().st_size
|
|
}
|
|
elif path.is_dir():
|
|
return {
|
|
"exists": True,
|
|
"type": "directory",
|
|
"path": str(path.relative_to(project_dir))
|
|
}
|
|
else:
|
|
return {
|
|
"exists": True,
|
|
"type": "other",
|
|
"path": str(path.relative_to(project_dir))
|
|
}
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
|
|
@tool(
|
|
name="file_mkdir",
|
|
description="Create a directory within the project workspace. Creates parent directories if needed. Use when you need to create a folder.",
|
|
parameters={
|
|
"type": "object",
|
|
"properties": {
|
|
"path": {
|
|
"type": "string",
|
|
"description": "Directory path to create (relative to project root or absolute within project)"
|
|
}
|
|
},
|
|
"required": ["path"]
|
|
},
|
|
category="file"
|
|
)
|
|
def file_mkdir(arguments: dict) -> dict:
|
|
"""
|
|
Create directory
|
|
|
|
Args:
|
|
arguments: {
|
|
"path": "new/folder",
|
|
"project_id": "project-uuid"
|
|
}
|
|
|
|
Returns:
|
|
{"success": true}
|
|
"""
|
|
try:
|
|
path, project_dir = _resolve_path(arguments["path"], arguments.get("project_id"))
|
|
|
|
created = not path.exists()
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
|
|
return {
|
|
"success": True,
|
|
"path": str(path.relative_to(project_dir)),
|
|
"created": created
|
|
}
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|