data_factory/utils/path_utils.py

135 lines
5.5 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2025/3/15 15:26
# @Author : AngesZhu
# @File : path_utils.py
# @Desc : 路径相关操作封装
import os
from pathlib import Path
class PathOperator:
"""
封装路径相关操作的工具类。
"""
def __init__(self, file_path: str = None):
"""
初始化 PathOperator 实例。
:param file_path: 当前文件的路径(可选)。如果未提供,则默认为当前工作目录。
"""
try:
self.file_path = Path(file_path) if file_path else Path.cwd()
except Exception as e:
raise ValueError(f"初始化路径时发生错误: {e}")
def get_full_path(self) -> str:
"""
获取当前文件的完整路径。
:return: 完整路径字符串。
:raises RuntimeError: 如果获取路径失败。
"""
try:
return str(self.file_path.resolve())
except Exception as e:
raise RuntimeError(f"获取完整路径时发生错误: {e}")
def join_path(self, *paths: str) -> str:
"""
拼接路径并返回完整路径。
:param paths: 要拼接的路径片段。
:return: 拼接后的完整路径字符串。
:raises ValueError: 如果路径拼接失败。
"""
try:
if not all(isinstance(p, str) for p in paths):
raise ValueError("所有路径片段必须是字符串")
return str(self.file_path.joinpath(*paths).resolve())
except Exception as e:
raise ValueError(f"路径拼接时发生错误: {e}")
def create_directory(self, directory_path: str) -> str:
"""
在指定路径下创建文件夹。如果文件夹已存在,则不会重复创建。
:param directory_path: 要创建的文件夹路径。
:return: 创建的文件夹路径。
:raises PermissionError: 如果没有权限创建文件夹。
:raises ValueError: 如果路径无效。
"""
try:
if not isinstance(directory_path, str):
raise ValueError("directory_path 必须是一个字符串")
path = Path(directory_path)
path.mkdir(parents=True, exist_ok=True)
return str(path.resolve())
except PermissionError:
raise PermissionError(f"无法创建文件夹 '{directory_path}',请检查权限")
except Exception as e:
raise RuntimeError(f"创建文件夹时发生错误: {e}")
def is_file_exists(self, file_path: str) -> bool:
"""
判断文件是否存在。
:param file_path: 文件路径。
:return: 如果文件存在则返回 True否则返回 False。
:raises ValueError: 如果路径无效。
"""
try:
if not isinstance(file_path, str):
raise ValueError("file_path 必须是一个字符串")
return Path(file_path).is_file()
except Exception as e:
raise RuntimeError(f"判断文件是否存在时发生错误: {e}")
def get_parent_path(self, path: str = None) -> str:
"""
获取指定路径的父级路径。
:param path: 指定路径(可选)。如果未提供,则使用初始化时的路径。
:return: 父级路径字符串。
:raises ValueError: 如果路径无效。
"""
try:
target_path = Path(path) if path else self.file_path
if not isinstance(path, str) and path is not None:
raise ValueError("path 必须是一个字符串或 None")
return str(target_path.parent.resolve())
except Exception as e:
raise RuntimeError(f"获取父级路径时发生错误: {e}")
def get_all_files(self, directory_path: str = None) -> list:
"""
获取当前路径或指定路径下的所有文件名称。
:param directory_path: 指定路径(可选)。如果未提供,则使用初始化时的路径。
:return: 文件名称列表。
:raises NotADirectoryError: 如果路径不是有效目录。
"""
try:
target_path = Path(directory_path) if directory_path else self.file_path
if not target_path.is_dir():
raise NotADirectoryError(f"路径 '{target_path}' 不是有效的目录")
return [file.name for file in target_path.iterdir() if file.is_file()]
except Exception as e:
raise RuntimeError(f"获取文件列表时发生错误: {e}")
def get_all_paths_recursive(self, directory_path: str = None) -> dict:
"""
获取指定路径下的所有层级以及文件路径。
:param directory_path: 指定路径(可选)。如果未提供,则使用初始化时的路径。
:return: 包含所有文件和文件夹路径的列表。
:raises NotADirectoryError: 如果路径不是有效目录。
"""
try:
target_path = Path(directory_path) if directory_path else self.file_path
if not target_path.is_dir():
raise NotADirectoryError(f"路径 '{target_path}' 不是有效的目录")
all_levels = {}
if os.path.isdir(directory_path):
for root, dirs, files in os.walk(directory_path):
all_levels[Path(root).name] = {"path": root, "files": files}
return all_levels
except Exception as e:
raise RuntimeError(f"递归获取路径时发生错误: {e}")