Class Zip

client/util/zipmgr.py:30–165

class Zip(object):
    def __init__(self):
        """
        检查工具的存在与是否可用
        """
        # logger.info("初始化zip模块中...")
        if not os.environ.get(ZIP_HOME, False):
            raise NodeConfigError("the node does not have ZIP tool, please check.")
        self.ZIP_TOOL_NAME = "7z"
        if platform.platform().find("debian") >= 0:
            logger.info("platform: %s" % platform.platform())
            logger.info(
                "检测到在ubuntu系统中调用zip，因此将调用7z_ubuntu。（7z_ubuntu可能存在兼容问题，如果异常，请自行安装:sudo apt-get install p7zip-full）"
            )
            self.ZIP_TOOL_NAME = "7z_ubuntu"

    def compress(self, path, zip_file):
        """
        基于7z实现的压缩，默认使用16线程极速压缩
        :param path: 被压缩的资源路径
        :param zip_file: 压缩完成生成的文件
        :return:
        """
        logger.info("zip模块执行压缩操作...")
        args = [
            os.path.join(os.environ[ZIP_HOME], self.ZIP_TOOL_NAME),
            "a",
            "-t7z",
            zip_file,
            "%s/*" % path,
            "-mmt16",
            "-mx1",
        ]
        if os.path.exists(zip_file):
            PathMgr().rmpath(zip_file)
        try:
            args = PathMgr().format_cmd_arg_list(args)
            process = SubProcController(
                args,
                print_enable=False,
                shell=False,
                stdout_filepath=None,
                stderr_filepath=None,
                stdout_line_callback=self.compress_subprocc_log,
                stderr_line_callback=self.compress_subprocc_log,
                env=EnvSet().get_origin_env(),
            )
            process.wait(ZIP_TIME_LIMIT)
        except FileNotFoundError:
            self.zipfile_compress(path, zip_file)
        if not os.path.exists(zip_file):
            err_msg = "compress error! zip file is empty, please check! 如果是ubuntu环境，7z_ubuntu可能存在兼容问题，" \
                      "请自行安装:sudo apt-get install p7zip-full"
            logger.error(err_msg)
            raise ZIPError(err_msg)
        return True

    def decompress_by_7z(self, zip_file, path, print_enable=False):