| 1 | from helpers.api import ApiHandler, Input, Output, Request |
| 2 | from helpers.file_browser import FileBrowser |
| 3 | from helpers import runtime, extension |
| 4 | from api import get_work_dir_files |
| 5 | import posixpath |
| 6 | |
| 7 | |
| 8 | class RenameWorkDirFile(ApiHandler): |
| 9 | async def process(self, input: Input, request: Request) -> Output: |
| 10 | try: |
| 11 | action = input.get("action", "rename") |
| 12 | current_path = input.get("currentPath", "") |
| 13 | |
| 14 | if action == "move": |
| 15 | file_paths = input.get("paths", []) |
| 16 | destination_path = input.get("destinationPath", "") |
| 17 | if not isinstance(file_paths, list) or not all( |
| 18 | isinstance(path, str) and path for path in file_paths |
| 19 | ): |
| 20 | return {"error": "Paths are required"} |
| 21 | if not isinstance(destination_path, str) or not destination_path: |
| 22 | return {"error": "Destination path is required"} |
| 23 | file_paths = [ |
| 24 | path if path.startswith("/") else f"/{path}" |
| 25 | for path in file_paths |
| 26 | ] |
| 27 | if not destination_path.startswith("/"): |
| 28 | destination_path = f"/{destination_path}" |
| 29 | moved_paths = await runtime.call_development_function( |
| 30 | move_items, file_paths, destination_path |
| 31 | ) |
| 32 | res = bool(moved_paths) |
| 33 | changed_paths = [*file_paths, *moved_paths] |
| 34 | elif action in {"rename", "create-folder"}: |
| 35 | new_name = (input.get("newName", "") or "").strip() |
| 36 | if not new_name: |
| 37 | return {"error": "New name is required"} |
| 38 | |
| 39 | if action == "create-folder": |
| 40 | parent_path = input.get("parentPath", current_path) |
| 41 | if not parent_path: |
| 42 | return {"error": "Parent path is required"} |
| 43 | res = await runtime.call_development_function( |
| 44 | create_folder, parent_path, new_name |
| 45 | ) |
| 46 | changed_paths = [ |
| 47 | posixpath.join(str(parent_path).rstrip("/"), new_name) |
| 48 | ] |
| 49 | else: |
| 50 | file_path = input.get("path", "") |
| 51 | if not file_path: |
| 52 | return {"error": "Path is required"} |
| 53 | if not file_path.startswith("/"): |
| 54 | file_path = f"/{file_path}" |
| 55 | res = await runtime.call_development_function( |
| 56 | rename_item, file_path, new_name |
| 57 | ) |
| 58 | changed_paths = [ |
| 59 | file_path, |
| 60 | posixpath.join(posixpath.dirname(file_path), new_name), |
| 61 | ] |
| 62 | else: |
| 63 | return {"error": "Unsupported file operation"} |
| 64 | |
| 65 | if res: |
| 66 | await extension.call_extensions_async( |
| 67 | "workdir_file_mutation_after", |
| 68 | agent=None, |
| 69 | data={ |
| 70 | "action": action, |
| 71 | "path": changed_paths[-1], |
| 72 | "paths": changed_paths, |
| 73 | "current_path": current_path, |
| 74 | }, |
| 75 | ) |
| 76 | result = await runtime.call_development_function( |
| 77 | get_work_dir_files.get_files, current_path |
| 78 | ) |
| 79 | return {"data": result} |
| 80 | |
| 81 | error_msg = { |
| 82 | "create-folder": "Failed to create folder", |
| 83 | "move": "Move failed", |
| 84 | }.get(action, "Rename failed") |
| 85 | return {"error": error_msg} |
| 86 | |
| 87 | except Exception as e: |
| 88 | return {"error": str(e)} |
| 89 | |
| 90 | |
| 91 | async def rename_item(file_path: str, new_name: str) -> bool: |
| 92 | browser = FileBrowser() |
| 93 | return browser.rename_item(file_path, new_name) |
| 94 | |
| 95 | |
| 96 | async def create_folder(parent_path: str, folder_name: str) -> bool: |
| 97 | browser = FileBrowser() |
| 98 | return browser.create_folder(parent_path, folder_name) |
| 99 | |
| 100 | |
| 101 | async def move_items(file_paths: list[str], destination_path: str) -> list[str]: |
| 102 | browser = FileBrowser() |
| 103 | return browser.move_items(file_paths, destination_path) |