from subprocess import Popen, PIPE from datetime import datetime, timedelta from collections import OrderedDict from operator import itemgetter import re import logging class FeatureFlagInfo(object): def __init__(self, date_added, feature_name, story_num, date_enabled, team_prefix): """Contains relevant info about a feature flag""" self.date_added = date_added self.feature_name = feature_name self.story_num = story_num self.date_enabled = date_enabled self.team_prefix = team_prefix class Parser(object): COUNT_BY_PREFIX = {} @staticmethod def execute_git_blame(): return Popen(['git', 'blame', '../../features.yml'], stdout=PIPE).stdout.readlines() @staticmethod def __get_date(output_line): return datetime.strptime(re.search(r'\d{4}-\d{2}-\d{2}', output_line).group(), '%Y-%m-%d').date() @staticmethod def __feature_name(output_line): return output_line.split('name: ')[1].split('\\')[0] @staticmethod def __story_number(output_line): try: return output_line.split('#')[1].split('\\')[0].strip() except IndexError: return None @staticmethod def __enabled_date(output_line): if 'enabled' in output_line and '1' in output_line.split('enabled: ')[1]: return Parser.__get_date(output_line) else: return None @staticmethod def __get_prefix(story_number): if story_number: return story_number.split('-')[0] else: return None @staticmethod def create_feature_info_objects(lines): feature_flag_info_objects = [] for idx, line in enumerate(lines): if 'name: ' in str(line): name = Parser.__feature_name(str(line)) story_line = str(lines[idx - 1]) story_num = Parser.__story_number(story_line) date_added = Parser.__get_date(story_line) enabled_date = Parser.__enabled_date(str(lines[idx + 2])) feature_flag_info_objects.append(FeatureFlagInfo( date_added, name, story_num, enabled_date, Parser.__get_prefix(story_num))) return feature_flag_info_objects @staticmethod def printer(prefix, features): Parser.COUNT_BY_PREFIX[prefix] = len(features) print('----------------{}----------------'.format(prefix)) print('{} TOTAL\n'.format(str(len(features)))) for feature in features: print('NAME: {}'.format(feature.feature_name)) print('ENABLED ON: {}'.format(str(feature.date_enabled))) print('STORY: {}\n'.format(feature.story_num)) @staticmethod def do_output(): blame_output = Parser.execute_git_blame() feature_info_objects = Parser.create_feature_info_objects(blame_output) print('There are {} feature flags\n'.format(str(len(feature_info_objects)))) past_date_features = [] for feature_flag_obj in feature_info_objects: try: if (datetime.now().date() - feature_flag_obj.date_enabled) > timedelta(days=60): past_date_features.append(feature_flag_obj) except TypeError: if feature_flag_obj.date_enabled is None: logging.info('{} has not been enabled'.format(feature_flag_obj.feature_name)) else: logging.warning('{} has an unexpected issue'.format(feature_flag_obj.__dict__)) print('There are {} enabled feature flags older than 60 days:\n'.format(str(len(past_date_features)))) # Sort by team prefix and date enabled, if either is missing print it at the end past_date_features.sort( key=lambda x: ((x.team_prefix is None, x.team_prefix), (x.date_enabled is None, x.date_enabled)), reverse=False) last_prefix = None list_by_prefix = [] for feature_flag_obj in past_date_features: if last_prefix != feature_flag_obj.team_prefix: if len(list_by_prefix) > 0: Parser.printer(last_prefix, list_by_prefix) list_by_prefix = [] last_prefix = feature_flag_obj.team_prefix list_by_prefix.append(feature_flag_obj) else: list_by_prefix.append(feature_flag_obj) Parser.printer(last_prefix, list_by_prefix) # Print team with total number of stale flags in descending order print('Count of stale flags by team:\n') for key, value in OrderedDict(sorted(Parser.COUNT_BY_PREFIX.items(), key=itemgetter(1), reverse=True)).items(): print(key, value) Parser.do_output()