src/eric7/Plugins/CheckerPlugins/SyntaxChecker/pyCheckSyntax.py

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

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

"""
Module implementing the syntax check for Python 3.
"""

import ast
import builtins
import contextlib
import multiprocessing
import queue
import re
import traceback
import warnings

with contextlib.suppress(ImportError):
    from pyflakes.checker import Checker
    from pyflakes.messages import ImportStarUsage, ImportStarUsed

VcsConflictMarkerRegExpList = (
    re.compile(
        r"""^<<<<<<< .*?\|\|\|\|\|\|\| .*?=======.*?>>>>>>> .*?$""",
        re.MULTILINE | re.DOTALL,
    ),
    re.compile(r"""^<<<<<<< .*?=======.*?>>>>>>> .*?$""", re.MULTILINE | re.DOTALL),
)


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

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


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

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


def extractLineFlags(line, startComment="#", endComment="", flagsLine=False):
    """
    Function to extract flags starting and ending with '__' from a line
    comment.

    @param line line to extract flags from
    @type str
    @param startComment string identifying the start of the comment
    @type str
    @param endComment string identifying the end of a comment
    @type str
    @param flagsLine flag indicating to check for a flags only line
    @type bool
    @return list containing the extracted flags
    @rtype list of str
    """
    flags = []

    if not flagsLine or (flagsLine and line.strip().startswith(startComment)):
        pos = line.rfind(startComment)
        if pos >= 0:
            comment = line[pos + len(startComment) :].strip()
            if endComment:
                endPos = line.rfind(endComment)
                if endPos >= 0:
                    comment = comment[:endPos]
            flags = [
                f.strip()
                for f in comment.split()
                if (f.startswith("__") and f.endswith("__"))
            ]
            flags += [
                f.strip().lower() for f in comment.split() if f in ("noqa", "NOQA")
            ]
    return flags


def pySyntaxAndPyflakesCheck(
    filename,
    codestring,
    checkFlakes=True,
    ignoreStarImportWarnings=False,
    additionalBuiltins=None,
):
    """
    Function to compile one Python source file to Python bytecode
    and to perform a pyflakes check.

    @param filename source filename
    @type str
    @param codestring string containing the code to compile
    @type str
    @param checkFlakes flag indicating to do a pyflakes check
    @type bool
    @param ignoreStarImportWarnings flag indicating to ignore 'star import' warnings
    @type bool
    @param additionalBuiltins list of names pyflakes should consider as builtins
    @type list of str
    @return dictionary with the keys 'error' and 'warnings' which
            hold a list containing details about the error/warnings
            (file name, line number, column, codestring (only at syntax
            errors), the message, a list with arguments for the message)
    @rtype dict
    """
    return __pySyntaxAndPyflakesCheck(
        filename, codestring, checkFlakes, ignoreStarImportWarnings, additionalBuiltins
    )


def pySyntaxAndPyflakesBatchCheck(argumentsList, send, fx, cancelled, maxProcesses=0):
    """
    Module function to check syntax for a batch of files.

    @param argumentsList list of arguments tuples as given for pySyntaxAndPyflakesCheck
    @type list
    @param send reference to send function
    @type func
    @param fx registered service name
    @type str
    @param cancelled reference to function checking for a cancellation
    @type func
    @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 times the number of processes)
    tasks = len(argumentsList)
    initialTasks = min(2 * NumberOfProcesses, tasks)
    for _ in range(initialTasks):
        taskQueue.put(argumentsList.pop(0))

    # 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
    for _ in range(tasks):
        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 argumentsList:
            taskQueue.put(argumentsList.pop(0))

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

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

    taskQueue.close()
    doneQueue.close()


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

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


def __pySyntaxAndPyflakesCheck(
    filename,
    codestring,
    checkFlakes=True,
    ignoreStarImportWarnings=False,
    additionalBuiltins=None,
):
    """
    Function to compile one Python source file to Python bytecode
    and to perform a pyflakes check.

    @param filename source filename
    @type str
    @param codestring string containing the code to compile
    @type str
    @param checkFlakes flag indicating to do a pyflakes check
    @type bool
    @param ignoreStarImportWarnings flag indicating to
        ignore 'star import' warnings
    @type bool
    @param additionalBuiltins list of names pyflakes should consider as builtins
    @type list of str
    @return dictionary with the keys 'error' and 'warnings' which
            hold a list containing details about the error/ warnings
            (file name, line number, column, codestring (only at syntax
            errors), the message, a list with arguments for the message)
    @rtype dict
    """
    if codestring:
        warnings.filterwarnings("error")
        errorDict = {}
        try:
            # Check for VCS conflict markers
            for conflictMarkerRe in VcsConflictMarkerRegExpList:
                conflict = conflictMarkerRe.search(codestring)
                if conflict is not None:
                    start, i = conflict.span()
                    lineindex = 1 + codestring.count("\n", 0, start)
                    return [
                        {
                            "error": (
                                filename,
                                lineindex,
                                0,
                                "",
                                "VCS conflict marker found",
                            )
                        }
                    ]

            if filename.endswith(".ptl"):
                try:
                    import quixote.ptl_compile  # __IGNORE_WARNING_I10__
                except ImportError:
                    return [
                        {"error": (filename, 0, 0, "", "Quixote plugin not found.")}
                    ]
                template = quixote.ptl_compile.Template(codestring, filename)
                template.compile()
            else:
                module = builtins.compile(
                    codestring, filename, "exec", ast.PyCF_ONLY_AST
                )
        except SyntaxError as detail:
            index = 0
            code = ""
            error = ""
            lines = traceback.format_exception_only(SyntaxError, detail)
            match = re.match(
                r'\s*File "(.+)", line (\d+)', lines[0].replace("<string>", filename)
            )
            if match is not None:
                fn, line = match.group(1, 2)
                if lines[1].startswith("SyntaxError:"):
                    error = re.match("SyntaxError: (.+)", lines[1]).group(1)
                else:
                    code = re.match("(.+)", lines[1]).group(1)
                    for seLine in lines[2:]:
                        if seLine.startswith("SyntaxError:"):
                            error = re.match("SyntaxError: (.+)", seLine).group(1)
                        elif seLine.rstrip().endswith("^"):
                            index = len(seLine.rstrip()) - 4
            else:
                fn = detail.filename
                line = detail.lineno or 1
                error = detail.msg
            errorDict = {"error": (fn, int(line), index, code.strip(), error)}
        except ValueError as detail:
            try:
                fn = detail.filename
                line = detail.lineno
                error = detail.msg
            except AttributeError:
                fn = filename
                line = 1
                error = str(detail)
            errorDict = {"error": (fn, line, 0, "", error)}
        except Exception as detail:
            with contextlib.suppress(AttributeError):
                fn = detail.filename
                line = detail.lineno
                error = detail.msg
                errorDict = {"error": (fn, line, 0, "", error)}
        finally:
            warnings.resetwarnings()

        # return the syntax error or warning, if one was detected
        if errorDict:
            return [errorDict]

        # pyflakes
        if not checkFlakes:
            return [{}]

        results = []
        lines = codestring.splitlines()
        try:
            flakesWarnings = Checker(
                module, filename, builtins=additionalBuiltins, withDoctest=True
            )
            flakesWarnings.messages.sort(key=lambda a: a.lineno)
            for flakesWarning in flakesWarnings.messages:
                if ignoreStarImportWarnings and isinstance(
                    flakesWarning, (ImportStarUsed, ImportStarUsage)
                ):
                    continue

                _fn, lineno, col, message, msg_args = flakesWarning.getMessageData()
                lineFlags = extractLineFlags(lines[lineno - 1].strip())
                with contextlib.suppress(IndexError):
                    lineFlags += extractLineFlags(lines[lineno].strip(), flagsLine=True)
                if (
                    "__IGNORE_WARNING__" not in lineFlags
                    and "__IGNORE_FLAKES_WARNING__" not in lineFlags
                    and "noqa" not in lineFlags
                ):
                    results.append((_fn, lineno, col, "", message, msg_args))
        except SyntaxError as err:
            msg = err.text.strip() if err.text.strip() else err.msg
            results.append((filename, err.lineno, 0, "FLAKES_ERROR", msg, []))

        return [{"warnings": results}]

    else:
        return [{}]