Libraries.cache() - Code Metrics - Inspection of "Merge branch 'issue/266' into develop" - fuzeman/Plex-Trakt-Scrobbler - Measure and Improve Code Quality continuously with Scrutinizer

Test Setup Failed

Push — develop ( d0ed0b...b83390 )

by Dean

created 2016-01-25 04:28 UTC

Libraries.cache() B

↳ Parent: Libraries

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	5
dl	0
loc	33
rs	8.0894

from plugin.core.environment import Environment

import os
import platform
import shutil
import sys


# Create dummy `Log` (for tests)
try:
    Log.Debug('Using framework "Log" handler')
except NameError:
    from plex_mock.framework import Logger

    Log = Logger()
    Log.Debug('Using dummy "Log" handler')


class Libraries(object):
    contents_path = os.path.abspath(os.path.join(Environment.path.code, '..'))

    native_directories = [
        'Libraries\\FreeBSD',
        'Libraries\\Linux',
        'Libraries\\MacOSX',
        'Libraries\\Windows'
    ]

    unicode_map = {
        65535:      'ucs2',
        1114111:    'ucs4'
    }

    @classmethod
    def cache(cls):
        """Cache native libraries into the plugin data directory"""

        # Retrieve library platforms
        libraries_path = os.path.join(cls.contents_path, 'Libraries')
        platforms = os.listdir(libraries_path)

        if 'Shared' in platforms:
            platforms.remove('Shared')

        # Create destination directory
        destination = os.path.join(Environment.path.plugin_data, 'Libraries')

        # Ensure destination exists
        StorageHelper.create_directories(destination)

        # Delete existing libraries
        for name in os.listdir(destination):
            path = os.path.join(destination, name)

            StorageHelper.delete_tree(path)

        # Copy libraries to directory
        for name in platforms:
            p_source = os.path.join(libraries_path, name)
            p_destination = os.path.join(destination, name)

            if not StorageHelper.copy_tree(p_source, p_destination):
                return None

        Log.Debug('Cached native libraries to %r', StorageHelper.to_relative_path(destination))
        return destination

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

        if not cache:
            return Environment.path.libraries

        # Cache native libraries
        libraries_path = cls.cache()

        if libraries_path:
            # Reset native library directories in `sys.path`
            cls.reset()

            return libraries_path

        return Environment.path.libraries

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

        # Retrieve libraries path
        libraries_path = cls.get_libraries_path(cache)

        Log.Info('Using native libraries at %r', StorageHelper.to_relative_path(libraries_path))

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

        if not system_architecture:
            return

        Log.Debug('System: %r, Architecture: %r', system, system_architecture)

        architectures = [system_architecture]

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

        for architecture in reversed(architectures + ['universal']):
            # Common
            PathHelper.insert(libraries_path, system, architecture)

            # UCS
            if sys.maxunicode in cls.unicode_map:
                PathHelper.insert(libraries_path, system, architecture, cls.unicode_map[sys.maxunicode])

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

            if not StorageHelper.is_relative_path(path):
                continue

            Log.Info('[PATH] %s', StorageHelper.to_relative_path(path))

    @staticmethod
    def test():
        Log.Info('Testing native library support...')

        # Check "apsw" availability
        try:
            import apsw

            Log.Info(' - apsw: %r, sqlite: %r', apsw.apswversion(), apsw.SQLITE_VERSION_NUMBER)
        except Exception, ex:
            Log.Error(' - Unable to import "apsw": %s', ex)

        # Check "llist" availability
        try:
            import llist

            Log.Info(' - llist: available')
        except Exception, ex:
            Log.Warn(' - Unable to import "llist": %s', ex)

    @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(cls.contents_path.lower()):
                continue

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

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

            # Ignore non-native library directories
            if path_rel not in cls.native_directories:
                continue

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


class PathHelper(object):
    @classmethod
    def insert(cls, base, system, architecture, *args):
        """Insert a new path into `sys.path` if it passes basic validation"""

        path = os.path.join(base, system, architecture, *args)

        if path in sys.path:
            return False

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

        sys.path.insert(0, path)

        Log.Debug('Inserted path: %r', StorageHelper.to_relative_path(path))
        return True

    @classmethod
    def remove(cls, path):
        """Remove path from `sys.path` if it exists"""

        if path not in sys.path:
            return False

        sys.path.remove(path)

        Log.Debug('Removed path: %r', StorageHelper.to_relative_path(path))
        return True


class StorageHelper(object):
    base_names = [
        'plug-ins',
        'plug-in support',
        'trakttv.bundle'
    ]

    @classmethod
    def create_directories(cls, path, *args, **kwargs):
        """Create directory at `path` include any parent directories"""

        try:
            os.makedirs(path, *args, **kwargs)
            return True
        except OSError, ex:
            if ex.errno == 17:
                # Directory already exists
                return True

            Log.Warn('Unable to create directories: %r - (%s) %s', cls.to_relative_path(path), ex.errno, ex)
        except Exception, ex:
            Log.Warn('Unable to create directories: %r - (%s) %s', cls.to_relative_path(path), type(ex), ex)

        return False

    @classmethod
    def copy_tree(cls, source, destination):
        """Copy the directory at `source` to `destination`"""

        try:
            shutil.copytree(source, destination)

            Log.Debug('Copied %r to %r', cls.to_relative_path(source), cls.to_relative_path(destination))
            return True
        except Exception, ex:
            Log.Warn('Unable to copy %r to %r - %s', cls.to_relative_path(source), cls.to_relative_path(destination), ex)

        return False

    @classmethod
    def delete_tree(cls, path):
        """Delete the directory (at `path`"""

        try:
            shutil.rmtree(path)

            Log.Debug('Deleted %r', cls.to_relative_path(path))
            return True
        except Exception, ex:
            Log.Warn('Unable to delete directory: %r - %s', cls.to_relative_path(path), ex)

        return False

    @classmethod
    def to_relative_path(cls, path):
        """Convert `path` to be relative to `StorageHelper.base_names`"""

        path_lower = path.lower()

        # Find base path
        base_path = None

        for base in cls.base_names:
            if base not in path_lower:
                continue

            base_path = path[:path_lower.find(base)]
            break

        # Check if `base_path` was found
        if not base_path:
            Log.Warn('Unable to find base path in %r', path)
            return path

        # Return relative path
        return os.path.relpath(path, base_path)

    @classmethod
    def is_relative_path(cls, path):
        """Check if `path` is relative to `StorageHelper.base_names`"""

        path_lower = path.lower()

        # Ignore framework paths
        if 'framework.bundle' in path_lower:
            return False

        # Find base path
        for base in cls.base_names:
            if base not in path_lower:
                continue

            return True

        return False


class SystemHelper(object):
    bits_map = {
        '32bit': 'i386',
        '64bit': 'x86_64'
    }

    machine_map = {
        ('32bit', 'i686'): 'i686'
    }

    name_map = {
        'Darwin': 'MacOSX'
    }

    @classmethod
    def architecture(cls):
        """Retrieve system architecture (i386, i686, x86_64)"""

        bits, _ = platform.architecture()
        machine = platform.machine()

        # Check for ARM machine
        if bits == '32bit' and machine.startswith('armv'):
            return cls.arm(machine)

        # Check (bits, machine) map
        machine_key = (bits, machine)

        if machine_key in cls.machine_map:
            return cls.machine_map[machine_key]

        # Check (bits) map
        if bits in cls.bits_map:
            return cls.bits_map[bits]

        Log.Info('Unable to determine system architecture - bits: %r, machine: %r', bits, machine)
        return None

    @classmethod
    def name(cls):
        """Retrieve system name (Windows, Linux, FreeBSD, MacOSX)"""

        system = platform.system()

        # Apply system map
        if system in cls.name_map:
            system = cls.name_map[system]

        return system

    @classmethod
    def arm(cls, machine):
        # Determine floating-point type
        float_type = cls.arm_float_type()

        if float_type is None:
            Log.Warn('Unable to use ARM libraries, unsupported floating-point type?')
            return None

        # Determine ARM version
        version = cls.arm_version()

        if version is None:
            Log.Warn('Unable to use ARM libraries, unsupported ARM version (%r)?' % machine)
            return None

        return '%s_%s' % (version, float_type)

    @classmethod
    def arm_version(cls, machine=None):
        # Read `machine` name if not provided
        if machine is None:
            machine = platform.machine()

        # Ensure `machine` is valid
        if not machine:
            return None

        # ARMv6
        if machine.startswith('armv6'):
            return 'armv6'

        # ARMv7
        if machine.startswith('armv7'):
            return 'armv7'

        return None

    @classmethod
    def arm_float_type(cls):
        if os.path.exists('/lib/arm-linux-gnueabihf'):
            return 'hf'

        if os.path.exists('/lib/arm-linux-gnueabi'):
            return 'sf'

        return None


1			from plugin.core.environment import Environment
2
3			import os
4			import platform
5			import shutil
6			import sys
7
8
9			# Create dummy `Log` (for tests)
10			try:
11			Log.Debug('Using framework "Log" handler')
12			except NameError:
13			from plex_mock.framework import Logger
14
15			Log = Logger()
16			Log.Debug('Using dummy "Log" handler')
17
18
19			class Libraries(object):
20			contents_path = os.path.abspath(os.path.join(Environment.path.code, '..'))
21
22			native_directories = [
23			'Libraries\\FreeBSD',
24			'Libraries\\Linux',
25			'Libraries\\MacOSX',
26			'Libraries\\Windows'
27			]
28
29			unicode_map = {
30			65535: 'ucs2',
31			1114111: 'ucs4'
32			}
33
34			@classmethod
35			def cache(cls):
36			"""Cache native libraries into the plugin data directory"""
37
38			# Retrieve library platforms
39			libraries_path = os.path.join(cls.contents_path, 'Libraries')
40			platforms = os.listdir(libraries_path)
41
42			if 'Shared' in platforms:
43			platforms.remove('Shared')
44
45			# Create destination directory
46			destination = os.path.join(Environment.path.plugin_data, 'Libraries')
47
48			# Ensure destination exists
49			StorageHelper.create_directories(destination)
50
51			# Delete existing libraries
52			for name in os.listdir(destination):
53			path = os.path.join(destination, name)
54
55			StorageHelper.delete_tree(path)
56
57			# Copy libraries to directory
58			for name in platforms:
59			p_source = os.path.join(libraries_path, name)
60			p_destination = os.path.join(destination, name)
61
62			if not StorageHelper.copy_tree(p_source, p_destination):
63			return None
64
65			Log.Debug('Cached native libraries to %r', StorageHelper.to_relative_path(destination))
66			return destination
67
68			@classmethod
69			def get_libraries_path(cls, cache=False):
70			"""Retrieve the native libraries base directory (and caching the libraries if enabled)"""
71
72			if not cache:
73			return Environment.path.libraries
74
75			# Cache native libraries
76			libraries_path = cls.cache()
77
78			if libraries_path:
79			# Reset native library directories in `sys.path`
80			cls.reset()
81
82			return libraries_path
83
84			return Environment.path.libraries
85
86			@classmethod
87			def setup(cls, cache=False):
88			"""Setup native library directories"""
89
90			# Retrieve libraries path
91			libraries_path = cls.get_libraries_path(cache)
92
93			Log.Info('Using native libraries at %r', StorageHelper.to_relative_path(libraries_path))
94
95			# Retrieve system details
96			system = SystemHelper.name()
97			system_architecture = SystemHelper.architecture()
98
99			if not system_architecture:
100			return
101
102			Log.Debug('System: %r, Architecture: %r', system, system_architecture)
103
104			architectures = [system_architecture]
105
106			if system_architecture == 'i686':
107			# Fallback to i386
108			architectures.append('i386')
109
110			for architecture in reversed(architectures + ['universal']):
111			# Common
112			PathHelper.insert(libraries_path, system, architecture)
113
114			# UCS
115			if sys.maxunicode in cls.unicode_map:
116			PathHelper.insert(libraries_path, system, architecture, cls.unicode_map[sys.maxunicode])
117
118			# Log library paths
119			for path in sys.path:
120			path = os.path.abspath(path)
121
122			if not StorageHelper.is_relative_path(path):
123			continue
124
125			Log.Info('[PATH] %s', StorageHelper.to_relative_path(path))
126
127			@staticmethod
128			def test():
129			Log.Info('Testing native library support...')
130
131			# Check "apsw" availability
132			try:
133			import apsw
134
135			Log.Info(' - apsw: %r, sqlite: %r', apsw.apswversion(), apsw.SQLITE_VERSION_NUMBER)
136			except Exception, ex:
137			Log.Error(' - Unable to import "apsw": %s', ex)
138
139			# Check "llist" availability
140			try:
141			import llist
142
143			Log.Info(' - llist: available')
144			except Exception, ex:
145			Log.Warn(' - Unable to import "llist": %s', ex)
146
147			@classmethod
148			def reset(cls):
149			"""Remove all the native library directives from `sys.path`"""
150
151			for path in sys.path:
152			path = os.path.abspath(path)
153
154			if not path.lower().startswith(cls.contents_path.lower()):
155			continue
156
157			# Convert to relative path
158			path_rel = os.path.relpath(path, cls.contents_path)
159
160			# Take the first two fragments
161			path_rel = os.path.sep.join(path_rel.split(os.path.sep)[:2])
162
163			# Ignore non-native library directories
164			if path_rel not in cls.native_directories:
165			continue
166
167			# Remove from `sys.path`
168			PathHelper.remove(path)
169
170
171			class PathHelper(object):
172			@classmethod
173			def insert(cls, base, system, architecture, *args):
174			"""Insert a new path into `sys.path` if it passes basic validation"""
175
176			path = os.path.join(base, system, architecture, *args)
177
178			if path in sys.path:
179			return False
180
181			if not os.path.exists(path):
182			return False
183
184			sys.path.insert(0, path)
185
186			Log.Debug('Inserted path: %r', StorageHelper.to_relative_path(path))
187			return True
188
189			@classmethod
190			def remove(cls, path):
191			"""Remove path from `sys.path` if it exists"""
192
193			if path not in sys.path:
194			return False
195
196			sys.path.remove(path)
197
198			Log.Debug('Removed path: %r', StorageHelper.to_relative_path(path))
199			return True
200
201
202			class StorageHelper(object):
203			base_names = [
204			'plug-ins',
205			'plug-in support',
206			'trakttv.bundle'
207			]
208
209			@classmethod
210			def create_directories(cls, path, args, *kwargs):
211			"""Create directory at `path` include any parent directories"""
212
213			try:
214			os.makedirs(path, args, *kwargs)
215			return True
216			except OSError, ex:
217			if ex.errno == 17:
218			# Directory already exists
219			return True
220
221			Log.Warn('Unable to create directories: %r - (%s) %s', cls.to_relative_path(path), ex.errno, ex)
222			except Exception, ex:
223			Log.Warn('Unable to create directories: %r - (%s) %s', cls.to_relative_path(path), type(ex), ex)
224
225			return False
226
227			@classmethod
228			def copy_tree(cls, source, destination):
229			"""Copy the directory at `source` to `destination`"""
230
231			try:
232			shutil.copytree(source, destination)
233
234			Log.Debug('Copied %r to %r', cls.to_relative_path(source), cls.to_relative_path(destination))
235			return True
236			except Exception, ex:
237			Log.Warn('Unable to copy %r to %r - %s', cls.to_relative_path(source), cls.to_relative_path(destination), ex)
238
239			return False
240
241			@classmethod
242			def delete_tree(cls, path):
243			"""Delete the directory (at `path`"""
244
245			try:
246			shutil.rmtree(path)
247
248			Log.Debug('Deleted %r', cls.to_relative_path(path))
249			return True
250			except Exception, ex:
251			Log.Warn('Unable to delete directory: %r - %s', cls.to_relative_path(path), ex)
252
253			return False
254
255			@classmethod
256			def to_relative_path(cls, path):
257			"""Convert `path` to be relative to `StorageHelper.base_names`"""
258
259			path_lower = path.lower()
260
261			# Find base path
262			base_path = None
263
264			for base in cls.base_names:
265			if base not in path_lower:
266			continue
267
268			base_path = path[:path_lower.find(base)]
269			break
270
271			# Check if `base_path` was found
272			if not base_path:
273			Log.Warn('Unable to find base path in %r', path)
274			return path
275
276			# Return relative path
277			return os.path.relpath(path, base_path)
278
279			@classmethod
280			def is_relative_path(cls, path):
281			"""Check if `path` is relative to `StorageHelper.base_names`"""
282
283			path_lower = path.lower()
284
285			# Ignore framework paths
286			if 'framework.bundle' in path_lower:
287			return False
288
289			# Find base path
290			for base in cls.base_names:
291			if base not in path_lower:
292			continue
293
294			return True
295
296			return False
297
298
299			class SystemHelper(object):
300			bits_map = {
301			'32bit': 'i386',
302			'64bit': 'x86_64'
303			}
304
305			machine_map = {
306			('32bit', 'i686'): 'i686'
307			}
308
309			name_map = {
310			'Darwin': 'MacOSX'
311			}
312
313			@classmethod
314			def architecture(cls):
315			"""Retrieve system architecture (i386, i686, x86_64)"""
316
317			bits, _ = platform.architecture()
318			machine = platform.machine()
319
320			# Check for ARM machine
321			if bits == '32bit' and machine.startswith('armv'):
322			return cls.arm(machine)
323
324			# Check (bits, machine) map
325			machine_key = (bits, machine)
326
327			if machine_key in cls.machine_map:
328			return cls.machine_map[machine_key]
329
330			# Check (bits) map
331			if bits in cls.bits_map:
332			return cls.bits_map[bits]
333
334			Log.Info('Unable to determine system architecture - bits: %r, machine: %r', bits, machine)
335			return None
336
337			@classmethod
338			def name(cls):
339			"""Retrieve system name (Windows, Linux, FreeBSD, MacOSX)"""
340
341			system = platform.system()
342
343			# Apply system map
344			if system in cls.name_map:
345			system = cls.name_map[system]
346
347			return system
348
349			@classmethod
350			def arm(cls, machine):
351			# Determine floating-point type
352			float_type = cls.arm_float_type()
353
354			if float_type is None:
355			Log.Warn('Unable to use ARM libraries, unsupported floating-point type?')
356			return None
357
358			# Determine ARM version
359			version = cls.arm_version()
360
361			if version is None:
362			Log.Warn('Unable to use ARM libraries, unsupported ARM version (%r)?' % machine)
363			return None
364
365			return '%s_%s' % (version, float_type)
366
367			@classmethod
368			def arm_version(cls, machine=None):
369			# Read `machine` name if not provided
370			if machine is None:
371			machine = platform.machine()
372
373			# Ensure `machine` is valid
374			if not machine:
375			return None
376
377			# ARMv6
378			if machine.startswith('armv6'):
379			return 'armv6'
380
381			# ARMv7
382			if machine.startswith('armv7'):
383			return 'armv7'
384
385			return None
386
387			@classmethod
388			def arm_float_type(cls):
389			if os.path.exists('/lib/arm-linux-gnueabihf'):
390			return 'hf'
391
392			if os.path.exists('/lib/arm-linux-gnueabi'):
393			return 'sf'
394
395			return None
396

fuzeman / Plex-Trakt-Scrobbler

Push — develop ( d0ed0b...b83390 )

Libraries.cache() B

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like