BackupMaintenanceManager.process_period() - Code Metrics - Inspection of "Fixed migration schema tests" - fuzeman/Plex-Trakt-Scrobbler - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( 6fb855...7cfc9a )

by Dean

created 2016-02-10 07:33 UTC

BackupMaintenanceManager.process_period() F

↳ Parent: BackupMaintenanceManager

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	224.3143

Metric	Value
cc	15
dl	0
loc	78
ccs	1
cts	42
cp	0.0238
crap	224.3143
rs	2.1566

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, 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
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):
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, ex:
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

Push — develop ( 6fb855...7cfc9a )

BackupMaintenanceManager.process_period() F

Complexity

Size

Duplication

Code Coverage

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like