Mercurial Repositories > Vulture Checker / file revision

# -*- coding: utf-8 -*-

# Copyright (c) 2015 - 2025 Detlev Offenbach <detlev@die-offenbachs.de>
#

"""
Module implementing the cyclomatic complexity service.
"""

import multiprocessing
import queue

from vulture import Vulture


def initService():
    """
    Initialize the service and return the entry point.

    @return the entry point for the background client
    @rtype function
    """
    return vultureCheck


def initBatchService():
    """
    Initialize the batch service and return the entry point.

    @return the entry point for the background client
    @rtype function
    """
    return batchVultureCheck


def vultureCheck(file, text=""):
    """
    Private function to analyze one file.

    @param file source filename
    @type str
    @param text source text
    @type str
    @return tuple containing the result dictionary
    @rtype (tuple of dict)
    """
    return __analyze(file, text)


def batchVultureCheck(argumentsList, send, fx, cancelled, maxProcesses=0):
    """
    Module function to analyze a batch of files.

    @param argumentsList list of arguments tuples as given for vultureCheck
    @type list
    @param send reference to send function
    @type function
    @param fx registered service name
    @type str
    @param cancelled reference to function checking for a cancellation
    @type function
    @param maxProcesses number of processes to be used
    @type int
    """
    if maxProcesses == 0:
        # determine based on CPU count
        try:
            NumberOfProcesses = multiprocessing.cpu_count()
            if NumberOfProcesses >= 1:
                NumberOfProcesses -= 1
        except NotImplementedError:
            NumberOfProcesses = 1
    else:
        NumberOfProcesses = maxProcesses

    # Create queues
    taskQueue = multiprocessing.Queue()
    doneQueue = multiprocessing.Queue()

    # Submit tasks (initially two time number of processes
    initialTasks = 2 * NumberOfProcesses
    for task in argumentsList[:initialTasks]:
        taskQueue.put(task)

    # Start worker processes
    workers = [
        multiprocessing.Process(target=workerTask, args=(taskQueue, doneQueue))
        for _ in range(NumberOfProcesses)
    ]
    for worker in workers:
        worker.start()

    # Get and send results
    endIndex = len(argumentsList) - initialTasks
    for i in range(len(argumentsList)):
        resultSent = False
        wasCancelled = False

        while not resultSent:
            try:
                # get result (waiting max. 3 seconds and send it to frontend
                filename, result = doneQueue.get()
                send(fx, filename, result)
                resultSent = True
            except queue.Empty:
                # ignore empty queue, just carry on
                if cancelled():
                    wasCancelled = True
                    break

        if wasCancelled or cancelled():
            # just exit the loop ignoring the results of queued tasks
            break

        if i < endIndex:
            taskQueue.put(argumentsList[i + initialTasks])

    # Tell child processes to stop
    for _ in range(NumberOfProcesses):
        taskQueue.put("STOP")

    for worker in workers:
        worker.join()
        worker.close()


def workerTask(inputQueue, outputQueue):
    """
    Module function acting as the parallel worker for the vulture check.

    @param inputQueue input queue
    @type multiprocessing.Queue
    @param outputQueue output queue
    @type multiprocessing.Queue
    """
    for filename, source in iter(inputQueue.get, "STOP"):
        result = __analyze(filename, source)
        outputQueue.put((filename, result))


def __analyze(filename, text=""):
    """
    Private function to analyze one Python file.

    @param filename source file name
    @type str
    @param text source text
    @type str
    @return tuple containing the result dictionary
    @rtype (tuple of dict)
    """
    try:
        v = EricVulture()
        v.scan(text, filename=filename)
        res = v.getResults()
    except Exception as err:
        res = {"error": str(err)}
    return (res,)


class EricVulture(Vulture):
    """
    Class to adopt the Vulture class to the eric plug-in functionality.
    """

    def __item2Dict(self, item):
        """
        Private method to convert a vulture item to a dictionary.

        @param item vulture item
        @type vulture.Item
        @return item dictionary
        @rtype dict
        """
        return {
            "name": item.name,
            "type": item.typ,
            "file": str(item.filename),
            "first_line": item.first_lineno,
            "last_line": item.last_lineno,
            "confidence": item.confidence,
        }

    def getResults(self):
        """
        Public method to get the scan results.

        @return scan results
        @rtype dict
        """
        return {
            "UnusedAttributes": [self.__item2Dict(i) for i in self.unused_attrs],
            "UnusedClasses": [self.__item2Dict(i) for i in self.unused_classes],
            "UnusedFunctions": [self.__item2Dict(i) for i in self.unused_funcs],
            "UnusedMethods": [self.__item2Dict(i) for i in self.unused_methods],
            "UnusedImports": [self.__item2Dict(i) for i in self.unused_imports],
            "UnusedProperties": [self.__item2Dict(i) for i in self.unused_props],
            "UnusedVariables": [self.__item2Dict(i) for i in self.unused_vars],
        }