I have built a housekeeping script and I would like to improve it. I have a JSON file with the details such as directory path, number of days after which the directory or file should be zipped, deletion period for the zip as well as if it's files or directories that we are zipping.
Config JSON file:
{
"foo": [
{
"path": "/the/directory/to/be/zipped/",
"nr_files_not_zip": "7",
"deletion_period": "7",
"file_or_directory": "files"
}
]
}
For housekeeping, I have tried to use a factory design pattern as there are potentially different file selection requirements but for the moment I am only selecting the files to be zipped based on their mod date. The script loads the config and creates the housekeeping object through a factory method. Then based on the zip period, deletion period as well as the zip-type it zips and deletes the appropriate files or directories.
I have been using python for a couple of years and I trying to improve. I would really appreciate any feedback regarding the code improvements and best practices. The housekeeping script:
#!/usr/bin/python
import abc
import os
import sys
import json
import datetime
import time
import zipfile
import shutil
#abstract housekeeping
class housekeeping(object):
__metaclass__ = abc.ABCMeta
@abc.abstractmethod
def pick_files(self):
pass
@abc.abstractmethod
def files_to_be_zipped(self):
pass
@abc.abstractmethod
def sort_deletion(self):
pass
@abc.abstractmethod
def get_type(self):
pass
@abc.abstractmethod
def get_directory(self):
pass
@abc.abstractmethod
def get_deletion_period(self):
pass
#houskeeping foo concrete class
class housekeeping_foo(housekeeping):
def __init__(self,directory, normal_files,zipped_files,nr_to_leave,deletion_period,type_to_zip):
self.normal_files = normal_files
self.zipped_files = zipped_files
self.nr_to_leave = nr_to_leave
self.deletion_period = deletion_period
self.directory = directory
self.type_to_zip = type_to_zip
#grabs the files, directories and zip files and stores them in a specif hashmap
def pick_files(self):
for dir_name, subdirList, fileList in os.walk(self.directory):
for file_name in fileList:
full_path = dir_name + "/" + file_name
if os.path.exists(full_path) and not file_name.startswith('.'):
if not full_path.endswith('.zip') and not full_path.endswith('.gz'):
mod_time = os.path.getmtime(full_path)
last_modified_date = datetime.datetime.fromtimestamp(mod_time)
self.normal_files.update({full_path: last_modified_date})
else:
mod_time = os.path.getmtime(full_path)
last_modified_date = datetime.datetime.fromtimestamp(mod_time)
self.zipped_files.update({full_path: last_modified_date})
return self.normal_files, self.zipped_files
#removes the files which not be zipped. Based on the mod date for this particular object
def files_to_be_zipped(self):
for key, value in self.normal_files.items():
str_date = value.date()
today = datetime.datetime.now().date()
difference = str(today - str_date)
if "days" in difference:
date = difference.split(" ", 1)[0]
else:
date = 0
if int(date) > int(self.nr_to_leave):
print("file to be zipped")
else:
del self.normal_files[key]
return self.normal_files
#removes the files which should not be included in deletion
def sort_deletion(self):
for key, value in self.zipped_files.items():
time_since_insertion = str(datetime.datetime.now() - value)
if "days" in time_since_insertion:
days = int(time_since_insertion.split(' ',1)[0])
else:
days = 0
if days < int(self.deletion_period):
del self.zipped_files[key]
return self.zipped_files
#return zip type
def get_type(self):
return self.type_to_zip
#return directory
def get_directory(self):
return self.directory
#return deletion period
def get_deletion_period(self):
return self.deletion_period
class housekeeping_factory:
def create_housekeeping(self,name, directory, normal_files,zipped_files,nr_to_leave,deletion_period,type_to_zip):
if name == "foo":
return housekeeping_foo(directory,normal_files,zipped_files,nr_to_leave,deletion_period,type_to_zip)
#loading JSON config file
def load_config(directory_name):
with open('housekeeping.config.json') as f:
data = json.load(f)
for config_items in data[directory_name]:
directory_name_wih_full_path = config_items["path"]
nr_to_leave = config_items["nr_files_not_zip"]
deletion_period = config_items["deletion_period"]
type_to_zip = config_items["file_or_directory"]
return directory_name_wih_full_path, nr_to_leave, deletion_period, type_to_zip
#creates the housekeeping object strips the files and directories which should not be included in the zipping or deletion
def housekeeping_client():
normal_files = {}
zipped_files = {}
if len(sys.argv) == 2:
directory_name = sys.argv[1]
directory_name_wih_full_path, nr_to_leave, deletion_period, type_to_zip = load_config(directory_name)
factory = housekeeping_factory()
housekeeping_shape = factory.create_housekeeping(directory_name, directory_name_wih_full_path,normal_files,zipped_files,nr_to_leave,deletion_period,type_to_zip)
housekeeping_shape.pick_files()
files_to_zip = housekeeping_shape.files_to_be_zipped()
to_delete = housekeeping_shape.get_deletion_period()
if to_delete != "NA":
file_to_del = housekeeping_shape.sort_deletion()
else:
file_to_del = {}
zip_type = housekeeping_shape.get_type()
file_directory = housekeeping_shape.get_directory()
run_housekeeping_client(files_to_zip,file_to_del,zip_type, file_directory)
else:
print("Wrong number of arguments...")
#depending on the zip type a zip directory or zip file methods will be called
def run_housekeeping_client(files_to_zip,file_to_del,zip_type, file_directory):
for key, value in files_to_zip.items():
directory_full_path_with_file = key
if zip_type == "files":
zip_the_file(key, key)
os.remove(directory_full_path_with_file)
elif zip_type == "directory":
zip_the_directory(key)
shutil.rmtree(directory_full_path_with_file)
####################################################
if len(file_to_del.keys()) != 0:
for key, value in file_to_del.items():
directory_full_path_with_file = key
os.remove(directory_full_path_with_file)
#zipping directories
def zip_the_directory(dir_name):
file_with_zip = dir_name
file_to_zip = dir_name + ".zip"
zip_File = zipfile.ZipFile(file_to_zip,'w', zipfile.ZIP_DEFLATED)
directory_path = dir_name.split("/")[-1]
lenDirPath = len(directory_path)
for root, dirs, files in os.walk(file_with_zip):
for file in files:
filePath = os.path.join(root, file)
zip_File.write(filePath , filePath[lenDirPath :])
zip_File.close()
#zipping files
def zip_the_file(file_name):
file_with_zip = file_name + ".zip"
file_to_zip = file_name
directory_path = file_name.split("/")[-1]
len_dir_path = len(directory_path)
with zipfile.ZipFile(file_with_zip, 'w', compression=zipfile.ZIP_DEFLATED) as zip_file:
zip_file.write(file_to_zip, file_to_zip[len_dir_path :])
print("file has been zipped")
if __name__== '__main__':
housekeeping_client()