contentmonster/classes/shorethread.py

from classes.config import MonsterConfig
from classes.doghandler import DogHandler
from classes.directory import Directory
from classes.database import Database

from watchdog.observers import Observer

from multiprocessing import Process, Queue
from typing import NoReturn

import time
import os.path


class ShoreThread(Process):
    """Thread handling the discovery of shore-side file changes
    """
    def __init__(self, state: dict) -> None:
        """Create a new ShoreThread object

        Args:
            state (dict): Dictionary containing the application state
        """
        super().__init__()
        self._dogs = []
        self._state = state
        self.queue = Queue()

    def getAllFiles(self) -> list:
        """Return File objects for all files in all Directories

        Returns:
            list: List of all File objects discovered
        """
        files = []

        for directory in self._state["config"].directories:
            for f in directory.getFiles():
                files.append(f)

        return files

    def clearFiles(self) -> None:
        """Clear the files variable in the application state
        """
        del self._state["files"][:]

    def monitor(self) -> None:
        """Initialize monitoring of Directories specified in configuration
        """
        for directory in self._state["config"].directories:
            print("Creating dog for " + str(directory.location))
            handler = DogHandler(directory, self.queue)
            dog = Observer()
            dog.schedule(handler, str(directory.location), False)
            dog.start()
            self._dogs.append(dog)

    def run(self) -> NoReturn:
        """Launch the ShoreThread and start monitoring for file changes
        """
        print("Launched Shore Thread")
        self.getAllFiles()
        self.monitor()

        while True:
            self.joinDogs()
            self.processQueue()

    def joinDogs(self) -> None:
        """Join observers to receive updates on the queue
        """
        for dog in self._dogs:
            dog.join(1)

    def purgeFile(self, directory: Directory, name: str) -> None:
        """Purge a removed File from the processing queue and database

        Args:
            directory (Directory): Directory (previously) containing the File
            name (str): Filename of the deleted File
        """
        # Remove file from processing queue
        for f in self._state["files"]:
            if f.directory == directory and f.name == name:
                del(f)

        # Remove file from database        
        db = Database()
        db.removeFile(directory, name)

    def addFile(self, fileobj):
        """Add a File object to the processing queue, if not already there

        Args:
            fileobj (classes.file.File): File object to add to the queue
        """
        found = False

        for f in self._state["files"]:
            if f.directory == fileobj.directory and f.name == fileobj.name:
                if f.uuid != fileobj.uuid:
                    del(f)
                else:
                    found = True

        if not found:
            # Do not queue file if it is of size 0
            if os.path.getsize(str(fileobj.getFullPath())):
                self._state["files"].append(fileobj)

    def processFile(self, directory: Directory, name: str) -> None:
        """Process a file entry from the observer queue

        Args:
            directory (classes.directory.Directory): Directory containing the
              created, deleted, modified or moved File
            name (str): Filename of the created, deleted, modified or moved 
              File
        """
        if (fileobj := directory.getFile(name)):
            self.addFile(fileobj)
        else:
            self.purgeFile(directory, name)

    def processQueue(self) -> None:
        """Handle events currently on the queue

        N.B.: An event on the queue is a (directory, basename) tuple, where
        "directory" is a Directory object, and "basename" is the name of a
        File that has been created, moved, modified or deleted.
        """
        directory, basename = self.queue.get() # Will block until an event is found
        self.processFile(directory, basename)

    def terminate(self, *args, **kwargs) -> NoReturn:
        """Terminate observer threads, then terminate self
        """
        for dog in self._dogs:
            dog.terminate()
            dog.join()

        super().terminate(*args, **kwargs)