newestuser / github-workflow-dashboard Goto Github PK

    def cleanup_workflow_runs(self, repo, workflow_selection=None, run_selection=None, interactive:bool=True):
        ''' Delete workflow runs.
        Actual you have to do this manual, because a good-usable functionality to clean up the workflow runs on the github page is not implemented... This is lame.
        This function will help you and saves your time.
        see: https://support.github.com/ticket/personal/0/1260794

        In case of interactive-mode
        1. List workflows
            -> Select workflow: Name, >LOST-RUNS< or all
        2. List selectable runs
            -> Select runs to delete (black list): Run Number(s), all, all-except-last, cancelled, failure or success
            -> Select runs to not delete (white list): Run Number(s)
        3. List selected runs (to delete)
            -> Confirm

        Parameter:
            repo: Repository with type 'Github.get_repo()' or as str.
            workflow_selection: default = None, e.g. 'Self Hosted CI'
            run_selection: default = None, e.g. 'all', 'all-except-last', 'cancelled', 'failure' or 'success'
            interactive: default = True. If True the function will expact yout input/selection. If False all workflow-runs will be deleted.
        '''
        # input management
        if isinstance(repo, str):
            repo = self.get_repo(repo)
        if repo == None:
            raise Exception("Selected Repository is not in your organization!")

        # get runs
        totalRunList = repo.get_workflow_runs()

        # get a list of all workflows in the repo
        workflowList = repo.get_workflows()
        if workflowList.totalCount == 0:
            print("There are no workflows in your selected repository: '{}'".format(repo.name))
            return

        # print list
        print("List of all workflows of your selected repository: '{}'".format(repo.name))
        run_count = 0
        for workflow in workflowList:
            runs = workflow.get_runs()
            run_count = run_count + runs.totalCount
            print(" ID: '{}', Name: '{}', State: '{}': Run(s): '{}'".format(workflow.id, workflow.name, workflow.state, runs.totalCount))
        
        # calculate lost runs
        lost_run_count = totalRunList.totalCount-run_count
        if lost_run_count > 0:
            print(" LOST-RUNS: There are '{}' workflow run(s) without an active yaml-file!".format(lost_run_count))

        # if no selection of a workflow is initiated by function-call
        if workflow_selection == None and interactive:
            workflow_selection = input("Please select a workflow by Name (e.g. >CI< or >LOST-RUNS< or press [Return] for all):")
        # if no interactive and no selection is done: choose the first workflow of the list
        elif workflow_selection == None and not interactive:
            print("No workflow chosen.")

        # search workflow in list
        found = False

        # if no workflow chosen
        if workflow_selection == None or workflow_selection == '':
            found = True
            runList = totalRunList
            print("{} run(s) found.".format(totalRunList.totalCount))
        # in case of lost runs
        elif workflow_selection == 'LOST-RUNS':
            found = True
            runList = []
            for run in totalRunList:
                run_found = False
                # check if run has a workflow
                for workflow in workflowList:
                    if run.workflow_id == workflow.id:
                        run_found = True
                # if not, add it to the run list
                if run_found == False:
                    runList.append(run)
            # raise exception if something went wrong
            if len(runList) != lost_run_count:
                raise Exception("Something went wrong with get lost runs!")
            print("{} run(s) found.".format(lost_run_count))
        # if workflow chosen
        else:
            for workflow in workflowList:
                if workflow.name == workflow_selection:
                    print("Workflow found in list above: {}".format(workflow_selection))
                    found = True
                    # get runs
                    runList = workflow.get_runs()
                    print("{} run(s) found.".format(runList.totalCount))
                    break
        
        # if workflow found
        if found:
            print("Run List: ")
            for run in runList:
                #print(" Run Number: '{}', Satus: '{}', Conclusion: '{}', ID: '{}'".format(run.run_number, run.status, run.conclusion, run.id))
                print(" Run Number: '{}', Satus: '{}', Conclusion: '{}', ID: '{}', Branch: '{}', SHA: '{}'".format(run.run_number, run.status, run.conclusion, run.id, run.head_branch, run.head_sha[0:7]))   
            # get input
            if run_selection == None and interactive:
                ans = input("Select a special run number (e.g. 123), a list of workflow run numbers seperated by comma (e.g. 123, 111, 112) or type >all<, >all-except-last<, >cancelled<, >failure< or >success<:")
            # choose all if no selection is done an not interaction is used
            elif run_selection == None and not interactive:
                ans = 'all'
            # choose selection if special run is selected
            else:
                ans = run_selection
            # manage input: choose selection
            selectedRuns = []
            # special run number is selected
            if ans not in ['all', 'cancelled', 'failure', 'success', 'all-except-last']:
                ansSplit = ans.split(',')
                for id in ansSplit:
                    if id.isnumeric():
                        selectedRuns.append(int(id))
                    else:
                        raise Exception("Invalid input '{}'!".format(id))
            else:
                for run in runList:
                    # delete all runs
                    if ans == 'all':
                        selectedRuns.append(run.run_number)
                    # check conclusion
                    elif ans in ['cancelled', 'failure', 'success'] and run.conclusion == ans:
                        selectedRuns.append(run.run_number)
                    # in case of "all-except-last"
                    elif ans == 'all-except-last' and not run == runList[0]:
                        selectedRuns.append(run.run_number)

            # whitelist
            if interactive:
                ans = input("Do you want to keep runs? Select a special run number (e.g. 123) or a list of workflow run numbers seperated by comma (e.g. 123, 111, 112). For no run keeping press [Return]:")
                if ans != '':
                    keepRuns = []
                    # convert run slection
                    ansSplit = ans.split(',')
                    for id in ansSplit:
                        keepRuns.append(int(id))

                    # step over keep runs
                    for keepRun in keepRuns:
                        # search in list and change List
                        for selectedRun in selectedRuns:
                            if selectedRun == keepRun:
                                # delete keepRunId from run List (keeping)
                                selectedRuns.remove(keepRun)
                                break

            # list runs to delete
            if interactive:
                print("Selected Run List: ")
                for run in runList:
                    if run.run_number in selectedRuns:
                        #print(" Run Number: '{}', Satus: '{}', Conclusion: '{}', ID: '{}'".format(run.run_number, run.status, run.conclusion, run.id))
                        print(" Run Number: '{}', Satus: '{}', Conclusion: '{}', ID: '{}', Branch: '{}', SHA: '{}'".format(run.run_number, run.status, run.conclusion, run.id, run.head_branch, run.head_sha[0:7]))          

                ans = input("Confirm selection! Yes [default] or No:")
                if ans in ['y', 'Y', 'yes', 'YES', 'Yes', '']:
                    pass
                elif ans in ['n', 'N', 'no', 'NO', 'No']:
                    return
                else:
                    print("Unkown input! Skip.")
                    return

            # interate over list and 
            for run in runList:
                if run.run_number in selectedRuns:
                    # delete completed workflow run
                    if run.status == 'completed':
                        print(" delete run with number: '{}'".format(run.run_number))
                        repo._requester.requestJson("DELETE", run.url)
                    # ask for cancel workflow run
                    else:
                        print("It is not possible to delete run: '{}'. This is not completed.".format(run.run_number))
                        if interactive:
                            ans = input("Do you want to cancel it? Yes [default] or No:")
                            if ans in ['y', 'Y', 'yes', 'YES', 'Yes', '']:
                                run.cancel()
                                print("Please rerun cleanup")
                            elif ans in ['n', 'N', 'no', 'NO', 'No']:
                                print("No cancel selected!")
                            else:
                                print("Unkown input! Skip.")
        else:
            print("Workflow '{}' not found in list above.".format(workflow_selection))```