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:
- Look for git tag visit env URL (alfa tag, beta tag)
- Run command line interface to retrieve git log between two tags to get all commit info
- Run GitHub command line interface to retrieve PR information by issue number
- 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