Clean up directory and file artifacts

Question

I have some directories with artifacts in them, and I want to clean them up by deleting them. The requirements are:

• Keep the 3 most recent directories
• Keep only one snapshot folder
• Keep only one rc folder
• Delete everything else
• The print commands should be written to a log file for posterity

This code is in the root of a Linux Server and it will be run every morning.

Lib import In this case we only have basic libraries, so we do not need to raise the environment first

import os, sys, glob
import datetime
import re
import shutil
import subprocess

from xml.etree import ElementTree
from xml.etree.ElementTree import Element, SubElement, Comment
from xml.dom import minidom

Environment

env = 'cd environment/bin/activate'

PATHs

mypath = '/home/directories' #test
path_log = '/home/directories/delete-versions.log' #test

Gobal VAR

percent = 50
versions = 3
snapshots = 1
rcs = 1

Security exclude directories

exclude = ['a', 'b', 'c', 'd']

PREP

def start_var():
    now = datetime.datetime.now()
    return now

def raise_environment(env):
    try:
        subprocess.run(env, shell=True)
        print('Environment raised')
    except:
        print('Error: Environment not found. Please, run again manualy')

def info_log(path_log, message):
  with open(path_log,'a') as f:
      f.write(f'\n{message}\n')


###############################
######### Check space #########
###############################      
def bash_commands(command):
    ocup = str(subprocess.check_output(command, shell=True))
    ocup = int(str(re.findall('\d+', ocup)).replace("['", "").replace("']", ""))
    return ocup   

###############################
######### Acquisition #########
###############################
def getting_routes(mypath, exclude):  
    # Getting the list of the directories I am going to iter
    roots = routes = []
    # Let outside the exclude routes
    roots = os.listdir(mypath)
    roots = [mypath + '/' + x for x in roots if x not in exclude]
    # Looking for directories with more than one version and with xx.xx.xx
    # When I found a version directory, i get the up route
    for root in roots:
        for (dirpath, _, _) in os.walk(root):
            find = re.findall('\d+\.\d+\.\d+', dirpath)
            if len(find) >= 1:
                directory = str(re.findall('^(.+)\/[^\/]+$', dirpath)).replace("['", "").replace("']", "")
                if directory not in routes:
                    routes.append(directory)
    print(f'Routes ready')
    info_log(path_log, 'Routes ready')
    return(routes)

############################
######### Wrangling #########
############################
def delete(path, delete_time):
    if len(delete_time) > 0:
        for item in delete_time:
            #shutil.rmtree(path + '/' + item, ignore_errors=True)
            #I want to know if I delete or not the directories, so I do not use ignore_erros and I create a try/except
            try:
                shutil.rmtree(path + '/' + item)
                message08 = ' Deleting: '+ path  + '/' + item
                print(f'\n{message08}\n')
                info_log(path_log, message08)
            except:
                message09 = item + ' read only. We do not delete'
                print(f'\n{message09}\n')
                info_log(path_log, message08)
                
                
def prettify(elem):
    #Return a pretty-printed XML string for the Element.
    rough_string = ElementTree.tostring(elem, 'utf-8')
    reparsed = minidom.parseString(rough_string)
    return reparsed.toprettyxml(indent="  ") # For each element
    

def create_modify_xmls(path, all, keep_directories, keep_snapshots):
          now = str(datetime.datetime.now())
          top = Element('metadata')
          child1 = SubElement(top, 'Id')
          child1.text = '.'.join(path.replace(mypath + '/', '').split('/')[:-1])
          child2 = SubElement(top, 'Id02')
          child2.text = path.split('/')[-1]
          child3 = SubElement(top, 'versioning')
          current_group = SubElement(child3, 'versions')
          lastupdated = SubElement(child3, 'lasUpdated')
          lastupdated.text = now
          # metadata-local
          for a in all:
              version = SubElement(current_group, 'version')
              version.text = a
          xml = str(prettify(top))
          with open(path + '/-local.xml','w') as f:
              f.write(xml)
          # metadata-releases
          for k in keep_directories:
              version = SubElement(current_group, 'version')
              version.text = k
          xml = str(prettify(top))
          with open(path + '/-releases.xml','w') as f:
              f.write(xml)
          for s in keep_snapshots:
              version = SubElement(current_group, 'version')
              version.text = s
          xml = str(prettify(top))
          with open(path + '/-snapshots.xml','w') as f:
              f.write(xml)

############################
######### Analysis #########
############################
def find_directories_snapshots_rcs(routes, snapshots, rcs, versions):           
    for path in routes:# List of routes to find
        files = os.listdir(path) #List with all inside path
        snapshots = keep_snapshorts = delete_snapshots = []
        rcs = keep_rcs = delete_rcs = xmls = []
        all_directories = keep_directories = delete_directories = []
        message03 = '----------------------------------------------------\nGo to:'+ path +'\n----------------------------------------------------'
        print(f'\n{message03}\n')
        info_log(path_log, message03)
        for f in files: # For each element
            is_directory = os.path.isdir(path + '/' + f)
            if is_directory == True:
                all_directories.append(f)
        all_directories.sort(reverse=True)
        message04 = ' All directories: '+ str(all_directories)
        print(f'\n{message04}\n') 
        info_log(path_log, message04)
        # We are going to find here snapshot, redhat and RCs
        # Everything else is going to be treated as the same
        snapshots = [w for w in all_directories if 'SNAPSHOT' in w]
        snapshots.sort(reverse=True)
        if len(snapshots) > 0:
            keep_snapshots = snapshots[:snapshots]
            delete_snapshots = snapshots[snapshots:]
            message05 = ' All snapshots:'+ str(snapshots) +'\n Snapshots to keep: ' + str(keep_snapshots) + '\
            \n Snapshots to delete: ' + str(delete_snapshots)
            print(f'\n{message05}\n')
            info_log(path_log, message05)
        # Now RCs
        rcs = [w for w in all_directories if 'RC' in w]
        rcs.sort(reverse=True)
        if len(rcs) > 0:   
            keep_rcs = rcs[:rcs]
            delete_rcs = rcs[rcs:]
            message06 = ' All RCs:'+ str(rcs) + '\n RCs to keep: ' + str(keep_rcs) + '\n RCs to delete: '+ str(delete_rcs)
            print(f'\n{message06}\n')
            info_log(path_log, message06)
        # Now redhats
        # We want to delete all redhats
        redhats = [w for w in all_directories if 'redhat' in w]   
        # Preparamos
        all_directories = [x for x in all_directories if x not in snapshots]
        all_directories = [x for x in all_directories if x not in rcs]
        all_directories = [x for x in all_directories if x not in redhats]           
        keep_directories = all_directories[:versions]
        delete_directories = all_directories[versions:] + redhats
        delete_time = delete_snapshots + delete_rcs + delete_directories
        all = keep_directories + keep_rcs + keep_snapshots
        all.sort()
        message07 = ' Directories:'+ str(all_directories) +'\n Directories to keep: '+ str(keep_directories) +'\n Directories to delete: '+ str(delete_directories)
        print(f'\n{message07}\n')
        info_log(path_log, message07)
        # Now is when delete for real
        delete(path, delete_time)
        # Create XML
        create_modify_xmls(path, all, keep_directories, keep_snapshots)

def duration(start):
    end = datetime.datetime.now()
    duration = end - start
    message10 = 'Duracion del proceso: '+ str(duration)
    print(f'\n{message10}\n')
    info_log(path_log, message10)

#################################################################################################

if __name__ == '__main__':
    raise_environment(paradigma_env)
    start = start_var()
    message01 = '--------------------------------- Ejecution ' + str(start)+' ------------------'
    info_log(path_log, message01)
    command01 = "df -k | grep root | awk '{print $5}'"
    ocup01 = bash_commands(command01)
    if ocup01 < percent:
        # If the ocupation of the server ies less tahan the percent we did, out and log
        message02 = 'Ocu is ' + str(ocup01) + '%, less than '+ str(percent) +'%.\
        \nOut'
        print(f'\n{message02}\n')
        info_log(path_log, message02)
    else:
        # It the ocupation is high or equal to percent, start
        message03 = 'Ocup is '+ str(ocup01) +'%, higher or equal to '+ str(percent) +'%.\nStart delete process'
        print(f'\n{message03}\n')
        info_log(path_log, message03)
        routes = getting_routes(mypath, exclude)
        find_directories_snapshots_rcs(routes, snapshots, rcs, versions)  
    duration(start)

Answer 1

Stream of consciousness review from top-to-bottom.

• You should make sure you're adhering to PEP8 and general styling guides. I like to use Black to format my code
• start_var just obfuscates datetime.datetime.now() and doesn't need to exist. duration is similar. I honestly would rather see this wrapped up in your info logger class (see below) - have it log start/end time on __enter__ and __exit__.
• I don't see the point of raise_environment - you could just do os.chdir(path)

info_log

I don't like that this repeatedly opens a file everytime you need it, and now everything needs to know/assume the path_log variable everywhere. Instead, I would prefer that you construct a logger object and pass that around. This also lets you avoid the messy duplication of printing and logging.

I would do something like this (untested):

import sys
class InfoLogger:

    def __init__(self, log_file, print_loc=sys.stdout):
        self.log_file = log_file
        self.print_here = print_loc

    def __enter__(self):
        self.open_file = open(self.log_file, 'a')

    def __exit__(self):
        self.open_file.close()

    def log_message(message):
        to_write = f"\n{message}\n"
        self.open_file.write(to_write)
        self.print_here.write(to_write)

This lets you do something like this:

with InfoLogger(path_log) as logger:
    getting_routes(base_path, exclude_list, logger)
    # etc

If you don't do this, please at least factor your print statements into this as well to avoid duplication of formatting.

getting_routes

Throughout this function, you use low-level os APIs, or you do direct string operations. In Python 3.4+, you can use pathlib instead.

A few additional notes:

• re.findall('\d+\.\d+\.\d+') is pretty magic, and doesn't actually get used (we don't care about the versions, we just want to check if they exist). I would probably wrap this in a helper function
• directory = str(re.findall('^(.+)\/[^\/]+$', dirpath)).replace("['", "").replace("']", "") is not what you want - you want re.findall('pattern').join(",") (I don't know why - this line doesn't make sense to me, which is a good indicator that you need a helper function, to split it onto multiple lines, and maybe some comments)
• if directory not in routes can get very expensive for long lists. Consider using a set instead

I ended up with something like this:

def getting_routes(mypath, exclude):
    routes = set()
    get_routes_recursive(mypath, exclude, routes)
    return routes

def get_routes_recursive(base_path, exclude_list, routes):
    for path in base_path.iterdir():
        if path.name in exclude_list:
            continue
        if path.is_dir():
            if is_versioned_path(path.name):
                add_all_children(path, exclude_list, routes)
            else:
                get_routes_recursive(path, exclude_list, routes)

def add_all_children(base_path, exclude_list, routes):
    routes.update(
        path
        for path in base_path.glob("**\*")
        if path.name not in exclude_list
    )            

def is_versioned_path(path):
    return re.findall(r"\d+\.\d+\.\d+", path.name) 

delete

You should not use length to identify non-empty lists (this applies elsewhere) - instead, you can just do if my_list. Even better, if the only thing you want to do is loop, then just loop - it won't do anything if empty.

When you handle exceptions, you should never use a bare except: - always catch the specific list of exceptions you want to do something with.

Additionally, for safety, you should avoid just concatenating paths. Instead, you can use the slash operator: shutil.rmtree(path / item) (this assumes you're using pathlib.

I didn't notice it until here, but you don't need a unique messageX variable for each message (I don't think you need them at all - see InfoLogger above). Just use message (or whatever) each time.

create_modify_xmls

This line is very suspect:

child1.text = '.'.join(path.replace(mypath + '/', '').split('/')[:-1])

I don't know exactly what you're trying to do - I think you're trying to remove your root path, and then get the path without the final component? There are a number of APIs that I think will work better for this. Specifically, you would do something like this (just guessing):

mypath = Path("/home/directories")
childpath = Path(<something>)
if childpath.is_relative_to(mypath):
    mypath_parents_length = len(mypath.parents)
    child1.text = ".".join(
        parent.name
        for i, parent in enumerate(childpath.parents)
        if i >= mypath_parents_length
    )
else:
    child1.text = ".".join(childpath.parents.name)

Similarly, child2.text = path.split('/')[-1] should become child2.text = path.name

find_directories_snapshots_rcs

Again, this will be cleaner with pathlib instead of os.

There's a lot of code here, and I'm hitting review fatigue, so I won't touch on everything:

This pattern (some_list = list[:list]) doesn't work - you can't slice using a list, unless there is something I'm missing about how you've defined this.

I suspect that you'll be better off not using list comprehensions and just looping over all_directories once to accumulate your other lists.

Avoid using the names of builtins (all) as variable names

Final thoughts

I don't think you need to use subprocess ("df -k | grep root | awk '{print $5}'"); I think you can just use os.statvfs (I'm on a Windows machine, so I can't test).