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

BackupMaintenanceManager.process_period() F
last analyzed 2018-03-05 01:11 UTC

↳ Parent: BackupMaintenanceManager

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	285.8526

Importance

Changes

Metric	Value
dl	0
loc	78
ccs	1
cts	42
cp	0.0238
rs	2.1974
c	0
b	0
f	0
cc	17
crap	285.8526

How to fix Long Method Complexity

from plugin.core.backup.constants import BACKUP_NAME_REGEX, BACKUP_PATH, BACKUP_PERIODS, BACKUP_RETENTION

from plugin.core.backup.models import BackupGroup, BackupRevision
from plugin.core.backup.tasks import ArchiveTask, CompactTask
from plugin.core.helpers.variable import try_convert

from datetime import datetime
from fnmatch import fnmatch
import logging
import os

log = logging.getLogger(__name__)


class BackupMaintenanceManager(object):
    def __init__(self):
        self.now = datetime.now()

    def run(self):
        # Run maintenance on backup groups
        for group in BackupGroup.list(max_depth=1):
            self.process_group(group)

    #
    # Process methods
    #

    def process_group(self, group):
        log.debug('Running maintenance on group: %r', group)

        # Process policy periods
        for period in BACKUP_PERIODS:
            self.process_period(period, group)

    def process_period(self, period, group):
        policy = BACKUP_RETENTION[period]

        # Retrieve options
        p_files = policy.get('files')

        if p_files is None:
            raise ValueError('Policy "%s" is missing the "files" attribute')

        # Build lists of revisions, grouped by period
        revisions_grouped = {}

        for base_path, dirs, files in os.walk(group.path):

            # Strip UNC prefix from `base_path`
            if base_path.startswith('\\\\?\\'):
                base_path = base_path[4:]

            # Ensure directory starts with a year
            rel_path = os.path.relpath(base_path, group.path)

            try:
                year = rel_path[:rel_path.index(os.path.sep)]
            except ValueError:
                year = rel_path

            if len(year) != 4 or try_convert(year, int) is None:
                continue

            # Search for revision metadata files
            for name in files:
                # Ensure file name matches the policy "files" filter
                if not fnmatch(name, p_files):
                    continue

                # Build path
                path = os.path.join(base_path, name)

                # Match revision metadata against regex pattern
                if not BACKUP_NAME_REGEX.match(name):
                    continue

                # Load metadata from file
                try:
                    revision = BackupRevision.load(path)
                except Exception as ex:

                    log.warn('Unable to load revision at %r: %s', path, ex, exc_info=True)
                    continue

                # Retrieve timestamp period
                key = self.timestamp_period(period, revision.timestamp)

                if key == self.timestamp_period(period, self.now):
                    # Backup occurred in the current period
                    continue

                # Store details in `revisions_grouped` dictionary
                if key not in revisions_grouped:
                    revisions_grouped[key] = []

                revisions_grouped[key].append(revision)

        # Ensure revisions have been found
        if not revisions_grouped:
            return

        log.debug('Processing period: %r (group: %r)', period, group)

        # Search for weeks exceeding the policy
        for key, revisions in revisions_grouped.items():
            # Compact period
            if policy.get('compact'):
                CompactTask.run(period, key, revisions, policy['compact'])

            # Archive revisions (if enabled)
            if policy.get('archive'):
                ArchiveTask.run(group, period, key, revisions, policy['archive'])

        log.debug('Done')

    #
    # Helpers
    #

    @staticmethod
    def timestamp_period(period, timestamp):
        if period == 'day':
            return timestamp.date()

        if period == 'week':
            return tuple(timestamp.isocalendar()[:2])

        if period == 'month':
            return timestamp.year, timestamp.month

        if period == 'year':
            return timestamp.year,

        raise ValueError('Unknown period: %r' % period)


1	1	from plugin.core.backup.constants import BACKUP_NAME_REGEX, BACKUP_PATH, BACKUP_PERIODS, BACKUP_RETENTION
		0 ignored issues – show Unused Code introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report Unused BACKUP_PATH imported from plugin.core.backup.constants Loading history...
2	1	from plugin.core.backup.models import BackupGroup, BackupRevision
3	1	from plugin.core.backup.tasks import ArchiveTask, CompactTask
4	1	from plugin.core.helpers.variable import try_convert
5
6	1	from datetime import datetime
7	1	from fnmatch import fnmatch
8	1	import logging
9	1	import os
10
11	1	log = logging.getLogger(__name__)
12
13
14	1	class BackupMaintenanceManager(object):
15	1	def __init__(self):
16		self.now = datetime.now()
17
18	1	def run(self):
19		# Run maintenance on backup groups
20		for group in BackupGroup.list(max_depth=1):
21		self.process_group(group)
22
23		#
24		# Process methods
25		#
26
27	1	def process_group(self, group):
28		log.debug('Running maintenance on group: %r', group)
29
30		# Process policy periods
31		for period in BACKUP_PERIODS:
32		self.process_period(period, group)
33
34	1	def process_period(self, period, group):
35		policy = BACKUP_RETENTION[period]
36
37		# Retrieve options
38		p_files = policy.get('files')
39
40		if p_files is None:
41		raise ValueError('Policy "%s" is missing the "files" attribute')
42
43		# Build lists of revisions, grouped by period
44		revisions_grouped = {}
45
46		for base_path, dirs, files in os.walk(group.path):
		0 ignored issues – show Unused Code introduced 2016-11-18 22:00 UTC by Report Bug Copy Issue Report The variable `dirs` seems to be unused. Loading history...
47		# Strip UNC prefix from `base_path`
48		if base_path.startswith('\\\\?\\'):
49		base_path = base_path[4:]
50
51		# Ensure directory starts with a year
52		rel_path = os.path.relpath(base_path, group.path)
53
54		try:
55		year = rel_path[:rel_path.index(os.path.sep)]
56		except ValueError:
57		year = rel_path
58
59		if len(year) != 4 or try_convert(year, int) is None:
60		continue
61
62		# Search for revision metadata files
63		for name in files:
64		# Ensure file name matches the policy "files" filter
65		if not fnmatch(name, p_files):
66		continue
67
68		# Build path
69		path = os.path.join(base_path, name)
70
71		# Match revision metadata against regex pattern
72		if not BACKUP_NAME_REGEX.match(name):
73		continue
74
75		# Load metadata from file
76		try:
77		revision = BackupRevision.load(path)
78		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...
79		log.warn('Unable to load revision at %r: %s', path, ex, exc_info=True)
80		continue
81
82		# Retrieve timestamp period
83		key = self.timestamp_period(period, revision.timestamp)
84
85		if key == self.timestamp_period(period, self.now):
86		# Backup occurred in the current period
87		continue
88
89		# Store details in `revisions_grouped` dictionary
90		if key not in revisions_grouped:
91		revisions_grouped[key] = []
92
93		revisions_grouped[key].append(revision)
94
95		# Ensure revisions have been found
96		if not revisions_grouped:
97		return
98
99		log.debug('Processing period: %r (group: %r)', period, group)
100
101		# Search for weeks exceeding the policy
102		for key, revisions in revisions_grouped.items():
103		# Compact period
104		if policy.get('compact'):
105		CompactTask.run(period, key, revisions, policy['compact'])
106
107		# Archive revisions (if enabled)
108		if policy.get('archive'):
109		ArchiveTask.run(group, period, key, revisions, policy['archive'])
110
111		log.debug('Done')
112
113		#
114		# Helpers
115		#
116
117	1	@staticmethod
118		def timestamp_period(period, timestamp):
119		if period == 'day':
120		return timestamp.date()
121
122		if period == 'week':
123		return tuple(timestamp.isocalendar()[:2])
124
125		if period == 'month':
126		return timestamp.year, timestamp.month
127
128		if period == 'year':
129		return timestamp.year,
130
131		raise ValueError('Unknown period: %r' % period)
132

fuzeman / Plex-Trakt-Scrobbler

BackupMaintenanceManager.process_period() F last analyzed 2018-03-05 01:11 UTC

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

BackupMaintenanceManager.process_period() F
last analyzed 2018-03-05 01:11 UTC