from __future__ import annotations import asyncio from functools import partial from pathlib import Path from typing import Any from app.agents.tools.types import RegisteredTool, ToolResult PROJECT_ROOT = Path(__file__).resolve().parents[5] def _validate_path(raw_path: str) -> Path: """경로를 검증하고 절대 경로로 변환합니다.""" resolved = (PROJECT_ROOT / raw_path).resolve() if not resolved.is_relative_to(PROJECT_ROOT): raise ValueError(f"프로젝트 외부 경로 접근 불가: {raw_path}") return resolved def _edit_sync(path: Path, old_string: str, new_string: str) -> str: content = path.read_text(encoding="utf-8") count = content.count(old_string) if count == 0: raise ValueError("old_string을 파일에서 찾을 수 없습니다") if count > 1: raise ValueError(f"old_string이 {count}회 발견됨 — 고유한 문자열을 제공해 주세요") updated = content.replace(old_string, new_string, 1) path.write_text(updated, encoding="utf-8") return "파일 수정 완료" def create_edit_file_tool() -> RegisteredTool: """파일 수정 도구를 생성합니다.""" async def execute(params: dict[str, Any]) -> ToolResult: raw_path: str = params["path"] old_string: str = params["old_string"] new_string: str = params["new_string"] try: path = _validate_path(raw_path) except ValueError as e: return ToolResult(data=f"오류: {e}") if not path.exists(): return ToolResult(data=f"파일 없음: {raw_path}") if not path.is_file(): return ToolResult(data=f"파일이 아님: {raw_path}") loop = asyncio.get_running_loop() try: result = await loop.run_in_executor( None, partial(_edit_sync, path, old_string, new_string) ) except ValueError as e: return ToolResult(data=f"수정 실패: {e}") except Exception as e: return ToolResult(data=f"파일 수정 실패: {e}") return ToolResult(data=result) return RegisteredTool( name="edit_file", description="파일의 특정 부분을 수정합니다. 정확한 old_string을 제공해야 합니다.", compact_description="파일 수정", concurrency_safe=False, execute=execute, )