Moving Artifacts

VICTOR-OSEGHALE

This script helps to move artifacts from one location to another without overwriting any file.

• Compatible with Python 2.7+
• Moves artifacts from an old location STALE_ARTIFACT_DIRECTORY to a new location ARTIFACT_DIRECTORY
• Only moves the artifact which doesn't exist in the ARTIFACT_DIRECTORY to avoid overwriting artifacts

import os, shutil, re
STALE_ARTIFACT_DIRECTORY = "/home/user/atlassian-home/artifacts/" # Path to the Home directory artifacts
ARTIFACT_DIRECTORY = "/home/user/atlassian-home/shared/artifacts" # Path to Shared directory artifact

list_of_migrated_artifacts =[]

# If the path to an artifact location does not exist in {BAMBOO-HOME}/shared directory but exist {BAMBOO-HOME}/artifact directory, we consider it as an artifact that needs to be moved
def is_artifacts_present_in_current_plans(storage_tag='',plan_artifact='', type=''):
    # Construct a path to Bamboo artifact storage in plan storag tag
    storage_path = '{0}/{1}/{2}'.format(ARTIFACT_DIRECTORY,storage_tag,type)
    if os.path.exists(storage_path):
        # check if artifact exist in the path
        artifact_path = '{0}/{1}/{2}/{3}'.format(ARTIFACT_DIRECTORY,storage_tag,type,plan_artifact)
        stale_artifact_path = '{0}{1}/{2}/{3}'.format(STALE_ARTIFACT_DIRECTORY,storage_tag,type,plan_artifact)
        # If the artifact exist in {BAMBOO-HOME}/shared directory, do nothing
        if os.path.exists(artifact_path): 
            x = None
        else:
            shutil.move(stale_artifact_path,storage_path)
            
            result = {"Moved": stale_artifact_path, "to":storage_path}
            list_of_migrated_artifacts.append(result)
     
    else:
        stale_plan_storage= '{0}{1}'.format(STALE_ARTIFACT_DIRECTORY,storage_tag)
        main_plan_storage = '{0}/{1}'.format(ARTIFACT_DIRECTORY,storage_tag) 
        # Move the storage path and all its directories to {BAMBOO-HOME}/shared/artifacts/... 
        if not os.path.exists(main_plan_storage):
            tmp_stale_storage = '{0}{1}'.format(STALE_ARTIFACT_DIRECTORY,storage_tag)
            tmp_main_plan_storage = '{0}/{1}'.format(ARTIFACT_DIRECTORY,storage_tag)
           
            shutil.move(tmp_stale_storage,tmp_main_plan_storage)
            result = {"Moved":tmp_stale_storage, "to":tmp_main_plan_storage}
            list_of_migrated_artifacts.append(result)
     
        else:
            _tmp_stale_storage = '{0}{1}/{2}'.format(STALE_ARTIFACT_DIRECTORY,storage_tag,type)
            tmp_main_plan_storage = '{0}/{1}'.format(ARTIFACT_DIRECTORY,storage_tag)
            shutil.move(_tmp_stale_storage,tmp_main_plan_storage)
            result = {"Moved":_tmp_stale_storage,"to":tmp_main_plan_storage}
            list_of_migrated_artifacts.append(result)
            
    return


def prepare_artifacts_directories():
    try:
        for storage_tag in os.listdir(STALE_ARTIFACT_DIRECTORY): # list all storage tags in artifact directory # {storage_tag} 
            storage_path = os.path.join(STALE_ARTIFACT_DIRECTORY, storage_tag) # create a path to the storage tags /Users/voseghale/atlassian-home/bamboo803/artifacts/{storage_tag} 
            if os.path.isdir(storage_path):
                for x in os.listdir(storage_path): # get all shared and none shared artifacts
                    is_shared_artifacts = re.match(r'shared',x)
                    if is_shared_artifacts:
                        #
                        # Get all shared artifacts
                        shared_artifact_path = os.path.join(storage_path,x) # create the path to shared
                        if os.path.isdir(shared_artifact_path):
                            for artifact_ in os.listdir(shared_artifact_path): # list each artifact in a shared path
                                path_to_each_artifact = os.path.join(shared_artifact_path,artifact_)
                                if os.path.isdir(path_to_each_artifact):
                                    result = is_artifacts_present_in_current_plans(storage_tag=storage_tag,plan_artifact=artifact_, type=x)           
                                    
                    else:
                        # Get all none shared artifacts
                        share_artifact_path = os.path.join(storage_path,x) # create the path to shared
                        if os.path.isdir(share_artifact_path):
                            for artifact_ in os.listdir(share_artifact_path): # list each artifact in a shared path
                                path_to_each_artifact = os.path.join(share_artifact_path,artifact_)
                                if os.path.isdir(path_to_each_artifact):   
                                    result = is_artifacts_present_in_current_plans(storage_tag=storage_tag,plan_artifact=artifact_, type=x)
                                                               


        print("---- Artifacts Copied ----")
        print(list_of_migrated_artifacts)
    except FileNotFoundError as err:
        print(err)
    return

if __name__== "__main__":
    prepare_artifacts_directories()

 

Danwin Ambros-Francis

Great script @VICTOR-OSEGHALE it helped a ton to resolve our issue!