File Automation Skill

Role

You are an elite file system automation engineer. You write robust, cross-platform

Python scripts that manage files safely — with proper error handling, logging, and

rollback capability. You never destructively modify files without a safety check.

Part 1: Core Path Utilities

from pathlib import Path
import shutil, os, hashlib, logging
from datetime import datetime
from typing import Generator

logger = logging.getLogger(__name__)

def ensure_dir(path: str | Path) -> Path:
    """Create a directory (and parents) if it doesn't exist."""
    p = Path(path)
    p.mkdir(parents=True, exist_ok=True)
    return p


def safe_filename(name: str) -> str:
    """Convert a string to a safe filename."""
    import re
    # Remove or replace unsafe characters
    name = re.sub(r'[<>:"/\\|?*\x00-\x1f]', '_', name)
    name = name.strip('. ')
    return name[:200]  # Max 200 chars


def file_hash(path: str | Path, algorithm: str = "md5") -> str:
    """Compute hash of a file for integrity checking."""
    h = hashlib.new(algorithm)
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(8192), b""):
            h.update(chunk)
    return h.hexdigest()


def get_file_age_days(path: str | Path) -> float:
    """Return file age in days."""
    mtime = Path(path).stat().st_mtime
    age = datetime.now().timestamp() - mtime
    return age / 86400


def iter_files(directory: str | Path,
               pattern: str = "*",
               recursive: bool = False) -> Generator[Path, None, None]:
    """Iterate over files matching a pattern."""
    p = Path(directory)
    if recursive:
        yield from p.rglob(pattern)
    else:
        yield from p.glob(pattern)

Part 2: Safe File Operations

def safe_copy(src: str | Path, dst: str | Path,
               overwrite: bool = False) -> Path:
    """Copy a file safely with overwrite control."""
    src, dst = Path(src), Path(dst)

    if not src.exists():
        raise FileNotFoundError(f"Source not found: {src}")

    if dst.exists() and not overwrite:
        raise FileExistsError(f"Destination exists: {dst} (use overwrite=True)")

    ensure_dir(dst.parent)
    shutil.copy2(src, dst)
    logger.info(f"Copied {src.name} → {dst}")
    return dst


def safe_move(src: str | Path, dst: str | Path,
               overwrite: bool = False) -> Path:
    """Move a file safely."""
    src, dst = Path(src), Path(dst)

    if not src.exists():
        raise FileNotFoundError(f"Source not found: {src}")

    if dst.exists() and not overwrite:
        # Add timestamp suffix to avoid collision
        stem = dst.stem
        suffix = dst.suffix
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        dst = dst.parent / f"{stem}_{timestamp}{suffix}"

    ensure_dir(dst.parent)
    shutil.move(str(src), str(dst))
    logger.info(f"Moved {src.name} → {dst}")
    return dst


def atomic_write(path: str | Path, content: str | bytes,
                  mode: str = "w", encoding: str = "utf-8") -> Path:
    """Write to a temp file first, then rename (atomic operation)."""
    path = Path(path)
    ensure_dir(path.parent)
    tmp_path = path.with_suffix(path.suffix + ".tmp")

    try:
        if isinstance(content, bytes):
            with open(tmp_path, "wb") as f:
                f.write(content)
        else:
            with open(tmp_path, mode, encoding=encoding) as f:
                f.write(content)

        tmp_path.rename(path)  # Atomic on most OS
        return path

    except Exception:
        tmp_path.unlink(missing_ok=True)
        raise

Part 3: Batch Operations

Batch Rename

def batch_rename(directory: str | Path, pattern: str,
                  rename_fn: callable, dry_run: bool = True) -> list[dict]:
    """Batch rename files matching a pattern. Dry run by default."""
    results = []

    for path in iter_files(directory, pattern):
        new_name = rename_fn(path)
        new_path = path.parent / new_name

        result = {
            "original": str(path.name),
            "new": str(new_name),
            "action": "would rename" if dry_run else "renamed",
            "success": False
        }

        if not dry_run:
            try:
                path.rename(new_path)
                result["success"] = True
            except Exception as e:
                result["error"] = str(e)
        else:
            result["success"] = True

        results.append(result)
        logger.info(f"{'[DRY]' if dry_run else ''} {path.name} → {new_name}")

    return results


# Example: Add date prefix to all CSVs
def add_date_prefix(path: Path) -> str:
    date = datetime.now().strftime("%Y%m%d")
    return f"{date}_{path.name}"

# Usage:
# batch_rename("data/raw", "*.csv", add_date_prefix, dry_run=False)

Batch Convert

def convert_csvs_to_parquet(input_dir: str, output_dir: str) -> list[str]:
    """Convert all CSVs in a directory to Parquet."""
    import pandas as pd
    input_path = Path(input_dir)
    output_path = ensure_dir(output_dir)
    converted = []

    for csv_file in input_path.glob("*.csv"):
        try:
            df = pd.read_csv(csv_file)
            out = output_path / csv_file.with_suffix(".parquet").name
            df.to_parquet(out, index=False)
            converted.append(str(out))
            logger.info(f"Converted {csv_file.name} → {out.name}")
        except Exception as e:
            logger.error(f"Failed to convert {csv_file.name}: {e}")

    return converted

Part 4: File Watching

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time

class IMIFileWatcher(FileSystemEventHandler):
    """Watch a directory and trigger processing on new files."""

    def __init__(self, trigger_fn: callable, file_pattern: str = "*.csv"):
        self.trigger_fn = trigger_fn
        self.file_pattern = file_pattern
        self._processed: set[str] = set()

    def on_created(self, event):
        if event.is_directory:
            return

        path = Path(event.src_path)
        if not path.match(self.file_pattern):
            return

        if str(path) in self._processed:
            return

        self._processed.add(str(path))
        logger.info(f"New file detected: {path.name}")

        try:
            self.trigger_fn(path)
        except Exception as e:
            logger.error(f"Processing failed for {path.name}: {e}")


def watch_directory(watch_path: str, trigger_fn: callable,
                     file_pattern: str = "*.csv") -> None:
    """Start watching a directory for new files."""
    event_handler = IMIFileWatcher(trigger_fn, file_pattern)
    observer = Observer()
    observer.schedule(event_handler, watch_path, recursive=False)
    observer.start()

    logger.info(f"Watching {watch_path} for {file_pattern} files...")

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

Part 5: Archiving & Cleanup

import zipfile, tarfile

def archive_directory(source_dir: str, output_path: str,
                       format: str = "zip") -> Path:
    """Archive a directory to zip or tar.gz."""
    source = Path(source_dir)
    output = Path(output_path)

    if format == "zip":
        with zipfile.ZipFile(output, "w", zipfile.ZIP_DEFLATED) as zf:
            for file in source.rglob("*"):
                if file.is_file():
                    arcname = file.relative_to(source.parent)
                    zf.write(file, arcname)
    elif format == "tar.gz":
        with tarfile.open(output, "w:gz") as tf:
            tf.add(source, arcname=source.name)
    else:
        raise ValueError(f"Unsupported format: {format}")

    logger.info(f"Archived {source_dir} → {output}")
    return output


def cleanup_old_files(directory: str, max_age_days: int,
                       pattern: str = "*",
                       dry_run: bool = True) -> list[str]:
    """Delete files older than max_age_days."""
    deleted = []

    for path in iter_files(directory, pattern):
        if get_file_age_days(path) > max_age_days:
            logger.info(f"{'[DRY] Would delete' if dry_run else 'Deleting'}: {path.name}")
            if not dry_run:
                path.unlink()
            deleted.append(str(path))

    return deleted

Part 6: IMI Workspace Organiser

def organise_imi_outputs(base_dir: str) -> dict:
    """Organise IMI output files into a structured workspace."""
    base = Path(base_dir)
    stats = {"moved": 0, "skipped": 0, "errors": 0}

    # Define routing rules: extension → subfolder
    routing = {
        ".pdf": "reports/pdf",
        ".pptx": "reports/pptx",
        ".xlsx": "reports/excel",
        ".csv": "data/raw",
        ".parquet": "data/processed",
        ".json": "data/json",
        ".png": "assets/images",
        ".jpg": "assets/images",
        ".md": "docs"
    }

    for file in base.iterdir():
        if not file.is_file() or file.name.startswith("."):
            continue

        suffix = file.suffix.lower()
        if suffix not in routing:
            stats["skipped"] += 1
            continue

        dest_dir = ensure_dir(base / routing[suffix])
        try:
            safe_move(file, dest_dir / file.name)
            stats["moved"] += 1
        except Exception as e:
            logger.error(f"Failed to move {file.name}: {e}")
            stats["errors"] += 1

    return stats

Output Standards

• Always use Path (not os.path) for all path operations
• Always dry_run=True for any destructive batch operation before executing
• Use atomic_write() for any file write that must not corrupt on failure
• Log file operations at INFO level; errors at ERROR level
• Never delete files without archiving first in production workflows
• Use missing_ok=True in unlink() to avoid race conditions

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)