poyee-data-warehouse/dw_base/utils/file_utils.py

import json
import os
import sys
import re
from os import path
from typing import Dict, List

abspath = os.path.abspath(__file__)
root_path = re.sub(r"poyee-data-warehouse.*", "poyee-data-warehouse", abspath)
sys.path.append(root_path)
from dw_base import NORM_GRN, NORM_RED
from dw_base import PROJECT_ROOT_PATH
from dw_base.utils.log_utils import pretty_print


def append_file(content: str, file_path: str):
    """
    将字符串写入文件（追加写入）
    Args:
        content: 字符串内容
        file_path: 文件路径
    Returns:
    """
    with open(file_path, "a", encoding='utf-8') as file:
        file.write(content)


def delete_file(file_path: str):
    """
    删除文件
    Args:
        file_path: 文件路径
    Returns:
    """
    os.system(f'rm -f {file_path}')


def get_abs_path(fs_path: str, check_exist: bool = True) -> str:
    """
        获取文件的绝对路径，注意传入的路径必须是``项目``目录下的相对路径或绝对路径（多此一举）
    Args:
        fs_path: 文件的路径
        check_exist: 是否对文件存在情况进行判定, 一般情况下, 读文件为True, 写文件为False
    Returns:
        文件的绝对路径
    """
    if fs_path.startswith(PROJECT_ROOT_PATH):
        if not check_exist or os.path.exists(fs_path):
            return fs_path
        raise FileNotFoundError('绝对路径 %s 不存在' % fs_path)
    abs_path = None
    for sub_dir in os.listdir(PROJECT_ROOT_PATH):
        if fs_path.startswith(sub_dir):
            abs_path = os.path.join(PROJECT_ROOT_PATH, fs_path)
            break
    if abs_path is None:
        raise FileNotFoundError('相对路径 `%s` 不存在' % fs_path)
    if not check_exist or os.path.exists(abs_path):
        return abs_path
    raise FileNotFoundError('绝对路径 %s 不存在' % abs_path)


def list_files(fs_dir: str, recursive: bool = False, extension: str = None) -> List[str]:
    """
    列出给定目录下所有文件
    Args:
        fs_dir: 目录（绝对路径或项目下的相对路径）
        recursive: 是否递归
        extension: 文件扩展名
    Returns:
        所有子文件（绝对路径）
    """
    fs_dir = get_abs_path(fs_dir, check_exist=True)
    files_with_abspath = []
    for child in os.listdir(fs_dir):
        abs_child = f'{fs_dir}/{child}'
        if extension is not None and not abs_child.endswith(extension):
            continue
        if os.path.isfile(abs_child):
            files_with_abspath.append(abs_child)
        elif recursive:
            for sub_child in list_files(abs_child, recursive):
                files_with_abspath.append(path.abspath(sub_child))
    return files_with_abspath


def load_json_file(file_path: str) -> Dict[str, any]:
    """
    读取文件，返回JSON
    Args:
        file_path: 文件路径
    Returns:
    """
    try:
        with open(get_abs_path(file_path), 'r') as file:
            return json.load(file)
    except Exception as e:
        pretty_print(f'{NORM_RED}加载json文件 {NORM_GRN}{file_path}{NORM_RED} 失败：{str(e)}')
        raise e


def read_file_content(file_path: str) -> str:
    """
    读取文件内容，作为一个字符串返回
    Args:
        file_path: 文件路径
    Returns: 文件内容
    """
    with open(get_abs_path(file_path), 'r') as file:
        return file.read()


def read_file_lines(file_path: str) -> List[str]:
    """
    读取文件内容，作为多行字符串返回（每行依然保留有换行符）
    Args:
        file_path: 文件路径
    Returns:
    """
    with open(get_abs_path(file_path), 'r', encoding='utf-8') as file:
        return file.readlines()


def write_file(content: str, file_path: str):
    """
    将字符串写入文件（覆盖写入）
    Args:
        content: 字符串内容
        file_path: 文件路径
    Returns:
    """
    with open(file_path, "w", encoding='utf-8') as file:
        file.write(content)


if __name__ == '__main__':
    ol = list_files('launch-pad/contacts/00new_contact/03fix/04mid_pre', False)
    for e in ol:
        print(e)
    print('-----------------------------------------------------')
    for e in sorted(ol):
        print(e)