RefactoringRope/CodeAssistClient.py

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

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

"""
Module implementing the code assist client interface to rope.
"""

import sys
import contextlib

modulePath = sys.argv[-1]   # it is always the last parameter
sys.path.insert(1, modulePath)

try:
    import rope.base.project
    import rope.base.libutils
    import rope.contrib.codeassist
    from rope.base.exceptions import BadIdentifierError, ModuleSyntaxError
except ImportError:
    sys.exit(42)

from eric7.EricNetwork.EricJsonClient import EricJsonClient


class CodeAssistClient(EricJsonClient):
    """
    Class implementing the code assist client interface to rope.
    """
    IdProject = "Project"
    
    def __init__(self, host, port, idString, projectPath):
        """
        Constructor
        
        @param host ip address the background service is listening
        @type str
        @param port port of the background service
        @type int
        @param idString assigned client id to be sent back to the server in
            order to identify the connection
        @type str
        @param projectPath path to the project
        @type str
        """
        super().__init__(host, port, idString)
        
        self.__methodMapping = {
            "getConfig": self.__getConfig,
            "configChanged": self.__configChanged,
            "closeProject": self.__closeProject,
            "getCompletions": self.__getCompletions,
            "getCallTips": self.__getCallTips,
            "getDocumentation": self.__getDocumentation,
            "gotoDefinition": self.__gotoDefinition,
            "gotoReferences": self.__getReferences,
            "reportChanged": self.__reportChanged,
        }
        
        from FileSystemCommands import RefactoringClientFileSystemCommands
        self.__fsCommands = RefactoringClientFileSystemCommands(self)
        
        self.__projectpath = projectPath
        self.__project = rope.base.project.Project(
            self.__projectpath, fscommands=self.__fsCommands)
        self.__project.validate(self.__project.root)
        
        self.__id = idString
    
    def handleCall(self, method, params):
        """
        Public method to handle a method call from the server.
        
        @param method requested method name
        @type str
        @param params dictionary with method specific parameters
        @type dict
        """
        self.__methodMapping[method](params)
    
    def __handleRopeError(self, err):
        """
        Private method to process a rope error.
        
        @param err rope exception object
        @type Exception
        @return dictionary containing the error information
        @rtype dict
        """
        ropeError = str(type(err)).split()[-1]
        ropeError = ropeError[1:-2].split('.')[-1]
        errorDict = {
            "Error": ropeError,
            "ErrorString": str(err),
        }
        if ropeError == 'ModuleSyntaxError':
            errorDict["ErrorFile"] = err.filename
            errorDict["ErrorLine"] = err.lineno
        
        return errorDict
    
    def __getConfig(self, params):
        """
        Private method to send some configuration data to the server.
        
        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        result = {
            "RopeFolderName": self.__project.ropefolder.real_path,
            "Id": self.__id,
        }
        
        self.sendJson("Config", result)
    
    def __configChanged(self, params):
        """
        Private method to handle a change of the configuration file.
        
        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        self.__project.close()
        self.__project = rope.base.project.Project(
            self.__projectpath, fscommands=self.__fsCommands)
        self.__project.validate(self.__project.root)
    
    def __closeProject(self, params):
        """
        Private slot to validate the project.
        
        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        self.__project.close()
    
    def __getCompletions(self, params):
        """
        Private method to calculate possible completions.
        
        @param params dictionary containing the method parameters
        @type dict
        """
        filename = params["FileName"]
        source = params["Source"]
        offset = params["Offset"]
        maxfixes = params["MaxFixes"]
        
        self.__project.prefs.set("python_path", params["SysPath"])
        resource = (
            rope.base.libutils.path_to_resource(self.__project, filename)
            if filename else
            None
        )
        
        errorDict = {}
        completions = []
        
        try:
            proposals = rope.contrib.codeassist.code_assist(
                self.__project, source, offset, resource, maxfixes=maxfixes)
            for proposal in proposals:
                proposalType = proposal.type
                if proposal.name.startswith("__"):
                    proposalType = "__" + proposalType
                elif proposal.name.startswith("_"):
                    proposalType = "_" + proposalType
                completions.append({
                    "Name": proposal.name,
                    "CompletionType": proposalType,
                })
        except Exception as err:
            errorDict = self.__handleRopeError(err)
        
        result = {
            "Completions": completions,
            "CompletionText": params["CompletionText"],
            "FileName": filename,
        }
        result.update(errorDict)
        
        self.sendJson("CompletionsResult", result)
    
    def __getCallTips(self, params):
        """
        Private method to calculate possible calltips.
        
        @param params dictionary containing the method parameters
        @type dict
        """
        filename = params["FileName"]
        source = params["Source"]
        offset = params["Offset"]
        maxfixes = params["MaxFixes"]
        
        self.__project.prefs.set("python_path", params["SysPath"])
        resource = (
            rope.base.libutils.path_to_resource(self.__project, filename)
            if filename else
            None
        )
        
        errorDict = {}
        calltips = []
        
        try:
            cts = rope.contrib.codeassist.get_calltip(
                self.__project, source, offset, resource, maxfixes=maxfixes,
                remove_self=True)
            if cts is not None:
                calltips = [cts]
        except Exception as err:
            errorDict = self.__handleRopeError(err)
        
        result = {
            "CallTips": calltips,
        }
        result.update(errorDict)
        
        self.sendJson("CallTipsResult", result)
    
    def __getDocumentation(self, params):
        """
        Private method to get some source code documentation.
        
        @param params dictionary containing the method parameters
        @type dict
        """
        filename = params["FileName"]
        source = params["Source"]
        offset = params["Offset"]
        maxfixes = params["MaxFixes"]
        
        if self.__id != CodeAssistClient.IdProject:
            self.__project.prefs.set("python_path", params["SysPath"])
        resource = (
            rope.base.libutils.path_to_resource(self.__project, filename)
            if filename else
            None
        )
        
        errorDict = {}
        documentation = ""
        cts = None
        
        with contextlib.suppress(Exception):
            cts = rope.contrib.codeassist.get_calltip(
                self.__project, source, offset, resource, maxfixes=maxfixes,
                remove_self=True)
        
        if cts is not None:
            while '..' in cts:
                cts = cts.replace('..', '.')
            if '(.)' in cts:
                cts = cts.replace('(.)', '(...)')
        
        try:
            documentation = rope.contrib.codeassist.get_doc(
                self.__project, source, offset, resource, maxfixes=maxfixes)
        except Exception as err:
            errorDict = self.__handleRopeError(err)
        
        typeName = self.__getObjectTypeAndName(
            self.__project, source, offset, resource, maxfixes=maxfixes)
        
        documentationDict = self.__processDocumentation(cts, documentation,
                                                        typeName)
        result = {
            "DocumentationDict": documentationDict,
        }
        result.update(errorDict)
        
        self.sendJson("DocumentationResult", result)
    
    def __processDocumentation(self, cts, documentation, typeName):
        """
        Private method to process the call-tips and documentation.
        
        @param cts call-tips
        @type str
        @param documentation extracted source code documentation
        @type str
        @param typeName type and name of the object
        @type tuple of (str, str)
        @return dictionary containing document information
        @rtype dictionary with keys "name", "argspec", "module" and
            "docstring"
        """
        objectFullname = ""
        calltip = ""
        argspec = ""
        module = ""

        if cts:
            cts = cts.replace(".__init__", "")
            parenthesisPos = cts.find("(")
            if parenthesisPos != -1:
                objectFullname = cts[:parenthesisPos]
                objectName = objectFullname.split('.')[-1]
                cts = cts.replace(objectFullname, objectName)
                calltip = cts
            else:
                objectFullname = cts
        
        if objectFullname and not objectFullname.startswith("self."):
            if calltip:
                argspecStart = calltip.find("(")
                argspec = calltip[argspecStart:]
            
            moduleEnd = objectFullname.rfind('.')
            module = objectFullname[:moduleEnd]
        
        if not objectFullname and typeName[1] not in ["", "<unknown>"]:
            objectFullname = typeName[1]

        return {
            "name": objectFullname,
            "argspec": argspec,
            "module": module,
            "docstring": documentation,
            "typ": typeName[0]
        }
    
    def __getObjectTypeAndName(self, project, sourceCode, offset,
                               resource=None, maxfixes=1):
        """
        Private method to determine an object type and name for the given
        location.
        
        @param project reference to the rope project object
        @type rope.base.project.Project
        @param sourceCode source code
        @type str
        @param offset offset to base the calculation on
        @type int
        @param resource reference to the rope resource object
        @type rope.base.resources.Resource
        @param maxfixes number of fixes to be done
        @type int
        @return tuple containing the object type and name
        @rtype tuple of (str, str)
        """
        from rope.base import pyobjects, pyobjectsdef, pynames
        from rope.contrib import fixsyntax
        
        try:
            fixer = fixsyntax.FixSyntax(project, sourceCode, resource,
                                        maxfixes)
            pyname = fixer.pyname_at(offset)
        except BadIdentifierError:
            pyname = None
        except ModuleSyntaxError:
            pyname = None
        except IndexError:
            pyname = None
        if pyname is None:
            return "<unknown>", "<unknown>"
        
        pyobject = pyname.get_object()
        if isinstance(pyobject, pyobjectsdef.PyPackage):
            typ = "package"
            if isinstance(pyname, pynames.ImportedModule):
                name = pyname.module_name
            else:
                name = "<unknown>"
        elif isinstance(pyobject, pyobjectsdef.PyModule):
            typ = "module"
            name = pyobject.get_name()
        elif isinstance(pyobject, pyobjectsdef.PyClass):
            typ = "class"
            name = pyobject.get_name()
        elif isinstance(pyobject, pyobjectsdef.PyFunction):
            typ = pyobject.get_kind()
            name = pyobject.get_name()
        elif isinstance(pyobject, pyobjects.PyObject):
            typ = "object"
            name = ""
        else:
            typ = ""
            name = ""
        
        return typ, name
    
    def __gotoDefinition(self, params):
        """
        Private method to handle the Goto Definition action.
        
        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        import rope.base.libutils
        
        filename = params["FileName"]
        offset = params["Offset"]
        source = params["Source"]
        uid = params["Uuid"]
        
        self.__project.prefs.set("python_path", params["SysPath"])
        resource = (
            rope.base.libutils.path_to_resource(self.__project, filename)
            if filename else
            None
        )
        
        errorDict = {}
        result = {}
        
        import rope.contrib.findit
        try:
            location = rope.contrib.findit.find_definition(
                self.__project, source, offset, resource)
        except Exception as err:
            location = None
            errorDict = self.__handleRopeError(err)
        
        if location is not None:
            result["Location"] = {
                "ModulePath": location.resource.real_path,
                "Line": location.lineno,
            }
        result["Uuid"] = uid
        result.update(errorDict)
        
        self.sendJson("GotoDefinitionResult", result)
    
    def __getReferences(self, params):
        """
        Private method to get the places a parameter is referenced.
        
        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        import rope.base.libutils
        
        filename = params["FileName"]
        offset = params["Offset"]
        line = params["Line"]
        uid = params["Uuid"]
        
        self.__project.prefs.set("python_path", params["SysPath"])
        resource = (
            rope.base.libutils.path_to_resource(self.__project, filename)
            if filename else
            None
        )
        
        errorDict = {}
        gotoReferences = []
        
        import rope.contrib.findit
        try:
            occurrences = rope.contrib.findit.find_occurrences(
                self.__project, resource, offset, in_hierarchy=True)
            for occurrence in occurrences:
                if (
                    occurrence.lineno == line and
                    occurrence.resource.real_path == filename
                ):
                    continue
                gotoReferences.append({
                    'ModulePath': occurrence.resource.real_path,
                    'Line': occurrence.lineno,
                    'Code': occurrence.resource.read().splitlines()[
                        occurrence.lineno - 1],
                })
        except Exception as err:
            errorDict = self.__handleRopeError(err)
        
        result = {
            "GotoReferencesList": gotoReferences,
            "Uuid": uid,
        }
        result.update(errorDict)
        
        self.sendJson("GotoReferencesResult", result)
    
    def __reportChanged(self, params):
        """
        Private method to register some changed sources.
        
        @param params dictionary containing the method parameters sent by
            the server
        @type dict
        """
        filename = params["FileName"]
        oldSource = params["OldSource"]
        
        with contextlib.suppress(Exception):
            rope.base.libutils.report_change(
                self.__project, filename, oldSource)


if __name__ == '__main__':
    if len(sys.argv) != 6:
        print('Host, port, id, project path and module path parameters are'
              ' missing. Abort.')
        sys.exit(1)
    
    host, port, idString, projectPath = sys.argv[1:-1]
    
    # Create a Qt6 application object in order to allow the processing of
    # modules containing Qt stuff.
    try:
        from PyQt6.QtCore import QCoreApplication
    except ImportError:
        QCoreApplication = None
    if QCoreApplication is not None:
        app = QCoreApplication(sys.argv)
    
    client = CodeAssistClient(host, int(port), idString, projectPath)
    # Start the main loop
    client.run()
    
    sys.exit(0)

#
# eflag: noqa = M801