.config/coc/extensions/node_modules/coc-pyright/pythonFiles/refactor.py

   1 # Arguments are:
   2 # 1. Working directory.
   3 # 2. Rope folder
   4
   5 import difflib
   6 import io
   7 import json
   8 import os
   9 import sys
  10 import traceback
  11
  12 try:
  13     import rope
  14     import rope.base.project
  15     import rope.base.taskhandle
  16     from rope.base import libutils
  17     from rope.refactor.rename import Rename
  18     from rope.refactor.extract import ExtractMethod, ExtractVariable
  19     from rope.refactor.importutils import FromImport, NormalImport
  20     from rope.refactor.importutils.module_imports import ModuleImports
  21 except ImportError:
  22     jsonMessage = {
  23         "error": True,
  24         "message": "Rope not installed",
  25         "traceback": "",
  26         "type": "ModuleNotFoundError",
  27     }
  28     sys.stderr.write(json.dumps(jsonMessage))
  29     sys.stderr.flush()
  30
  31 WORKSPACE_ROOT = sys.argv[1]
  32 ROPE_PROJECT_FOLDER = '.vim/.ropeproject'
  33
  34
  35 class RefactorProgress:
  36     """
  37     Refactor progress information
  38     """
  39
  40     def __init__(self, name="Task Name", message=None, percent=0):
  41         self.name = name
  42         self.message = message
  43         self.percent = percent
  44
  45
  46 class ChangeType:
  47     """
  48     Change Type Enum
  49     """
  50
  51     EDIT = 0
  52     NEW = 1
  53     DELETE = 2
  54
  55
  56 class Change:
  57     """"""
  58
  59     EDIT = 0
  60     NEW = 1
  61     DELETE = 2
  62
  63     def __init__(self, filePath, fileMode=ChangeType.EDIT, diff=""):
  64         self.filePath = filePath
  65         self.diff = diff
  66         self.fileMode = fileMode
  67
  68
  69 def x_diff(x):
  70     new = x["new_contents"]
  71     old = x["old_contents"]
  72     old_lines = old.splitlines(True)
  73     if not old_lines[-1].endswith("\n"):
  74         old_lines[-1] = old_lines[-1] + os.linesep
  75         new = new + os.linesep
  76
  77     result = difflib.unified_diff(
  78         old_lines,
  79         new.splitlines(True),
  80         "a/" + x["path"],
  81         "b/" + x["path"],
  82     )
  83     return "".join(list(result))
  84
  85
  86 def get_diff(changeset):
  87     """This is a copy of the code form the ChangeSet.get_description method found in Rope."""
  88     new = changeset.new_contents
  89     old = changeset.old_contents
  90     if old is None:
  91         if changeset.resource.exists():
  92             old = changeset.resource.read()
  93         else:
  94             old = ""
  95
  96     # Ensure code has a trailing empty lines, before generating a diff.
  97     # https://github.com/Microsoft/vscode-python/issues/695.
  98     old_lines = old.splitlines(True)
  99     if not old_lines[-1].endswith("\n"):
 100         old_lines[-1] = old_lines[-1] + os.linesep
 101         new = new + os.linesep
 102
 103     result = difflib.unified_diff(
 104         old_lines,
 105         new.splitlines(True),
 106         "a/" + changeset.resource.path,
 107         "b/" + changeset.resource.path,
 108     )
 109     return "".join(list(result))
 110
 111
 112 class BaseRefactoring(object):
 113     """
 114     Base class for refactorings
 115     """
 116
 117     def __init__(self, project, resource, name="Refactor", progressCallback=None):
 118         self._progressCallback = progressCallback
 119         self._handle = rope.base.taskhandle.TaskHandle(name)
 120         self._handle.add_observer(self._update_progress)
 121         self.project = project
 122         self.resource = resource
 123         self.changes = []
 124
 125     def _update_progress(self):
 126         jobset = self._handle.current_jobset()
 127         if jobset and not self._progressCallback is None:
 128             progress = RefactorProgress()
 129             # getting current job set name
 130             if jobset.get_name() is not None:
 131                 progress.name = jobset.get_name()
 132             # getting active job name
 133             if jobset.get_active_job_name() is not None:
 134                 progress.message = jobset.get_active_job_name()
 135             # adding done percent
 136             percent = jobset.get_percent_done()
 137             if percent is not None:
 138                 progress.percent = percent
 139             if not self._progressCallback is None:
 140                 self._progressCallback(progress)
 141
 142     def stop(self):
 143         self._handle.stop()
 144
 145     def refactor(self):
 146         try:
 147             self.onRefactor()
 148         except rope.base.exceptions.InterruptedTaskError:
 149             # we can ignore this exception, as user has cancelled refactoring
 150             pass
 151
 152     def onRefactor(self):
 153         """
 154         To be implemented by each base class
 155         """
 156         pass
 157
 158
 159 class RenameRefactor(BaseRefactoring):
 160     def __init__(
 161         self,
 162         project,
 163         resource,
 164         name="Rename",
 165         progressCallback=None,
 166         startOffset=None,
 167         newName="new_Name",
 168     ):
 169         BaseRefactoring.__init__(self, project, resource, name, progressCallback)
 170         self._newName = newName
 171         self.startOffset = startOffset
 172
 173     def onRefactor(self):
 174         renamed = Rename(self.project, self.resource, self.startOffset)
 175         changes = renamed.get_changes(self._newName, task_handle=self._handle)
 176         for item in changes.changes:
 177             if isinstance(item, rope.base.change.ChangeContents):
 178                 self.changes.append(
 179                     Change(item.resource.real_path, ChangeType.EDIT, get_diff(item))
 180                 )
 181             else:
 182                 raise Exception("Unknown Change")
 183
 184
 185 class ExtractVariableRefactor(BaseRefactoring):
 186     def __init__(
 187         self,
 188         project,
 189         resource,
 190         name="Extract Variable",
 191         progressCallback=None,
 192         startOffset=None,
 193         endOffset=None,
 194         newName="new_Name",
 195         similar=False,
 196         global_=False,
 197     ):
 198         BaseRefactoring.__init__(self, project, resource, name, progressCallback)
 199         self._newName = newName
 200         self._startOffset = startOffset
 201         self._endOffset = endOffset
 202         self._similar = similar
 203         self._global = global_
 204
 205     def onRefactor(self):
 206         renamed = ExtractVariable(
 207             self.project, self.resource, self._startOffset, self._endOffset
 208         )
 209         changes = renamed.get_changes(self._newName, self._similar, self._global)
 210         for item in changes.changes:
 211             if isinstance(item, rope.base.change.ChangeContents):
 212                 self.changes.append(
 213                     Change(item.resource.real_path, ChangeType.EDIT, get_diff(item))
 214                 )
 215             else:
 216                 raise Exception("Unknown Change")
 217
 218
 219 class ExtractMethodRefactor(ExtractVariableRefactor):
 220     def __init__(
 221         self,
 222         project,
 223         resource,
 224         name="Extract Method",
 225         progressCallback=None,
 226         startOffset=None,
 227         endOffset=None,
 228         newName="new_Name",
 229         similar=False,
 230         global_=False,
 231     ):
 232         ExtractVariableRefactor.__init__(
 233             self,
 234             project,
 235             resource,
 236             name,
 237             progressCallback,
 238             startOffset=startOffset,
 239             endOffset=endOffset,
 240             newName=newName,
 241             similar=similar,
 242             global_=global_,
 243         )
 244
 245     def onRefactor(self):
 246         renamed = ExtractMethod(
 247             self.project, self.resource, self._startOffset, self._endOffset
 248         )
 249         changes = renamed.get_changes(self._newName, self._similar, self._global)
 250         for item in changes.changes:
 251             if isinstance(item, rope.base.change.ChangeContents):
 252                 self.changes.append(
 253                     Change(item.resource.real_path, ChangeType.EDIT, get_diff(item))
 254                 )
 255             else:
 256                 raise Exception("Unknown Change")
 257
 258
 259 class ImportRefactor(BaseRefactoring):
 260     def __init__(
 261         self,
 262         project,
 263         resource,
 264         text = None,
 265         name = None,
 266         parent = None,
 267     ):
 268         BaseRefactoring.__init__(self, project, resource, name='Add Import', progressCallback=None)
 269         self._name = name
 270         self._text = text
 271         self._parent = parent
 272
 273     def onRefactor(self):
 274         if self._parent:
 275             import_info = FromImport(self._parent, 0, [(self._name, None)])
 276         else:
 277             import_info = NormalImport([(self._name, None)])
 278
 279         pymodule = self.project.get_pymodule(self.resource)
 280         module_imports = ModuleImports(self.project, pymodule)
 281         module_imports.add_import(import_info)
 282         changed_source = module_imports.get_changed_source()
 283         if changed_source:
 284             changeset = {
 285                 "old_contents": self._text,
 286                 "new_contents": changed_source,
 287                 "path": self.resource.path
 288             }
 289             self.changes.append(Change(self.resource.path, ChangeType.EDIT, x_diff(changeset)))
 290         else:
 291             raise Exception('Unknown Change')
 292
 293
 294 class RopeRefactoring(object):
 295     def __init__(self):
 296         self.default_sys_path = sys.path
 297         self._input = io.open(sys.stdin.fileno(), encoding="utf-8")
 298
 299     def _rename(self, filePath, start, newName, indent_size):
 300         """
 301         Renames a variable
 302         """
 303         project = rope.base.project.Project(
 304             WORKSPACE_ROOT,
 305             ropefolder=ROPE_PROJECT_FOLDER,
 306             save_history=False,
 307             indent_size=indent_size,
 308         )
 309         resourceToRefactor = libutils.path_to_resource(project, filePath)
 310         refactor = RenameRefactor(
 311             project, resourceToRefactor, startOffset=start, newName=newName
 312         )
 313         refactor.refactor()
 314         changes = refactor.changes
 315         project.close()
 316         valueToReturn = []
 317         for change in changes:
 318             valueToReturn.append({"diff": change.diff})
 319         return valueToReturn
 320
 321     def _extractVariable(self, filePath, start, end, newName, indent_size):
 322         """
 323         Extracts a variable
 324         """
 325         project = rope.base.project.Project(
 326             WORKSPACE_ROOT,
 327             ropefolder=ROPE_PROJECT_FOLDER,
 328             save_history=False,
 329             indent_size=indent_size,
 330         )
 331         resourceToRefactor = libutils.path_to_resource(project, filePath)
 332         refactor = ExtractVariableRefactor(
 333             project,
 334             resourceToRefactor,
 335             startOffset=start,
 336             endOffset=end,
 337             newName=newName,
 338             similar=True,
 339         )
 340         refactor.refactor()
 341         changes = refactor.changes
 342         project.close()
 343         valueToReturn = []
 344         for change in changes:
 345             valueToReturn.append({"diff": change.diff})
 346         return valueToReturn
 347
 348     def _extractMethod(self, filePath, start, end, newName, indent_size):
 349         """
 350         Extracts a method
 351         """
 352         project = rope.base.project.Project(
 353             WORKSPACE_ROOT,
 354             ropefolder=ROPE_PROJECT_FOLDER,
 355             save_history=False,
 356             indent_size=indent_size,
 357         )
 358         resourceToRefactor = libutils.path_to_resource(project, filePath)
 359         refactor = ExtractMethodRefactor(
 360             project,
 361             resourceToRefactor,
 362             startOffset=start,
 363             endOffset=end,
 364             newName=newName,
 365             similar=True,
 366         )
 367         refactor.refactor()
 368         changes = refactor.changes
 369         project.close()
 370         valueToReturn = []
 371         for change in changes:
 372             valueToReturn.append({"diff": change.diff})
 373         return valueToReturn
 374
 375     def _add_import(self, filePath, text, name, parent, indent_size):
 376         """
 377         Add import
 378         """
 379         project = rope.base.project.Project(
 380             WORKSPACE_ROOT,
 381             ropefolder=ROPE_PROJECT_FOLDER,
 382             save_history=False,
 383             indent_size=indent_size,
 384         )
 385         resourceToRefactor = libutils.path_to_resource(project, filePath)
 386         refactor = ImportRefactor(
 387             project,
 388             resourceToRefactor,
 389             text,
 390             name,
 391             parent
 392         )
 393         refactor.refactor()
 394         changes = refactor.changes
 395         project.close()
 396         valueToReturn = []
 397         for change in changes:
 398             valueToReturn.append({"diff": change.diff})
 399         return valueToReturn
 400
 401     def _serialize(self, identifier, results):
 402         """
 403         Serializes the refactor results
 404         """
 405         return json.dumps({"id": identifier, "results": results})
 406
 407     def _deserialize(self, request):
 408         """Deserialize request from VSCode.
 409
 410         Args:
 411             request: String with raw request from VSCode.
 412
 413         Returns:
 414             Python dictionary with request data.
 415         """
 416         return json.loads(request)
 417
 418     def _process_request(self, request):
 419         """Accept serialized request from VSCode and write response."""
 420         request = self._deserialize(request)
 421         lookup = request.get("lookup", "")
 422
 423         if lookup == "":
 424             pass
 425         elif lookup == "rename":
 426             changes = self._rename(
 427                 request["file"],
 428                 int(request["start"]),
 429                 request["name"],
 430                 int(request["indent_size"]),
 431             )
 432             return self._write_response(self._serialize(request["id"], changes))
 433         elif lookup == "add_import":
 434             changes = self._add_import(
 435                 request["file"],
 436                 request["text"],
 437                 request["name"],
 438                 request.get("parent", None),
 439                 int(request["indent_size"]),
 440             )
 441             return self._write_response(self._serialize(request["id"], changes))
 442         elif lookup == "extract_variable":
 443             changes = self._extractVariable(
 444                 request["file"],
 445                 int(request["start"]),
 446                 int(request["end"]),
 447                 request["name"],
 448                 int(request["indent_size"]),
 449             )
 450             return self._write_response(self._serialize(request["id"], changes))
 451         elif lookup == "extract_method":
 452             changes = self._extractMethod(
 453                 request["file"],
 454                 int(request["start"]),
 455                 int(request["end"]),
 456                 request["name"],
 457                 int(request["indent_size"]),
 458             )
 459             return self._write_response(self._serialize(request["id"], changes))
 460
 461     def _write_response(self, response):
 462         sys.stdout.write(response + "\n")
 463         sys.stdout.flush()
 464
 465     def watch(self):
 466         self._write_response("STARTED")
 467         while True:
 468             try:
 469                 self._process_request(self._input.readline())
 470             except:
 471                 exc_type, exc_value, exc_tb = sys.exc_info()
 472                 tb_info = traceback.extract_tb(exc_tb)
 473                 jsonMessage = {
 474                     "error": True,
 475                     "message": str(exc_value),
 476                     "traceback": str(tb_info),
 477                     "type": str(exc_type),
 478                 }
 479                 sys.stderr.write(json.dumps(jsonMessage))
 480                 sys.stderr.flush()
 481
 482
 483 if __name__ == "__main__":
 484     RopeRefactoring().watch()