redirect() - Code Metrics - Inspection of "Fixed an issue where the plugin could crash if Ple..." - fuzeman/Plex-Trakt-Scrobbler - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( 31cd0f...eceb88 )

by Dean

created 2016-04-13 01:24 UTC

redirect() C

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
dl	0
loc	24
rs	5.5
cc	7

from core.logger import Logger

from plugin.core.constants import PLUGIN_PREFIX

import base64
import cerealizer
import hashlib
import inspect
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 add_attribute(target, source, key, value_type=str, func=None, target_key=None):
    if target_key is None:
        target_key = key

    value = try_convert(source.get(key, None), value_type)

    if value:
        target[target_key] = func(value) if func else value


def merge(a, b):
    a.update(b)
    return a


def all(items):
    for item in items:
        if not item:
            return False
    return True


def any(items):
    for item in items:
        if item:
            return True

    return False


def json_import():
    try:
        import simplejson as json

        log.info("Using 'simplejson' module for JSON serialization")
        return json, 'json'
    except ImportError:
        pass

    # Try fallback to 'json' module
    try:
        import json

        log.info("Using 'json' module for JSON serialization")
        return json, 'json'
    except ImportError:
        pass

    # Try fallback to 'demjson' module
    try:
        import demjson

        log.info("Using 'demjson' module for JSON serialization")
        return demjson, 'demjson'
    except ImportError:
        log.warn("Unable to find json module for serialization")
        raise Exception("Unable to find json module for serialization")

# Import json serialization module
JSON, JSON_MODULE = json_import()


# JSON serialization wrappers to simplejson/json or demjson
def json_decode(s):
    if JSON_MODULE == 'json':
        return JSON.loads(s)

    if JSON_MODULE == 'demjson':
        return JSON.decode(s)

    raise NotImplementedError()


def json_encode(obj):
    if JSON_MODULE == 'json':
        return JSON.dumps(obj)

    if JSON_MODULE == 'demjson':
        return JSON.encode(obj)

    raise NotImplementedError()


def str_format(s, *args, **kwargs):
    """Return a formatted version of S, using substitutions from args and kwargs.

    (Roughly matches the functionality of str.format but ensures compatibility with Python 2.5)
    """

    args = list(args)

    x = 0
    while x < len(s):
        # Skip non-start token characters
        if s[x] != '{':
            x += 1
            continue

        end_pos = s.find('}', x)

        # If end character can't be found, move to next character
        if end_pos == -1:
            x += 1
            continue

        name = s[x + 1:end_pos]

        # Ensure token name is alpha numeric
        if not name.isalnum():
            x += 1
            continue

        # Try find value for token
        value = args.pop(0) if args else kwargs.get(name)

        if value:
            value = str(value)

            # Replace token with value
            s = s[:x] + value + s[end_pos + 1:]

            # Update current position
            x = x + len(value) - 1

        x += 1

    return s


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 total_seconds(span):
    return (span.microseconds + (span.seconds + span.days * 24 * 3600) * 1e6) / 1e6


def sum(values):
    result = 0

    for x in values:
        result = result + x

    return result


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)

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

    try:
        th.start()
        log.debug("Spawned thread with name '%s'" % thread_name)
    except thread.error, 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, ex:
        log.error('Exception raised in thread "%s": %s', thread_name, ex, exc_info=True)


def schedule(func, seconds, *args, **kwargs):
    def schedule_sleep():
        time.sleep(seconds)
        func(*args, **kwargs)

    spawn(schedule_sleep)


def build_repr(obj, keys):
    key_part = ', '.join([
        ('%s: %s' % (key, repr(getattr(obj, key))))
        for key in keys
    ])

    cls = getattr(obj, '__class__')

    return '<%s %s>' % (getattr(cls, '__name__'), key_part)


def plural(value):
    if type(value) is list:
        value = len(value)

    if value == 1:
        return ''

    return 's'


def join_attributes(**kwargs):
    fragments = [
        (('%s: %s' % (key, value)) if value else None)
        for (key, value) in kwargs.items()
    ]

    return ', '.join([x for x in fragments if x])


def md5(value):
    # Generate MD5 hash of key
    m = hashlib.md5()
    m.update(value)

    return m.hexdigest()


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

    if kwargs:
        location += '?' + urllib.urlencode(kwargs)

    try:
        request = Core.sandbox.context.request

        # Retrieve protocol
        protocol = request.protocol

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

        # Build URL
        if request and request.host and location[0] == "/":
            location = protocol + "://" + request.host + location
    except Exception, ex:
        log.warn('Redirect - %s', str(ex), exc_info=True)

    # Return redirect response
    return Redirect(location)


1			from core.logger import Logger
2
3			from plugin.core.constants import PLUGIN_PREFIX
4
5			import base64
6			import cerealizer
7			import hashlib
8			import inspect
9			import sys
10			import threading
11			import thread
12			import time
13			import urllib
14
15			log = Logger('core.helpers')
16
17
18			PY25 = sys.version_info[0] == 2 and sys.version_info[1] == 5
19
20
21			def try_convert(value, value_type, default=None):
22			try:
23			return value_type(value)
24			except ValueError:
25			return default
26			except TypeError:
27			return default
28
29
30			def add_attribute(target, source, key, value_type=str, func=None, target_key=None):
31			if target_key is None:
32			target_key = key
33
34			value = try_convert(source.get(key, None), value_type)
35
36			if value:
37			target[target_key] = func(value) if func else value
38
39
40			def merge(a, b):
41			a.update(b)
42			return a
43
44
45			def all(items):
46			for item in items:
47			if not item:
48			return False
49			return True
50
51
52			def any(items):
53			for item in items:
54			if item:
55			return True
56
57			return False
58
59
60			def json_import():
61			try:
62			import simplejson as json
63
64			log.info("Using 'simplejson' module for JSON serialization")
65			return json, 'json'
66			except ImportError:
67			pass
68
69			# Try fallback to 'json' module
70			try:
71			import json
72
73			log.info("Using 'json' module for JSON serialization")
74			return json, 'json'
75			except ImportError:
76			pass
77
78			# Try fallback to 'demjson' module
79			try:
80			import demjson
81
82			log.info("Using 'demjson' module for JSON serialization")
83			return demjson, 'demjson'
84			except ImportError:
85			log.warn("Unable to find json module for serialization")
86			raise Exception("Unable to find json module for serialization")
87
88			# Import json serialization module
89			JSON, JSON_MODULE = json_import()
90
91
92			# JSON serialization wrappers to simplejson/json or demjson
93			def json_decode(s):
94			if JSON_MODULE == 'json':
95			return JSON.loads(s)
96
97			if JSON_MODULE == 'demjson':
98			return JSON.decode(s)
99
100			raise NotImplementedError()
101
102
103			def json_encode(obj):
104			if JSON_MODULE == 'json':
105			return JSON.dumps(obj)
106
107			if JSON_MODULE == 'demjson':
108			return JSON.encode(obj)
109
110			raise NotImplementedError()
111
112
113			def str_format(s, args, *kwargs):
114			"""Return a formatted version of S, using substitutions from args and kwargs.
115
116			(Roughly matches the functionality of str.format but ensures compatibility with Python 2.5)
117			"""
118
119			args = list(args)
120
121			x = 0
122			while x < len(s):
123			# Skip non-start token characters
124			if s[x] != '{':
125			x += 1
126			continue
127
128			end_pos = s.find('}', x)
129
130			# If end character can't be found, move to next character
131			if end_pos == -1:
132			x += 1
133			continue
134
135			name = s[x + 1:end_pos]
136
137			# Ensure token name is alpha numeric
138			if not name.isalnum():
139			x += 1
140			continue
141
142			# Try find value for token
143			value = args.pop(0) if args else kwargs.get(name)
144
145			if value:
146			value = str(value)
147
148			# Replace token with value
149			s = s[:x] + value + s[end_pos + 1:]
150
151			# Update current position
152			x = x + len(value) - 1
153
154			x += 1
155
156			return s
157
158
159			def str_pad(s, length, align='left', pad_char=' ', trim=False):
160			if not s:
161			return s
162
163			if not isinstance(s, (str, unicode)):
164			s = str(s)
165
166			if len(s) == length:
167			return s
168			elif len(s) > length and not trim:
169			return s
170
171			if align == 'left':
172			if len(s) > length:
173			return s[:length]
174			else:
175			return s + (pad_char * (length - len(s)))
176			elif align == 'right':
177			if len(s) > length:
178			return s[len(s) - length:]
179			else:
180			return (pad_char * (length - len(s))) + s
181			else:
182			raise ValueError("Unknown align type, expected either 'left' or 'right'")
183
184
185			def pad_title(value):
186			"""Pad a title to 30 characters to force the 'details' view."""
187			return str_pad(value, 30, pad_char=' ')
188
189
190			def total_seconds(span):
191			return (span.microseconds + (span.seconds + span.days * 24 * 3600) * 1e6) / 1e6
192
193
194			def sum(values):
195			result = 0
196
197			for x in values:
198			result = result + x
199
200			return result
201
202
203			def timestamp():
204			return str(time.time())
205
206
207			# <bound method type.start of <class 'Scrobbler'>>
208			RE_BOUND_METHOD = Regex(r"<bound method (type\.)?(?P<name>.?) of <(class '(?P<class>.?)')?")
209
210
211			def get_func_name(obj):
212			if inspect.ismethod(obj):
213			match = RE_BOUND_METHOD.match(repr(obj))
214
215			if match:
216			cls = match.group('class')
217			if not cls:
218			return match.group('name')
219
220			return '%s.%s' % (
221			match.group('class'),
222			match.group('name')
223			)
224
225			return None
226
227
228			def get_class_name(cls):
229			if not inspect.isclass(cls):
230			cls = getattr(cls, '__class__')
231
232			return getattr(cls, '__name__')
233
234
235			def spawn(func, args, *kwargs):
236			thread_name = kwargs.pop('thread_name', None) or get_func_name(func)
237
238			th = threading.Thread(target=thread_wrapper, name=thread_name, kwargs={
239			'func': func,
240			'args': args,
241			'kwargs': kwargs,
242			'thread_name': thread_name
243			})
244
245			try:
246			th.start()
247			log.debug("Spawned thread with name '%s'" % thread_name)
248			except thread.error, ex:
249			log.error('Unable to spawn thread: %s', ex, exc_info=True, extra={
250			'data': {
251			'active_count': threading.active_count()
252			}
253			})
254			return None
255
256			return th
257
258
259			def thread_wrapper(func, args=None, kwargs=None, thread_name=None):
260			if args is None:
261			args = ()
262
263			if kwargs is None:
264			kwargs = {}
265
266			try:
267			func(args, *kwargs)
268			except Exception, ex:
269			log.error('Exception raised in thread "%s": %s', thread_name, ex, exc_info=True)
270
271
272			def schedule(func, seconds, args, *kwargs):
273			def schedule_sleep():
274			time.sleep(seconds)
275			func(args, *kwargs)
276
277			spawn(schedule_sleep)
278
279
280			def build_repr(obj, keys):
281			key_part = ', '.join([
282			('%s: %s' % (key, repr(getattr(obj, key))))
283			for key in keys
284			])
285
286			cls = getattr(obj, '__class__')
287
288			return '<%s %s>' % (getattr(cls, '__name__'), key_part)
289
290
291			def plural(value):
292			if type(value) is list:
293			value = len(value)
294
295			if value == 1:
296			return ''
297
298			return 's'
299
300
301			def join_attributes(**kwargs):
302			fragments = [
303			(('%s: %s' % (key, value)) if value else None)
304			for (key, value) in kwargs.items()
305			]
306
307			return ', '.join([x for x in fragments if x])
308
309
310			def md5(value):
311			# Generate MD5 hash of key
312			m = hashlib.md5()
313			m.update(value)
314
315			return m.hexdigest()
316
317
318			def safe_encode(string):
319			string = str(string)
320			return base64.b64encode(string).replace('/', '@').replace('+', '*').replace('=', '_')
321
322
323			def pack(obj):
324			serialized_obj = cerealizer.dumps(obj)
325			encoded_string = safe_encode(serialized_obj)
326			return urllib.quote(encoded_string)
327
328
329			def function_path(name, ext=None, **kwargs):
330			return '%s/:/function/%s%s?%s' % (
331			PLUGIN_PREFIX,
332			name,
333			('.%s' % ext) if ext else '',
334
335			urllib.urlencode({
336			'function_args': pack(kwargs)
337			})
338			)
339
340
341			def redirect(path, **kwargs):
342			location = PLUGIN_PREFIX + path
343
344			if kwargs:
345			location += '?' + urllib.urlencode(kwargs)
346
347			try:
348			request = Core.sandbox.context.request
349
350			# Retrieve protocol
351			protocol = request.protocol
352
353			if request.host.endswith('.plex.direct:32400'):
354			# Secure connection
355			protocol = 'https'
356
357			# Build URL
358			if request and request.host and location[0] == "/":
359			location = protocol + "://" + request.host + location
360			except Exception, ex:
361			log.warn('Redirect - %s', str(ex), exc_info=True)
362
363			# Return redirect response
364			return Redirect(location)
365

fuzeman / Plex-Trakt-Scrobbler

Push — develop ( 31cd0f...eceb88 )

redirect() C

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like