hendrik/tools/coder.py

import os
import subprocess
import re
import glob as glob_module

schema_read_file = {
    "type": "function",
    "function": {
        "name"         : "read_file",
        "description"  : "Read the contents of a file. Returns the file content with line numbers prefixed.",
        "parameters"   : {
            "type"        : "object",
            "properties"  : {
                "path": {"type": "string", "description": "Absolute path to the file to read"}
            },
            "required"    : ["path"]
        }
    }
}

def read_file(path):
    try:
        with open(path, 'r', encoding='utf-8', errors='replace') as f:
            lines = f.readlines()
        return ''.join(f"{i+1}: {line}" for i, line in enumerate(lines))
    except Exception as e:
        return f"Error reading file: {str(e)}"

schema_write_file = {
    "type": "function",
    "function": {
        "name"         : "write_file",
        "description"  : "Write content to a file. Overwrites the file if it exists.",
        "parameters"   : {
            "type"        : "object",
            "properties"  : {
                "path": {"type": "string", "description": "Absolute path to the file to write"},
                "content": {"type": "string", "description": "Content to write to the file"}
            },
            "required"    : ["path", "content"]
        }
    }
}

def write_file(path, content):
    try:
        with open(path, 'w', encoding='utf-8') as f:
            f.write(content)
        return f"Successfully wrote to {path}"
    except Exception as e:
        return f"Error writing file: {str(e)}"

schema_edit_file = {
    "type": "function",
    "function": {
        "name"         : "edit_file",
        "description"  : "Replace old_string with new_string in a file. If old_string is not found, returns error. If multiple matches, replaces all.",
        "parameters"   : {
            "type"        : "object",
            "properties"  : {
                "path": {"type": "string", "description": "Absolute path to the file to edit"},
                "old_string": {"type": "string", "description": "String to replace"},
                "new_string": {"type": "string", "description": "Replacement string"}
            },
            "required"    : ["path", "old_string", "new_string"]
        }
    }
}

def edit_file(path, old_string, new_string):
    try:
        with open(path, 'r', encoding='utf-8', errors='replace') as f:
            content = f.read()
        if old_string not in content:
            return f"Error: old_string not found in {path}"
        new_content = content.replace(old_string, new_string)
        with open(path, 'w', encoding='utf-8') as f:
            f.write(new_content)
        return f"Successfully edited {path}"
    except Exception as e:
        return f"Error editing file: {str(e)}"

schema_run_bash = {
    "type": "function",
    "function": {
        "name"         : "run_bash",
        "description"  : "Run a bash command. Returns stdout, stderr, and return code.",
        "parameters"   : {
            "type"        : "object",
            "properties"  : {
                "command": {"type": "string", "description": "Bash command to execute"}
            },
            "required"    : ["command"]
        }
    }
}

def run_bash(command):
    try:
        result = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=30)
        return f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}\nreturn code: {result.returncode}"
    except Exception as e:
        return f"Error running command: {str(e)}"

schema_search_code = {
    "type": "function",
    "function": {
        "name": "search_code",
        "description": "Search for files or content in files. Use type 'glob' to find files matching a pattern, 'content' to search file contents for a regex pattern.",
        "parameters": {
            "type": "object",
            "properties": {
                "pattern": {"type": "string", "description": "Glob pattern (for type 'glob') or regex pattern (for type 'content')"},
                "path": {"type": "string", "description": "Directory to search in (default: current working directory)", "default": "."},
                "search_type": {"type": "string", "enum": ["glob", "content"], "description": "Type of search: 'glob' for files, 'content' for file contents"}
            },
            "required": ["pattern", "search_type"]
        }
    }
}

def search_code(pattern, search_type, path="."):
    try:
        if search_type == "glob":
            files = glob_module.glob(f"{path}/**/{pattern}", recursive=True)
            return "\n".join(files) if files else "No files found"
        elif search_type == "content":
            results = []
            for root, dirs, files in os.walk(path):
                for file in files:
                    file_path = os.path.join(root, file)
                    try:
                        with open(file_path, 'r', encoding='utf-8', errors='replace') as f:
                            for i, line in enumerate(f.readlines(), 1):
                                if re.search(pattern, line):
                                    results.append(f"{file_path}:{i}: {line.strip()}")
                    except:
                        continue
            return "\n".join(results[:50]) if results else "No matches found"
        else:
            return "Invalid search_type. Use 'glob' or 'content'."
    except Exception as e:
        return f"Error searching: {str(e)}"
    
schema_git_operation = {
    "type": "function",
    "function": {
        "name": "git_operation",
        "description": "Run a git command. Pass the git arguments as a list (e.g., ['status', '--short'] for 'git status --short'). "
                       "POLICY: Never run 'git add' or 'git commit' without explicit user permission. "
                       "Safe to run without asking: git status, git diff, git log. "
                       "Always ask first before committing.",
        "parameters": {
            "type": "object",
            "properties": {
                "args": {"type": "array", "items": {"type": "string"}, "description": "List of git command arguments (without 'git' prefix)"}
            },
            "required": ["args"]
        }
    }
}

def git_operation(args):
    try:
        result = subprocess.run(["git", *args], capture_output=True, text=True, timeout=10)
        return f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}\nreturn code: {result.returncode}"
    except Exception as e:
        return f"Error running git command: {str(e)}"