CreateException._parse_trakt_error() - Code Metrics - Inspection of "Removed some unused helper methods" - fuzeman/Plex-Trakt-Scrobbler - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( c6fa42...950d5f )

by Dean

created 2016-12-04 09:36 UTC

CreateException._parse_trakt_error() A

↳ Parent: CreateException

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	14.0742

Importance

Changes

Metric	Value
c	0
b	0
f	0
dl	0
loc	15
ccs	1
cts	7
cp	0.1429
rs	9.2
cc	4
crap	14.0742

from plugin.core.constants import PLUGIN_VERSION_BASE, PLUGIN_VERSION_BRANCH, PMS_PATH
from plugin.core.exceptions import ConnectionError

from plugin.core.helpers.error import ErrorHasher
from plugin.managers.core.base import Manager, Create
from plugin.managers.message import MessageManager
from plugin.models.exception import Exception

from plugin.models.message import Message

from datetime import datetime
from requests import exceptions as requests_exceptions
from requests.packages.urllib3 import exceptions as urllib3_exceptions
from six.moves.urllib.parse import urlparse
# .scrutinizer.yml
before_commands:
    - sudo pip install abc # Python2
    - sudo pip3 install abc # Python3
import logging
import os
import re
import socket
import ssl

import sys
import trakt

VERSION_BASE = '.'.join([str(x) for x in PLUGIN_VERSION_BASE])
VERSION_BRANCH = PLUGIN_VERSION_BRANCH

RE_TRACEBACK = re.compile(r"\w+ \(most recent call last\)\:\n(?P<traceback>(?:.*?\n)*)(?P<type>\w+)\: (?P<message>.*?)(?:\n|$)", re.IGNORECASE)


FINAL_EXCEPTION_TYPES = [
    urllib3_exceptions.ProxyError
]

log = logging.getLogger(__name__)


class CreateException(Create):
    #
    # exc_info
    #

    def from_exc_info(self, exc_info=None):
        if exc_info is None:
            # Retrieve `exc_info` of last exception
            exc_info = sys.exc_info()

        # Parse exception
        message_type = Message.Type.Exception

        try:
            message_type, exc_info = self._parse_exception(exc_info)
        except Exception as ex:
            log.warn('Unable to parse exception: %s', ex, exc_info=True)

        # Create exception
        exception = self.model(
            type=ErrorHasher.exc_type(exc_info[0]),
            message=ErrorHasher.exc_message(exc_info[1]),
            traceback=ErrorHasher.exc_traceback(exc_info[2]),

            timestamp=datetime.utcnow(),
            version_base=VERSION_BASE,
            version_branch=VERSION_BRANCH
        )

        # Calculate exception hash
        exception.hash = ErrorHasher.hash(exception)

        # Create/Lookup message for exception
        exception.error = MessageManager.get.from_exception(

            exception,
            message_type=message_type
        )

        # Save exception details
        exception.save()

        return exception, exception.error

    #
    # message
    #

    def from_message(self, message):
        match = RE_TRACEBACK.match(message)

        if match is None:
            return

        # Create exception
        exception = self.model(
            type=match.group('type'),
            message=match.group('message'),
            traceback=self.strip_traceback(match.group('traceback')),

            timestamp=datetime.utcnow(),
            version_base=VERSION_BASE,
            version_branch=VERSION_BRANCH
        )

        # Calculate exception hash
        exception.hash = ErrorHasher.hash(exception)

        # Create/Lookup message for exception
        exception.error = MessageManager.get.from_exception(exception)


        # Save exception details
        exception.save()

        return exception, exception.error

    @staticmethod
    def strip_traceback(tb):
        lines = tb.split('\n')

        for x in xrange(len(lines)):

            line = lines[x]

            if not line.startswith('  File'):
                continue

            try:
                # Try find path start/end quotes
                path_start = line.index('"') + 1
                path_end = line.index('"', path_start)
            except ValueError:
                # Unable to find path quotes
                continue

            # Convert path to relative
            path = os.path.relpath(line[path_start:path_end], PMS_PATH)

            # Update line
            lines[x] = line[:path_start] + path + line[path_end:]

        return '\n'.join(lines)

    def _parse_exception(self, exc_info):
        if type(exc_info) is not tuple or len(exc_info) != 3:
            return Message.Type.Exception, exc_info

        # Parse exception
        _, ex, tb = exc_info


        if isinstance(ex, trakt.RequestError):
            return self._parse_trakt_error(exc_info)

        if isinstance(ex, requests_exceptions.RequestException):
            return self._parse_request_exception(exc_info)

        return Message.Type.Exception, exc_info

    def _parse_trakt_error(self, exc_info):
        if type(exc_info) is not tuple or len(exc_info) != 3:
            return Message.Type.Exception, exc_info

        # Parse exception information
        _, ex, tb = exc_info

        if not isinstance(ex, trakt.RequestError):
            return Message.Type.Exception, exc_info

        # Construct connection error
        return Message.Type.Trakt, (
            ConnectionError,
            ConnectionError(self._get_exception_message(ex)),
            tb
        )

    def _parse_request_exception(self, exc_info):
        if type(exc_info) is not tuple or len(exc_info) != 3:
            return Message.Type.Exception, exc_info

        # Parse exception information
        _, ex, tb = exc_info

        if not isinstance(ex, requests_exceptions.RequestException) or not ex.request:
            return Message.Type.Exception, exc_info

        # Parse request url
        url = urlparse(ex.request.url)

        if not url:
            return Message.Type.Exception, exc_info

        # Retrieve service title
        if url.netloc.endswith('.plex.tv'):
            message_type = Message.Type.Plex
        elif url.netloc.endswith('.sentry.skipthe.net'):
            message_type = Message.Type.Sentry
        elif url.netloc.endswith('.trakt.tv'):
            message_type = Message.Type.Trakt
        else:
            return Message.Type.Exception, exc_info

        # Construct connection error
        return message_type, (
            ConnectionError,
            ConnectionError(self._get_exception_message(ex)),
            tb
        )

    def _get_exception_message(self, ex):
        if not issubclass(ex.__class__, BaseException):
            return ex

        if type(ex) in FINAL_EXCEPTION_TYPES:
            return ex.message

        # Return exception messages
        if isinstance(ex, trakt.RequestError):
            return '%s (code: %r)' % (
                ex.error[1],
                ex.status_code
            )

        if isinstance(ex, socket.error):
            if ex.errno is None:
                return '%s' % (
                    ex.message or ex.strerror
                )

            return '%s (code: %r)' % (
                ex.message or ex.strerror,
                ex.errno
            )

        # Expand "requests" and "urllib3" exceptions
        if isinstance(ex, requests_exceptions.RequestException):
            if len(ex.args) < 1:
                return ex.message

            return self._get_exception_message(ex.args[0])

        if isinstance(ex, urllib3_exceptions.MaxRetryError):
            return self._get_exception_message(ex.reason)

        if isinstance(ex, urllib3_exceptions.HTTPError):
            if issubclass(ex.message.__class__, BaseException):
                return self._get_exception_message(ex.message)

            if len(ex.args) < 1:
                return ex.message

            return self._get_exception_message(ex.args[0])

        # Unknown exception
        return ex.message


class ExceptionManager(Manager):
    create = CreateException

    model = Exception


Push — develop ( c6fa42...950d5f )

CreateException._parse_trakt_error() A

Complexity

Size

Duplication

Code Coverage

Importance

1. Missing Dependencies

2. Missing init.py files

1	1	from plugin.core.constants import PLUGIN_VERSION_BASE, PLUGIN_VERSION_BRANCH, PMS_PATH
2	1	from plugin.core.exceptions import ConnectionError
		0 ignored issues – show Bug Best Practice introduced 2016-12-04 09:39 UTC by Report Bug Copy Issue Report This seems to re-define the built-in `ConnectionError`. It is generally discouraged to redefine built-ins as this makes code very hard to read. Loading history...
3	1	from plugin.core.helpers.error import ErrorHasher
4	1	from plugin.managers.core.base import Manager, Create
5	1	from plugin.managers.message import MessageManager
6	1	from plugin.models.exception import Exception
		0 ignored issues – show Bug Best Practice introduced 2015-12-19 01:32 UTC by Report Bug Copy Issue Report This seems to re-define the built-in `Exception`. It is generally discouraged to redefine built-ins as this makes code very hard to read. Loading history...
7	1	from plugin.models.message import Message
8
9	1	from datetime import datetime
10	1	from requests import exceptions as requests_exceptions
11	1	from requests.packages.urllib3 import exceptions as urllib3_exceptions
12	1	from six.moves.urllib.parse import urlparse
		0 ignored issues – show Configuration introduced 2016-12-04 09:39 UTC by Report Bug Copy Issue Report The import `six.moves.urllib.parse` could not be resolved. This can be caused by one of the following: 1. Missing Dependencies This error could indicate a configuration issue of Pylint. Make sure that your libraries are available by adding the necessary commands. # .scrutinizer.yml before_commands: - sudo pip install abc # Python2 - sudo pip3 install abc # Python3 Tip: We are currently not using virtualenv to run pylint, when installing your modules make sure to use the command for the correct version. 2. Missing __init__.py files This error could also result from missing `__init__.py` files in your module folders. Make sure that you place one file in each sub-folder. Loading history...
13	1	import logging
14	1	import os
15	1	import re
16	1	import socket
17	1	import ssl
		0 ignored issues – show Unused Code introduced 2016-12-04 09:39 UTC by Report Bug Copy Issue Report The import `ssl` seems to be unused. Loading history...
18	1	import sys
19	1	import trakt
20
21	1	VERSION_BASE = '.'.join([str(x) for x in PLUGIN_VERSION_BASE])
22	1	VERSION_BRANCH = PLUGIN_VERSION_BRANCH
23
24	1	RE_TRACEBACK = re.compile(r"\w+ \(most recent call last\)\:\n(?P<traceback>(?:.?\n))(?P<type>\w+)\: (?P<message>.*?)(?:\n\|$)", re.IGNORECASE)
		0 ignored issues – show Coding Style introduced 2016-02-05 03:55 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (143/120). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
25
26	1	FINAL_EXCEPTION_TYPES = [
27		urllib3_exceptions.ProxyError
28		]
29
30	1	log = logging.getLogger(__name__)
31
32
33	1	class CreateException(Create):
34		#
35		# exc_info
36		#
37
38	1	def from_exc_info(self, exc_info=None):
39		if exc_info is None:
40		# Retrieve `exc_info` of last exception
41		exc_info = sys.exc_info()
42
43		# Parse exception
44		message_type = Message.Type.Exception
45
46		try:
47		message_type, exc_info = self._parse_exception(exc_info)
48		except Exception as ex:
49		log.warn('Unable to parse exception: %s', ex, exc_info=True)
50
51		# Create exception
52		exception = self.model(
53		type=ErrorHasher.exc_type(exc_info[0]),
54		message=ErrorHasher.exc_message(exc_info[1]),
55		traceback=ErrorHasher.exc_traceback(exc_info[2]),
56
57		timestamp=datetime.utcnow(),
58		version_base=VERSION_BASE,
59		version_branch=VERSION_BRANCH
60		)
61
62		# Calculate exception hash
63		exception.hash = ErrorHasher.hash(exception)
64
65		# Create/Lookup message for exception
66		exception.error = MessageManager.get.from_exception(
		0 ignored issues – show Bug introduced 2015-12-19 01:32 UTC by Report Bug Copy Issue Report It seems like a value for argument `exception` is missing in the unbound method call. Loading history...
67		exception,
68		message_type=message_type
69		)
70
71		# Save exception details
72		exception.save()
73
74		return exception, exception.error
75
76		#
77		# message
78		#
79
80	1	def from_message(self, message):
81		match = RE_TRACEBACK.match(message)
82
83		if match is None:
84		return
85
86		# Create exception
87		exception = self.model(
88		type=match.group('type'),
89		message=match.group('message'),
90		traceback=self.strip_traceback(match.group('traceback')),
91
92		timestamp=datetime.utcnow(),
93		version_base=VERSION_BASE,
94		version_branch=VERSION_BRANCH
95		)
96
97		# Calculate exception hash
98		exception.hash = ErrorHasher.hash(exception)
99
100		# Create/Lookup message for exception
101		exception.error = MessageManager.get.from_exception(exception)
		0 ignored issues – show Bug introduced 2015-12-19 01:32 UTC by Report Bug Copy Issue Report It seems like a value for argument `exception` is missing in the unbound method call. Loading history...
102
103		# Save exception details
104		exception.save()
105
106		return exception, exception.error
107
108	1	@staticmethod
109		def strip_traceback(tb):
110		lines = tb.split('\n')
111
112		for x in xrange(len(lines)):
		0 ignored issues – show Comprehensibility Best Practice introduced 2015-12-19 01:32 UTC by Report Bug Copy Issue Report Undefined variable 'xrange' Loading history...
113		line = lines[x]
114
115		if not line.startswith(' File'):
116		continue
117
118		try:
119		# Try find path start/end quotes
120		path_start = line.index('"') + 1
121		path_end = line.index('"', path_start)
122		except ValueError:
123		# Unable to find path quotes
124		continue
125
126		# Convert path to relative
127		path = os.path.relpath(line[path_start:path_end], PMS_PATH)
128
129		# Update line
130		lines[x] = line[:path_start] + path + line[path_end:]
131
132		return '\n'.join(lines)
133
134	1	def _parse_exception(self, exc_info):
135		if type(exc_info) is not tuple or len(exc_info) != 3:
136		return Message.Type.Exception, exc_info
137
138		# Parse exception
139		_, ex, tb = exc_info
		0 ignored issues – show Unused Code introduced 2016-12-04 09:39 UTC by Report Bug Copy Issue Report The variable `tb` seems to be unused. Loading history...
140
141		if isinstance(ex, trakt.RequestError):
142		return self._parse_trakt_error(exc_info)
143
144		if isinstance(ex, requests_exceptions.RequestException):
145		return self._parse_request_exception(exc_info)
146
147		return Message.Type.Exception, exc_info
148
149	1	def _parse_trakt_error(self, exc_info):
150		if type(exc_info) is not tuple or len(exc_info) != 3:
151		return Message.Type.Exception, exc_info
152
153		# Parse exception information
154		_, ex, tb = exc_info
155
156		if not isinstance(ex, trakt.RequestError):
157		return Message.Type.Exception, exc_info
158
159		# Construct connection error
160		return Message.Type.Trakt, (
161		ConnectionError,
162		ConnectionError(self._get_exception_message(ex)),
163		tb
164		)
165
166	1	def _parse_request_exception(self, exc_info):
167		if type(exc_info) is not tuple or len(exc_info) != 3:
168		return Message.Type.Exception, exc_info
169
170		# Parse exception information
171		_, ex, tb = exc_info
172
173		if not isinstance(ex, requests_exceptions.RequestException) or not ex.request:
174		return Message.Type.Exception, exc_info
175
176		# Parse request url
177		url = urlparse(ex.request.url)
178
179		if not url:
180		return Message.Type.Exception, exc_info
181
182		# Retrieve service title
183		if url.netloc.endswith('.plex.tv'):
184		message_type = Message.Type.Plex
185		elif url.netloc.endswith('.sentry.skipthe.net'):
186		message_type = Message.Type.Sentry
187		elif url.netloc.endswith('.trakt.tv'):
188		message_type = Message.Type.Trakt
189		else:
190		return Message.Type.Exception, exc_info
191
192		# Construct connection error
193		return message_type, (
194		ConnectionError,
195		ConnectionError(self._get_exception_message(ex)),
196		tb
197		)
198
199	1	def _get_exception_message(self, ex):
200		if not issubclass(ex.__class__, BaseException):
201		return ex
202
203		if type(ex) in FINAL_EXCEPTION_TYPES:
204		return ex.message
205
206		# Return exception messages
207		if isinstance(ex, trakt.RequestError):
208		return '%s (code: %r)' % (
209		ex.error[1],
210		ex.status_code
211		)
212
213		if isinstance(ex, socket.error):
214		if ex.errno is None:
215		return '%s' % (
216		ex.message or ex.strerror
217		)
218
219		return '%s (code: %r)' % (
220		ex.message or ex.strerror,
221		ex.errno
222		)
223
224		# Expand "requests" and "urllib3" exceptions
225		if isinstance(ex, requests_exceptions.RequestException):
226		if len(ex.args) < 1:
227		return ex.message
228
229		return self._get_exception_message(ex.args[0])
230
231		if isinstance(ex, urllib3_exceptions.MaxRetryError):
232		return self._get_exception_message(ex.reason)
233
234		if isinstance(ex, urllib3_exceptions.HTTPError):
235		if issubclass(ex.message.__class__, BaseException):
236		return self._get_exception_message(ex.message)
237
238		if len(ex.args) < 1:
239		return ex.message
240
241		return self._get_exception_message(ex.args[0])
242
243		# Unknown exception
244		return ex.message
245
246
247	1	class ExceptionManager(Manager):
248	1	create = CreateException
249
250		model = Exception
251

fuzeman / Plex-Trakt-Scrobbler

Push — develop ( c6fa42...950d5f )

CreateException._parse_trakt_error() A

Complexity

Size

Duplication

Code Coverage

Importance

1. Missing Dependencies

2. Missing __init__.py files

Duplication Side-by-Side

Filter issues like

2. Missing init.py files