LibrariesManager - Code Metrics - Inspection of "Merge branch 'release/1.3.0'" - fuzeman/Plex-Trakt-Scrobbler - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 6e8656...9792e6 )

by Dean

created 2016-12-30 04:51 UTC

LibrariesManager C

↳ Parent: Project

Complexity

Total Complexity

Size/Duplication

Total Lines	337
Duplicated Lines	0 %

Test Coverage

Coverage

67.09%

Importance

Changes	2
Bugs	0	Features	1

Metric	Value
wmc	57
c	2
b	0
f	1
dl	0
loc	337
ccs	106
cts	158
cp	0.6709
rs	5.1724

10 Methods

Rating	Name	Size	Complexity
C	_insert_paths_unix()	34	7
A	_insert_paths_windows()	18	2
B	setup()	36	4
B	_libraries_path()	37	5
F	_read_distribution_metadata()	33	9
B	reset()	25	4
D	_insert_paths()	43	8
A	_insert_architecture_paths()	19	3
A	_cache_libraries()	13	2
F	test()	67	13

How to fix Complexity

from plugin.core.configuration import Configuration
from plugin.core.environment import Environment
from plugin.core.message import InterfaceMessages
from plugin.core.helpers.variable import merge
from plugin.core.libraries.cache import CacheManager
from plugin.core.libraries.constants import CONTENTS_PATH, NATIVE_DIRECTORIES, UNICODE_MAP
from plugin.core.libraries.helpers import PathHelper, StorageHelper, SystemHelper
from plugin.core.libraries.tests import LIBRARY_TESTS
from plugin.core.logger.handlers.error_reporter import ErrorReporter

import json
import logging
import os
import platform
import sys

log = logging.getLogger(__name__)


class LibrariesManager(object):
    @classmethod
    def setup(cls, cache=False):
        """Setup native library directories

        :param cache: Enable native library caching
        :type cache: bool
        """

        # Read distribution metadata
        distribution = cls._read_distribution_metadata()

        # Use `cache` value from advanced configuration
        cache = Configuration.advanced['libraries'].get_boolean('cache', cache)

        # Retrieve libraries path (and cache libraries, if enabled)
        libraries_path = cls._libraries_path(cache)

        if not libraries_path:
            return

        log.info('Using native libraries at %r', StorageHelper.to_relative_path(libraries_path))

        # Remove current native library directories from `sys.path`
        cls.reset()

        # Insert platform specific library paths
        cls._insert_paths(distribution, libraries_path)

        # Display library paths in logfile
        for path in sys.path:
            path = os.path.abspath(path)

            if StorageHelper.is_framework_path(path):
                continue

            log.info('[PATH] %s', StorageHelper.to_relative_path(path))

    @staticmethod
    def test():
        """Test native libraries to ensure they can be correctly loaded"""
        log.info('Testing native library support...')

        # Retrieve library directories
        search_paths = []

        for path in sys.path:
            path_lower = path.lower()

            if 'trakttv.bundle' not in path_lower and 'com.plexapp.plugins.trakttv' not in path_lower:
                continue

            search_paths.append(path)

        # Run library tests
        metadata = {}

        for test in LIBRARY_TESTS:
            # Run tests
            result = test.run(search_paths)

            if not result.get('success'):
                log_func = logging.warn if test.optional else logging.error

                # Write message to logfile
                if 'traceback' in result:
                    log_func('%s: unavailable - %s\n%%s' % (test.name, result.get('message')), result['traceback'])
                else:
                    log_func('%s: unavailable - %s' % (test.name, result.get('message')), exc_info=result.get('exc_info'))


                if not test.optional:
                    return

                continue

            # Test successful
            t_metadata = result.get('metadata') or {}
            t_versions = t_metadata.get('versions')

            if t_versions:
                expanded = len(t_versions) > 1 or (
                    t_versions and t_versions.keys()[0] != test.name
                )

                if expanded:
                    log.info('%s: available (%s)', test.name, ', '.join([
                        '%s: %s' % (key, value)
                        for key, value in t_versions.items()
                    ]))
                else:
                    key = t_versions.keys()[0]

                    log.info('%s: available (%s)', test.name, t_versions[key])
            else:
                log.info('%s: available', test.name)

            # Merge result into `metadata`
            merge(metadata, t_metadata, recursive=True)

        # Include versions in error reports
        versions = metadata.get('versions') or {}

        ErrorReporter.set_tags(dict([
            ('%s.version' % key, value)
            for key, value in versions.items()
        ]))

    @classmethod
    def reset(cls):
        """Remove all the native library directives from `sys.path`"""

        for path in sys.path:
            path = os.path.abspath(path)

            if not path.lower().startswith(CONTENTS_PATH.lower()):
                continue

            # Convert to relative path
            path_rel = os.path.relpath(path, CONTENTS_PATH)

            # Take the first two fragments
            path_rel = os.path.sep.join(path_rel.split(os.path.sep)[:2])

            # Convert to unix-style separators (/)
            path_rel = path_rel.replace('\\', '/')

            # Ignore non-native library directories
            if path_rel not in NATIVE_DIRECTORIES:
                continue

            # Remove from `sys.path`
            PathHelper.remove(path)

    @classmethod
    def _read_distribution_metadata(cls):
        metadata_path = os.path.join(Environment.path.contents, 'distribution.json')

        if not os.path.exists(metadata_path):
            return None

        # Read distribution metadata
        try:
            with open(metadata_path, 'r') as fp:
                distribution = json.load(fp)
        except Exception as ex:

            log.warn('Unable to read distribution metadata: %s', ex, exc_info=True)
            return None

        if not distribution or not distribution.get('name'):
            return

        # Set distribution name tag
        ErrorReporter.set_tags({
            'distribution.name': distribution['name']
        })

        # Set distribution release tags
        release = distribution.get('release')

        if release and release.get('version') and release.get('branch'):
            ErrorReporter.set_tags({
                'distribution.version': release['version'],
                'distribution.branch': release['branch']
            })

        return distribution

    @classmethod
    def _libraries_path(cls, cache=False):
        """Retrieve the native libraries base directory (and cache the libraries if enabled)

        :param cache: Enable native library caching
        :type cache: bool
        """

        # Use specified libraries path (from "advanced.ini')
        libraries_path = Configuration.advanced['libraries'].get('libraries_path')

        if libraries_path and os.path.exists(libraries_path):
            log.info('Using libraries at %r', StorageHelper.to_relative_path(libraries_path))
            ErrorReporter.set_tags({
                'libraries.source': 'custom'
            })
            return libraries_path

        # Use system libraries (if bundled libraries have been disabled in "advanced.ini")
        if not Configuration.advanced['libraries'].get_boolean('bundled', True):
            log.info('Bundled libraries have been disabled, using system libraries')
            ErrorReporter.set_tags({
                'libraries.source': 'system'
            })
            return None

        # Cache libraries (if enabled)
        if cache:
            ErrorReporter.set_tags({
                'libraries.source': 'cache'
            })
            return cls._cache_libraries()

        ErrorReporter.set_tags({
            'libraries.source': 'bundle'
        })
        return Environment.path.libraries

    @classmethod
    def _cache_libraries(cls):
        cache_path = Configuration.advanced['libraries'].get('cache_path')

        # Try cache libraries to `cache_path`
        libraries_path = CacheManager.sync(cache_path)

        if not libraries_path:
            log.info('Unable to cache libraries, using bundled libraries directly')
            return Environment.path.libraries

        log.info('Cached libraries to %r', StorageHelper.to_relative_path(libraries_path))
        return libraries_path

    @classmethod
    def _insert_paths(cls, distribution, libraries_path):
        # Display platform details
        p_bits, _ = platform.architecture()
        p_machine = platform.machine()

        log.debug('Bits: %r, Machine: %r', p_bits, p_machine)

        # Retrieve system details
        system = SystemHelper.name()
        architecture = SystemHelper.architecture()

        if not architecture:
            InterfaceMessages.add(60, 'Unable to retrieve system architecture')
            return False

        log.debug('System: %r, Architecture: %r', system, architecture)

        # Build architecture list
        architectures = [architecture]

        if architecture == 'i686':
            # Fallback to i386
            architectures.append('i386')

        # Insert library paths
        found = False

        for arch in architectures + ['universal']:
            if cls._insert_architecture_paths(libraries_path, system, arch):
                log.debug('Inserted libraries path for system: %r, arch: %r', system, arch)
                found = True

        # Display interface message if no libraries were found
        if not found:
            if distribution and distribution.get('name'):
                message = 'Unable to find compatible native libraries in the %s distribution' % distribution['name']
            else:
                message = 'Unable to find compatible native libraries'

            InterfaceMessages.add(60, '%s (system: %r, architecture: %r)', message, system, architecture)

        return found

    @classmethod
    def _insert_architecture_paths(cls, libraries_path, system, architecture):
        architecture_path = os.path.join(libraries_path, system, architecture)

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

        # Architecture libraries
        PathHelper.insert(libraries_path, system, architecture)

        # System libraries
        if system == 'Windows':
            # Windows libraries (VC++ specific)
            cls._insert_paths_windows(libraries_path, system, architecture)
        else:
            # Darwin/FreeBSD/Linux libraries
            cls._insert_paths_unix(libraries_path, system, architecture)

        return True

    @staticmethod
    def _insert_paths_unix(libraries_path, system, architecture):
        # UCS specific libraries
        ucs = UNICODE_MAP.get(sys.maxunicode)
        log.debug('UCS: %r', ucs)

        if ucs:
            PathHelper.insert(libraries_path, system, architecture, ucs)

        # CPU specific libraries
        cpu_type = SystemHelper.cpu_type()
        page_size = SystemHelper.page_size()

        log.debug('CPU Type: %r', cpu_type)
        log.debug('Page Size: %r', page_size)

        if cpu_type:
            PathHelper.insert(libraries_path, system, architecture, cpu_type)

            if page_size:
                PathHelper.insert(libraries_path, system, architecture, '%s_%s' % (cpu_type, page_size))

        # UCS + CPU specific libraries
        if cpu_type and ucs:
            PathHelper.insert(libraries_path, system, architecture, cpu_type, ucs)

            if page_size:
                PathHelper.insert(libraries_path, system, architecture, '%s_%s' % (cpu_type, page_size), ucs)

        # Include attributes in error reports
        ErrorReporter.set_tags({
            'cpu.type': cpu_type,
            'memory.page_size': page_size,
            'python.ucs': ucs
        })

    @staticmethod
    def _insert_paths_windows(libraries_path, system, architecture):
        vcr = SystemHelper.vcr_version() or 'vc12'  # Assume "vc12" if call fails
        ucs = UNICODE_MAP.get(sys.maxunicode)

        log.debug('VCR: %r, UCS: %r', vcr, ucs)

        # VC++ libraries
        PathHelper.insert(libraries_path, system, architecture, vcr)

        # UCS libraries
        if ucs:
            PathHelper.insert(libraries_path, system, architecture, vcr, ucs)

        # Include attributes in error reports
        ErrorReporter.set_tags({
            'python.ucs': ucs,
            'vcr.version': vcr
        })


1	1	from plugin.core.configuration import Configuration
2	1	from plugin.core.environment import Environment
3	1	from plugin.core.message import InterfaceMessages
4	1	from plugin.core.helpers.variable import merge
5	1	from plugin.core.libraries.cache import CacheManager
6	1	from plugin.core.libraries.constants import CONTENTS_PATH, NATIVE_DIRECTORIES, UNICODE_MAP
7	1	from plugin.core.libraries.helpers import PathHelper, StorageHelper, SystemHelper
8	1	from plugin.core.libraries.tests import LIBRARY_TESTS
9	1	from plugin.core.logger.handlers.error_reporter import ErrorReporter
10
11	1	import json
12	1	import logging
13	1	import os
14	1	import platform
15	1	import sys
16
17	1	log = logging.getLogger(__name__)
18
19
20	1	class LibrariesManager(object):
21	1	@classmethod
22	1	def setup(cls, cache=False):
23		"""Setup native library directories
24
25		:param cache: Enable native library caching
26		:type cache: bool
27		"""
28
29		# Read distribution metadata
30	1	distribution = cls._read_distribution_metadata()
31
32		# Use `cache` value from advanced configuration
33	1	cache = Configuration.advanced['libraries'].get_boolean('cache', cache)
34
35		# Retrieve libraries path (and cache libraries, if enabled)
36	1	libraries_path = cls._libraries_path(cache)
37
38	1	if not libraries_path:
39		return
40
41	1	log.info('Using native libraries at %r', StorageHelper.to_relative_path(libraries_path))
42
43		# Remove current native library directories from `sys.path`
44	1	cls.reset()
45
46		# Insert platform specific library paths
47	1	cls._insert_paths(distribution, libraries_path)
48
49		# Display library paths in logfile
50	1	for path in sys.path:
51	1	path = os.path.abspath(path)
52
53	1	if StorageHelper.is_framework_path(path):
54		continue
55
56	1	log.info('[PATH] %s', StorageHelper.to_relative_path(path))
57
58	1	@staticmethod
59		def test():
60		"""Test native libraries to ensure they can be correctly loaded"""
61	1	log.info('Testing native library support...')
62
63		# Retrieve library directories
64	1	search_paths = []
65
66	1	for path in sys.path:
67	1	path_lower = path.lower()
68
69	1	if 'trakttv.bundle' not in path_lower and 'com.plexapp.plugins.trakttv' not in path_lower:
70	1	continue
71
72	1	search_paths.append(path)
73
74		# Run library tests
75	1	metadata = {}
76
77	1	for test in LIBRARY_TESTS:
78		# Run tests
79	1	result = test.run(search_paths)
80
81	1	if not result.get('success'):
82		log_func = logging.warn if test.optional else logging.error
83
84		# Write message to logfile
85		if 'traceback' in result:
86		log_func('%s: unavailable - %s\n%%s' % (test.name, result.get('message')), result['traceback'])
87		else:
88		log_func('%s: unavailable - %s' % (test.name, result.get('message')), exc_info=result.get('exc_info'))
		0 ignored issues – show Coding Style introduced 2016-06-06 12:03 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (122/120). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
89
90		if not test.optional:
91		return
92
93		continue
94
95		# Test successful
96	1	t_metadata = result.get('metadata') or {}
97	1	t_versions = t_metadata.get('versions')
98
99	1	if t_versions:
100	1	expanded = len(t_versions) > 1 or (
101		t_versions and t_versions.keys()[0] != test.name
102		)
103
104	1	if expanded:
105	1	log.info('%s: available (%s)', test.name, ', '.join([
106		'%s: %s' % (key, value)
107		for key, value in t_versions.items()
108		]))
109		else:
110	1	key = t_versions.keys()[0]
111
112	1	log.info('%s: available (%s)', test.name, t_versions[key])
113		else:
114	1	log.info('%s: available', test.name)
115
116		# Merge result into `metadata`
117	1	merge(metadata, t_metadata, recursive=True)
118
119		# Include versions in error reports
120	1	versions = metadata.get('versions') or {}
121
122	1	ErrorReporter.set_tags(dict([
123		('%s.version' % key, value)
124		for key, value in versions.items()
125		]))
126
127	1	@classmethod
128		def reset(cls):
129		"""Remove all the native library directives from `sys.path`"""
130
131	1	for path in sys.path:
132	1	path = os.path.abspath(path)
133
134	1	if not path.lower().startswith(CONTENTS_PATH.lower()):
135	1	continue
136
137		# Convert to relative path
138	1	path_rel = os.path.relpath(path, CONTENTS_PATH)
139
140		# Take the first two fragments
141	1	path_rel = os.path.sep.join(path_rel.split(os.path.sep)[:2])
142
143		# Convert to unix-style separators (/)
144	1	path_rel = path_rel.replace('\\', '/')
145
146		# Ignore non-native library directories
147	1	if path_rel not in NATIVE_DIRECTORIES:
148	1	continue
149
150		# Remove from `sys.path`
151		PathHelper.remove(path)
152
153	1	@classmethod
154		def _read_distribution_metadata(cls):
155	1	metadata_path = os.path.join(Environment.path.contents, 'distribution.json')
156
157	1	if not os.path.exists(metadata_path):
158		return None
159
160		# Read distribution metadata
161	1	try:
162	1	with open(metadata_path, 'r') as fp:
163	1	distribution = json.load(fp)
164		except Exception as ex:
		0 ignored issues – show Best Practice introduced 2016-12-30 03:05 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...
165		log.warn('Unable to read distribution metadata: %s', ex, exc_info=True)
166		return None
167
168	1	if not distribution or not distribution.get('name'):
169		return
170
171		# Set distribution name tag
172	1	ErrorReporter.set_tags({
173		'distribution.name': distribution['name']
174		})
175
176		# Set distribution release tags
177	1	release = distribution.get('release')
178
179	1	if release and release.get('version') and release.get('branch'):
180		ErrorReporter.set_tags({
181		'distribution.version': release['version'],
182		'distribution.branch': release['branch']
183		})
184
185	1	return distribution
186
187	1	@classmethod
188	1	def _libraries_path(cls, cache=False):
189		"""Retrieve the native libraries base directory (and cache the libraries if enabled)
190
191		:param cache: Enable native library caching
192		:type cache: bool
193		"""
194
195		# Use specified libraries path (from "advanced.ini')
196	1	libraries_path = Configuration.advanced['libraries'].get('libraries_path')
197
198	1	if libraries_path and os.path.exists(libraries_path):
199		log.info('Using libraries at %r', StorageHelper.to_relative_path(libraries_path))
200		ErrorReporter.set_tags({
201		'libraries.source': 'custom'
202		})
203		return libraries_path
204
205		# Use system libraries (if bundled libraries have been disabled in "advanced.ini")
206	1	if not Configuration.advanced['libraries'].get_boolean('bundled', True):
207		log.info('Bundled libraries have been disabled, using system libraries')
208		ErrorReporter.set_tags({
209		'libraries.source': 'system'
210		})
211		return None
212
213		# Cache libraries (if enabled)
214	1	if cache:
215		ErrorReporter.set_tags({
216		'libraries.source': 'cache'
217		})
218		return cls._cache_libraries()
219
220	1	ErrorReporter.set_tags({
221		'libraries.source': 'bundle'
222		})
223	1	return Environment.path.libraries
224
225	1	@classmethod
226		def _cache_libraries(cls):
227		cache_path = Configuration.advanced['libraries'].get('cache_path')
228
229		# Try cache libraries to `cache_path`
230		libraries_path = CacheManager.sync(cache_path)
231
232		if not libraries_path:
233		log.info('Unable to cache libraries, using bundled libraries directly')
234		return Environment.path.libraries
235
236		log.info('Cached libraries to %r', StorageHelper.to_relative_path(libraries_path))
237		return libraries_path
238
239	1	@classmethod
240		def _insert_paths(cls, distribution, libraries_path):
241		# Display platform details
242	1	p_bits, _ = platform.architecture()
243	1	p_machine = platform.machine()
244
245	1	log.debug('Bits: %r, Machine: %r', p_bits, p_machine)
246
247		# Retrieve system details
248	1	system = SystemHelper.name()
249	1	architecture = SystemHelper.architecture()
250
251	1	if not architecture:
252		InterfaceMessages.add(60, 'Unable to retrieve system architecture')
253		return False
254
255	1	log.debug('System: %r, Architecture: %r', system, architecture)
256
257		# Build architecture list
258	1	architectures = [architecture]
259
260	1	if architecture == 'i686':
261		# Fallback to i386
262		architectures.append('i386')
263
264		# Insert library paths
265	1	found = False
266
267	1	for arch in architectures + ['universal']:
268	1	if cls._insert_architecture_paths(libraries_path, system, arch):
269	1	log.debug('Inserted libraries path for system: %r, arch: %r', system, arch)
270	1	found = True
271
272		# Display interface message if no libraries were found
273	1	if not found:
274		if distribution and distribution.get('name'):
275		message = 'Unable to find compatible native libraries in the %s distribution' % distribution['name']
276		else:
277		message = 'Unable to find compatible native libraries'
278
279		InterfaceMessages.add(60, '%s (system: %r, architecture: %r)', message, system, architecture)
280
281	1	return found
282
283	1	@classmethod
284		def _insert_architecture_paths(cls, libraries_path, system, architecture):
285	1	architecture_path = os.path.join(libraries_path, system, architecture)
286
287	1	if not os.path.exists(architecture_path):
288	1	return False
289
290		# Architecture libraries
291	1	PathHelper.insert(libraries_path, system, architecture)
292
293		# System libraries
294	1	if system == 'Windows':
295		# Windows libraries (VC++ specific)
296		cls._insert_paths_windows(libraries_path, system, architecture)
297		else:
298		# Darwin/FreeBSD/Linux libraries
299	1	cls._insert_paths_unix(libraries_path, system, architecture)
300
301	1	return True
302
303	1	@staticmethod
304		def _insert_paths_unix(libraries_path, system, architecture):
305		# UCS specific libraries
306	1	ucs = UNICODE_MAP.get(sys.maxunicode)
307	1	log.debug('UCS: %r', ucs)
308
309	1	if ucs:
310	1	PathHelper.insert(libraries_path, system, architecture, ucs)
311
312		# CPU specific libraries
313	1	cpu_type = SystemHelper.cpu_type()
314	1	page_size = SystemHelper.page_size()
315
316	1	log.debug('CPU Type: %r', cpu_type)
317	1	log.debug('Page Size: %r', page_size)
318
319	1	if cpu_type:
320		PathHelper.insert(libraries_path, system, architecture, cpu_type)
321
322		if page_size:
323		PathHelper.insert(libraries_path, system, architecture, '%s_%s' % (cpu_type, page_size))
324
325		# UCS + CPU specific libraries
326	1	if cpu_type and ucs:
327		PathHelper.insert(libraries_path, system, architecture, cpu_type, ucs)
328
329		if page_size:
330		PathHelper.insert(libraries_path, system, architecture, '%s_%s' % (cpu_type, page_size), ucs)
331
332		# Include attributes in error reports
333	1	ErrorReporter.set_tags({
334		'cpu.type': cpu_type,
335		'memory.page_size': page_size,
336		'python.ucs': ucs
337		})
338
339	1	@staticmethod
340		def _insert_paths_windows(libraries_path, system, architecture):
341		vcr = SystemHelper.vcr_version() or 'vc12' # Assume "vc12" if call fails
342		ucs = UNICODE_MAP.get(sys.maxunicode)
343
344		log.debug('VCR: %r, UCS: %r', vcr, ucs)
345
346		# VC++ libraries
347		PathHelper.insert(libraries_path, system, architecture, vcr)
348
349		# UCS libraries
350		if ucs:
351		PathHelper.insert(libraries_path, system, architecture, vcr, ucs)
352
353		# Include attributes in error reports
354		ErrorReporter.set_tags({
355		'python.ucs': ucs,
356		'vcr.version': vcr
357		})
358

fuzeman / Plex-Trakt-Scrobbler

Push — master ( 6e8656...9792e6 )

LibrariesManager C

Complexity

Size/Duplication

Test Coverage

Importance

10 Methods

How to fix Complexity

Complex Class

Duplication Side-by-Side

Filter issues like