【Python小工具】文件解压

文件解压

# -*- coding: utf-8 -*-

import os
import sys
import time
import shutil
import filetype
import gzip
import tarfile
import zipfile


class FileUtils:
    """一些文件处理相关的功能"""

    @classmethod
    def list_paths(cls, dir_path: str, depth: int = 0, suffix=None, key_str: str = None):
        """
        1) Generator。
        2) 遍历 dir_path 目录下的文件的路径。
        3) 注意：这里的路径使用'/'。
        :param dir_path:    str     要遍历的目录路径
        :param depth:       int     扫描的深度 0:当前目录，1：当前目录的下一级目录
        :param suffix:      str     返回的路径中包含特定后缀，如 ".py" 或者 "py"，默认None，没有后缀限制
        :param key_str:     str     返回的路径中包含特定的关键词
        """

        # 设定当前目录的表示值
        current_dir_level = 0

        if suffix:
            if not suffix.startswith('.'):
                suffix = '.' + suffix

        for _path in os.listdir(dir_path):
            tmp_path = os.path.join(dir_path, _path)

            if os.path.isdir(tmp_path):
                if current_dir_level < depth:
                    yield from cls.list_paths(tmp_path, depth - 1, suffix, key_str)

            else:
                found = []
                if suffix:
                    if os.path.splitext(tmp_path)[-1] == suffix:
                        found.append(True)
                    else:
                        found.append(False)

                if key_str:
                    if key_str in tmp_path:
                        found.append(True)
                    else:
                        found.append(False)

                if all(found):
                    yield tmp_path

    @classmethod
    def copy_files(cls, src_dir: str, dst_dir: str, depth=0, suffix=None, key_str: str = None):
        """
        复制指定目录 src_dir下的文件到 dst_dir目录
            depth：  src_dir递归查找深度，0: 当前目录，1:子目录，...
            suffix: 搜索指定后缀的文件,None,表示所有文件
            key_str: 搜索路径中含特定字符的路径，None,表示没有特定字符
        """
        for file in cls.list_paths(src_dir, depth=depth, suffix=suffix, key_str=key_str):
            # 复制文件到目录
            shutil.copy(file, dst_dir)

    @classmethod
    def move_files(cls, src_dir: str, dst_dir: str, depth=0, suffix=None):
        """
        移动指定目录 src_dir下的文件到 dst_dir目录
            depth：  src_dir递归查找深度，0: 当前目录，1:子目录，...
                    注意：所有文件都在 dst_dir下同一级
            suffix: 搜索指定后缀的文件,None,表示所有文件
        """
        for file in cls.list_paths(src_dir, depth, suffix):
            # 复制文件到目录
            shutil.move(file, dst_dir)

    @classmethod
    def remove_files(cls, src_dir: str, depth=0, suffix=None):
        """
        删除指定目录 src_dir下的所有文件
            depth：  src_dir递归查找深度，0: 当前目录，1:子目录，...
                    注意：所有文件都在 dst_dir下同一级
            suffix: 搜索指定后缀的文件,None,表示所有文件
        """
        for file in cls.list_paths(src_dir, depth, suffix):
            # 复制文件到目录
            os.remove(file)

    @staticmethod
    def clean_expiry_files_dirs(dir_path, expiry_days: int or float):
        """
            根据创建时间，清理 dir_path下超期的文件和文件夹。expiry_days默认为None，不清理
        """
        if expiry_days < 0:
            return

        current_date_seconds = time.time()
        expiry_seconds = expiry_days * 24 * 60 ** 2

        for i in os.listdir(dir_path):
            path_ = os.path.join(dir_path, i)
            create_date_seconds = os.path.getctime(path_)

            if current_date_seconds - create_date_seconds > expiry_seconds:
                shutil.rmtree(path_)
                print(f"path: {path_}, deleted.")

    @staticmethod
    def extract_zip(zip_path: str, dst_dir: str):
        """
        解压缩zip文件 zip_path下的文件到dst_dir目录
            zip_path:  zip文件
            dst_dir: 存放解压后文件路径
        """
        zip_file = zipfile.ZipFile(zip_path)
        for name in zip_file.namelist():
            try:
                new_name = name.encode('cp437').decode('gbk')
                new_name = os.path.join(dst_dir, new_name)
                if os.path.isdir(new_name) and not os.path.exists(new_name):
                    os.makedirs(new_name)
                with open(new_name, 'wb') as file:
                    file.write(zip_file.read(name))
            except:
                zip_file.extract(name, dst_dir)
        zip_file.close()

    @staticmethod
    def extract_rar(rar_path: str, dst_dir: str):
        """
       解压缩rar文件 rar_path下的文件到dst_dir目录
           rar_path:  rar文件
           dst_dir: 存放解压后文件路径
       """
        try:
            from unrar import rarfile
        except LookupError:
            print("[Error]Please install unrar library")
            sys.exit(1)
        rar_file = rarfile.RarFile(rar_path)
        rar_file.extractall(dst_dir)

    @staticmethod
    def extract_tar(tar_path: str, dst_dir: str):
        """
       解压缩tar文件 tar_path下的文件到dst_dir目录，支持'tar.gz', 'tar.bz2', 'tar.xz' 格式的TAR文件
           tar_path:  tar文件
           dst_dir: 存放解压后文件路径
       """
        if filetype.guess(tar_path).extension in ['gz', 'bz2', 'xz']:
            mode = "r:{}".format(filetype.guess(tar_path).extension)
            print(mode)
        else:
            raise ValueError("暂不支持解压此格式的文件。")
        tar_file = tarfile.open(tar_path, mode)
        tar_file.extractall(dst_dir)
        tar_file.close()

    @staticmethod
    def extract_gz(gz_path: str, dst_dir: str):
        """
       解压缩gz文件 gz_path下的文件到dst_dir目录
           gz_path:  gz文件
           dst_dir: 存放解压后文件路径
       """
        # 获取文件的名称，去掉后缀名
        f_name = gz_path.split('/')[-1].replace(".gz", "")
        # 开始解压
        g_file = gzip.GzipFile(gz_path)
        # 读取解压后的文件，并写入去掉后缀名的同名文件（即得到解压后的文件）
        with open(os.path.join(dst_dir, f_name), "wb+") as f:
            f.write(g_file.read())
        g_file.close()


if __name__ == '__main__':
    tar_path = "ARP01.fp.2012010118.f5.srm.gz"
    dst_dir = ""
    FileUtils.extract_gz(tar_path, dst_dir)