To run this:

- `pip3 install -r requirements.txt`

- `python3 controller.py --help`

**Output:**

- For example, watching the path `E:\watchdog` recursively for log and text files:

```

import datetime

from pygtail import Pygtail

import re

class FileChecker:

def __init__(self, exceptionPattern):

self.exceptionPattern = exceptionPattern

def checkForException(self, event, path):

# Get current date and time according to the specified format.

now = (datetime.datetime.now()).strftime("%Y-%m-%d %H:%M:%S")

# Read the lines of the file (specified in the path) that have not been read yet

# Meaning by that it will start from the point where it was last stopped.

for num, line in enumerate(Pygtail(path), 1):

# Remove leading and trailing whitespaces including newlines.

line = line.strip()

# Return all non-overlapping matches of the values specified in the Exception Pattern.

# The line is scanned from left to right and matches are returned in the oder found.

if line and any(re.findall('|'.join(self.exceptionPattern), line, flags=re.I | re.X)):

# Observation Detected

type = 'observation'

msg = f"{now} -- {event.event_type} -- File: {path} -- Observation: {line}"

yield type, msg

elif line:

# No Observation Detected

type = 'msg'

msg = f"{now} -- {event.event_type} -- File: {path}"

yield type, msg

WATCH_DIRECTORY = "C:\\SCRIPTS"

WATCH_DELAY = 1

WATCH_RECURSIVELY = False

DO_WATCH_DIRECTORIES = True

WATCH_PATTERN = '.txt,.trc,.log'

LOG_FILES_EXTENSIONS = ('.txt', '.log', '.trc')

EXCEPTION_PATTERN = ['EXCEPTION', 'FATAL', 'ERROR']

from watchdog.observers import Observer

from watchdog.events import FileSystemEventHandler

import time

import config

import os

from checker import FileChecker

import datetime

from colorama import Fore, Style, init

init()

GREEN = Fore.GREEN

BLUE = Fore.BLUE

RED = Fore.RED

YELLOW = Fore.YELLOW

event2color = {

"created": GREEN,

"modified": BLUE,

"deleted": RED,

"moved": YELLOW,

}

def print_with_color(s, color=Fore.WHITE, brightness=Style.NORMAL, **kwargs):

"""Utility function wrapping the regular `print()` function

print(f"{brightness}{color}{s}{Style.RESET_ALL}", **kwargs)

class LogHandler(FileSystemEventHandler):

def __init__(self, watchPattern, exceptionPattern, doWatchDirectories):

self.watchPattern = watchPattern

self.exceptionPattern = exceptionPattern

self.doWatchDirectories = doWatchDirectories

# Instantiate the checker

self.fc = FileChecker(self.exceptionPattern)

def on_any_event(self, event):

now = (datetime.datetime.now()).strftime("%Y-%m-%d %H:%M:%S")

# print("event happened:", event)

# To Observe files only not directories

if not event.is_directory:

# To cater for the on_move event

path = event.src_path

if hasattr(event, 'dest_path'):

path = event.dest_path

# Ensure that the file extension is among the pre-defined ones.

if path.endswith(self.watchPattern):

msg = f"{now} -- {event.event_type} -- File: {path}"

if event.event_type in ('modified', 'created', 'moved'):

# check for exceptions in log files

if path.endswith(config.LOG_FILES_EXTENSIONS):

for type, msg in self.fc.checkForException(event=event, path=path):

print_with_color(

msg, color=event2color[event.event_type], brightness=Style.BRIGHT)

else:

print_with_color(

msg, color=event2color[event.event_type])

else:

print_with_color(msg, color=event2color[event.event_type])

elif self.doWatchDirectories:

msg = f"{now} -- {event.event_type} -- Folder: {event.src_path}"

print_with_color(msg, color=event2color[event.event_type])

def on_modified(self, event):

pass

def on_deleted(self, event):

pass

def on_created(self, event):

pass

def on_moved(self, event):

pass

class LogWatcher:

# Initialize the observer

observer = None

# Initialize the stop signal variable

stop_signal = 0

# The observer is the class that watches for any file system change and then dispatches the event to the event handler.

def __init__(self, watchDirectory, watchDelay, watchRecursively, watchPattern, doWatchDirectories, exceptionPattern):

# Initialize variables in relation

self.watchDirectory = watchDirectory

self.watchDelay = watchDelay

self.watchRecursively = watchRecursively

self.watchPattern = watchPattern

self.doWatchDirectories = doWatchDirectories

self.exceptionPattern = exceptionPattern

# Create an instance of watchdog.observer

self.observer = Observer()

# The event handler is an object that will be notified when something happens to the file system.

self.event_handler = LogHandler(

watchPattern, exceptionPattern, self.doWatchDirectories)

def schedule(self):

print("Observer Scheduled:", self.observer.name)

# Call the schedule function via the Observer instance attaching the event

self.observer.schedule(

self.event_handler, self.watchDirectory, recursive=self.watchRecursively)

def start(self):

print("Observer Started:", self.observer.name)

self.schedule()

# Start the observer thread and wait for it to generate events

now = (datetime.datetime.now()).strftime("%Y-%m-%d %H:%M:%S")

msg = f"Observer: {self.observer.name} - Started On: {now}"

print(msg)

msg = (

f"Watching {'Recursively' if self.watchRecursively else 'Non-Recursively'}: {self.watchPattern}"

f" -- Folder: {self.watchDirectory} -- Every: {self.watchDelay}(sec) -- For Patterns: {self.exceptionPattern}"

)

print(msg)

self.observer.start()

def run(self):

print("Observer is running:", self.observer.name)

self.start()

try:

while True:

time.sleep(self.watchDelay)

if self.stop_signal == 1:

print(

f"Observer stopped: {self.observer.name} stop signal:{self.stop_signal}")

self.stop()

break

except:

self.stop()

self.observer.join()

def stop(self):

print("Observer Stopped:", self.observer.name)

now = (datetime.datetime.now()).strftime("%Y-%m-%d %H:%M:%S")

msg = f"Observer: {self.observer.name} - Stopped On: {now}"

print(msg)

self.observer.stop()

self.observer.join()

def info(self):

info = {

'observerName': self.observer.name,

'watchDirectory': self.watchDirectory,

'watchDelay': self.watchDelay,

'watchRecursively': self.watchRecursively,

'watchPattern': self.watchPattern,

}

return info

def is_dir_path(path):

"""Utility function to check whether a path is an actual directory"""

if os.path.isdir(path):

return path

else:

raise NotADirectoryError(path)

if __name__ == "__main__":

import argparse

parser = argparse.ArgumentParser(

description="Watchdog script for watching for files & directories' changes")

parser.add_argument("path",

default=config.WATCH_DIRECTORY,

type=is_dir_path,

)

parser.add_argument("-d", "--watch-delay",

help=f"Watch delay, default is {config.WATCH_DELAY}",

default=config.WATCH_DELAY,

type=int,

)

parser.add_argument("-r", "--recursive",

action="store_true",

help=f"Whether to recursively watch for the path's children, default is {config.WATCH_RECURSIVELY}",

default=config.WATCH_RECURSIVELY,

)

parser.add_argument("-p", "--pattern",

help=f"Pattern of files to watch, default is {config.WATCH_PATTERN}",

default=config.WATCH_PATTERN,

)

parser.add_argument("--watch-directories",

action="store_true",

help=f"Whether to watch directories, default is {config.DO_WATCH_DIRECTORIES}",

default=config.DO_WATCH_DIRECTORIES,

)

# parse the arguments

args = parser.parse_args()

# define & launch the log watcher

log_watcher = LogWatcher(

watchDirectory=args.path,

watchDelay=args.watch_delay,

watchRecursively=args.recursive,

watchPattern=tuple(args.pattern.split(",")),

doWatchDirectories=args.watch_directories,

exceptionPattern=config.EXCEPTION_PATTERN,

)

log_watcher.run()