Calculating Maintainability Index for a whole project

import datetime
import json
import tempfile
import webbrowser
from collections import OrderedDict

import plotly
from git import Repo
from progress.bar import Bar
from radon.cli import Config, MIHarvester, RawHarvester

from django.conf import settings
from django.core.management.base import BaseCommand
from django.utils import timezone


class WorseMaintainabilityIndexError(Exception):
    pass


class Command(BaseCommand):
    help = "Calculate current Maintainability Index"

    def add_arguments(self, parser):
        parser.add_argument(
            '--commit',
            action='store_true',
            dest='commit',
            default=False,
            help="Commit result.",
        )
        parser.add_argument(
            '--fail',
            action='store_true',
            dest='fail',
            default=False,
            help="Raise Error, if MI is worse than before.",
        )
        parser.add_argument(
            '--showhistory',
            action='store_true',
            dest='showhistory',
            default=False,
            help="Show history of MI in browser.",
        )
        parser.add_argument(
            '--init',
            action='store_true',
            dest='init',
            default=False,
            help="Initialize history for past commits.",
        )

    def handle(self, *args, **options):
        if options.get('init', False):
            build_for_past('bump version')
            return

        mi_calculator = MaintainabilityIndexCalculater()
        if options.get('showhistory', False):
            mi_calculator.show_history()
            return
        if options.get('fail', False):
            mi_calculator.check_for_fail()
        if options.get('commit', False):
            mi_calculator.commit_new_mi()
        return str(mi_calculator.maintainability_index)


class MaintainabilityIndexCalculater(object):
    def __init__(self):
        self.radon_settings = getattr(settings, 'RADON_MI_SETTINGS', {})

        self.history_file_path = self.radon_settings.get(
            'history_file_path', 'mi_history.json'
        )
        self.current_state_file_path = self.radon_settings.get(
            'current_state_file_path', 'mi_current_state.json'
        )
        self.__init_harvesters()

    def get_current_history(self) -> dict:
        """
        Return dict of history of MIs by given path to history file.
        """
        mi_history = getattr(self, '_mi_history', False)
        if mi_history:
            return mi_history
        try:
            with open(self.history_file_path, 'r') as mi_history_file:
                current_history = json.loads(mi_history_file.read())
        except FileNotFoundError:
            current_history = {}
        setattr(self, '_mi_history', current_history)
        return current_history

    def check_for_fail(self) -> None:
        """
        Raise an Error if new_maintainability_index is worse than recent from history.
        """
        mi_history = self.get_current_history()
        if not mi_history:
            # No old data. No worse result possible.
            return
        last_history = mi_history[sorted(mi_history.keys())[-1]]
        last_maintainability_index = last_history['mi']
        if self.maintainability_index < last_maintainability_index:
            raise WorseMaintainabilityIndexError(
                f'From {last_maintainability_index} ' f'to {self.maintainability_index}!'
            )

    def commit_new_mi(self) -> None:
        """
        Save new maintainability index and current state of all files in a commit.
        """
        repo = Repo.init()
        index = Repo.init().index
        current_commit = repo.head.commit
        mi_history = self.get_current_history()
        with open(self.history_file_path, 'w') as mi_history_file:
            mi_history[timezone.now().isoformat()] = {
                'sha': current_commit.hexsha,
                'mi': self.maintainability_index,
                'lloc': self.lloc,
                'loc': self.loc,
                'sloc': self.sloc,
            }
            mi_history_file.write(json.dumps(mi_history))
        with open(self.current_state_file_path, 'w') as mi_current_state:
            mi_current_state.write(self.mi_harvester.as_json())
        index.add([self.history_file_path, self.current_state_file_path])
        index.commit('chore: Update maintainability index')
        repo.close()

    def __init_harvesters(self) -> None:
        """
        Initialize needed harvesters with settings.
        """
        paths = self.radon_settings.get('paths', [])
        config = Config(
            min='A',
            max='C',
            exclude=self.radon_settings.get('exclude'),
            ignore=self.radon_settings.get('ignore'),
            multi=self.radon_settings.get('multi', False),
            show=True,
            sort=True,
        )
        raw_harvester = RawHarvester(paths, config)
        mi_harvester = MIHarvester(paths, config)
        self.raw_harvester = raw_harvester
        self.mi_harvester = mi_harvester

    @property
    def maintainability_index(self) -> float:
        """
        Return maintainability index of given scope.

        1. Count logical lines of code (lloc) on every file.
        2. Calculate Maintainability Index on every file.
        3. Merge Maintainability Index of every file by rating files higher by their
           logical lines of code.
        """
        maintainability_index = getattr(self, '_maintainability_index', None)
        if maintainability_index:
            return maintainability_index

        loc_data = self.loc_data

        lloc_count = 0
        global_mi = 0

        for file_name, mi_data in self.mi_harvester.results:
            try:
                lloc = loc_data[file_name]['lloc']
                mi = mi_data['mi']
            except KeyError:
                continue
            lloc_count += lloc
            global_mi += mi * lloc

        maintainability_index = global_mi / lloc_count

        setattr(self, '_maintainability_index', maintainability_index)

        return maintainability_index

    @property
    def loc_data(self) -> dict:
        """
        Return data collected by RAW Harvester.
        """
        loc_data = getattr(self, '_loc_data', None)
        if loc_data:
            return loc_data
        loc_data = dict(self.raw_harvester.results)
        setattr(self, '_loc_data', loc_data)
        return loc_data

    @property
    def loc(self) -> int:
        """
        Return lines of code of given scope.
        """
        loc_data = self.loc_data
        return sum(file_loc_data.get('loc', 0) for file_loc_data in loc_data.values())

    @property
    def lloc(self) -> int:
        """
        Return logical lines of code of given scope.
        """
        loc_data = self.loc_data
        return sum(file_loc_data.get('lloc', 0) for file_loc_data in loc_data.values())

    @property
    def sloc(self) -> int:
        """
        Return logical lines of code of given scope.
        """
        loc_data = self.loc_data
        return sum(file_loc_data.get('sloc', 0) for file_loc_data in loc_data.values())

    @property
    def comments(self) -> int:
        """
        Return comment lines of code of given scope.
        """
        loc_data = self.loc_data
        return sum(
            file_loc_data.get('comments', 0) for file_loc_data in loc_data.values()
        )

    def show_history(self) -> None:
        """
        Show history of the maintainability index in graph in browser.
        """
        mi_history = self.get_current_history()
        mi_history = OrderedDict(sorted(mi_history.items(), key=lambda x: x[0]))
        temporary_file = tempfile.NamedTemporaryFile(prefix='mi_history', suffix='.html')
        file_name = temporary_file.name
        temporary_file.close()

        mi_data = plotly.graph_objs.Scatter(
            x=list(mi_history.keys()),
            y=[values['mi'] for values in mi_history.values()],
            name='MI',
        )
        loc_data = plotly.graph_objs.Scatter(
            x=list(mi_history.keys()),
            y=[values['loc'] for values in mi_history.values()],
            yaxis='y2',
            name='LOC',
        )
        lloc_data = plotly.graph_objs.Scatter(
            x=list(mi_history.keys()),
            y=[values['lloc'] for values in mi_history.values()],
            yaxis='y2',
            name='LLOC',
        )
        sloc_data = plotly.graph_objs.Scatter(
            x=list(mi_history.keys()),
            y=[values['sloc'] for values in mi_history.values()],
            yaxis='y2',
            name='SLOC',
        )
        plotly.offline.plot(
            {
                'data': [mi_data, loc_data, lloc_data, sloc_data],
                'layout': plotly.graph_objs.Layout(
                    yaxis={'rangemode': 'tozero'},
                    yaxis2={
                        'side': 'right',
                        'overlaying': 'y',
                        'showgrid': False,
                        'rangemode': 'tozero',
                    },
                ),
            },
            auto_open=False,
            filename=file_name,
        )
        if file_name[0] != '/':
            # windows ...
            url = 'file:///' + file_name
        else:
            url = 'file://' + file_name
        webbrowser.open_new(url)


def build_for_past(commit_lookup_message: str) -> None:
    repo = Repo.init()
    current_branch = repo.active_branch

    if repo.is_dirty():
        raise Exception('Dirty Repo! Stash/Commit unchanged files!')

    history = {}
    revisions = []
    for commit in repo.iter_commits(current_branch):
        if commit_lookup_message in commit.message:
            revisions.append(
                (
                    commit.hexsha,  # key
                    datetime.datetime.fromtimestamp(commit.committed_date),  # date
                )
            )
    bar = Bar("Processing", max=len(revisions))
    error = None
    try:
        for revision in revisions:
            repo.git.checkout(revision[0])
            mic = MaintainabilityIndexCalculater()
            mi = mic.maintainability_index
            lloc = mic.lloc
            loc = mic.loc
            sloc = mic.sloc

            history[revision[1].isoformat()] = {
                'sha': revision[0],
                'mi': mi,
                'lloc': lloc,
                'loc': loc,
                'sloc': sloc,
            }
            bar.next()
    except Exception as e:
        error = e
    finally:
        bar.finish()
        repo.git.checkout(current_branch)
        repo.close()

        radon_settings = getattr(settings, 'RADON_MI_SETTINGS', {})
        history_file_path = radon_settings.get('history_file_path', 'mi_history.json')
        with open(history_file_path, 'w') as f:
            f.write(json.dumps(history))
        if error:
            raise error