qinglong/file_util.py

import json
import os


class FileUtil:
    """
    文件操作工具类。
    """

    @staticmethod
    def list_files(folder_path, recursive=False):
        """
        列出指定文件夹下的所有文件。

        :param folder_path: 文件夹路径
        :param recursive: 是否递归遍历子文件夹，默认为 False
        :return: 文件路径列表
        :raises FileNotFoundError: 如果文件夹不存在
        :raises PermissionError: 如果无权限访问文件夹
        """
        if not os.path.exists(folder_path):
            raise FileNotFoundError(f"文件夹未找到: {folder_path}")
        if not os.path.isdir(folder_path):
            raise ValueError(f"路径不是文件夹: {folder_path}")

        files = []
        try:
            if recursive:
                for root, _, filenames in os.walk(folder_path):
                    for filename in filenames:
                        files.append(os.path.join(root, filename))
            else:
                for entry in os.listdir(folder_path):
                    full_path = os.path.join(folder_path, entry)
                    if os.path.isfile(full_path):
                        files.append(full_path)
        except PermissionError as e:
            raise PermissionError(f"无权限访问文件夹: {e}")

        return files

    @staticmethod
    def read_file(file_path):
        """
        从指定文件中读取内容。

        :param file_path: 文件路径
        :return: 文件内容（字符串）
        :raises FileNotFoundError: 如果文件不存在
        :raises IOError: 如果读取文件时发生错误
        """
        try:
            with open(file_path, 'r', encoding='utf-8') as file:
                return file.read()
        except FileNotFoundError:
            raise FileNotFoundError(f"文件未找到: {file_path}")
        except IOError as e:
            raise IOError(f"读取文件时发生错误: {e}")

    @staticmethod
    def write_file(file_path, content):
        """
        将内容写入到指定文件中。

        :param file_path: 文件路径
        :param content: 要写入的内容（字符串）
        :raises IOError: 如果写入文件时发生错误
        """
        # 确保文件夹存在
        directory = os.path.dirname(file_path)
        if directory and not os.path.exists(directory):
            os.makedirs(directory)
        try:
            with open(file_path, 'w', encoding='utf-8') as file:
                file.write(content)
        except IOError as e:
            raise IOError(f"写入文件时发生错误: {e}")

    @staticmethod
    def read_json(file_path):
        """
        从指定文件中读取内容并解析为 JSON。

        :param file_path: 文件路径
        :return: 解析后的 JSON 对象（字典或列表）
        :raises FileNotFoundError: 如果文件不存在
        :raises IOError: 如果读取文件时发生错误
        :raises json.JSONDecodeError: 如果 JSON 解析失败
        """
        # 检查文件是否存在且非空
        if not os.path.exists(file_path):
            raise FileNotFoundError(f"文件未找到: {file_path}")
        if os.path.getsize(file_path) == 0:
            raise ValueError(f"文件内容为空: {file_path}")
        try:
            with open(file_path, 'r', encoding='utf-8') as file:
                return json.load(file)
        except FileNotFoundError:
            raise FileNotFoundError(f"文件未找到: {file_path}")
        except IOError as e:
            raise IOError(f"读取文件时发生错误: {e}")
        except json.JSONDecodeError as e:
            raise json.JSONDecodeError(f"JSON 解析失败: {e.msg}", e.doc, e.pos)

    @staticmethod
    def write_json(file_path, data):
        """
        将 JSON 数据写入到指定文件中。

        :param file_path: 文件路径
        :param data: 要写入的 JSON 数据（字典、列表等）
        :raises IOError: 如果写入文件时发生错误
        :raises TypeError: 如果数据无法序列化为 JSON
        """
        # 确保文件夹存在
        directory = os.path.dirname(file_path)
        if directory and not os.path.exists(directory):
            os.makedirs(directory)
        try:
            with open(file_path, 'w', encoding='utf-8') as file:
                json.dump(data, file, ensure_ascii=False, indent=4)
        except IOError as e:
            raise IOError(f"写入文件时发生错误: {e}")
        except TypeError as e:
            raise TypeError(f"数据无法序列化为 JSON: {e}")