toolsets/batch-ops/toolset.py

"""
batch‑ops — Bulk file operations: rename, find‑replace, and file stats.
===========================================================================

Pure‑stdlib utilities for batch processing files and directories —
the repetitive tasks agents do across multiple files.
"""

import os
import re
import shutil
from pathlib import Path

try:
    from toolstore.toolset import tool
except ImportError:
    def tool(fn):
        return fn


# ── batch_rename ───────────────────────────────────────────────────────

@tool
def batch_rename(*, directory: str, pattern: str,
                 replacement: str, dry_run: bool = True,
                 recursive: bool = False) -> dict:
    """Rename files matching a regex pattern, with dry‑run preview.

    Args:
        directory:   Path to the directory.
        pattern:     Regex pattern to match in filenames.
        replacement: Replacement string (can use \\1, \\2 for groups).
        dry_run:     If True, only shows what would change (default True).
        recursive:   If True, descend into subdirectories.

    Returns:
        dict with "changes" list of {old_name, new_name} and "count".
    """
    p = Path(directory).expanduser().resolve()
    if not p.is_dir():
        return {"error": f"Not a directory: {p}"}

    try:
        compiled = re.compile(pattern)
    except re.error as exc:
        return {"error": f"Invalid regex: {exc}"}

    changes = []
    glob_fn = p.rglob if recursive else p.glob

    for fpath in glob_fn("*"):
        if not fpath.is_file():
            continue
        old_name = fpath.name
        new_name = compiled.sub(replacement, old_name)
        if new_name != old_name:
            new_path = fpath.parent / new_name
            if new_path.exists():
                changes.append({"old_name": old_name, "new_name": new_name,
                                "error": "target exists"})
            else:
                changes.append({"old_name": old_name, "new_name": new_name})
                if not dry_run:
                    fpath.rename(new_path)

    return {"changes": changes, "count": len(changes),
            "dry_run": dry_run, "directory": str(p)}


# ── batch_find_replace ─────────────────────────────────────────────────

@tool
def batch_find_replace(*, directory: str, pattern: str,
                       find_text: str, replace_text: str,
                       dry_run: bool = True,
                       recursive: bool = False) -> dict:
    """Find‑and‑replace across multiple text files in a directory.

    Args:
        directory:    Path to the directory.
        pattern:      Glob pattern for files (e.g. "*.py", "*.md").
        find_text:    Text to find (literal, not regex).
        replace_text: Replacement text.
        dry_run:      If True, only reports what would change.
        recursive:    If True, descend into subdirectories.

    Returns:
        dict with "changes" list of {file, matches} and "total_files", "total_matches".
    """
    p = Path(directory).expanduser().resolve()
    if not p.is_dir():
        return {"error": f"Not a directory: {p}"}

    glob_fn = p.rglob if recursive else p.glob
    changes = []
    total_files = 0
    total_matches = 0

    for fpath in glob_fn(pattern):
        if not fpath.is_file():
            continue
        try:
            content = fpath.read_text(encoding="utf-8", errors="replace")
        except Exception:
            continue

        count = content.count(find_text)
        if count > 0:
            total_files += 1
            total_matches += count
            changes.append({"file": str(fpath.relative_to(p)), "matches": count})
            if not dry_run:
                new_content = content.replace(find_text, replace_text)
                fpath.write_text(new_content, encoding="utf-8")

    return {"changes": changes, "total_files": total_files,
            "total_matches": total_matches, "dry_run": dry_run,
            "directory": str(p)}


# ── batch_stats ────────────────────────────────────────────────────────

@tool
def batch_stats(*, directory: str, pattern: str = "*",
                recursive: bool = False) -> dict:
    """Get aggregate file statistics for a directory.

    Args:
        directory: Path to the directory.
        pattern:   Glob pattern (default "*" = all files).
        recursive: If True, include subdirectories.

    Returns:
        dict with: file_count, total_size_bytes, extensions ({ext: count}),
                   largest_file, smallest_file
    """
    p = Path(directory).expanduser().resolve()
    if not p.is_dir():
        return {"error": f"Not a directory: {p}"}

    glob_fn = p.rglob if recursive else p.glob
    files = []
    for fpath in glob_fn(pattern):
        if fpath.is_file():
            files.append(fpath)

    if not files:
        return {"file_count": 0, "total_size_bytes": 0, "extensions": {},
                "largest_file": None, "smallest_file": None,
                "directory": str(p)}

    sizes = [(f, f.stat().st_size) for f in files]
    largest = max(sizes, key=lambda x: x[1])
    smallest = min(sizes, key=lambda x: x[1])

    extensions = {}
    for fpath, _ in sizes:
        ext = fpath.suffix.lower() or "(none)"
        extensions[ext] = extensions.get(ext, 0) + 1

    return {
        "file_count": len(files),
        "total_size_bytes": sum(s for _, s in sizes),
        "extensions": extensions,
        "largest_file": {"name": largest[0].name, "size_bytes": largest[1]},
        "smallest_file": {"name": smallest[0].name, "size_bytes": smallest[1]},
        "directory": str(p),
    }


# ── batch_copy ─────────────────────────────────────────────────────────

@tool
def batch_copy(*, source_dir: str, dest_dir: str,
               pattern: str = "*", recursive: bool = False,
               overwrite: bool = False) -> dict:
    """Copy files matching a pattern from source to destination directory.

    Args:
        source_dir: Source directory.
        dest_dir:   Destination directory.
        pattern:    Glob pattern to match (default "*").
        recursive:  If True, descend into subdirectories.
        overwrite:  If True, overwrite existing files.

    Returns:
        dict with "copied" list of filenames and "count".
    """
    src = Path(source_dir).expanduser().resolve()
    dst = Path(dest_dir).expanduser().resolve()

    if not src.is_dir():
        return {"error": f"Not a directory: {src}"}

    dst.mkdir(parents=True, exist_ok=True)

    glob_fn = src.rglob if recursive else src.glob
    copied = []
    skipped = []

    for fpath in glob_fn(pattern):
        if not fpath.is_file():
            continue
        dest_path = dst / fpath.name
        if dest_path.exists() and not overwrite:
            skipped.append(fpath.name)
            continue
        shutil.copy2(fpath, dest_path)
        copied.append(fpath.name)

    return {"copied": copied, "count": len(copied),
            "skipped": skipped, "dest_dir": str(dst)}