"""Application script.""" import json import os import statistics from datetime import datetime import github from github import Github import requests JENKINS_USER = os.getenv('JENKINS_USER') JENKINS_TOKEN = os.getenv('JENKINS_TOKEN') GITHUB_TOKEN = os.getenv('GITHUB_TOKEN') LINK_TO_POST = os.getenv('LINK_TO_POST') g = Github(GITHUB_TOKEN) def get_pipelines(): """Get all pipelines.""" response = requests.get('https://pipeline.theorchard.io/api/json', auth=(JENKINS_USER, JENKINS_TOKEN)) arr = response.json() jobs = arr.get('jobs') return jobs def get_red_jobs(jobs): """Get red jobs from orchard pipeline API.""" red_jobs = [] for each_job in jobs: if 'color' in each_job: if each_job['color'] == 'red': if each_job['name'].find('pipeline') != -1: red_jobs.append(each_job) return red_jobs def filter_pipelines(red_jobs): """Filter out WIP jobs from orchard pipeline API.""" filtered_jobs = [] for each_job in red_jobs: if 'url' in each_job: response = requests.get(each_job['url'] + 'api/json/', auth=(JENKINS_USER, JENKINS_TOKEN)) pipeline = response.json() if 'description' in pipeline: if pipeline['description'].find('wip') == -1 \ and pipeline['description'].find('WIP') == -1: filtered_jobs.append(each_job) if 'name' in pipeline: if pipeline['name'].find('clone') != -1 \ or pipeline['name'].find('delete') != -1 \ or pipeline['name'].find('test') != -1: filtered_jobs.remove(each_job) return filtered_jobs def get_pipeline_git_urls(red_jobs): """Get github urls from pipelines.""" for each_job in red_jobs: if 'url' in each_job: response = requests.get(each_job['url'] + 'api/json/', auth=(JENKINS_USER, JENKINS_TOKEN)) pipeline = response.json() search_last_build_git_link(pipeline, each_job) def search_last_build_git_link(pipeline, each_job): """Search last unsuccessful for git links.""" if 'lastUnsuccessfulBuild' in pipeline: if 'url' in pipeline['lastUnsuccessfulBuild']: response = requests.get( pipeline['lastUnsuccessfulBuild']['url'] + 'api/json/', auth=(JENKINS_USER, JENKINS_TOKEN)) last_unsuccessful_build = response.json() if 'actions' in last_unsuccessful_build: search_for_url_in_last_unsuccessful_build( last_unsuccessful_build, each_job) def search_for_url_in_last_unsuccessful_build(last_unsuccessful_build, each_job): """Search for git link in unsuccessful build.""" get_pipeline_date(last_unsuccessful_build, each_job) i = 0 while i < len(last_unsuccessful_build['actions']): if 'remoteUrls' \ in last_unsuccessful_build['actions'][i]: each_job['GitLink'] = last_unsuccessful_build['actions'][i]['remoteUrls'][0] # noqa: E501 i += 1 def get_pipeline_date(last_unsuccessful_build, each_job): """Search for pipeline last execution time.""" if 'timestamp' in last_unsuccessful_build: timestamp = datetime\ .fromtimestamp(last_unsuccessful_build['timestamp']/1000)\ .strftime('%Y-%m-%d') each_job['LastExecution'] = timestamp return timestamp def git_related_work(red_jobs): """Make github related processes.""" for each_job in red_jobs: if 'GitLink' in each_job: format_git_links(each_job) get_github_info(each_job) else: each_job['GitLink'] = 'No link provided' each_job['repo'] = 'No link provided' each_job['author'] = 'cannot find author of this commit' def format_git_links(each_job): """Format git links.""" each_job['GitLink'] = each_job['GitLink'].replace('git@github.com:', '') each_job['GitLink'] = each_job['GitLink'].replace('.git', '') def get_github_info(each_job): """Get each pipeline github info.""" try: repo = g.get_repo(each_job['GitLink']) commit = repo.get_commits()[0] if commit.author is None: author = 'cannot find author of this commit' each_job['author'] = author each_job['repo'] = repo.html_url else: each_job['author'] = commit.author.html_url each_job['repo'] = repo.html_url except github.GithubException: print('oops') def get_days_count(each_job, days_count_list): """Count days of failing pipelines.""" f_date = datetime.strptime(each_job['LastExecution'], '%Y-%m-%d') l_date = datetime.today() delta = l_date - f_date each_job['days'] = str(delta.days) days_count_list.append(delta.days) return delta.days def post_header_to_slack(red_jobs): """Post message header to slack.""" days_count_list = [] for each_job in red_jobs: get_days_count(each_job, days_count_list) filtered_jobs = delete_todays_pipelines(red_jobs) now = datetime.today() today_date = now.strftime('%Y-%m-%d') red_builds_number = len(filtered_jobs) days_over_2_weeks_list = [] get_all_days_over_2_weeks(filtered_jobs, days_over_2_weeks_list) average_days_count = statistics.mean(days_count_list) payload = { 'blocks': [ { 'type': 'section', 'text': { 'type': 'mrkdwn', 'text': 'Today’s Date: *' + today_date + '*\n' + 'Number of Red Builds: *' + str(red_builds_number) + '*\n' + 'Number of Red Builds Broken More Than 2 Weeks: *' # noqa: E501 + str(len(days_over_2_weeks_list)) + '*\n' + 'Average Days Since Successful Run: *' + str(int(average_days_count)) + '*\n' + 'Median Days Pipelines Have Been Broken: *' + str(int(statistics.median(days_count_list))) + '*' } } ] } response = requests.post(LINK_TO_POST, data=json.dumps(payload), headers={'Content-Type': 'application/json'}) print(response.text) return filtered_jobs def get_all_days_over_2_weeks(red_jobs, days_over_2_weeks_list): """Get all pipelines days > 2 weeks .""" for each_job in red_jobs: if int(each_job['days']) >= 14: days_over_2_weeks_list.append(each_job) return days_over_2_weeks_list def delete_todays_pipelines(red_jobs): """Remove pipelines 0 days long.""" filtered_jobs = [] for each_job in red_jobs: if int(each_job['days']) != 0: filtered_jobs.append(each_job) return filtered_jobs def post_to_slack(red_jobs): """Post to slack.""" ordered_jobs = sorted(red_jobs, key=lambda x: x['LastExecution'], reverse=False) i = 0 for each_job in ordered_jobs: pipeline_name = each_job['url'] pipeline_name = \ pipeline_name.replace('https://pipeline.theorchard.io/job/', '') pipeline_name = pipeline_name.replace('/', '') payload = { 'blocks': [ { 'type': 'section', 'text': { 'type': 'mrkdwn', 'text': '<' + each_job['url'] + '|*' + pipeline_name + '*>' + ' ( <' + each_job['repo'] + '|repo> - ' + each_job['author'] + ' ) \n Date - *' + each_job['LastExecution'] + '( ' + each_job['days'] + ' days)*' } } ] } response = requests.post(LINK_TO_POST, data=json.dumps(payload), headers={'Content-Type': 'application/json'}) i += 1 print(response.text) def main(): """Start script.""" pipelines = get_pipelines() red_pipelines = get_red_jobs(pipelines) filtered_pipelines = filter_pipelines(red_pipelines) get_pipeline_git_urls(filtered_pipelines) git_related_work(filtered_pipelines) jobs_to_post = post_header_to_slack(filtered_pipelines) post_to_slack(jobs_to_post) if __name__ == '__main__': main()