AskaEth
e6875f0b4b
- 13-service async plugin framework - Textual TUI with CLI fallback - Plugin hot-reload + permission system - Web management panel (aiohttp) - Bridge-based inter-module communication - 10 regression tests Fixes applied: - PBKDF2-SHA256 auth (was plain SHA256) - Auth bypass removed (was allow-all on fail) - Bare excepts replaced with logged errors - CatFramework/DreamSu -> SenSu naming unified - ServiceManager: health checks + startup_order - Env var credentials (SENSU_ADMIN_PASSWORD etc)
145 lines
5.2 KiB
Python
145 lines
5.2 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import logging
|
|
import os
|
|
import shutil
|
|
from pathlib import Path
|
|
from typing import List, Optional
|
|
import hashlib
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class FileUtils:
|
|
"""文件操作工具类"""
|
|
|
|
@staticmethod
|
|
def ensure_directory(directory_path: str) -> bool:
|
|
"""确保目录存在"""
|
|
try:
|
|
path = Path(directory_path)
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
logger.debug(f"目录已确保存在: {directory_path}")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"创建目录时出错 {directory_path}: {str(e)}", exc_info=True)
|
|
return False
|
|
|
|
@staticmethod
|
|
def safe_write(file_path: str, content: str, backup: bool = True) -> bool:
|
|
"""安全写入文件（支持备份）"""
|
|
try:
|
|
path = Path(file_path)
|
|
|
|
# 备份原文件
|
|
if backup and path.exists():
|
|
backup_path = path.with_suffix(path.suffix + '.bak')
|
|
shutil.copy2(path, backup_path)
|
|
logger.debug(f"文件已备份: {backup_path}")
|
|
|
|
# 写入新内容
|
|
with open(path, 'w', encoding='utf-8') as f:
|
|
f.write(content)
|
|
|
|
logger.debug(f"文件写入成功: {file_path}, 大小: {len(content)} 字节")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"写入文件时出错 {file_path}: {str(e)}", exc_info=True)
|
|
return False
|
|
|
|
@staticmethod
|
|
def safe_read(file_path: str, default: str = "") -> str:
|
|
"""安全读取文件"""
|
|
try:
|
|
path = Path(file_path)
|
|
if not path.exists():
|
|
logger.warning(f"文件不存在: {file_path}")
|
|
return default
|
|
|
|
with open(path, 'r', encoding='utf-8') as f:
|
|
content = f.read()
|
|
|
|
logger.debug(f"文件读取成功: {file_path}, 大小: {len(content)} 字节")
|
|
return content
|
|
|
|
except Exception as e:
|
|
logger.error(f"读取文件时出错 {file_path}: {str(e)}", exc_info=True)
|
|
return default
|
|
|
|
@staticmethod
|
|
def list_files(directory: str, pattern: str = "*", recursive: bool = False) -> List[Path]:
|
|
"""列出目录中的文件"""
|
|
try:
|
|
path = Path(directory)
|
|
if not path.exists():
|
|
logger.warning(f"目录不存在: {directory}")
|
|
return []
|
|
|
|
if recursive:
|
|
files = list(path.rglob(pattern))
|
|
else:
|
|
files = list(path.glob(pattern))
|
|
|
|
# 过滤出文件（非目录）
|
|
files = [f for f in files if f.is_file()]
|
|
|
|
logger.debug(f"列出文件: {directory}, 模式: {pattern}, 找到 {len(files)} 个文件")
|
|
return files
|
|
|
|
except Exception as e:
|
|
logger.error(f"列出文件时出错 {directory}: {str(e)}", exc_info=True)
|
|
return []
|
|
|
|
@staticmethod
|
|
def calculate_file_hash(file_path: str, algorithm: str = "md5") -> Optional[str]:
|
|
"""计算文件哈希值"""
|
|
try:
|
|
path = Path(file_path)
|
|
if not path.exists():
|
|
logger.warning(f"文件不存在: {file_path}")
|
|
return None
|
|
|
|
hash_func = getattr(hashlib, algorithm)()
|
|
|
|
with open(path, 'rb') as f:
|
|
for chunk in iter(lambda: f.read(4096), b""):
|
|
hash_func.update(chunk)
|
|
|
|
file_hash = hash_func.hexdigest()
|
|
logger.debug(f"文件哈希计算完成: {file_path} -> {algorithm}:{file_hash}")
|
|
return file_hash
|
|
|
|
except Exception as e:
|
|
logger.error(f"计算文件哈希时出错 {file_path}: {str(e)}", exc_info=True)
|
|
return None
|
|
|
|
@staticmethod
|
|
def cleanup_old_files(directory: str, pattern: str, keep_count: int) -> int:
|
|
"""清理旧文件，保留指定数量的最新文件"""
|
|
try:
|
|
files = FileUtils.list_files(directory, pattern)
|
|
if len(files) <= keep_count:
|
|
logger.debug(f"文件数量未超过限制，无需清理: {directory}")
|
|
return 0
|
|
|
|
# 按修改时间排序
|
|
files.sort(key=lambda x: x.stat().st_mtime, reverse=True)
|
|
|
|
# 删除旧文件
|
|
removed_count = 0
|
|
for file_to_remove in files[keep_count:]:
|
|
try:
|
|
file_to_remove.unlink()
|
|
removed_count += 1
|
|
logger.debug(f"删除旧文件: {file_to_remove}")
|
|
except Exception as e:
|
|
logger.error(f"删除文件时出错 {file_to_remove}: {str(e)}", exc_info=True)
|
|
|
|
logger.info(f"文件清理完成: {directory}, 删除 {removed_count} 个文件")
|
|
return removed_count
|
|
|
|
except Exception as e:
|
|
logger.error(f"清理旧文件时出错 {directory}: {str(e)}", exc_info=True)
|
|
return 0
|