lambdagent/agentexample/agent67/tools/file_manager.py

"""
agent67.tools.file_manager — 文件管理器

Lambda: λ action_json. file_op(action)
"""
import json
import os
import shlex
import subprocess
from datetime import datetime
from pathlib import Path

from .safety import ask_user_confirm


def manage_file(input_json: str) -> str:
    """
    文件系统操作。

    支持: read / write / list / search / info / mkdir / move / delete / tree
    """
    try:
        data = json.loads(input_json)
    except json.JSONDecodeError:
        return f"❌ 无效的 JSON 输入: {input_json}"

    action = data.get("action", "")
    path = os.path.expanduser(data.get("path", ""))

    try:
        if action == "read":
            p = Path(path)
            if not p.exists():
                return f"❌ 文件不存在: {path}"
            content = p.read_text(encoding="utf-8", errors="replace")
            if len(content) > 5000:
                return content[:5000] + f"\n... (文件共 {len(content)} 字符，已截断)"
            return content

        elif action == "write":
            content = data.get("content", "")
            p = Path(path)
            p.parent.mkdir(parents=True, exist_ok=True)
            p.write_text(content, encoding="utf-8")
            return f"✅ 已写入 {path} ({len(content)} 字符)"

        elif action == "list":
            p = Path(path) if path else Path.home()
            if not p.exists():
                return f"❌ 目录不存在: {path}"
            items = sorted(p.iterdir(), key=lambda x: (not x.is_dir(), x.name))
            lines = []
            for item in items[:50]:
                prefix = "📁" if item.is_dir() else "📄"
                size = ""
                if item.is_file():
                    s = item.stat().st_size
                    size = f" ({s} B)" if s < 1024 else f" ({s // 1024} KB)"
                lines.append(f"  {prefix} {item.name}{size}")
            header = f"📂 {p} ({len(items)} 项)"
            if len(items) > 50:
                header += " (只显示前50项)"
            return header + "\n" + "\n".join(lines)

        elif action == "search":
            pattern = data.get("pattern", "*")
            p = Path(path) if path else Path.home()
            matches = list(p.rglob(pattern))[:20]
            if not matches:
                return f"未找到匹配 '{pattern}' 的文件"
            return f"找到 {len(matches)} 个匹配:\n" + "\n".join(
                f"  {m}" for m in matches
            )

        elif action == "info":
            p = Path(path)
            if not p.exists():
                return f"❌ 路径不存在: {path}"
            stat = p.stat()
            return (
                f"📋 {p.name}\n"
                f"  类型: {'目录' if p.is_dir() else '文件'}\n"
                f"  大小: {stat.st_size:,} bytes\n"
                f"  修改: {datetime.fromtimestamp(stat.st_mtime).strftime('%Y-%m-%d %H:%M:%S')}\n"
                f"  权限: {oct(stat.st_mode)[-3:]}"
            )

        elif action == "mkdir":
            Path(path).mkdir(parents=True, exist_ok=True)
            return f"✅ 目录已创建: {path}"

        elif action == "move":
            src = os.path.expanduser(data.get("src", ""))
            dst = os.path.expanduser(data.get("dst", ""))
            if not os.path.exists(src):
                return f"❌ 源路径不存在: {src}"
            os.rename(src, dst)
            return f"✅ 已移动: {src} → {dst}"

        elif action == "delete":
            if not ask_user_confirm(f"删除: {path}"):
                return "❌ 用户取消了删除操作"
            p = Path(path)
            if p.is_file():
                p.unlink()
                return f"✅ 文件已删除: {path}"
            elif p.is_dir():
                import shutil
                shutil.rmtree(path)
                return f"✅ 目录已删除: {path}"
            return f"❌ 路径不存在: {path}"

        elif action == "tree":
            depth = data.get("depth", 2)
            result = subprocess.run(
                f"find {shlex.quote(path)} -maxdepth {depth} -type f -o -type d",
                shell=True, capture_output=True, text=True, timeout=10,
            )
            return result.stdout.strip()[:3000] or f"目录为空: {path}"

        else:
            return f"❌ 未知操作: {action}. 支持: read/write/list/search/info/mkdir/move/delete/tree"

    except PermissionError:
        return f"🔒 权限不足: {path}"
    except Exception as e:
        return f"❌ 错误: {e}"