cyclop/tools/log_tool.py

# tools/log_tool.py

from .base_tool import BaseTool
import logging
import os
from datetime import datetime, timedelta

class LogTool(BaseTool):
    def __init__(self):
        # Set up logging
        self.logger = logging.getLogger(__name__)
        self.logger.setLevel(logging.INFO)
        
        # Create a file handler
        file_handler = logging.FileHandler('log_tool.log')
        file_handler.setLevel(logging.INFO)
        
        # Create a console handler
        console_handler = logging.StreamHandler()
        console_handler.setLevel(logging.INFO)
        
        # Create a formatting for the logs
        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        file_handler.setFormatter(formatter)
        console_handler.setFormatter(formatter)
        
        # Add the handlers to the logger
        self.logger.addHandler(file_handler)
        self.logger.addHandler(console_handler)

    def clear(self):
        pass

    def get_functions(self):
        return [
            {
                "name": "get_log_contents",
                "description": "Get the contents of the log file.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "line_count": {
                            "type": "integer",
                            "description": "Number of lines from the end of the log file to retrieve"
                        }
                    },
                    "required": []
                }
            }
        ]

    def execute(self, function_name, **kwargs):
        self.logger.info(f"Executing: {function_name}")
        
        if function_name == "get_log_contents":
            return self._get_log_contents(kwargs.get("line_count"))
        else:
            error_message = f"Unknown function: {function_name}"
            self.logger.error(error_message)
            return error_message

    def _get_log_contents(self, line_count=None):
        log_file_path = 'logs/output.log'
        
        if not os.path.exists(log_file_path):
            error_message = "Log file does not exist."
            self.logger.error(error_message)
            return error_message

        try:
            with open(log_file_path, 'r') as log_file:
                log_lines = log_file.readlines()

                if line_count is not None:
                    log_lines = log_lines[-line_count:]
                else:
                    now = datetime.now()
                    twenty_four_hours_ago = now - timedelta(days=1)
                    log_lines = [line for line in log_lines if datetime.strptime(line.split(' - ')[0], '%Y-%m-%d %H:%M:%S,%f') > twenty_four_hours_ago]

                return "".join(log_lines)
        except Exception as e:
            error_message = f"An error occurred while reading the log file: {e}"
            self.logger.error(error_message)
            return error_message