Building an Intelligent Log Analyzer in Python with argparse, Ollama, and Mistral AI

ollama pull mistral

mkdir smart-tool
cd smart-tool
python3 -m venv venv
source venv activate

pip install requests

import argparse

def main():
    parser = argparse.ArgumentParser(description='Example CLI')
    parser.add_argument('-n', '--name', help='Your name')
    parser.add_argument('--age', type=int, help='Your age')

    args = parser.parse_args()

    print(f"Hello, {args.name}! You are {args.age} years old.")

if __name__ == '__main__':
    main()

python3 age.py -n Your_name --age Your_age

import argparse
import requests
import json

def analyze_log(file_path, custom_message=None):
    # Default prompt message
    default_message = "Hi. Please, take a look at this log content and find error lines if any. If possible, give some clues on how to resolve these issues."
    prompt_message = custom_message if custom_message else default_message

    # Read the log file
    with open(file_path, 'r') as file:
        log_content = file.read()

    # Prepare the data to be sent to the Ollama API
    payload = {
        "model": "mistral",
        "prompt": f"{log_content} \n\n {prompt_message}",
    }

    # Send the content to Ollama API and handle streaming response
    response = requests.post(
        "http://localhost:11434/api/generate",  # Replace with your Ollama API endpoint
        headers={"Content-Type": "application/json"},
        data=json.dumps(payload),
        stream=True
    )

    # Check if the request was successful
    if response.status_code == 200:
        print("Analysis Result:")
        for line in response.iter_lines():
            if line:
                decoded_line = line.decode('utf-8')
                json_response = json.loads(decoded_line)
                print(json_response["response"], end='', flush=True)
                if json_response.get("done"):
                    break
        print()  # Ensure the final output ends with a newline
    else:
        print(f"Failed to analyze log file. Status code: {response.status_code}")
        print(response.text)

def main():
    parser = argparse.ArgumentParser(description="Intelligent Log Analyzer")
    parser.add_argument('file', type=str, help='Path to the log file')
    parser.add_argument('--message', type=str, help='Custom message to include in the prompt', default=None)

    args = parser.parse_args()
    analyze_log(args.file, args.message)

if __name__ == "__main__":
    main()

python smartlog.py 2023-09-09.log

Analysis Result:
 Based on the given log content, it appears that there is an issue with sending webhooks, specifically that they are returning a `<nil>` value. This could indicate a few different things:

1. The webhook URLs are incorrect or inaccessible. Check that the URLs are valid and reachable from your server. You can try accessing them directly in a web browser to see if you get an error.
2. Authentication is required but not being provided. Make sure that any authentication tokens or keys required by the webhooks are being passed correctly.
3. The data being sent to the webhooks is invalid. Check that the format of the data being sent matches what is expected by the webhook.
4. Network connectivity issues. Check that your server has network connectivity to the webhook servers, and that any firewalls or security groups are configured correctly.
5. Maximum retries reached. This warning indicates that the script is attempting to send the same webhook multiple times (up to a maximum of 5), but it's still failing. You may need to investigate why the webhooks are not being sent successfully, and consider implementing error handling or retry logic in your code.

To help diagnose the issue further, you could add some debug statements to your code to print out more details about what is happening when the webhook is being sent. For example, you could print out the URL being used, any authentication tokens or keys, and any error messages that are being returned. This information could provide additional clues as to what is causing the issue.

python smartlog.py 2023-09-09.log --message "What is the structure of these logs"

Analysis Result:
 These logs are written in a plain text format with each line representing an event or message. The first part of each line is a timestamp indicating when the event occurred. Following the timestamp, there are one or more messages enclosed in parentheses and prefixed by their respective severity level (ERROR or WARNING).

For example:

ERROR - 2023-09-09 22:58:45 - error sending webhook: %!s(<nil>)


The above log entry indicates an ERROR event that occurred on September 9, 2023 at 22:58:45. The message associated with this event is "error sending webhook: ".

Similarly, the following line is a WARNING event that occurred on September 9, 2023 at 23:04:26:


WARNING - 2023-09-09 23:04:26 - maximum retries reached: %!s(int=5)


This warning message indicates that the maximum number of retries (5) have been reached for an operation related to a webhook.

import argparse
import requests
import json
import threading
import time
from colorama import init, Fore, Style

# Initialize colorama
init(autoreset=True)

def analyze_log(file_path, custom_message=None):
    # Default prompt message
    default_message = "Hi. Please, take a look at this log content and find error lines if any. If possible, give some clues on how to resolve these issues."
    prompt_message = custom_message if custom_message else default_message

    # Read the log file
    with open(file_path, 'r') as file:
        log_content = file.read()

    # Prepare the data to be sent to the Ollama API
    payload = {
        "model": "mistral",
        "prompt": f"{prompt_message}\n\n{log_content}"
    }

    # Function to display a loading animation
    def loading_animation(stop_event):
        while not stop_event.is_set():
            for char in "|/-\\":
                print(Fore.YELLOW + f'\rLoading {char}', end='', flush=True)
                time.sleep(0.1)
                if stop_event.is_set():
                    break
        print('\r', end='', flush=True)  # Clear the loading line

    # Event to signal the loading animation to stop
    stop_event = threading.Event()

    # Start the loading animation in a separate thread
    loader_thread = threading.Thread(target=loading_animation, args=(stop_event,))
    loader_thread.start()

    # Send the content to Ollama API and handle streaming response
    try:
        response = requests.post(
            "http://localhost:11434/api/generate",  # Replace with your Ollama API endpoint
            headers={"Content-Type": "application/json"},
            data=json.dumps(payload),
            stream=True
        )

        # Check if the request was successful
        if response.status_code == 200:
            stop_event.set()
            loader_thread.join()
            print(Fore.GREEN + "\nAnalysis Result:")
            for line in response.iter_lines():
                if line:
                    decoded_line = line.decode('utf-8')
                    json_response = json.loads(decoded_line)
                    print(Fore.CYAN + json_response["response"], end='', flush=True)
                    if json_response.get("done"):
                        break
            print()  # Ensure the final output ends with a newline
        else:
            stop_event.set()
            loader_thread.join()
            print(Fore.RED + f"\nFailed to analyze log file. Status code: {response.status_code}")
            print(Fore.RED + response.text)
    except Exception as e:
        stop_event.set()
        loader_thread.join()
        print(Fore.RED + f"\nAn error occurred: {str(e)}")
    finally:
        # Ensure the loading animation stops in case of any unexpected errors
        stop_event.set()
        loader_thread.join()

def main():
    parser = argparse.ArgumentParser(description="Intelligent Log Analyzer")
    parser.add_argument('file', type=str, help='Path to the log file')
    parser.add_argument('--message', type=str, help='Custom message to include in the prompt', default=None)

    args = parser.parse_args()
    analyze_log(args.file, args.message)

if __name__ == "__main__":
    main()