123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130 |
- """
- 通用工具函数
- Common utilities for Cd Prediction Integrated System
- """
- import os
- import shutil
- import logging
- from typing import List, Union
- def setup_logging(log_file: str, level: int = logging.INFO) -> None:
- """
- 设置日志配置
-
- @param log_file: 日志文件路径
- @param level: 日志级别
- """
- # 确保日志目录存在
- os.makedirs(os.path.dirname(log_file), exist_ok=True)
-
- # 配置日志格式
- formatter = logging.Formatter(
- '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
- )
-
- # 文件处理器
- file_handler = logging.FileHandler(log_file, encoding='utf-8')
- file_handler.setFormatter(formatter)
-
- # 控制台处理器
- console_handler = logging.StreamHandler()
- console_handler.setFormatter(formatter)
-
- # 根日志配置
- logging.basicConfig(
- level=level,
- handlers=[file_handler, console_handler],
- format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
- )
- def validate_files(file_paths: List[str]) -> bool:
- """
- 验证文件是否存在
-
- @param file_paths: 文件路径列表
- @return: 所有文件都存在返回True,否则返回False
- """
- missing_files = []
- for file_path in file_paths:
- if not os.path.exists(file_path):
- missing_files.append(file_path)
-
- if missing_files:
- logging.error(f"以下文件不存在: {missing_files}")
- return False
-
- return True
- def copy_file(src: str, dst: str) -> bool:
- """
- 复制文件
-
- @param src: 源文件路径
- @param dst: 目标文件路径
- @return: 复制成功返回True,否则返回False
- """
- try:
- # 确保目标目录存在
- os.makedirs(os.path.dirname(dst), exist_ok=True)
- shutil.copy2(src, dst)
- logging.info(f"文件复制成功: {src} -> {dst}")
- return True
- except Exception as e:
- logging.error(f"文件复制失败: {src} -> {dst}, 错误: {str(e)}")
- return False
- def ensure_directory(directory: str) -> None:
- """
- 确保目录存在,如果不存在则创建
-
- @param directory: 目录路径
- """
- if not os.path.exists(directory):
- os.makedirs(directory, exist_ok=True)
- logging.info(f"创建目录: {directory}")
- def get_file_size(file_path: str) -> Union[int, None]:
- """
- 获取文件大小(字节)
-
- @param file_path: 文件路径
- @return: 文件大小,如果文件不存在返回None
- """
- try:
- return os.path.getsize(file_path)
- except OSError:
- return None
- def format_file_size(size_bytes: int) -> str:
- """
- 格式化文件大小显示
-
- @param size_bytes: 文件大小(字节)
- @return: 格式化的文件大小字符串
- """
- if size_bytes == 0:
- return "0B"
-
- size_names = ["B", "KB", "MB", "GB", "TB"]
- import math
- i = int(math.floor(math.log(size_bytes, 1024)))
- p = math.pow(1024, i)
- s = round(size_bytes / p, 2)
- return f"{s} {size_names[i]}"
- def safe_remove_file(file_path: str) -> bool:
- """
- 安全删除文件
-
- @param file_path: 文件路径
- @return: 删除成功返回True,否则返回False
- """
- try:
- if os.path.exists(file_path):
- os.remove(file_path)
- logging.info(f"文件删除成功: {file_path}")
- return True
- except Exception as e:
- logging.error(f"文件删除失败: {file_path}, 错误: {str(e)}")
- return False
|