mcp-server/files_mcp/core/fs_ops.py

# coding=utf-8
"""
@Time   :2026/6/30
@Author :tao.chen

Pure file-system operations on already-resolved Path objects. Callers
are responsible for sandbox resolution (see path_guard.resolve_and_check).
No global state, no logging — kept trivially testable.
"""
import os
import shutil
import tempfile
from pathlib import Path

from files_mcp.models import FileEntry, OperationResult

MAX_BYTES = 1_048_576  # 1 MB cap, mirrors read_job_file / update_job_file


def _to_entry(p: Path) -> FileEntry:
    st = p.lstat()  # lstat: do not follow symlink for type detection
    if p.is_symlink():
        kind = "symlink"
    elif p.is_dir():
        kind = "directory"
    elif p.is_file():
        kind = "file"
    else:
        kind = "other"
    return FileEntry(
        path=str(p),
        name=p.name,
        type=kind,
        size=st.st_size if kind == "file" else 0,
        mtime=st.st_mtime,
        mode=st.st_mode,
    )


def _check_size(content: str) -> None:
    if len(content.encode("utf-8")) > MAX_BYTES:
        raise ValueError(
            f"content too large: {len(content)} bytes > {MAX_BYTES}"
        )


def _atomic_write(path: Path, content: str) -> None:
    fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=".tmp_", suffix=path.suffix)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            f.write(content)
        os.replace(tmp, path)
    except BaseException:
        Path(tmp).unlink(missing_ok=True)
        raise


def create(path: Path, content: str, overwrite: bool) -> OperationResult:
    if path.exists() and not overwrite:
        raise FileExistsError(f"file already exists: {path}")
    _check_size(content)
    _atomic_write(path, content)
    return OperationResult(path=str(path), status="CREATED", size=path.stat().st_size)


def read(path: Path) -> dict:
    if not path.exists():
        raise KeyError(f"path not found: {path}")
    if not path.is_file():
        raise ValueError(f"path {path!r} is not a regular file")
    raw = path.read_bytes()
    if len(raw) > MAX_BYTES:
        raise ValueError(f"file too large: {len(raw)} bytes > {MAX_BYTES}")
    try:
        text = raw.decode("utf-8")
    except UnicodeDecodeError as e:
        raise ValueError(f"not valid utf-8: {path}") from e
    return {"path": str(path), "content": text, "size": len(raw), "encoding": "utf-8"}


def update(path: Path, content: str) -> OperationResult:
    if not path.exists():
        raise KeyError(f"path not found: {path}")
    _check_size(content)
    _atomic_write(path, content)
    return OperationResult(path=str(path), status="UPDATED", size=path.stat().st_size)


def delete(path: Path, recursive: bool) -> OperationResult:
    if not path.exists():
        raise KeyError(f"path not found: {path}")
    if path.is_dir() and not recursive:
        # refuse to delete non-empty dirs unless recursive
        if any(path.iterdir()):
            raise ValueError(
                f"directory {path!r} is not empty (use recursive=True)"
            )
    if path.is_dir():
        if recursive:
            shutil.rmtree(path)
        else:
            # empty dir (non-empty case raised above)
            path.rmdir()
    else:
        path.unlink()
    return OperationResult(path=str(path), status="DELETED", size=None)


def list_dir(path: Path, recursive: bool, max_depth: int) -> list[FileEntry]:
    if not path.exists():
        raise KeyError(f"path not found: {path}")
    if not path.is_dir():
        raise ValueError(f"path {path!r} is not a directory")
    if max_depth < 0:
        raise ValueError("max_depth must be >= 0")
    out: list[FileEntry] = []
    if not recursive:
        max_depth = 0
    for cur, dirs, files in os.walk(path):
        rel = Path(cur).relative_to(path)
        depth = 0 if str(rel) == "." else len(rel.parts)
        if depth > max_depth:
            dirs[:] = []  # don't descend further
            continue
        for d in dirs:
            out.append(_to_entry(Path(cur) / d))
        for f in files:
            out.append(_to_entry(Path(cur) / f))
    return out


def stat(path: Path) -> FileEntry:
    if not path.exists() and not path.is_symlink():
        raise KeyError(f"path not found: {path}")
    return _to_entry(path)


def search(path: Path, pattern: str, recursive: bool) -> list[str]:
    if not path.exists():
        raise KeyError(f"path not found: {path}")
    if not path.is_dir():
        raise ValueError(f"path {path!r} is not a directory")
    glob_fn = path.rglob if recursive else path.glob
    return [str(p) for p in glob_fn(pattern)]


def move(src: Path, dst: Path, overwrite: bool) -> OperationResult:
    if not src.exists():
        raise KeyError(f"src not found: {src}")
    if dst.exists() and not overwrite:
        raise FileExistsError(f"dst already exists: {dst}")
    if dst.exists() and overwrite:
        if dst.is_dir() and not dst.is_symlink():
            shutil.rmtree(dst)
        else:
            dst.unlink()
    os.replace(src, dst)
    size = dst.stat().st_size if dst.is_file() else None
    return OperationResult(path=str(dst), status="MOVED", size=size)


def copy(src: Path, dst: Path, overwrite: bool) -> OperationResult:
    if not src.exists():
        raise KeyError(f"src not found: {src}")
    if dst.exists() and not overwrite:
        raise FileExistsError(f"dst already exists: {dst}")
    if src.is_dir():
        if dst.exists() and overwrite:
            shutil.rmtree(dst)
        shutil.copytree(src, dst)
        size = None
    else:
        if dst.exists() and overwrite:
            dst.unlink()
        shutil.copy2(src, dst)
        size = dst.stat().st_size
    return OperationResult(path=str(dst), status="COPIED", size=size)