#!/usr/bin/env python """ Migration Assistant [MA] Python Utility This utility performs some useful tasks on behalf of VMware-Migration-Assistant.exe. It allows more rapid development of features, and frees us from having to deal with Windows API quirks. The initial version deals with zip files (zip, unzip, test, compare). More capabilities can be added easily in the future if needed. # SYNOPSIS usage: ma_util.py [-h] [--zip zip_file source_dir] [--unzip zip_file target_dir] [--test zip_file] [--compare zip_file source_dir] [--result file] [--log file] [--config file] optional arguments: -h, --help show this help message and exit commands: --zip zip_file source_dir create zip file from directory --unzip zip_file target_dir extract zip file to directory --test zip_file test integrity of zip file --compare zip_file source_dir compare zip file against directory options: --result file result json path --log file log file path --config file config file path # EXAMPLE ma_util.py --zip file.zip source_dir ma_util.py --unzip file.zip target_dir ma_util.py --test file.zip ma_util.py --compare file.zip source_dir # NOTES By default it logs to the same MA log, %temp%/migration-assistant.log, and output status in the JSON %temp%/ma_util.json. """ import sys import logging import argparse import os import json import datetime import shutil import mazipfile as zipfile import re import subprocess LOG_FORMAT = '%(asctime)s | %(module)s| %(levelname).1s: %(message)s' LOG_FILE = 'migration-assistant.log' RESULT_FILE = 'ma_util.json' MA_CONFIG = 'ma_config.json' is_win, is_mac, is_lin = (False, False, False) # pylint: disable=invalid-name exe_name = os.path.basename(sys.argv[0]) # pylint: disable=invalid-name exe_dir = os.path.dirname(__file__) # pylint: disable=invalid-name class ZipFile(object): """Class that handles zip files""" def __init__(self, result): """ Constructor @param result: result object @type result: ResultOutput """ self._result = result def create(self, zip_file, source_dir, include_dirpath=False): """ Create a zip file from source directory @param zip_file: zip file to create @param source_dir: source directory to compress @param include_dirpath: should source dir be in zip path @return: True if operation is successful """ logging.info("Creating zip file from directory ...") logging.info(" zip: " + zip_file) logging.info(" dir: " + source_dir) try: # chop off extension if any, as shutil.make_archive adds it zip_file = os.path.splitext(zip_file)[0] # if any of these special dir are specified as input, # don't include dir path, just standard zip approach n_path = os.path.normpath(source_dir) if n_path in ['.', '..', '/', '\\']: logging.info('Special dir, not using dir name in zip') include_dirpath = False # if path is c:\ or c: if is_win and re.match(r'^[a-zA-Z]:(\\)?$', n_path): logging.info('Root drive, not using dir name in zip') include_dirpath = False if not include_dirpath: # this puts the contents of source_dir in to zip file shutil.make_archive(zip_file, 'zip', source_dir) else: # this puts source_dir/* into the zip file shutil.make_archive(zip_file, 'zip', os.path.join(source_dir, os.path.pardir), # from parent source_dir) # zip this folder logging.info('Zip file created.') return True except Exception as ex: # pylint: disable=broad-except # msg = str(ex).decode('string_escape') # py2 only msg = str(ex) logging.error(msg) self._result.error = msg my_print(msg) def extract(self, zip_file, dest_dir): """ Extract the content of the zip file. @param zip_file: zip file to extract from @param dest_dir: directory to extract to @return: True if operation is successful """ logging.info("Extracting zip file to directory ...") logging.info(" zip: " + zip_file) logging.info(" dir: " + dest_dir) try: with zipfile.ZipFile(zip_file) as zfile: zfile.extractall(dest_dir) logging.info('Zip file extracted.') return True except Exception as ex: # pylint: disable=broad-except # msg = str(ex).decode('string_escape') # py2 only msg = str(ex) logging.error(msg) self._result.error = msg my_print(msg) def test(self, zip_file): """ Check the integrity of the zip file. @param zip_file: input zip file @return: True if zip file is valid. """ logging.info("Testing zip file ...") logging.info(" " + zip_file) try: with zipfile.ZipFile(zip_file) as zfile: retval = zfile.testzip() # return first bad file if not retval: logging.info('No errors detected.') return True msg = "Corrupt file in zip: {}".format(retval) self._result.error = msg logging.error(msg) my_print(msg) except Exception as ex: # pylint: disable=broad-except # msg = str(ex).decode('string_escape') # py2 only msg = str(ex) logging.error(msg) self._result.error = msg my_print(msg) @staticmethod def _list_files_from_dir(path, add_dir_name=True): """ Get all the files from specific path recursively. @param path: directory to get files from @return: list of all files from all subdirectories @rtype: list """ retval = [] for root, dirs, files in os.walk(path): # pylint: disable=W0612 # just process files, not directories for filename in files: f_name = os.path.join(root, filename) # first remove the input 'path' prefix if f_name.startswith(path): f_name = f_name[len(path) + 1:] # also remove / after prefix path retval.append(f_name) # If path is c:\a\b\c\d, add 'd' to the generated list of # files, since the zip file content contains 'd' in its path. if add_dir_name: prefix = os.path.basename(os.path.normpath(path)) retval = [os.path.join(prefix, x) for x in retval] # Replace all backslash with slash. Need to do this at the end # since join() above may introduce \ as well. if is_win: retval = [x.replace('\\', '/') for x in retval] logging.info("Directory has %d files.", len(retval)) return retval def _list_files_from_zip(self, zip_file, include_dirs=False): """ Return the list of files from zip file. @param zip_file: input zip file @param include_dirs: should directory entry be included @return: list of files @rtype: list """ if not os.path.exists(zip_file): msg = "Zip file does not exist: {}".format(zip_file) logging.error(msg) self._result.error = msg my_print(msg) return try: with zipfile.ZipFile(zip_file) as zfile: # this returns file path with / in them file_list = zfile.namelist() if not include_dirs: # filter out directories, those that ends with / file_list = [x for x in file_list if not x.endswith('/')] logging.info("Zip file has %d files.", len(file_list)) return file_list except Exception as ex: # pylint: disable=broad-except # msg = str(ex).decode('string_escape') # py2 only msg = str(ex) msg = "Failed to get file list from zip: {}".format(msg) logging.error(msg) self._result.error = msg my_print(ex) def compare(self, zip_file, source_dir): """ Compare zip_file content with directory. @param zip_file: input zip file @param source_dir: directory to compare against @return: True if zip_file content matches directory content. @rtype: bool """ logging.info("Compare zip file with directory ...") logging.info(" " + zip_file) logging.info(" " + source_dir) zip_file_sn = os.path.basename(zip_file) logging.info("Getting list of files from zip ...") file_list = self._list_files_from_zip(zip_file) if not file_list: msg = 'No files in zip: {}'.format(zip_file) my_print(msg) self._result.error = msg logging.error(msg) return logging.info("Getting list of files from directory ...") dir_list = ZipFile._list_files_from_dir(source_dir) if not dir_list: msg = 'No files in directory: {}'.format(source_dir) my_print(msg) self._result.error = msg logging.error(msg) return len1 = len(file_list) len2 = len(dir_list) if len1 != len2: msg = 'Number of files do not match: zip ({}), directory ({})'.format(len1, len2) my_print(msg) self._result.warning = msg logging.warning(msg) # return else: msg = 'Number of files match: zip ({}), directory ({})'.format(len1, len2) my_print(msg) logging.info(msg) # pprint.pprint(dir_list) # pprint.pprint(file_list) # check if zipfile is missing any files that exist in dir dir_set = set(dir_list) file_set = set(file_list) # missing: files in directory but not in zip file # extra: files in zip file but not in directory missing_set = dir_set - file_set extra_set = file_set - dir_set logging.info("Missing files in zip: %d, Extra files in zip: %d", len(missing_set), len(extra_set)) # log extra files data if any if extra_set: # print number of extra files msg = 'Extra files in {}: {}'.format(zip_file_sn, len(extra_set)) my_print(msg) logging.warning(msg) extra_list = sorted(extra_set) for i, d_file in enumerate(extra_list): msg = ' [{}] {}'.format(i + 1, d_file) my_print(msg) logging.info(msg) # if no missing files we are done. extra files are warning only, # though we do log those files. if not missing_set: msg = 'No missing files.' my_print(msg) logging.info(msg) return True # print number of missing files msg = 'Missing files in {}: {}'.format(zip_file_sn, len(missing_set)) my_print(msg) logging.error(msg) # now log the missing files: missing_list = sorted(missing_set) for i, d_file in enumerate(missing_list): msg = ' [{}] {}'.format(i + 1, d_file) my_print(msg) logging.info(msg) # finally report status back on missing files # extra files are not as important so leaving those in log files if len(missing_list) > 10: self._result.error = \ 'Missing files in {} (first 10): {}'.format( zip_file_sn, missing_list[:10]) else: self._result.error = 'Missing files in {}: {}'.format( zip_file_sn, missing_list) class ZipFileExternal(ZipFile): """Class to handle external zip tools""" def __init__(self, result, config): """ Constructor @param result: result object @type result: ResultOutput @param config: config object @type config: ConfigFile """ super(ZipFileExternal, self).__init__(result) self._config = config def _run(self, cmd, shell=False): """ Execute a command and return the output @param cmd: command to execute @type cmd: list @param shell: invoke via shell @return: True if command is successful """ try: msg = 'invoking: {}'.format(cmd) logging.info(msg) my_print(msg) subprocess.check_output(cmd, shell=shell) return True except subprocess.CalledProcessError as ex: logging.error(ex) my_print(ex) logging.error(ex.output) # print(ex.output) self._result.error = 'External tool failed: {}'.format(ex) def create(self, zip_file, source_dir, include_dirpath=False): """ Create a zip file @param zip_file: zip file we want to create @type zip_file: str @param source_dir: what to put into the zip file @type source_dir: str @param include_dirpath: should last part of source_dir be in the zip path? @type include_dirpath: bool @return: True if operation is successful @rtype: bool """ # get the external tool cmd line cmd_line = self._config.zip_external_cmdline() if not cmd_line: logging.error('No external tool defined, reverting to zipfile method.') return super(ZipFileExternal, self).create(zip_file, source_dir, include_dirpath) cmd_line = cmd_line.format(zip_file='"%s"' % zip_file, export_dir='"%s"' % source_dir) # print cmd_line return self._run(cmd_line, True) class Command(object): """Runs various commands this script supports""" def __init__(self, config, result): """ Constructor @type config: ConfigFile @type result: ResultOutput """ self._config = config self._result = result self._zip = ZipFile(self._result) self._zip_external = ZipFileExternal(self._result, config) def run_zip_create(self, zip_file, source_dir): """ Create a zip archive using python zip. @param zip_file: zip file to create @type zip_file: src @param source_dir: what to put in the zip file @type source_dir: src @return: 1 on failure, None/0 on success """ logging.info("zip file: %s", zip_file) logging.info("source dir: %s", source_dir) # logging.info("method: %s", 'python zip') my_print("Creating archive '{}.zip' from directory '{}' ...".format(zip_file, source_dir)) if not self._zip.create(zip_file, source_dir, include_dirpath=True): self._result.status = ResultOutput.ERROR return 1 my_print("Zip file created.") self._result.status = ResultOutput.SUCCESS def run_zip_create_tool(self, zip_file, source_dir): """ Create a zip archive using tools @param zip_file: zip file to create @type zip_file: str @param source_dir: what to put in the zip file @type source_dir: str @return: 1 on failure, None/0 on success """ logging.info("zip file: %s", zip_file) logging.info("source dir: %s", source_dir) logging.info("method: %s", self._config.zip_external_exe_name()) # If file have no extension, add .zip to it. Otherwise some # tools might create it in their native format. if not os.path.splitext(zip_file)[1]: zip_file += '.zip' my_print("Creating archive '{}' from directory '{}' ...".format(zip_file, source_dir)) if not self._zip_external.create(zip_file, source_dir, include_dirpath=True): self._result.status = ResultOutput.ERROR return 1 my_print("Zip file created.") self._result.status = ResultOutput.SUCCESS def run_zip_extract(self, zip_file, target_dir): """ Extract a zip archive @param zip_file: zip file to extract from @type zip_file: str @param target_dir: directory to extract to @type target_dir: str @return: 1 on failure, None/0 on success """ logging.info("zip file: %s", zip_file) logging.info("target dir: %s", target_dir) my_print("Extracting '{}' to '{}' ...".format(zip_file, target_dir)) if not self._zip.extract(zip_file, target_dir): self._result.status = ResultOutput.ERROR return 1 my_print("Extraction succeeded.") self._result.status = ResultOutput.SUCCESS def run_zip_test(self, zip_file): """ Test a zip archive @param zip_file: zip file to test @type zip_file: str @return: 1 on failure, None/0 on success """ my_print("Testing archive '{}' ...".format(zip_file)) if not self._zip.test(zip_file): self._result.status = ResultOutput.ERROR return 1 my_print("No errors detected.") self._result.status = ResultOutput.SUCCESS def run_zip_compare(self, zip_file, source_dir): """ Compare zip archive with source directory @param zip_file: zip file to compare @type zip_file: str @param source_dir: directory to compare against @type source_dir: str @return: 1 on failure, None/0 on success """ my_print("Comparing '{}' with '{}' ...".format(zip_file, source_dir)) if not self._zip.compare(zip_file, source_dir): self._result.status = ResultOutput.ERROR return 1 my_print("Comparison succeeded.") self._result.status = ResultOutput.SUCCESS class ResultOutput(object): """Class that deals with the result JSON file""" SUCCESS = 'success' ERROR = 'error' def __init__(self, config): """ Constructor @param config: config object @type config: ConfigFile """ self._config = config self._file = config.result_path() self._result = {} try: # delete the current status file if exist if os.path.exists(self._file): logging.info('Deleting %s ...', self._file) os.remove(self._file) except Exception as ex: # pylint: disable=broad-except logging.error(ex) @property def status(self): """Return current status""" return self._result.get('status') @status.setter def status(self, val): """Set status to new value""" if val not in [ResultOutput.SUCCESS, ResultOutput.ERROR]: raise ValueError(val) self._result['status'] = val @property def info(self): """Return info messages""" return self._result.get('info') @info.setter def info(self, val): """Add new info message""" curr = self._result.get('info') if curr: curr.append(val) else: self._result['info'] = [val] @property def warning(self): """Return warning messages""" return self._result.get('warning') @warning.setter def warning(self, val): curr = self._result.get('warning') if curr: curr.append(val) else: self._result['warning'] = [val] @property def error(self): """Return error information""" return self._result.get('error') @error.setter def error(self, val): """Set error information""" curr = self._result.get('error') if curr: curr.append(val) else: self._result['error'] = [val] @property def all_msg(self): """Get all the messages as one string""" i_list = self.info or [] w_list = self.warning or [] e_list = self.error or [] if i_list or w_list or e_list: return '\n'.join(i_list + w_list + e_list) def write(self): """Write status to JSON file""" logging.info('Writing %s ...', self._file) # combine warning/error list to form a message list. msg_str = self.all_msg if msg_str: self._result['message'] = msg_str try: write_json_file(self._file, self._result) except Exception as ex: # pylint: disable=broad-except logging.error(ex) class ConfigFile(object): """ Class that handles ma_util configuration """ def __init__(self, args): """ Constructor @param args: namespace returned from parse_args() """ self._args = args self._file = args.config or os.path.join(exe_dir, MA_CONFIG) try: # handle non exist config file gracefully with open(self._file) as data_file: self._config = json.load(data_file) except Exception as ex: # pylint: disable=broad-except, W0612 self._config = {} def __getitem__(self, item): return self._config.get(item) @staticmethod def _user_temp_non_session(): """Returns the non session specific %temp""" val = os.path.expandvars('$TEMP') if val == '$TEMP': return '' # if path ends with a digit, chop it off val = os.path.normpath(val) head, tail = os.path.split(val) if tail.isdigit() and len(tail) <= 5: # max session-id 65536 return head return val def log_path(self): """Returns the log path""" if self._args.log: return self._args.log return os.path.join(self._user_temp_non_session(), LOG_FILE) def result_path(self): """Return the result output json path""" if self._args.result: return self._args.result return os.path.join(self._user_temp_non_session(), RESULT_FILE) def zip_method(self): """Returns the zip.method value from config file""" try: method = self._config['zip']['method'] return str(method) except (TypeError, KeyError) as ex: # pylint: disable=W0612 # logging.info('Failed to get zip.method from config: %s', ex) return "1" def zip_external_exe_name(self): """Return the name of the exe""" try: exe = self._config['zip']['external_tool']['exe'] return os.path.basename(exe) except (TypeError, KeyError) as ex: # pylint: disable=W0612 # logging.info('Failed to get zip.external_tools from config: %s', ex) return def zip_external_cmdline(self): """Returns the external tool command line""" try: exe = self._config['zip']['external_tool']['exe'] cmd = self._config['zip']['external_tool']['cmd_line'] return exe + ' ' + cmd except (TypeError, KeyError) as ex: # pylint: disable=W0612 # logging.info('Failed to get zip.external_tools from config: %s', ex) return def read_json_file(filename): """Read JSON file and return content as python object. @param filename: Path to json file @type filename: str @return: Python dictionary represent JSON or None @rtype: dict """ try: with open(filename) as data_file: data = json.load(data_file) return data except EnvironmentError as ex: # Not exist is not necessarily an error logging.info('%s: %s', 'read_json_file', ex) except ValueError as ex: # File contains invalid JSON data logging.warning('%s: %s', 'read_json_file', ex) except Exception as ex: # pylint: disable=broad-except logging.error('%s: %s', 'read_json_file', ex) def write_json_file(filename, jsondata): """Write python object as JSON to file. @param filename: file to write @type filename: str @param jsondata: python dictionary to write to json @type jsondata: dict @return: True if write is successful, None otherwise """ try: with open(filename, 'w') as outfile: json.dump(jsondata, outfile, indent=2) outfile.write('\n') return True except EnvironmentError as ex: logging.error('%s: %s', 'write_json_file', ex) except Exception as ex: # pylint: disable=broad-except logging.error('%s: %s', 'write_json_file', ex) def my_print(msg): """ Print with ISO timestamp @param msg: message to print @type msg: str """ timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") print("[%s] %s" % (timestamp, msg)) def parse_args(): """ Parse command line arguments @return namespace from ArgumentParser.parse_args """ # create the parser object parser = argparse.ArgumentParser() group1 = parser.add_argument_group('commands') group1.add_argument('--zip', default=None, metavar=('zip_file', 'source_dir'), help='create zip file from directory', nargs=2) group1.add_argument('--unzip', default=None, metavar=('zip_file', 'target_dir'), help='extract zip file to directory', nargs=2) group1.add_argument('--test', metavar='zip_file', help='test integrity of zip file') group1.add_argument('--compare', default=None, metavar=('zip_file', 'source_dir'), help='compare zip file against directory', nargs=2) group2 = parser.add_argument_group('options') # remaining parameters group2.add_argument("--result", metavar='file', default=None, help="result json path") group2.add_argument("--log", metavar='file', default=None, help="log file path") group2.add_argument("--config", metavar='file', default=None, help="config file path") # parse the arguments args = parser.parse_args() # print(args) # return # any work to do? if args.zip or args.unzip or args.test or args.compare: return args else: parser.print_help() def setup_logging(config): """ Setup logging for the installer @param config: config object @type config: ConfigFile """ # initialize logging logging.basicConfig(format=LOG_FORMAT, level=logging.DEBUG, filename=config.log_path()) # write start header so we have clear separation between runs logging.info('***** %s %s *****', exe_name, sys.argv[1:]) def check_os(): """Check and set OS vars""" global is_win, is_mac, is_lin # pylint: disable=invalid-name,W0603 # detect platform if sys.platform == 'win32': is_win = True elif sys.platform == 'darwin': is_mac = True elif 'linux' in sys.platform: is_lin = True else: print('Unsupported platform: {}'.format(sys.platform)) return 50 # ERROR_NOT_SUPPORTED def main(): """Main entry point""" # parse argument before doing anything else args = parse_args() if not args: return check_os() # set os flags config = ConfigFile(args) # create the config object setup_logging(config) # initialize logs result = ResultOutput(config) # result json cmd = Command(config, result) # command object to do work # do work here retval = 0 if args.zip: # create zip archive method = config.zip_method() if method != "3": retval = cmd.run_zip_create(*args.zip) else: # retval = cmd.run_zip_create_tool(*args.zip) pass elif args.unzip: # extract zip archive retval = cmd.run_zip_extract(*args.unzip) elif args.test: # test zip integrity retval = cmd.run_zip_test(args.test) elif args.compare: # compare zip against folder retval = cmd.run_zip_compare(*args.compare) if retval is None: retval = 0 result.write() logging.info('***** %s exit [%d]', exe_name, retval) return retval if __name__ == '__main__': sys.exit(main())