common.py 3.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130
  1. """
  2. 通用工具函数
  3. Common utilities for Cd Prediction Integrated System
  4. """
  5. import os
  6. import shutil
  7. import logging
  8. from typing import List, Union
  9. def setup_logging(log_file: str, level: int = logging.INFO) -> None:
  10. """
  11. 设置日志配置
  12. @param log_file: 日志文件路径
  13. @param level: 日志级别
  14. """
  15. # 确保日志目录存在
  16. os.makedirs(os.path.dirname(log_file), exist_ok=True)
  17. # 配置日志格式
  18. formatter = logging.Formatter(
  19. '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
  20. )
  21. # 文件处理器
  22. file_handler = logging.FileHandler(log_file, encoding='utf-8')
  23. file_handler.setFormatter(formatter)
  24. # 控制台处理器
  25. console_handler = logging.StreamHandler()
  26. console_handler.setFormatter(formatter)
  27. # 根日志配置
  28. logging.basicConfig(
  29. level=level,
  30. handlers=[file_handler, console_handler],
  31. format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
  32. )
  33. def validate_files(file_paths: List[str]) -> bool:
  34. """
  35. 验证文件是否存在
  36. @param file_paths: 文件路径列表
  37. @return: 所有文件都存在返回True,否则返回False
  38. """
  39. missing_files = []
  40. for file_path in file_paths:
  41. if not os.path.exists(file_path):
  42. missing_files.append(file_path)
  43. if missing_files:
  44. logging.error(f"以下文件不存在: {missing_files}")
  45. return False
  46. return True
  47. def copy_file(src: str, dst: str) -> bool:
  48. """
  49. 复制文件
  50. @param src: 源文件路径
  51. @param dst: 目标文件路径
  52. @return: 复制成功返回True,否则返回False
  53. """
  54. try:
  55. # 确保目标目录存在
  56. os.makedirs(os.path.dirname(dst), exist_ok=True)
  57. shutil.copy2(src, dst)
  58. logging.info(f"文件复制成功: {src} -> {dst}")
  59. return True
  60. except Exception as e:
  61. logging.error(f"文件复制失败: {src} -> {dst}, 错误: {str(e)}")
  62. return False
  63. def ensure_directory(directory: str) -> None:
  64. """
  65. 确保目录存在,如果不存在则创建
  66. @param directory: 目录路径
  67. """
  68. if not os.path.exists(directory):
  69. os.makedirs(directory, exist_ok=True)
  70. logging.info(f"创建目录: {directory}")
  71. def get_file_size(file_path: str) -> Union[int, None]:
  72. """
  73. 获取文件大小(字节)
  74. @param file_path: 文件路径
  75. @return: 文件大小,如果文件不存在返回None
  76. """
  77. try:
  78. return os.path.getsize(file_path)
  79. except OSError:
  80. return None
  81. def format_file_size(size_bytes: int) -> str:
  82. """
  83. 格式化文件大小显示
  84. @param size_bytes: 文件大小(字节)
  85. @return: 格式化的文件大小字符串
  86. """
  87. if size_bytes == 0:
  88. return "0B"
  89. size_names = ["B", "KB", "MB", "GB", "TB"]
  90. import math
  91. i = int(math.floor(math.log(size_bytes, 1024)))
  92. p = math.pow(1024, i)
  93. s = round(size_bytes / p, 2)
  94. return f"{s} {size_names[i]}"
  95. def safe_remove_file(file_path: str) -> bool:
  96. """
  97. 安全删除文件
  98. @param file_path: 文件路径
  99. @return: 删除成功返回True,否则返回False
  100. """
  101. try:
  102. if os.path.exists(file_path):
  103. os.remove(file_path)
  104. logging.info(f"文件删除成功: {file_path}")
  105. return True
  106. except Exception as e:
  107. logging.error(f"文件删除失败: {file_path}, 错误: {str(e)}")
  108. return False