# Clean up directory and file artifacts

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:

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)


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.

• 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