Python脚本如何归档冷门缓存至低速存储

wen python案例 2026-06-26 235

本文目录导读：

核心脚本实现
配置文件示例
使用说明
最佳实践

我来为您提供一个归档冷门缓存到低速存储的Python脚本方案：

核心脚本实现

#!/usr/bin/env python3
"""
冷门缓存归档工具 - 将使用频率低的缓存文件迁移到低速存储
"""
import os
import shutil
import time
import json
import logging
from pathlib import Path
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import hashlib
# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('cache_archiver.log'),
        logging.StreamHandler()
    ]
)
class CacheArchiver:
    def __init__(self, config: Dict):
        """
        初始化归档器
        Args:
            config: 配置字典，包含以下键：
                - fast_cache_path: 高速缓存路径
                - slow_cache_path: 低速存储路径
                - metadata_file: 元数据文件路径
                - cold_threshold_days: 冷门判定天数
                - max_cache_size_gb: 高速缓存最大容量(GB)
                - archive_batch_size: 每次归档的文件数量
        """
        self.fast_path = Path(config['fast_cache_path'])
        self.slow_path = Path(config['slow_cache_path'])
        self.metadata_file = Path(config.get('metadata_file', 'cache_metadata.json'))
        self.cold_threshold = timedelta(days=config.get('cold_threshold_days', 30))
        self.max_cache_size = config.get('max_cache_size_gb', 100) * 1024 * 1024 * 1024
        self.batch_size = config.get('archive_batch_size', 100)
        # 确保目录存在
        self.fast_path.mkdir(parents=True, exist_ok=True)
        self.slow_path.mkdir(parents=True, exist_ok=True)
        # 加载元数据
        self.metadata = self._load_metadata()
    def _load_metadata(self) -> Dict:
        """加载缓存文件元数据"""
        if self.metadata_file.exists():
            try:
                with open(self.metadata_file, 'r') as f:
                    return json.load(f)
            except Exception as e:
                logging.error(f"加载元数据失败: {e}")
                return {}
        return {}
    def _save_metadata(self):
        """保存缓存文件元数据"""
        try:
            with open(self.metadata_file, 'w') as f:
                json.dump(self.metadata, f, indent=2)
        except Exception as e:
            logging.error(f"保存元数据失败: {e}")
    def _get_file_access_time(self, file_path: Path) -> Optional[datetime]:
        """获取文件最后访问时间"""
        try:
            stat = file_path.stat()
            return datetime.fromtimestamp(stat.st_atime)
        except Exception as e:
            logging.error(f"获取文件访问时间失败 {file_path}: {e}")
            return None
    def _calculate_file_hash(self, file_path: Path) -> str:
        """计算文件哈希值用于唯一标识"""
        try:
            hasher = hashlib.md5()
            with open(file_path, 'rb') as f:
                # 只读取文件头部和尾部来平衡性能和准确性
                hasher.update(f.read(8192))
                f.seek(-8192, 2)
                hasher.update(f.read())
            return hasher.hexdigest()
        except Exception as e:
            logging.error(f"计算文件哈希失败 {file_path}: {e}")
            return str(file_path.name)
    def scan_cache_files(self) -> List[Dict]:
        """扫描高速缓存中的文件"""
        files_info = []
        for file_path in self.fast_path.rglob('*'):
            if file_path.is_file() and not file_path.name.startswith('.'):
                file_info = {
                    'path': str(file_path),
                    'name': file_path.name,
                    'size': file_path.stat().st_size,
                    'last_access': self._get_file_access_time(file_path),
                    'hash': self._calculate_file_hash(file_path)
                }
                files_info.append(file_info)
        return files_info
    def identify_cold_files(self, files_info: List[Dict]) -> List[Dict]:
        """
        识别冷门缓存文件
        规则：
        1. 超过冷门阈值的文件
        2. 在元数据中标记为低访问频率的文件
        """
        cold_files = []
        now = datetime.now()
        for file_info in files_info:
            # 检查最后访问时间
            if file_info['last_access']:
                age = now - file_info['last_access']
                # 更新元数据中的访问信息
                file_hash = file_info['hash']
                if file_hash not in self.metadata:
                    self.metadata[file_hash] = {
                        'access_count': 0,
                        'first_seen': now.isoformat(),
                        'last_access': file_info['last_access'].isoformat()
                    }
                # 判断是否为冷门文件
                if age > self.cold_threshold:
                    cold_files.append(file_info)
        return cold_files
    def calculate_cache_size(self) -> int:
        """计算当前缓存总大小"""
        total_size = 0
        for file_path in self.fast_path.rglob('*'):
            if file_path.is_file():
                total_size += file_path.stat().st_size
        return total_size
    def archive_files(self, files_to_archive: List[Dict]):
        """
        归档文件到低速存储
        归档策略：
        1. 保持目录结构
        2. 创建符号链接或快捷方式
        3. 更新元数据
        """
        archived_count = 0
        for file_info in files_to_archive:
            try:
                src_path = Path(file_info['path'])
                # 计算相对路径以保持结构
                rel_path = src_path.relative_to(self.fast_path)
                dst_path = self.slow_path / rel_path
                # 确保目标目录存在
                dst_path.parent.mkdir(parents=True, exist_ok=True)
                # 移动文件到低速存储
                shutil.move(str(src_path), str(dst_path))
                # 在原始位置创建符号链接（可选）
                try:
                    os.symlink(dst_path, src_path)
                except:
                    # 如果符号链接失败，创建标记文件
                    marker_path = src_path.with_suffix('.archived')
                    marker_path.write_text(str(dst_path))
                # 更新元数据
                file_hash = file_info['hash']
                if file_hash in self.metadata:
                    self.metadata[file_hash]['archived'] = True
                    self.metadata[file_hash]['archived_path'] = str(dst_path)
                    self.metadata[file_hash]['archived_time'] = datetime.now().isoformat()
                archived_count += 1
                logging.info(f"归档文件: {file_info['name']} ({archived_count}/{len(files_to_archive)})")
            except Exception as e:
                logging.error(f"归档失败 {file_info['name']}: {e}")
        # 保存更新后的元数据
        self._save_metadata()
        return archived_count
    def run_archival_process(self):
        """运行完整的归档流程"""
        logging.info("开始冷门缓存归档流程")
        # 1. 扫描缓存文件
        logging.info("扫描高速缓存文件...")
        files_info = self.scan_cache_files()
        logging.info(f"发现 {len(files_info)} 个缓存文件")
        # 2. 检查缓存大小
        current_size = self.calculate_cache_size()
        logging.info(f"当前缓存大小: {current_size / (1024**3):.2f} GB")
        # 3. 识别冷门文件
        logging.info("识别冷门缓存文件...")
        cold_files = self.identify_cold_files(files_info)
        logging.info(f"发现 {len(cold_files)} 个冷门文件")
        # 4. 判断是否需要归档
        if cold_files and current_size > self.max_cache_size * 0.8:  # 达到80%容量触发
            # 根据最后访问时间排序
            cold_files.sort(key=lambda x: x['last_access'] if x['last_access'] else datetime.min)
            # 分批归档
            for i in range(0, len(cold_files), self.batch_size):
                batch = cold_files[i:i + self.batch_size]
                archived = self.archive_files(batch)
                logging.info(f"批次归档完成: {archived} 个文件")
                # 重新检查缓存大小
                current_size = self.calculate_cache_size()
                if current_size <= self.max_cache_size * 0.6:  # 降到60%以下停止
                    break
        else:
            logging.info("缓存空间充足，无需归档")
class SmartCacheManager:
    """智能缓存管理器 - 自动管理缓存策略"""
    def __init__(self):
        self.archiver = None
    def configure_smart_policy(self):
        """配置智能缓存策略"""
        # 根据文件类型设置不同的冷门阈值
        self.policies = {
            '.tmp': {'threshold_days': 1, 'priority': 'low'},
            '.log': {'threshold_days': 7, 'priority': 'low'},
            '.cache': {'threshold_days': 14, 'priority': 'medium'},
            '.thumb': {'threshold_days': 30, 'priority': 'high'},
            '*': {'threshold_days': 30, 'priority': 'normal'}
        }
    def adaptive_threshold(self, file_info: Dict) -> int:
        """自适应调整冷门阈值"""
        ext = Path(file_info['name']).suffix
        policy = self.policies.get(ext, self.policies['*'])
        return policy['threshold_days']
def create_default_config() -> Dict:
    """创建默认配置"""
    return {
        'fast_cache_path': '/path/to/fast/cache',
        'slow_cache_path': '/path/to/slow/storage',
        'metadata_file': '/path/to/metadata/cache_metadata.json',
        'cold_threshold_days': 30,
        'max_cache_size_gb': 100,
        'archive_batch_size': 100
    }
if __name__ == "__main__":
    # 使用示例
    config = create_default_config()
    # 根据实际情况修改路径
    config['fast_cache_path'] = './fast_cache'
    config['slow_cache_path'] = './slow_cache'
    archiver = CacheArchiver(config)
    archiver.run_archival_process()

配置文件示例

# config.yaml
cache_archiver:
  fast_cache_path: "/data/fast_cache"
  slow_cache_path: "/data/slow_cache" 
  metadata_file: "/data/metadata/cache_metadata.json"
  # 冷门判定参数
  cold_threshold_days: 30
  hot_threshold_hours: 24  # 热数据保留时间
  # 缓存大小限制
  max_cache_size_gb: 100
  min_free_space_gb: 20    # 保留的最小空闲空间
  # 归档策略
  archive_batch_size: 100
  preserve_symlinks: true   # 是否保留符号链接
  # 文件过滤
  exclude_patterns:
    - "*.tmp"
    - "*.lock"
  # 日志配置
  log_level: "INFO"
  log_file: "/var/log/cache_archiver.log"

使用说明

基本用法

# 命令行使用
python cache_archiver.py
# 定时任务配置 (crontab)
0 2 * * * /usr/bin/python3 /path/to/cache_archiver.py

高级功能

智能冷门识别
- 基于文件类型设置不同的阈值
- 分析访问模式（频率、时间分布）
渐进式归档
- 按文件大小、访问时间优先级排序
- 控制归档速度和网络/IO负载

回滚机制

def restore_archived_file(self, file_hash: str):
    """从低速存储恢复文件到高速缓存"""
    if file_hash in self.metadata and self.metadata[file_hash].get('archived'):
        src_path = Path(self.metadata[file_hash]['archived_path'])
        dst_path = Path(self.metadata[file_hash]['original_path'])
        shutil.copy2(src_path, dst_path)
        # 移除符号链接或标记