TryOpenThread.__find_prior_quotation_mark() - Code Metrics - Inspection of "Replace some statements with higher level abstract..." - thatsIch/sublime-rainmeter - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( abd3c1...8a5279 )

by thatsIch

created 2016-12-14 11:01 UTC

TryOpenThread.__find_prior_quotation_mark() A

↳ Parent: TryOpenThread

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	3
c	1
b	0
f	0
dl	0
loc	6
rs	9.4285

"""Module for opening paths containing Rainmeter-specific or Windows environment variables."""

import os
import re
import threading

import sublime
import sublime_plugin

from . import rainmeter
from . import logger


# Files to open with sublime instead of the system default
SETTINGS = sublime.load_settings("Rainmeter.sublime-settings")
DEF_EXTS = SETTINGS.get("rainmeter_default_open_sublime_extensions", "")

if DEF_EXTS is not None:
    DEF_EXTS = DEF_EXTS.strip().strip(r"|").strip()
else:
    DEF_EXTS = ""

ADD_EXTS = SETTINGS.get("rainmeter_open_sublime_extensions", "")

if ADD_EXTS is not None:
    ADD_EXTS = ADD_EXTS.strip().strip(r"|").strip()
else:
    ADD_EXTS = ""

SUBLIME_FILES = re.compile("(?i).*\\.(" + ADD_EXTS + "|" + DEF_EXTS + ")\\b")


def open_path(path, transient=False):
    """Try to open a path.

    A path could be opened either as:

    * a file or folder path or
    * URL in the system default application, or
    * in Sublime if it's a text file.

    Use transient=True to open a file in Sublime without assigning it a tab.
    A tab will be created once the buffer is modified.

    Will return False if the path doesn't exist in the file system, and
    True otherwise.
    """
    if not path:
        return False

    if not os.path.exists(path):
        return False

    sublime.set_timeout(lambda: sublime.status_message("Opening " + path), 10)
    if SUBLIME_FILES.search(path):
        if transient:
            sublime.set_timeout(
                lambda: sublime.active_window().open_file(path,
                                                          sublime.TRANSIENT),
                10)
        else:
            sublime.set_timeout(
                lambda: sublime.active_window().open_file(path),
                10)
    else:
        os.startfile(path)

    return True


def open_url(url):
    """Try opening a url with the system default for urls.

    Will return False if it's not a url, and True otherwise.
    """
    if re.match(r"(?i)(https?|ftp)://", url.strip()):
        os.startfile(url)
        sublime.set_timeout(lambda: sublime.status_message("Opening " + url),
                            10)
        return True
    else:
        return False


class TryOpenThread(threading.Thread):
    """A wrapper method to handle threading for opening line embedded URLs."""

    def __init__(self, line, region, opn):
        """Construct a thread for opening URL embedded in a line."""
        self.line = line
        self.region = region
        self.opn = opn
        threading.Thread.__init__(self)

    def __find_prior_quotation_mark(self):
        lastquote = self.region.a - 1
        while lastquote >= 0 and self.line[lastquote] != "\"":
            lastquote -= 1

        return lastquote

    def run(self):
        """Run the thread."""
        # 1. Selected text
        selected = self.line[self.region.a:self.region.b]
        if self.opn(selected):
            logger.info(
                __file__,
                "TryOpenThread.run(self)",
                "Open selected text"
            )
            return

        # 2. String enclosed in double quotes

        # Find the quotes before the current point (if any)
        lastquote = self.__find_prior_quotation_mark()

        if not lastquote < 0 and self.line[lastquote] == "\"":
            # Find the quote after the current point (if any)
            nextquote = self.region.b
            while nextquote == len(self.line) or self.line[nextquote] != "\"":
                nextquote += 1

            if not nextquote == len(self.line) \
                    and self.line[nextquote] == "\"":
                string = self.line[lastquote: nextquote].strip("\"")
                if self.opn(string):
                    logger.info(
                        __file__,
                        "TryOpenThread.run",
                        "Open string enclosed in quotes: " + string
                    )
                    return

        # 3. Region from last whitespace to next whitespace

        # Find the space before the current point (if any)
        lastspace = self.region.a - 1
        while lastspace >= 0 \
                and self.line[lastspace] != " " \
                and self.line[lastspace] != "\t":
            lastspace -= 1

        # Set to zero if nothing was found until the start of the line
        lastspace = max(lastspace, 0)

        if lastspace == 0 \
                or self.line[lastspace] == " " \
                or self.line[lastspace] == "\t":
            # Find the space after the current point (if any)
            nextspace = self.region.b
            while nextspace < len(self.line) \
                    and self.line[nextspace] != " " \
                    and self.line[nextspace] != "\t":
                nextspace += 1

            if nextspace >= len(self.line) \
                    or self.line[nextspace] == " " \
                    or self.line[nextspace] == "\t":
                string = self.line[lastspace: nextspace].strip()
                if self.opn(string):
                    logger.info(
                        __file__,
                        "TryOpenThread.run",
                        "Open string enclosed in whitespace: " + string
                    )
                    return

        # 4. Everything after the first \"=\" until the end
        # of the line (strip quotes)
        mtch = re.search(r"=\s*(.*)\s*$", self.line)
        if mtch and self.opn(mtch.group(1).strip("\"")):
            logger.info(
                __file__,
                "TryOpenThread.run",
                "Open text after \"=\": " +
                mtch.group(1).strip("\"")
            )
            return

        # 5. Whole line (strip comment character at start)
        stripmatch = re.search(r"^[ \t;]*?([^ \t;].*)\s*$", self.line)
        if self.opn(stripmatch.group(1)):
            logger.info(
                __file__,
                "TryOpenThread.run",
                "Open whole line: " +
                stripmatch.group(1)
            )
            return


class RainmeterOpenPathsCommand(sublime_plugin.TextCommand):
    """Try to open paths on lines in the current selection.

    Will try to open paths to files, folders or URLs on each line in the
    current selection. To achieve this, the following substrings of each line
    intersecting the selection are tested:

    1. The string inside the selection
    2. The string between possible quotes preceding and following the
       selection, if any
    3. The string between the preceding and following whitespace
    4. Everything after the first "=" on the line until the end of the line
    5. The whole line, stripped of preceding semicolons
    """

    def __split_selection_by_new_lines(self, selection):
        # Split all regions into individual segments on lines (using nicely
        # confusing python syntax).
        return [
            j for i in [
                self.view.split_by_newlines(region)
                for region in selection
            ]
            for j in i
        ]

    def __open_each_line_by_thread(self, lines):
        """Identify segments in selected lines and tries to open them each in a new thread.

        This can be resource intensive.
        """
        fnm = self.view.file_name()

        def opn(string):
            """Simple callback method to apply multiple opening operations.

            An URL can be either external or internal and thus is opened differently.
            """
            opened = open_path(rainmeter.make_path(string, fnm)) or open_url(string)
            if opened:
                logger.info(
                    __file__,
                    "run(self, edit)",
                    "found file or url '" + string + "' to open"
                )

        for linereg in lines:
            wholeline = self.view.line(linereg)
            thread = TryOpenThread(self.view.substr(wholeline),
                                   sublime.Region(linereg.a - wholeline.a,
                                                  linereg.b - wholeline.a),
                                   opn)
            thread.start()

    def run(self, _):
        """Detect various scenarios of file paths and try to open them one after the other.

        @param edit unused
        """
        selection = self.view.sel()
        lines = self.__split_selection_by_new_lines(selection)

        loaded_settings = sublime.load_settings("Rainmeter.sublime-settings")
        max_open_lines = loaded_settings.get("rainmeter_max_open_lines", 40)

        # Refuse if too many lines selected to avoid freezing

        if len(lines) > max_open_lines:
            accept = sublime.ok_cancel_dialog(
                "You are trying to open " +
                str(len(lines)) + " lines.\n" +
                "That's a lot, and could take some time. Try anyway?")
            if not accept:
                return

        self.__open_each_line_by_thread(lines)

    def is_enabled(self):
        """Check if current syntax is rainmeter."""
        israinmeter = self.view.score_selector(self.view.sel()[0].a,
                                               "source.rainmeter")

        return israinmeter > 0

    def is_visible(self):
        """It is visible if it is in Rainmeter scope."""
        return self.is_enabled()


1			"""Module for opening paths containing Rainmeter-specific or Windows environment variables."""
2
3			import os
4			import re
5			import threading
6
7			import sublime
8			import sublime_plugin
9
10			from . import rainmeter
11			from . import logger
12
13
14			# Files to open with sublime instead of the system default
15			SETTINGS = sublime.load_settings("Rainmeter.sublime-settings")
16			DEF_EXTS = SETTINGS.get("rainmeter_default_open_sublime_extensions", "")
17
18			if DEF_EXTS is not None:
19			DEF_EXTS = DEF_EXTS.strip().strip(r"\|").strip()
20			else:
21			DEF_EXTS = ""
22
23			ADD_EXTS = SETTINGS.get("rainmeter_open_sublime_extensions", "")
24
25			if ADD_EXTS is not None:
26			ADD_EXTS = ADD_EXTS.strip().strip(r"\|").strip()
27			else:
28			ADD_EXTS = ""
29
30			SUBLIME_FILES = re.compile("(?i).*\\.(" + ADD_EXTS + "\|" + DEF_EXTS + ")\\b")
31
32
33			def open_path(path, transient=False):
34			"""Try to open a path.
35
36			A path could be opened either as:
37
38			* a file or folder path or
39			* URL in the system default application, or
40			* in Sublime if it's a text file.
41
42			Use transient=True to open a file in Sublime without assigning it a tab.
43			A tab will be created once the buffer is modified.
44
45			Will return False if the path doesn't exist in the file system, and
46			True otherwise.
47			"""
48			if not path:
49			return False
50
51			if not os.path.exists(path):
52			return False
53
54			sublime.set_timeout(lambda: sublime.status_message("Opening " + path), 10)
55			if SUBLIME_FILES.search(path):
56			if transient:
57			sublime.set_timeout(
58			lambda: sublime.active_window().open_file(path,
59			sublime.TRANSIENT),
60			10)
61			else:
62			sublime.set_timeout(
63			lambda: sublime.active_window().open_file(path),
64			10)
65			else:
66			os.startfile(path)
67
68			return True
69
70
71			def open_url(url):
72			"""Try opening a url with the system default for urls.
73
74			Will return False if it's not a url, and True otherwise.
75			"""
76			if re.match(r"(?i)(https?\|ftp)://", url.strip()):
77			os.startfile(url)
78			sublime.set_timeout(lambda: sublime.status_message("Opening " + url),
79			10)
80			return True
81			else:
82			return False
83
84
85			class TryOpenThread(threading.Thread):
86			"""A wrapper method to handle threading for opening line embedded URLs."""
87
88			def __init__(self, line, region, opn):
89			"""Construct a thread for opening URL embedded in a line."""
90			self.line = line
91			self.region = region
92			self.opn = opn
93			threading.Thread.__init__(self)
94
95			def __find_prior_quotation_mark(self):
96			lastquote = self.region.a - 1
97			while lastquote >= 0 and self.line[lastquote] != "\"":
98			lastquote -= 1
99
100			return lastquote
101
102			def run(self):
103			"""Run the thread."""
104			# 1. Selected text
105			selected = self.line[self.region.a:self.region.b]
106			if self.opn(selected):
107			logger.info(
108			__file__,
109			"TryOpenThread.run(self)",
110			"Open selected text"
111			)
112			return
113
114			# 2. String enclosed in double quotes
115
116			# Find the quotes before the current point (if any)
117			lastquote = self.__find_prior_quotation_mark()
118
119			if not lastquote < 0 and self.line[lastquote] == "\"":
120			# Find the quote after the current point (if any)
121			nextquote = self.region.b
122			while nextquote == len(self.line) or self.line[nextquote] != "\"":
123			nextquote += 1
124
125			if not nextquote == len(self.line) \
126			and self.line[nextquote] == "\"":
127			string = self.line[lastquote: nextquote].strip("\"")
128			if self.opn(string):
129			logger.info(
130			__file__,
131			"TryOpenThread.run",
132			"Open string enclosed in quotes: " + string
133			)
134			return
135
136			# 3. Region from last whitespace to next whitespace
137
138			# Find the space before the current point (if any)
139			lastspace = self.region.a - 1
140			while lastspace >= 0 \
141			and self.line[lastspace] != " " \
142			and self.line[lastspace] != "\t":
143			lastspace -= 1
144
145			# Set to zero if nothing was found until the start of the line
146			lastspace = max(lastspace, 0)
147
148			if lastspace == 0 \
149			or self.line[lastspace] == " " \
150			or self.line[lastspace] == "\t":
151			# Find the space after the current point (if any)
152			nextspace = self.region.b
153			while nextspace < len(self.line) \
154			and self.line[nextspace] != " " \
155			and self.line[nextspace] != "\t":
156			nextspace += 1
157
158			if nextspace >= len(self.line) \
159			or self.line[nextspace] == " " \
160			or self.line[nextspace] == "\t":
161			string = self.line[lastspace: nextspace].strip()
162			if self.opn(string):
163			logger.info(
164			__file__,
165			"TryOpenThread.run",
166			"Open string enclosed in whitespace: " + string
167			)
168			return
169
170			# 4. Everything after the first \"=\" until the end
171			# of the line (strip quotes)
172			mtch = re.search(r"=\s(.)\s*$", self.line)
173			if mtch and self.opn(mtch.group(1).strip("\"")):
174			logger.info(
175			__file__,
176			"TryOpenThread.run",
177			"Open text after \"=\": " +
178			mtch.group(1).strip("\"")
179			)
180			return
181
182			# 5. Whole line (strip comment character at start)
183			stripmatch = re.search(r"^[ \t;]?([^ \t;].)\s*$", self.line)
184			if self.opn(stripmatch.group(1)):
185			logger.info(
186			__file__,
187			"TryOpenThread.run",
188			"Open whole line: " +
189			stripmatch.group(1)
190			)
191			return
192
193
194			class RainmeterOpenPathsCommand(sublime_plugin.TextCommand):
195			"""Try to open paths on lines in the current selection.
196
197			Will try to open paths to files, folders or URLs on each line in the
198			current selection. To achieve this, the following substrings of each line
199			intersecting the selection are tested:
200
201			1. The string inside the selection
202			2. The string between possible quotes preceding and following the
203			selection, if any
204			3. The string between the preceding and following whitespace
205			4. Everything after the first "=" on the line until the end of the line
206			5. The whole line, stripped of preceding semicolons
207			"""
208
209			def __split_selection_by_new_lines(self, selection):
210			# Split all regions into individual segments on lines (using nicely
211			# confusing python syntax).
212			return [
213			j for i in [
214			self.view.split_by_newlines(region)
215			for region in selection
216			]
217			for j in i
218			]
219
220			def __open_each_line_by_thread(self, lines):
221			"""Identify segments in selected lines and tries to open them each in a new thread.
222
223			This can be resource intensive.
224			"""
225			fnm = self.view.file_name()
226
227			def opn(string):
228			"""Simple callback method to apply multiple opening operations.
229
230			An URL can be either external or internal and thus is opened differently.
231			"""
232			opened = open_path(rainmeter.make_path(string, fnm)) or open_url(string)
233			if opened:
234			logger.info(
235			__file__,
236			"run(self, edit)",
237			"found file or url '" + string + "' to open"
238			)
239
240			for linereg in lines:
241			wholeline = self.view.line(linereg)
242			thread = TryOpenThread(self.view.substr(wholeline),
243			sublime.Region(linereg.a - wholeline.a,
244			linereg.b - wholeline.a),
245			opn)
246			thread.start()
247
248			def run(self, _):
249			"""Detect various scenarios of file paths and try to open them one after the other.
250
251			@param edit unused
252			"""
253			selection = self.view.sel()
254			lines = self.__split_selection_by_new_lines(selection)
255
256			loaded_settings = sublime.load_settings("Rainmeter.sublime-settings")
257			max_open_lines = loaded_settings.get("rainmeter_max_open_lines", 40)
258
259			# Refuse if too many lines selected to avoid freezing
260
261			if len(lines) > max_open_lines:
262			accept = sublime.ok_cancel_dialog(
263			"You are trying to open " +
264			str(len(lines)) + " lines.\n" +
265			"That's a lot, and could take some time. Try anyway?")
266			if not accept:
267			return
268
269			self.__open_each_line_by_thread(lines)
270
271			def is_enabled(self):
272			"""Check if current syntax is rainmeter."""
273			israinmeter = self.view.score_selector(self.view.sel()[0].a,
274			"source.rainmeter")
275
276			return israinmeter > 0
277
278			def is_visible(self):
279			"""It is visible if it is in Rainmeter scope."""
280			return self.is_enabled()
281

thatsIch / sublime-rainmeter

GitHub Access Token became invalid

Push — master ( abd3c1...8a5279 )

TryOpenThread.__find_prior_quotation_mark() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like