2
\$\begingroup\$

I am writing a small application to get commits between two tags from different env (apple, banana) and write the result into a csv file. For security, I have to replace some sensitive information with alfa, beta and application banana key words. The application is running fine from my local machine. Please help me to review the code.

Steps for application:

  1. Look for git tag visit env URL (alfa tag, beta tag)
  2. Run command line interface to retrieve git log between two tags to get all commit info
  3. Run GitHub command line interface to retrieve PR information by issue number
  4. Generate a csv file
#file directory: ~/main.py

import urllib.request
import re
import coloredlogs, logging
import json
import pandas as pd
import subprocess
from configs.config import ConfigPrivate
from env_enum import *
from utilities import core
from typing import Dict, List


logger = logging.getLogger(__name__)
SEPARATOR = '[^^^]'

coloredlogs.install(level=logging.DEBUG, logger=logger, fmt='%(asctime)s %(message)s')


def isMatched(tag):
    return len(set(tag.values())) == 1


class GitApp:
    pr_url = 'repos/<owner>/<repos>/issues'
    jira_url = '<jira_link>/browse'
    csv_title = ['col1', 'col2', 'col3', 'col4', 'col5', 'col6', 'col7', 'col8', 'col9']

    def __init__(self, application, config, alfa_tag, beta_tag):
        self.app_type = application
        self.config = config
        self.alfa_tag = alfa_tag
        self.beta_tag = beta_tag

    def get_git_log(self) -> List[str]:
        path = f'{self.config["project_dirs"]}/apps/{AppFullNameEnum[self.app_type].value}'
        cmd = f'git log {self.alfa_tag}..{self.beta_tag} --pretty=format:%h{SEPARATOR}%an' \
              f'{SEPARATOR}%s' \
              f'{SEPARATOR}%cd --date=short -- .'
        with core.pushd(path):
            try:
                log_info = subprocess.check_output(cmd, shell=True, bufsize=1, universal_newlines=True)
                return self.parse_logs_csv(log_info.split('\n'))
            except subprocess.CalledProcessError as e:
                logger.warning(e)
                logger.info(f'You may need to update repo {path} by git pull')
                exit()

    def parse_logs_csv(self, logs_info: List[str]) -> List[str]:
        logs = [i.split(SEPARATOR) for i in logs_info]
        logs_sets = list()
        for i, log in enumerate(logs):
            log_detail = dict(col1=log[0], col2=log[1], col3=log[2], col4=log[3])
            issue = re.findall(r'\(#\d+\)', log[2])
            issue_num = issue.pop()[2:-1]
            try:
                cmd = f'hub api {self.pr_url}/{issue_num}'
                loginfo = subprocess.check_output(cmd, shell=True, bufsize=1, universal_newlines=True)
                logs = json.loads(loginfo)
                qa_tag = re.findall(r'\[[Qq][Aa]:\s*(\w+)\s*\]', logs['body'])
                jira = re.findall(r'\[\s*(\w+-\d+)\s*\]', logs['title']) or re.findall(r'\[\s*(\w+-\d+)\s*\]',
                                                                                       logs['body'])
                log_detail['col5'] = logs['html_url']
                log_detail['col6'] = f'{self.jira_url}/{jira.pop()}' if jira else ''
                log_detail['col7'] = qa_tag.pop() if qa_tag else 'Ignore'
                log_detail['col8'] = 'Verify' if 'verify' == log_detail['qa_tag'].lower() else 'skip'
                logs_sets.append(
                    [log_detail['col1'], log_detail['col2'], log_detail['col3'], log_detail['col4'],
                     log_detail['col5'], log_detail['col6'], log_detail['col7'], log_detail['col8'], ''])
            except subprocess.CalledProcessError as e:
                logger.warning(e)
                exit()

        return logs_sets

    def generator_csv_file(self, logs: List[List[str]], filename: str):
        df = pd.DataFrame(logs, columns=self.csv_title)
        core.df_to_csv(df, filename)


class Applications:
    def __init__(self, app, config=ConfigPrivate().data):
        self.config = config
        self.app_type = AppEnum[app]
        self.url = self.generator_url()
        self.tag = self.generator_tag()
        self.git_logs = self.generator_git_log()

    def generator_url(self) -> Dict[EnvEnum, str]:
        return {env: f'http://{URLPrefixEnum[env.name]}.com/{URLPosfixEnum[self.app_type.name]}' for env in
                EnvEnum}

    def generator_tag(self) -> Dict[EnvEnum, str]:
        if (EnvEnum.alfa not in self.url) or (EnvEnum.beta not in self.url):
            logger.critical(f'Cannt find URL to get tag. Program Exits')

        tag = dict()
        for env in EnvEnum:
            with core.closing(urllib.request.urlopen(self.url[env])) as response:
                page = response.read().decode('utf-8')
                pattern = r'[\/(]eks-\d{4}-\d{2}-\d{2}-\d{2}-\d{2}-\w{7}[\/)]'
                t = re.findall(pattern, page)
                tag[env] = f'{AppFullNameEnum[self.app_type.name].value}/{t.pop()[1:-1]}'
        tagmsg = f'{EnvEnum.alfa.value} tag: {tag[EnvEnum.alfa]}---{EnvEnum.beta.value} tag: {tag[EnvEnum.beta]}'
        if isMatched(tag):
            logger.info(f'{tagmsg} matched. Nothing need to be vetted. Program exits')
            exit()
        logger.info(f'{tagmsg}')
        return tag

    def generator_git_log(self):
        git_app = GitApp(self.app_type.name, self.config, self.tag[EnvEnum.alfa], self.tag[EnvEnum.beta])
        return git_app.get_git_log()

    def generator_csv_file(self):
        from datetime import datetime
        now = datetime.now()
        git_app = GitApp(self.app_type.name, self.config, self.tag[EnvEnum.alfa], self.tag[EnvEnum.beta])
        git_app.generator_csv_file(self.git_logs,
                                   f'~/Desktop/{now.strftime("%Y_%m_%d_%H_%M_%S")}_'
                                   f'{AppFullNameEnum[self.app_type.name].value}.csv')


def main():
    apple = Applications('apple')
    apple.generator_csv_file()
    banana = Applications('banana')
    banana.generator_csv_file()


if __name__ == '__main__':
    main()

#file path: ~/configs/config.py
import os
import yaml


class ConfigPrivate:
    """Provides configuration information."""

    DEFAULT_CONFIG_DIR = os.path.abspath(os.path.dirname(__file__))
    DEFAULT_CONFIG_FILE = "conf.yaml"

    def __init__(self, config_path=None):
        """Initialize a configuration from a conf directory and conf file."""
        super(ConfigPrivate, self).__init__()
        failsafe_path = "/etc/config/{}.conf.yaml".format(self.DEFAULT_CONFIG_FILE)

        if config_path:
            path = config_path
        else:
            path = os.path.join(self.DEFAULT_CONFIG_DIR, self.DEFAULT_CONFIG_FILE)

        if not os.path.isfile(path):
            path = failsafe_path

        with open(path, 'r') as f:
            self.data = yaml.load(f, Loader=yaml.FullLoader)

#file path: ~/utilities/core.py
from contextlib import contextmanager

import os
import pandas as pd


def df_to_csv(data_frame: pd.DataFrame, f_name: str):
    data_frame.to_csv(f_name, encoding='utf-8', index=False)


@contextmanager
def closing(thing):
    try:
        yield thing
    finally:
        thing.close()


@contextmanager
def pushd(new_dir):
    old_dir = os.getcwd()
    os.chdir(new_dir)
    try:
        yield
    finally:
        os.chdir(old_dir)

#file path: ~/env_enum.py
from enum import Enum


class AppEnum(Enum):
    apple = 'apple'
    banana = 'banana'


class AppFullNameEnum(Enum):
    apple = 'apple_fullname'
    banana = 'banana_fullname'


class EnvEnum(Enum):
    alfa = 'alfa'
    beta = 'beta'


class URLPrefixEnum(Enum):
    alfa = 'alfa'
    beta = 'beta'

    def __str__(self):
        return self.value


class URLPosfixEnum(Enum):
    apple = 'apple_pos_url'
    banana = 'banana_pos_url'

    def __str__(self):
        return self.value


\$\endgroup\$

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Browse other questions tagged or ask your own question.