redirect() - Code Metrics - fuzeman/Plex-Trakt-Scrobbler - Measure and Improve Code Quality continuously with Scrutinizer

redirect() F
last analyzed 2018-03-05 01:11 UTC

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
c	2
b	0
f	0
dl	0
loc	48
rs	3.5294
cc	9

from core.logger import Logger

from plugin.core.constants import PLUGIN_PREFIX
from plugin.core.message import InterfaceMessages

import base64
import cerealizer
import functools
import hashlib

import inspect
import logging
import sys
import threading
import thread
import time
import urllib

log = Logger('core.helpers')


PY25 = sys.version_info[0] == 2 and sys.version_info[1] == 5


def try_convert(value, value_type, default=None):
    try:
        return value_type(value)
    except ValueError:
        return default
    except TypeError:
        return default


def all(items):

    for item in items:
        if not item:
            return False
    return True


def str_pad(s, length, align='left', pad_char=' ', trim=False):
    if not s:
        return s

    if not isinstance(s, (str, unicode)):

        s = str(s)

    if len(s) == length:
        return s
    elif len(s) > length and not trim:
        return s

    if align == 'left':
        if len(s) > length:
            return s[:length]
        else:
            return s + (pad_char * (length - len(s)))
    elif align == 'right':
        if len(s) > length:
            return s[len(s) - length:]
        else:
            return (pad_char * (length - len(s))) + s
    else:
        raise ValueError("Unknown align type, expected either 'left' or 'right'")


def pad_title(value):
    """Pad a title to 30 characters to force the 'details' view."""
    return str_pad(value, 30, pad_char=' ')


def timestamp():
    return str(time.time())


# <bound method type.start of <class 'Scrobbler'>>
RE_BOUND_METHOD = Regex(r"<bound method (type\.)?(?P<name>.*?) of <(class '(?P<class>.*?)')?")



def get_func_name(obj):
    if inspect.ismethod(obj):
        match = RE_BOUND_METHOD.match(repr(obj))

        if match:
            cls = match.group('class')
            if not cls:
                return match.group('name')

            return '%s.%s' % (
                match.group('class'),
                match.group('name')
            )

    return None


def get_class_name(cls):
    if not inspect.isclass(cls):
        cls = getattr(cls, '__class__')

    return getattr(cls, '__name__')


def spawn(func, *args, **kwargs):
    thread_name = kwargs.pop('thread_name', None) or get_func_name(func)

    # Construct thread
    th = threading.Thread(target=thread_wrapper, name=thread_name, kwargs={
        'func': func,
        'args': args,
        'kwargs': kwargs,
        'thread_name': thread_name
    })

    # Set daemon mode
    th.daemon = kwargs.pop('daemon', False)

    # Start thread
    try:
        th.start()
        log.debug("Spawned thread with name '%s'" % thread_name)
    except thread.error as ex:
        log.error('Unable to spawn thread: %s', ex, exc_info=True, extra={
            'data': {
                'active_count': threading.active_count()
            }
        })
        return None

    return th


def thread_wrapper(func, args=None, kwargs=None, thread_name=None):
    if args is None:
        args = ()

    if kwargs is None:
        kwargs = {}

    try:
        func(*args, **kwargs)
    except Exception as ex:

        log.error('Exception raised in thread "%s": %s', thread_name, ex, exc_info=True)


def safe_encode(string):
    string = str(string)
    return base64.b64encode(string).replace('/', '@').replace('+', '*').replace('=', '_')


def pack(obj):
    serialized_obj = cerealizer.dumps(obj)
    encoded_string = safe_encode(serialized_obj)
    return urllib.quote(encoded_string)



def function_path(name, ext=None, **kwargs):
    return '%s/:/function/%s%s?%s' % (
        PLUGIN_PREFIX,
        name,
        ('.%s' % ext) if ext else '',

        urllib.urlencode({

            'function_args': pack(kwargs)
        })
    )


def redirect(path, **kwargs):
    location = PLUGIN_PREFIX + path

    try:
        request = Core.sandbox.context.request


        # Add request parameters (required for authentication on some clients)
        kwargs.update({
            # Client
            'X-Plex-Client-Identifier': request.headers.get('X-Plex-Client-Identifier'),
            'X-Plex-Product': request.headers.get('X-Plex-Product'),
            'X-Plex-Version': request.headers.get('X-Plex-Version'),

            # Platform
            'X-Plex-Platform': request.headers.get('X-Plex-Platform'),
            'X-Plex-Platform-Version': request.headers.get('X-Plex-Platform-Version'),

            # Device
            'X-Plex-Device': request.headers.get('X-Plex-Device'),
            'X-Plex-Device-Name': request.headers.get('X-Plex-Device-Name'),
            'X-Plex-Device-Screen-Resolution': request.headers.get('X-Plex-Device-Screen-Resolution'),

            # Authentication
            'X-Plex-Token': request.headers.get('X-Plex-Token')
        })

        # Retrieve protocol
        protocol = request.protocol

        if request.host.endswith('.plex.direct:32400'):
            # Assume secure connection
            protocol = 'https'

        # Prepend protocol and host (if not already in `location`)
        if request and request.host and location[0] == "/":
            location = protocol + "://" + request.host + location
    except Exception as ex:

        log.warn('Redirect - %s', str(ex), exc_info=True)

    # Append parameters
    if kwargs:
        location += '?' + urllib.urlencode([

            (key, value) for key, value in kwargs.items()
            if value is not None
        ])

    # Return redirect response
    return Redirect(location, True)



def catch_errors(func):
    @functools.wraps(func)
    def inner(*args, **kwargs):
        if InterfaceMessages.critical:
            return error_record_view(logging.CRITICAL, InterfaceMessages.record)

        try:
            return func(*args, **kwargs)
        except Exception as ex:

            if InterfaceMessages.critical:
                return error_record_view(logging.CRITICAL, InterfaceMessages.record)

            log.error('Exception raised in view: %s', ex, exc_info=True)

            return error_view(
                'Exception',
                ex.message

            )

    return inner


def error_record_view(level, record):
    # Retrieve level name
    if level == logging.CRITICAL:
        level_name = 'Critical Error'
    else:
        level_name = logging.getLevelName(level).capitalize()

    # Build error view
    if not record:
        return error_view(level_name)

    return error_view(
        level_name,
        record.message
    )


def error_view(title, message=None):
    oc = ObjectContainer(

        title2=title,
        no_cache=True
    )

    oc.add(DirectoryObject(

        key=PLUGIN_PREFIX,
        title=pad_title('%s: %s' % (
            title,
            message or 'Unknown'
        ))
    ))

    return oc


1			from core.logger import Logger
2
3			from plugin.core.constants import PLUGIN_PREFIX
4			from plugin.core.message import InterfaceMessages
5
6			import base64
7			import cerealizer
8			import functools
9			import hashlib
			0 ignored issues – show Unused Code introduced 2016-12-04 09:39 UTC by Report Bug Copy Issue Report The import `hashlib` seems to be unused. Loading history...
10			import inspect
11			import logging
12			import sys
13			import threading
14			import thread
15			import time
16			import urllib
17
18			log = Logger('core.helpers')
19
20
21			PY25 = sys.version_info[0] == 2 and sys.version_info[1] == 5
22
23
24			def try_convert(value, value_type, default=None):
25			try:
26			return value_type(value)
27			except ValueError:
28			return default
29			except TypeError:
30			return default
31
32
33			def all(items):
			0 ignored issues – show Bug Best Practice introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report This seems to re-define the built-in `all`. It is generally discouraged to redefine built-ins as this makes code very hard to read. Loading history...
34			for item in items:
35			if not item:
36			return False
37			return True
38
39
40			def str_pad(s, length, align='left', pad_char=' ', trim=False):
41			if not s:
42			return s
43
44			if not isinstance(s, (str, unicode)):
			0 ignored issues – show Comprehensibility Best Practice introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report Undefined variable 'unicode' Loading history...
45			s = str(s)
46
47			if len(s) == length:
48			return s
49			elif len(s) > length and not trim:
50			return s
51
52			if align == 'left':
53			if len(s) > length:
54			return s[:length]
55			else:
56			return s + (pad_char * (length - len(s)))
57			elif align == 'right':
58			if len(s) > length:
59			return s[len(s) - length:]
60			else:
61			return (pad_char * (length - len(s))) + s
62			else:
63			raise ValueError("Unknown align type, expected either 'left' or 'right'")
64
65
66			def pad_title(value):
67			"""Pad a title to 30 characters to force the 'details' view."""
68			return str_pad(value, 30, pad_char=' ')
69
70
71			def timestamp():
72			return str(time.time())
73
74
75			# <bound method type.start of <class 'Scrobbler'>>
76			RE_BOUND_METHOD = Regex(r"<bound method (type\.)?(?P<name>.?) of <(class '(?P<class>.?)')?")
			0 ignored issues – show Comprehensibility Best Practice introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report Undefined variable 'Regex' Loading history...
77
78
79			def get_func_name(obj):
80			if inspect.ismethod(obj):
81			match = RE_BOUND_METHOD.match(repr(obj))
82
83			if match:
84			cls = match.group('class')
85			if not cls:
86			return match.group('name')
87
88			return '%s.%s' % (
89			match.group('class'),
90			match.group('name')
91			)
92
93			return None
94
95
96			def get_class_name(cls):
97			if not inspect.isclass(cls):
98			cls = getattr(cls, '__class__')
99
100			return getattr(cls, '__name__')
101
102
103			def spawn(func, args, *kwargs):
104			thread_name = kwargs.pop('thread_name', None) or get_func_name(func)
105
106			# Construct thread
107			th = threading.Thread(target=thread_wrapper, name=thread_name, kwargs={
108			'func': func,
109			'args': args,
110			'kwargs': kwargs,
111			'thread_name': thread_name
112			})
113
114			# Set daemon mode
115			th.daemon = kwargs.pop('daemon', False)
116
117			# Start thread
118			try:
119			th.start()
120			log.debug("Spawned thread with name '%s'" % thread_name)
121			except thread.error as ex:
122			log.error('Unable to spawn thread: %s', ex, exc_info=True, extra={
123			'data': {
124			'active_count': threading.active_count()
125			}
126			})
127			return None
128
129			return th
130
131
132			def thread_wrapper(func, args=None, kwargs=None, thread_name=None):
133			if args is None:
134			args = ()
135
136			if kwargs is None:
137			kwargs = {}
138
139			try:
140			func(args, *kwargs)
141			except Exception as ex:
			0 ignored issues – show Best Practice introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report Catching very general exceptions such as `Exception` is usually not recommended. Generally, you would want to handle very specific errors in the exception handler. This ensure that you do not hide other types of errors which should be fixed. So, unless you specifically plan to handle any error, consider adding a more specific exception. Loading history...
142			log.error('Exception raised in thread "%s": %s', thread_name, ex, exc_info=True)
143
144
145			def safe_encode(string):
146			string = str(string)
147			return base64.b64encode(string).replace('/', '@').replace('+', '*').replace('=', '_')
148
149
150			def pack(obj):
151			serialized_obj = cerealizer.dumps(obj)
152			encoded_string = safe_encode(serialized_obj)
153			return urllib.quote(encoded_string)
			0 ignored issues – show Bug introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report The Module `urllib` does not seem to have a member named `quote`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
154
155
156			def function_path(name, ext=None, **kwargs):
157			return '%s/:/function/%s%s?%s' % (
158			PLUGIN_PREFIX,
159			name,
160			('.%s' % ext) if ext else '',
161
162			urllib.urlencode({
			0 ignored issues – show Bug introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report The Module `urllib` does not seem to have a member named `urlencode`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
163			'function_args': pack(kwargs)
164			})
165			)
166
167
168			def redirect(path, **kwargs):
169			location = PLUGIN_PREFIX + path
170
171			try:
172			request = Core.sandbox.context.request
			0 ignored issues – show Comprehensibility Best Practice introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report Undefined variable 'Core' Loading history...
173
174			# Add request parameters (required for authentication on some clients)
175			kwargs.update({
176			# Client
177			'X-Plex-Client-Identifier': request.headers.get('X-Plex-Client-Identifier'),
178			'X-Plex-Product': request.headers.get('X-Plex-Product'),
179			'X-Plex-Version': request.headers.get('X-Plex-Version'),
180
181			# Platform
182			'X-Plex-Platform': request.headers.get('X-Plex-Platform'),
183			'X-Plex-Platform-Version': request.headers.get('X-Plex-Platform-Version'),
184
185			# Device
186			'X-Plex-Device': request.headers.get('X-Plex-Device'),
187			'X-Plex-Device-Name': request.headers.get('X-Plex-Device-Name'),
188			'X-Plex-Device-Screen-Resolution': request.headers.get('X-Plex-Device-Screen-Resolution'),
189
190			# Authentication
191			'X-Plex-Token': request.headers.get('X-Plex-Token')
192			})
193
194			# Retrieve protocol
195			protocol = request.protocol
196
197			if request.host.endswith('.plex.direct:32400'):
198			# Assume secure connection
199			protocol = 'https'
200
201			# Prepend protocol and host (if not already in `location`)
202			if request and request.host and location[0] == "/":
203			location = protocol + "://" + request.host + location
204			except Exception as ex:
			0 ignored issues – show Best Practice introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report Catching very general exceptions such as `Exception` is usually not recommended. Generally, you would want to handle very specific errors in the exception handler. This ensure that you do not hide other types of errors which should be fixed. So, unless you specifically plan to handle any error, consider adding a more specific exception. Loading history...
205			log.warn('Redirect - %s', str(ex), exc_info=True)
206
207			# Append parameters
208			if kwargs:
209			location += '?' + urllib.urlencode([
			0 ignored issues – show Bug introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report The Module `urllib` does not seem to have a member named `urlencode`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
210			(key, value) for key, value in kwargs.items()
211			if value is not None
212			])
213
214			# Return redirect response
215			return Redirect(location, True)
			0 ignored issues – show Comprehensibility Best Practice introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report Undefined variable 'Redirect' Loading history...
216
217
218			def catch_errors(func):
219			@functools.wraps(func)
220			def inner(args, *kwargs):
221			if InterfaceMessages.critical:
222			return error_record_view(logging.CRITICAL, InterfaceMessages.record)
223
224			try:
225			return func(args, *kwargs)
226			except Exception as ex:
			0 ignored issues – show Best Practice introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report Catching very general exceptions such as `Exception` is usually not recommended. Generally, you would want to handle very specific errors in the exception handler. This ensure that you do not hide other types of errors which should be fixed. So, unless you specifically plan to handle any error, consider adding a more specific exception. Loading history...
227			if InterfaceMessages.critical:
228			return error_record_view(logging.CRITICAL, InterfaceMessages.record)
229
230			log.error('Exception raised in view: %s', ex, exc_info=True)
231
232			return error_view(
233			'Exception',
234			ex.message
			0 ignored issues – show Bug introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report The Instance of `Exception` does not seem to have a member named `message`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
235			)
236
237			return inner
238
239
240			def error_record_view(level, record):
241			# Retrieve level name
242			if level == logging.CRITICAL:
243			level_name = 'Critical Error'
244			else:
245			level_name = logging.getLevelName(level).capitalize()
246
247			# Build error view
248			if not record:
249			return error_view(level_name)
250
251			return error_view(
252			level_name,
253			record.message
254			)
255
256
257			def error_view(title, message=None):
258			oc = ObjectContainer(
			0 ignored issues – show Comprehensibility Best Practice introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report Undefined variable 'ObjectContainer' Loading history...
259			title2=title,
260			no_cache=True
261			)
262
263			oc.add(DirectoryObject(
			0 ignored issues – show Comprehensibility Best Practice introduced 2016-11-10 04:18 UTC by Report Bug Copy Issue Report Undefined variable 'DirectoryObject' Loading history...
264			key=PLUGIN_PREFIX,
265			title=pad_title('%s: %s' % (
266			title,
267			message or 'Unknown'
268			))
269			))
270
271			return oc
272

fuzeman / Plex-Trakt-Scrobbler

redirect() F last analyzed 2018-03-05 01:11 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

redirect() F
last analyzed 2018-03-05 01:11 UTC