files.py

from typing import Any
from functools import partial

import logging as log
import os
import re
import subprocess

from autobot.funcs.base import Fn
from autobot.funcs.helpers import canonicalize_path

class CreateFile(Fn):
    def __init__(self):
        super().__init__(
            "create_file",
            "Create a file in the workspace",
            {
                "path": {
                    "type": "string",
                    "description": "Path of file to create relative to the workspace root"
                },
                "content": {
                    "type": "string",
                    "description": "Content of file to create"
                }
            },
            required=["path", "content"])

    def call(self,
             workspace_root: str,
             **params: dict[str, Any]):
        content = params["content"]
        rel_path = params["path"]
        path = canonicalize_path(workspace_root, rel_path)
        log.info("create_file: %s",
                 rel_path)
        print(f"create_file: {rel_path}")
        if os.path.exists(path):
            s = f"Path {rel_path} already exists."
            log.info(s)
            print(s)
            return s
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with open(path, "w", encoding="UTF-8") as f:
            f.write(content)
        s = f"Created file {rel_path}."
        log.info(s)
        print(s)
        return s

class ListFiles(Fn):
    def __init__(self):
        super().__init__(
            "list_files",
            "List files in the workspace",
            {
                "path": {
                    "type": "string",
                    "description": "Path to list files from relative to the workspace root"
                },
                "long": {
                    "type": "boolean",
                    "description": "Whether to list files with more information"
                }
            },
            required=["path"])

    def call(self, workspace_root, **params):
        rel_path = params["path"]
        path = canonicalize_path(workspace_root, rel_path)
        log.info(f"list_files: rel_path={rel_path}")
        log.info(f"list_files: canonical path is: {path}")
        print(f"list_files: {rel_path}")
        if not os.path.exists(path):
            s = f"Path {rel_path} does not exist."
            log.info(s)
            print(s)
            return s
        if not os.path.isdir(path):
            s = f"Path {rel_path} is not a directory."
            log.info(s)
            print(s)
            return s
        files = os.listdir(path)
        if not files:
            s =  f"No files in {rel_path}."
            log.info(s)
            return s
        log.info("Listed files at path: %s",
                 rel_path)
        print(f"Listed files at rel path: {rel_path}")
        print(f"Listed files at full path: {path}")
        path_minus_workspace = path[len(workspace_root):]
        pathed_files = path_files(path_minus_workspace, files)
        print(f"Pathed files: {pathed_files}")
        return "\n".join(map(partial(ls_dash_l, workspace_root), pathed_files))

def path_files(path, files):
    return [os.path.join(path, file)
            for file
            in files]

def ls_dash_l(workspace_root, path):
    #print(f"ls_dash_l: path => {path}")
    if path != "/":
        full_path = f"{workspace_root}/{path}"
    else:
        full_path = workspace_root
    #print(f"ls_dash_l: full_path => {full_path}")
    try:
        result = subprocess.run(["ls", "-l", full_path],
                            stdout=subprocess.PIPE,
                            check=True,
                            )
        output = result.stdout.decode() # "UTF-8") ?
        # TODO check for errors, etc.
        return output
    except subprocess.CalledProcessError as ex:
        print(f"ls_dash_l: error: {ex}")
        exception_to_output = str # TODO
        output = exception_to_output(ex)
        return f"FAILED:\n\n{output}\n"

class RemoveFile(Fn):
    def __init__(self):
        super().__init__(
            "remove_file",
            "Remove a file in the workspace",
            {
                "path": {
                    "type": "string",
                    "description": "Path of file to remove relative to the workspace root"
                }
            },
            required=["path"])

    def call(self, workspace_root, **params):
        rel_path = params["path"]
        path = canonicalize_path(workspace_root, rel_path)
        log.info("remove_file: %s",
                 rel_path)
        print(f"remove_file: {rel_path}")
        if not os.path.exists(path):
            s = f"Path {rel_path} does not exist."
            log.info(s)
            print(s)
            return s
        if not os.path.isfile(path):
            s = f"Path {rel_path} is not a file."
            log.info(s)
            print(s)
            return s
        os.remove(path)
        s = f"Removed file {rel_path}."
        log.info(s)
        print(s)
        return s

class UpdateFile(Fn):
    def __init__(self):
        super().__init__("update_file",
                       "Update a file in the workspace",
                       {
                           "path": {
                               "type": "string",
                               "description": ("Path of file to update relative"
                                               " to the workspace root")
                           },
                            "content": {
                                "type": "string",
                                "description": "New content of file"
                            }
                       },
                       required=["path"])

    def call(self,
             workspace_root: str,
             **params: dict[str, Any]):
        content = params["content"]
        rel_path = params["path"]
        path = canonicalize_path(workspace_root, rel_path)
        log.info("update_file: %s",
                 rel_path)
        print(f"update_file: {rel_path}")
        if not os.path.exists(path):
            s = f"Path {rel_path} does not exist."
            log.info(s)
            print(s)
            return s
        with open(path, "w", encoding="UTF-8") as f:
            f.write(content)
        s = f"Updated file {rel_path}."
        log.info(s)
        print(s)
        return s

class ViewFile(Fn):
    def __init__(self):
        super().__init__(
            "view_file",
            "View a file in the workspace",
            {
                "path": {
                    "type": "string",
                    "description": "Path of file to view relative to the workspace root"
                }
            },
            required=["path"])

    def call(self,
             workspace_root: str,
             **params: dict[str, Any]):
        rel_path: str = params["path"]
        path: str = canonicalize_path(workspace_root, rel_path)
        log.info("view_file: %s",
                 rel_path)
        print(f"view_file: {rel_path}")
        if not os.path.exists(path):
            s = f"Path {rel_path} does not exist."
            log.info(s)
            print(s)
            return s
        if not os.path.isfile(path):
            s = f"Path {rel_path} is not a file."
            log.info(s)
            print(s)
            return s
        with open(path, "r",  encoding="UTF-8") as f:
            content = f .read()
        log.info("Viewed file at path: %s",
                 rel_path)
        print(f"Viewed file at path: {rel_path}")
        return content

class SearchAndReplace(Fn):
    def __init__(self):
        super().__init__(
            "search_and_replace",
            "Search and replace text in a file",
            {
                "path": {
                    "type": "string",
                    "description": "Path of file to search and replace text relative to the workspace root"
                },
                "search_text": {
                    "type": "string",
                    "description": "Text to search for in the file"
                },
                "replace_text": {
                    "type": "string",
                    "description": "Text to replace the search text with in the file"
                }
            },
            required=["path", "search_text", "replace_text"])

    def call(self,
             workspace_root: str,
             **params: dict[str, Any]):
        search_text: str = params["search_text"]
        replace_text: str = params["replace_text"]
        rel_path: str = params["path"]
        path = canonicalize_path(workspace_root, rel_path)
        log.info("search_and_replace: %s",
                 rel_path)
        print(f"search_and_replace: {rel_path}")
        if not os.path.exists(path):
            s = f"Path {rel_path} does not exist."
            log.info(s)
            print(s)
            return s
        with open(path, "r", encoding="UTF-8") as f:
            content = f.read()
        search_text = re.escape(search_text)
        replace_text = re.escape(replace_text)
        new_content = re.sub(search_text, replace_text, content)
        with open(path, "w", encoding="UTF-8") as f:
            f.write(new_content)
        s = f"Replaced '{search_text}' with '{replace_text}' in file {rel_path}."
        log.info(s)
        print(s)
        return s

class AppendToFile(Fn):
    def __init__(self):
        super().__init__(
            "append_to_file",
            "Append content to a file in the workspace",
            {
                "path": {
                    "type": "string",
                    "description": "Path of file to append content to relative to the workspace root"
                },
                "content": {
                    "type": "string",
                    "description": "Content to append to the file"
                }
            },
            required=["path", "content"])

    def call(self,
             workspace_root: str,
             **params: dict[str, Any]):
        content: str = params["content"]
        rel_path: str = params["path"]
        path: str = canonicalize_path(workspace_root, rel_path)
        log.info("append_to_file: %s",
                 rel_path)
        print(f"append_to_file: {rel_path}")
        if not os.path.exists(path):
            s = f"Path {rel_path} does not exist."
            log.info(s)
            print(s)
            return s
        with open(path, "a", encoding="UTF-8") as f:
            f.write(content)
        s = f"Appended content to file {rel_path}."
        log.info(s)
        print(s)
        return s

class PrependToFile(Fn):
    def __init__(self):
        super().__init__(
            "prepend_to_file",
            "Prepend content to a file in the workspace",
            {
                "path": {
                    "type": "string",
                    "description": "Path of file to prepend content to relative to the workspace root"
                },
                "content": {
                    "type": "string",
                    "description": "Content to prepend to the file"
                }
            },
            required=["path", "content"])

    def call(self,
             workspace_root: str,
             **params: dict[str, Any]):
        content = params["content"]
        rel_path = params["path"]
        path = canonicalize_path(workspace_root, rel_path)
        log.info("prepend_to_file: %s",
                 rel_path)
        print(f"prepend_to_file: {rel_path}")
        if not os.path.exists(path):
            s = f"Path {rel_path} does not exist."
            log.info(s)
            print(s)
            return s
        with open(path, "r+", encoding="UTF-8") as f:
            old_content = f.read()
            f.seek(0)
            f.write(content + old_content)
        s = f"Prepended content to file {rel_path}."
        log.info(s)
        print(s)
        return s